A Production Agent API in Python
Chat wrappers are demos. Production agent APIs do one job with a schema.
Founding teams ship POST /chat and wonder why integrations break. Partners need structured data, timeouts, and idempotent retries — not conversation history.
One Endpoint, One Job
Bad: open-ended chat that returns markdown.
Good: POST /extract that returns a typed object.
Partner webhook arrives. You need three fields from the payload. The API extracts them or fails clearly. No parsing prose on the client.
from pydantic import BaseModel, Field
class WebhookExtractRequest(BaseModel):
partner_id: str
event_type: str
payload: dict
class WebhookExtractResult(BaseModel):
external_id: str
action: str = Field(description="create | update | delete")
severity: str = Field(description="low | medium | high")
The schema is the contract. Partners integrate against fields, not prompts.
The Endpoint
~80 lines. Structured output. Timeout. Log.
import asyncio
from fastapi import FastAPI, Header, HTTPException
from pydantic import BaseModel
app = FastAPI()
class ExtractRequest(BaseModel):
partner_id: str
event_type: str
payload: dict
class ExtractResult(BaseModel):
external_id: str
action: str
severity: str
async def llm_extract(req: ExtractRequest) -> ExtractResult:
# Call model with JSON schema / tool mode — return typed result
...
@app.post("/extract", response_model=ExtractResult)
async def extract(
req: ExtractRequest,
idempotency_key: str | None = Header(None, alias="Idempotency-Key"),
):
if not idempotency_key:
raise HTTPException(400, "Missing Idempotency-Key")
cached = cache_get(idempotency_key)
if cached:
return ExtractResult(**cached)
try:
result = await asyncio.wait_for(llm_extract(req), timeout=30.0)
except asyncio.TimeoutError:
raise HTTPException(503, "Extraction timed out")
cache_set(idempotency_key, result.model_dump(), ttl=3600)
log_extraction(
partner_id=req.partner_id,
event_type=req.event_type,
tokens=...,
latency_ms=...,
)
return result
No free text. No “please format as JSON.” The model returns the schema or the request fails.
Why Structure Wins
Integration partners retry on timeout. Idempotency keys prevent duplicate processing.
Structured output prevents:
- Markdown fences in API responses
- Missing fields that slip through regex parsing
- Silent schema drift when the model rephrases keys
At 2M+ events per day, one ambiguous field becomes thousands of bad rows.
Limits You Need on Day One
| Limit | Why |
|---|---|
| 30s timeout | Partners retry; hanging requests stack |
| Idempotency-Key | Same webhook delivered twice |
| Token cap on input | Oversized payloads blow cost and context |
| Schema validation | Reject malformed model output before DB write |
Log partner_id, event_type, tokens, latency_ms on every call. When finance asks why the bill doubled, you need the answer.
Chat vs Agent API
| Chat endpoint | Agent API |
|---|---|
| Open-ended | One job per route |
| Markdown | Typed schema |
| Session state | Stateless + idempotent |
| Demo-friendly | Integration-friendly |
Build the agent API first. Add chat later if users need it.
Key Takeaways
Production agent APIs are structured extraction endpoints, not chat.
Pydantic schema is the contract. Model output must match it or fail.
Idempotency and timeout are required for webhook-shaped workloads.
Log tokens and latency per request.
One endpoint, one job, one schema. That is what partners can integrate.