The Problem
You have a research agent that works beautifully on your laptop. It calls the Anthropic API, runs tool loops, and returns a detailed summary. You run it in a terminal and it works every time. Now a teammate wants to use it. Then a product manager wants to embed it in a dashboard. Then you want to run it in production for real users.
None of that is possible with a Python script that you run in a terminal. You need an HTTP endpoint. You need it to handle concurrent requests. You need it to stream progress back to the client in real time — because the agent takes 30 seconds and users will not stare at a spinner. And you need to ship it as a container so it runs identically in staging, production, and every engineer's machine.
That's the gap this lesson bridges. FastAPI turns your agent into an HTTP service. Docker packages that service into a container that runs anywhere. Uvicorn runs the service in production with the right concurrency model for agents. Together, these three components are the standard deployment stack for production Python AI services.
The gap between "it works in my terminal" and "it's deployed and serving requests" is not a devops problem — it's an architecture decision you make when you write the API layer and the Dockerfile. Make those decisions wrong and you'll be debugging production timeouts and blocking event loops for weeks.
Async is not optional for LLM services
LLM API calls take 5-60 seconds. If your endpoint is def instead of async def, every in-flight request blocks a thread. A single synchronous endpoint with 4 uvicorn workers can serve exactly 4 concurrent users before requests start queuing. With async def, a single worker handles hundreds of concurrent requests — waiting on I/O costs nothing while other requests are served. Async is not an optimization for agents; it's a correctness requirement.
FastAPI for Agent Endpoints
FastAPI is the right framework for agent services. It's async-first, generates OpenAPI docs automatically, validates requests with Pydantic, and has native StreamingResponse for SSE.
The request and response models:
from pydantic import BaseModel
class AgentRequest(BaseModel):
user_input: str
session_id: str | None = None
max_tokens: int = 4096
class AgentResponse(BaseModel):
output: str
steps: int
model: str
usage: dict
Pydantic validates every incoming request. If user_input is missing, FastAPI returns a 422 Unprocessable Entity before your agent code even runs. No manual validation needed.
A basic POST /agent/run endpoint:
@app.post("/agent/run", response_model=AgentResponse)
async def run_agent(request: AgentRequest):
result = await agent.run(request.user_input)
return AgentResponse(
output=result.output,
steps=result.step_count,
model=result.model,
usage=result.usage,
)
The async def is critical. The Anthropic SDK's async client (AsyncAnthropic) suspends the coroutine while the API call is in flight — the uvicorn event loop services other requests during the wait. No thread is blocked.
Streaming with SSE:
Most agent tasks take long enough that users need streaming. Use StreamingResponse with a generator that yields SSE-formatted strings:
from fastapi.responses import StreamingResponse
@app.post("/agent/run/stream")
async def run_agent_stream(request: AgentRequest):
async def event_generator():
async with client.messages.stream(
model="claude-opus-4-5",
max_tokens=request.max_tokens,
messages=[{"role": "user", "content": request.user_input}],
) as stream:
async for text in stream.text_stream:
data = json.dumps({"type": "text_delta", "text": text})
yield f"data: {data}\n\n"
yield "data: {\"type\": \"done\"}\n\n"
return StreamingResponse(
event_generator(),
media_type="text/event-stream",
headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"},
)
The X-Accel-Buffering: no header tells nginx (if you put it in front) not to buffer the stream — critical for SSE to reach the client in real time.
Background tasks for fire-and-forget:
For long jobs where you want to return a job ID immediately and let the agent run asynchronously:
from fastapi import BackgroundTasks
@app.post("/agent/run/async")
async def run_agent_async(request: AgentRequest, background_tasks: BackgroundTasks):
job_id = str(uuid.uuid4())
background_tasks.add_task(run_agent_job, job_id, request.user_input)
return {"job_id": job_id, "status": "queued"}
Health check:
@app.get("/health")
async def health():
return {"status": "ok", "model": "claude-opus-4-5", "timestamp": time.time()}
Health checks are used by Docker, Kubernetes, and load balancers to decide whether to send traffic to your container. If /health returns 200, traffic flows. If it returns 500, the container is taken out of rotation.
Your agent endpoint is def run_agent(request: AgentRequest) (not async). It calls the Anthropic API which takes 20 seconds. With 4 uvicorn workers, the maximum number of concurrent users before requests start queuing is:
Explanation: A synchronous def endpoint in uvicorn runs in a thread pool. Each in-flight request occupies one thread for its entire duration — 20 seconds for a 20-second LLM call. With 4 workers and sync endpoints, you have exactly 4 threads available, so only 4 requests can run simultaneously. The 5th user waits until one finishes. With async def and the async Anthropic client, a single worker can handle hundreds of concurrent requests — the event loop suspends the coroutine while waiting for the API, freeing it to serve other requests. Async is not optional for services that call LLMs.