Skip to content

Server

swerex.server

AUTH_TOKEN module-attribute 

AUTH_TOKEN = ''

api_key_header module-attribute 

api_key_header = APIKeyHeader(name='X-API-Key')

authenticate async 

authenticate(request: Request, call_next)

Authenticate requests with an API key (if set).

Source code in swerex/server.py 
40
41
42
43
44
45
46
47
@app.middleware("http")
async def authenticate(request: Request, call_next):
    """Authenticate requests with an API key (if set)."""
    if AUTH_TOKEN:
        api_key = await api_key_header(request)
        if api_key != AUTH_TOKEN:
            raise HTTPException(status_code=401, detail="Invalid API Key")
    return await call_next(request)

close async 

close()
Source code in swerex/server.py 
132
133
134
135
@app.post("/close")
async def close():
    await runtime.close()
    return CloseResponse()

close_session async 

close_session(request: CloseSessionRequest)
Source code in swerex/server.py 
87
88
89
@app.post("/close_session")
async def close_session(request: CloseSessionRequest):
    return serialize_model(await runtime.close_session(request))

create_session async 

create_session(request: CreateSessionRequest)
Source code in swerex/server.py 
77
78
79
@app.post("/create_session")
async def create_session(request: CreateSessionRequest):
    return serialize_model(await runtime.create_session(request))

exception_handler async 

exception_handler(request: Request, exc: Exception)

We catch exceptions that are thrown by the runtime, serialize them to JSON and return them to the client so they can reraise them in their own code.

Source code in swerex/server.py 
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
@app.exception_handler(Exception)
async def exception_handler(request: Request, exc: Exception):
    """We catch exceptions that are thrown by the runtime, serialize them to JSON and
    return them to the client so they can reraise them in their own code.
    """
    if isinstance(exc, (HTTPException, StarletteHTTPException)):
        return await http_exception_handler(request, exc)
    extra_info = getattr(exc, "extra_info", {})
    _exc = _ExceptionTransfer(
        message=str(exc),
        class_path=type(exc).__module__ + "." + type(exc).__name__,
        traceback=traceback.format_exc(),
        extra_info=extra_info,
    )
    return JSONResponse(status_code=511, content={"swerexception": _exc.model_dump()})

execute async 

execute(command: Command)
Source code in swerex/server.py 
92
93
94
@app.post("/execute")
async def execute(command: Command):
    return serialize_model(await runtime.execute(command))

is_alive async 

is_alive()
Source code in swerex/server.py 
72
73
74
@app.get("/is_alive")
async def is_alive():
    return serialize_model(await runtime.is_alive())

read_file async 

read_file(request: ReadFileRequest)
Source code in swerex/server.py 
97
98
99
@app.post("/read_file")
async def read_file(request: ReadFileRequest):
    return serialize_model(await runtime.read_file(request))

root async 

root()
Source code in swerex/server.py 
67
68
69
@app.get("/")
async def root():
    return {"message": "hello world"}

run async 

run(action: Action)
Source code in swerex/server.py 
82
83
84
@app.post("/run_in_session")
async def run(action: Action):
    return serialize_model(await runtime.run_in_session(action))

serialize_model 

serialize_model(model)
Source code in swerex/server.py 
36
37
def serialize_model(model):
    return model.model_dump() if hasattr(model, "model_dump") else model.dict()

upload async 

upload(file: UploadFile = File(...), target_path: str = Form(...), unzip: bool = Form(False))
Source code in swerex/server.py 
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
@app.post("/upload")
async def upload(
    file: UploadFile = File(...),
    target_path: str = Form(...),  # type: ignore
    unzip: bool = Form(False),
):
    target_path: Path = Path(target_path)
    target_path.parent.mkdir(parents=True, exist_ok=True)
    # First save the file to a temporary directory and potentially unzip it.
    with tempfile.TemporaryDirectory() as temp_dir:
        file_path = Path(temp_dir) / "temp_file_transfer"
        try:
            with open(file_path, "wb") as f:
                f.write(await file.read())
        finally:
            await file.close()
        if unzip:
            with zipfile.ZipFile(file_path, "r") as zip_ref:
                zip_ref.extractall(target_path)
            file_path.unlink()
        else:
            shutil.move(file_path, target_path)
    return UploadResponse()

write_file async 

write_file(request: WriteFileRequest)
Source code in swerex/server.py 
102
103
104
@app.post("/write_file")
async def write_file(request: WriteFileRequest):
    return serialize_model(await runtime.write_file(request))