Serving Layer

The augments.adk.serving module turns a local Agent into an ASGI app that any ASGI runtime can serve. Two surfaces are available: a plain-REST surface for generic HTTP clients and an A2A JSON-RPC surface for peer agents. Both are opt-in — the framework never serves a route the developer did not request.

Installation

pip install 'augments-adk[serve]'

This pulls in Starlette and sse-starlette. The A2A surface additionally requires the a2a extra:

pip install 'augments-adk[serve,a2a]'

Quick start: augments serve

augments serve is the fastest path from a local agent to a running HTTP endpoint. It exposes the REST surface and health routes by default:

# Serve the agent defined at my_pkg.agents:assistant
augments serve --agent my_pkg.agents:assistant

By default this binds 127.0.0.1:8000 and serves:

• POST /run — run the agent to completion, returns JSON
• POST /run_sse — run the agent and stream events as Server-Sent Events
• GET /healthz — liveness probe
• GET /readyz — readiness probe

Key flags

Enabling the A2A surface

Pass --card with a developer-authored AgentCard JSON to publish the A2A JSON-RPC and discovery surface alongside the REST surface:

augments serve --agent my_pkg.agents:assistant --card card.json

The card is published at GET /.well-known/agent-card.json. The CLI never synthesises a card; every field is intentional.

[!WARNING] The default A2A task store is in-memory. Tasks are lost on process restart, which breaks the A2A continuation token resume contract for any task that outlives the process. Pass --task-db tasks.sqlite for a durable store that recovers non-terminal tasks from prior processes on startup.

Container invocation

Inside a container, bind all interfaces and read the platform port:

augments serve --agent app.agents:assistant --host 0.0.0.0 --port "$PORT"

The augments deploy tooling bakes this command into the generated Dockerfile automatically.

REST request and response format

Both POST /run and POST /run_sse accept the same JSON body:

{
  "prompt": "What is the capital of France?",
  "max_turns": 10,
  "session": {
    "user_id": "alice",
    "session_id": "chat-123"
  }
}

Fields:

POST /run — synchronous response

Returns a JSON object with the run result when the agent finishes:

{
  "final_output": "Paris.",
  "items": [...],
  "usage": {
    "input_tokens": 42,
    "output_tokens": 3
  }
}

Response fields are Layer-1 and Layer-3 types only — the provider wire format never crosses this boundary.

POST /run_sse — streaming response

Returns a stream of text/event-stream frames. Each frame carries a JSON payload; the stream ends with an event: result frame carrying the final summary:

data: {"type": "text_delta", "text_delta": "Par"}

data: {"type": "text_delta", "text_delta": "is."}

event: result
data: {"final_output": "Paris.", "usage": {...}}

Programmatic API: build_app

For production deployments with custom ASGI runtimes or more control over the lifecycle, use build_app directly instead of the CLI:

import uvicorn
from augments.adk.agents import Agent
from augments.adk.serving import build_app
 
agent = Agent(name="assistant", system_prompt="You are a helpful assistant.")
 
app = build_app(
    agent,
    rest=True,
    health=True,
)
 
uvicorn.run(app, host="0.0.0.0", port=8080)

build_app raises ValueError if no surface is enabled.

Signature

def build_app(
    agent: Agent,
    *,
    rest: bool = False,
    a2a_server: A2AServer | None = None,
    health: bool = False,
    max_turns: int | None = None,
    run_config: RunConfig | None = None,
    session_factory: SessionFactory | None = None,
    readiness_probe: ReadinessProbe | None = None,
) -> Starlette: ...

Parameters:

Adding a session backend

Wire a SessionFactory to persist conversation history across requests.

For a single replica, SQLiteMultiSessions is sufficient:

from augments.adk.session import SQLiteMultiSessions
from augments.adk.serving import build_app
 
sessions = SQLiteMultiSessions(app_name="assistant", path="sessions.sqlite")
 
app = build_app(
    agent,
    rest=True,
    health=True,
    session_factory=lambda user_id, session_id: sessions.get_or_create(user_id, session_id),
)

For multiple replicas, use PostgresMultiSessions so every replica reads and writes the same conversation state, or pass --session-dsn to augments serve to have the CLI wire it automatically. See Horizontal scaling for the full multi-replica setup.

Adding a custom readiness probe

Provide an async predicate to signal when the service is ready for traffic:

async def my_readiness() -> bool:
    # Return False until external dependencies are reachable.
    return await db.ping()
 
app = build_app(agent, health=True, readiness_probe=my_readiness)

When the probe returns False, GET /readyz responds with HTTP 503.

See also

• Container contract — what every generated image must satisfy
• Kubernetes and Helm — Kubernetes manifests and Helm charts
• GCP Cloud Run — Cloud Run deployment
• AWS — ECS, App Runner, Lambda
• Horizontal scaling — multi-replica and shared backends
• A2A guide — full A2A protocol documentation