Other web frameworks

Available · pip install "agent-adapter[server]" for the mount

Any async framework can await executor.run(...) in its handlers, as the FastAPI guide does. To expose the whole registry over HTTP, mount the tool server (ASGI), or forward requests to ToolService (anything else).

File	Does
examples/backend_common.py	the shared service layer: build_backend, invoke, tool_server_options
examples/backend_other_asgi.py	a Starlette mount, an aiohttp shim over ToolService, a gRPC servicer
tests/docs/test_backend_handler_examples.py	runs the shim and the servicer offline; the mounts when the framework is installed

Data enters as an HTTP request (or a gRPC message), goes through ToolService or invoke to the executor, and leaves as JSON.

Mount the tool server (ASGI)

def starlette_app(backend: Backend | None = None) -> Any:
    """Starlette (and the same idea in Litestar, Quart, Sanic): mount the tool server."""
    from starlette.applications import Starlette
    from starlette.routing import Mount

    backend = backend or build_backend()
    tools = create_app(registry, mcp=False, **tool_server_options(backend))
    return Starlette(routes=[Mount("/agent", app=tools)])

Framework	Mount tools = create_app(registry, ...) with
Starlette	Mount("/agent", app=tools) (above)
Litestar	an @asgi("/agent", is_mount=True) handler that awaits tools(scope, receive, send)
Quart, Sanic (both ASGI apps)	an outer Starlette: Mount("/agent", app=tools), then Mount("/", app=your_app)
Strawberry GraphQL	resolvers await executor.run(...); mount tools beside the GraphQL app the same way

The mounted server brings its own auth, approvals, idempotency and problems (see HTTP tool server). With mcp=True (the default) its lifespan must run: mount it in a framework that forwards lifespan events, or pass mcp=False.

aiohttp and other non-ASGI servers

def tool_service(backend: Backend) -> ToolService:
    """The framework-free HTTP tool server: auth, approvals, problems; no web imports."""
    return ToolService(registry, **tool_server_options(backend))


async def forward(
    service: ToolService,
    method: str,
    path: str,
    headers: dict[str, str],
    body: bytes,
    client: str | None,
) -> tuple[int, dict[str, str], bytes]:
    """Run one request through `service`; JSON routes only (SSE needs a streaming reply)."""
    info = RequestInfo(method=method, path=path, headers=headers, query={}, client=client)
    response: ServiceResponse = await service.handle(info, body)
    if response.stream is not None:  # SSE: use aiohttp's StreamResponse, not this shim
        return 501, {"Content-Type": "application/json"}, b'{"title": "SSE not proxied"}'
    out = {"Content-Type": response.media_type, **response.headers}
    return response.status, out, json.dumps(response.body).encode()


def aiohttp_app(service: ToolService) -> Any:
    """aiohttp is not ASGI: a catch-all route that forwards to `service`."""
    from aiohttp import web  # type: ignore[import-not-found,unused-ignore]

    async def route(request: web.Request) -> web.Response:
        status, headers, body = await forward(
            service,
            request.method,
            request.path,
            dict(request.headers),
            await request.read(),
            request.remote,  # the peer: the server refuses remote anonymous callers
        )
        return web.Response(status=status, headers=headers, body=body)

    app = web.Application()
    app.router.add_route("*", "/{tail:.*}", route)
    return app

ToolService.handle(RequestInfo, body) is every route of the tool server with no web imports. The shim passes the real peer address: an anonymous server refuses remote callers by it. Event streams (SSE) need a streaming response, which this shim does not proxy (it answers 501).

gRPC

class ToolServicer:
    """A `grpc.aio` servicer method: unpack the request message, await the service layer."""

    def __init__(self, backend: Backend) -> None:
        self.backend = backend

    async def Invoke(self, request: Any, context: Any) -> dict[str, Any]:
        status, body = await invoke(self.backend, request.tool, json.loads(request.arguments))
        return {"status": status, "body": json.dumps(body)}  # build your reply message here

Servicer	Call
grpc.aio (async)	await invoke(...) as above
sync grpc.server (thread pool)	run_sync(invoke(...), timeout=...), or a long-lived loop (LoopRunner)

gRPC has no adapter auth: authenticate in a server interceptor and refuse unknown callers before the servicer runs. Map the reply's status to a gRPC status code.

Rules

Rule	Why
build one backend per process, at startup	executors and stores are shared by every request
await in async handlers, run_sync in sync ones	never asyncio.run inside a running loop
one auth path	the mounted server authenticates its own routes; your routes use yours

See also: FastAPI, Flask, Notebooks and apps.