Adding AI Agents to StreamShop

Adding AI Agents to StreamShop #

Wiring a Rust SSE gateway into StreamShop so operators and shoppers talk to the same platform through different personas.

StreamShop already had a commerce pipeline worth explaining: catalog → order → outbox → Redpanda → event-processor → ClickHouse. The AI work adds a front door for that story.

I added a Rust agent gateway that streams tool-using turns over SSE, calls the same service DNS the rest of Compose uses, and keeps session context in Redis. The basis for the agent came from this SitePoint article.

One gateway, multiple personas #

Every POST /v1/agent/stream carries a session_id, a user message, and an optional persona (operator by default, or shopper). Persona is not a separate process. It selects three things on the turn:

  1. System prompt re-injected every turn from code, never stored in Redis, so prompt updates apply mid-session
  2. Tool schemas advertised to the LLM
  3. Allowlist at execution time, even if the model invents a forbidden tool name the gateway refuses it
Persona Role Tools
operator Ops / SRE copilot get_order, get_orders_summary, check_service_health
shopper Catalog Q&A list_products

That split matters for the demo. The operator can ask “is analytics healthy?” and get a real /ready probe. The shopper can ask about products and prices, but cannot poke order status or invent a DLQ story. Capability is code-shaped, not prompt-shaped alone.

Turns stream as typed SSE events (turn_start, token_chunk, tool_call, tool_result, turn_end, error) so the admin UI can show tool traffic, not just final text

Conversation history lives under session:<id>:context in StreamShop Redis. System messages are stripped before save and reapplied on the next turn.

Holding many connections open #

Agent traffic is long-lived: one HTTP request, an open SSE response, keep-alives every 15s, and an async orchestrator task that may fan out tool calls while tokens stream. The gateway is built for that shape rather than for short CRUD RPCs.

  • Tokio + Axum accept concurrent streams
    • Each turn is tokio::spawn’d so the request handler returns the SSE body immediately
  • Tower ConcurrencyLimitLayer(256) caps in-flight requests
    • Overload surfaces as 503 with server at capacity instead of unbounded task growth
  • 120s request timeout and a 1 MB body limit bound hung LLM turns and oversized payloads
  • Multiplexed Redis and a shared reqwest client (pool_max_idle_per_host(20)) avoid opening a new TCP connection per tool call
  • Tool work runs concurrently per turn with cancellation tokens and a 30s per-tool timeout so a stuck upstream does not pin the stream forever

Video Walkthrough #

You can watch a video walkthrough of both personans asking questions and an attempt at prompt inject.

Watch the video walkthrough on YouTube