Server

swerex.server

AUTH_TOKEN module-attribute

AUTH_TOKEN = ''

api_key_header module-attribute

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

response_manager module-attribute

response_manager = ResponseManager()

ResponseManager

ResponseManager()

This stores the response of the last request, and is used in retries to return already executed requests.

Note that in the case of multiple concurrent clients, idempotency isn't guaranteed.

Source code in swerex/server.py

def __init__(self):
    self.last_processed_request_id = None
    self.last_processed_response = None

last_processed_request_id instance-attribute

last_processed_request_id = None

last_processed_response instance-attribute

last_processed_response = None

get_response

get_response(request_id)

Source code in swerex/server.py

def get_response(self, request_id):
    if request_id == self.last_processed_request_id:
        return self.last_processed_response
    return None

set_response

set_response(request_id, response)

Source code in swerex/server.py

def set_response(self, request_id, response):
    self.last_processed_request_id = request_id
    self.last_processed_response = response

authenticate async

authenticate(request: Request, call_next)

Authenticate requests with an API key (if set).

Source code in swerex/server.py

@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

@app.post("/close")
async def close():
    await runtime.close()
    return CloseResponse()

close_session async

close_session(request: CloseSessionRequest)

Source code in swerex/server.py

@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

@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

@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

@app.post("/execute")
async def execute(command: Command):
    return serialize_model(await runtime.execute(command))

handle_request_id async

handle_request_id(request: Request, call_next)

Handle request ID for idempotency.

Source code in swerex/server.py

@app.middleware("http")
async def handle_request_id(request: Request, call_next):
    """Handle request ID for idempotency."""
    request_id = request.headers.get("X-Request-ID")
    if request_id:
        response = response_manager.get_response(request_id)
        if response:
            return response

    response = await call_next(request)

    body_content = b""
    async for chunk in response.body_iterator:
        body_content += chunk

    new_response = Response(
        content=body_content,
        status_code=response.status_code,
        headers=dict(response.headers),
        media_type=response.media_type,
    )

    if request_id:
        response_manager.set_response(request_id, new_response)

    return new_response

is_alive async

is_alive()

Source code in swerex/server.py

@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

@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

@app.get("/")
async def root():
    return {"message": "hello world"}

run async

run(action: Action)

Source code in swerex/server.py

@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

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

@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

@app.post("/write_file")
async def write_file(request: WriteFileRequest):
    return serialize_model(await runtime.write_file(request))

SWE-agent Mini-SWE-Agent SWE-smith SWE-bench sb-cli