Lab 01 — Async Streaming Agent Service

Phase 12 · Lab 01 · Phase README · Warmup

The problem

Agents are I/O-bound (waiting on LLM and tool calls), so a production agent service is an async service: it serves many concurrent requests on few threads, streams tokens as they're generated, decouples work with events, dedupes retries with idempotency keys, and holds session state in an external cache. Build that service with stdlib asyncio only — deterministic and offline, with a real-FastAPI sketch left as an extension.

What you build

PieceWhat it does
stream_agent / sse_stream / format_sseasync token streaming in Server-Sent-Events shape
AgentServicerequest handling under an asyncio.Semaphore (concurrency + backpressure) + metrics
idempotencya repeated idempotency_key returns the cached result without re-running the model
TTLCachesession/result state with an injected clock TTL (Redis-shaped)
EventBus / Subscriptionin-process pub/sub with fan-out to multiple subscribers
health / graceful drainreadiness + shutdown() that waits for in-flight requests

File map

FileRole
lab.pyyour implementation (fill the # TODOs)
solution.pyreference + an asyncio.run demo (concurrency, idempotent replay, TTL expiry, pub/sub)
test_lab.py26 tests (wrapped in asyncio.run, no pytest-asyncio needed)
requirements.txtpytest only

Run it

pytest test_lab.py -v
LAB_MODULE=solution pytest test_lab.py -v
python solution.py

Success criteria

  • stream_agent yields token events then a done event, in SSE format.
  • The semaphore caps concurrency (max in-flight never exceeds the limit).
  • A repeated idempotency key replays the cached result without re-invoking the model.
  • TTLCache hits/misses/expires by the injected clock; the event bus fans out to all subscribers in order; graceful drain waits for in-flight work.
  • All 26 tests pass under both lab and solution.

How this maps to the real stack

  • stream_agent ≈ FastAPI's StreamingResponse over Server-Sent Events (the wire format chat UIs use); the async model is the asyncio event loop uvicorn runs.
  • The TTLCache ≈ Redis for session/conversation state and result caching; the EventBus ≈ Redis pub/sub, Kafka, or a cloud queue for event-driven agent workflows.
  • Idempotency keys ≈ the Stripe-style dedupe pattern (essential once agents retry — Phase 08).
  • The concurrency semaphore ≈ backpressure protecting downstream LLM rate limits; graceful drain ≈ Kubernetes readiness/termination handling.

Limits. A real service adds real network I/O, a real event broker, distributed state, and autoscaling; the async/streaming/idempotency/pub-sub mechanisms are exactly these.

Extensions (your own machine)

  • Wire the real-FastAPI app.py sketch: @app.post + StreamingResponse mapping to your stdlib version.
  • Swap the in-process bus for Redis pub/sub; swap the TTLCache for real Redis.
  • Add a rate limiter and per-tenant queues (Phase 13/14).

Interview / resume signal

"Built an async streaming agent service — SSE token streaming, an asyncio concurrency/backpressure layer, idempotency-key dedupe, a TTL state cache, and an event-bus pub/sub — the FastAPI/asyncio/ event-driven production shape, from first principles."