Build a Rust AI Agent Gateway with Tokio and Axum

Rust AI Agent Gateway: Build Low-Latency SSE Streaming in Axum

SitePoint TeamPublished inAI·Programming·APIs·
September 10, 2026

The AI briefing for Developers

Stay up to date with AI tools, model releases, and developer workflows that matter.

Weekly. Free. One click to leave.

SitePoint Premium
Stay Relevant and Grow Your Career in Tech

• Premium Results
• Publish articles on SitePoint
• Daily curated jobs
• Learning Paths
• Discounts to dev tools

7 Day Free Trial. Cancel Anytime.

When AI agents move from prototype to production, the gateway layer that orchestrates their interactions becomes a bottleneck. This article architects a complete agent gateway: an Axum-based HTTP layer that streams SSE events, a turn orchestrator that loops through LLM calls and tool dispatches as async pipeline stages, and a Redis integration layer for durable state and horizontal fan-out.

How to Build a Rust AI Agent Gateway with SSE Streaming

1. Scaffold a Rust project with Axum, Tokio, Redis, and reqwest dependencies organized into gateway, agent, streaming, and state modules.
2. Configure shared application state holding a Redis multiplexed connection, an HTTP client, and LLM credentials with a redacted Debug impl.
3. Implement an Axum SSE endpoint that validates requests, creates a bounded mpsc channel, and spawns the turn orchestrator as an async task.
4. Define a tagged AgentEvent enum covering the full turn lifecycle: start, token chunks, tool calls, tool results, turn end, and errors.
5. Build the turn orchestrator loop that streams LLM responses, accumulates tool-call deltas, and dispatches tool tasks concurrently with cancellation support.
6. Persist conversation context in Redis with TTL-based expiration and add pub/sub fan-out for multi-client observation across gateway instances.
7. Harden for production by adding concurrency limits, backpressure via bounded channels, client disconnect detection, and observability spans.
8. Benchmark with k6 against a mock LLM server, measuring time-to-first-byte, P99 latency, max connections, and memory footprint.

Table of Contents

Why Rust for AI Agent Gateways

When AI agents move from prototype to production, the gateway layer that orchestrates their interactions becomes a bottleneck. Every millisecond of overhead in the gateway compounds across multi-turn agent conversations, tool-call dispatches, and streaming token delivery. Interpreted-language gateways built in Python or Node.js introduce garbage collection pauses, event loop contention, and per-request memory overhead that grows with connection count (see the benchmark table below for memory at 1k connections). For teams building a Rust AI agent gateway with low-latency SSE streaming in Axum, the ownership model eliminates GC pauses entirely, and Tokio’s work-stealing async runtime keeps tail latencies tight under concurrent load.

An agent gateway does more than proxy HTTP. It routes incoming requests to the correct model provider, manages multi-turn conversation state, intercepts tool-call requests embedded in streamed LLM responses, dispatches those tool calls concurrently, feeds results back into the next turn, and streams incremental token output to clients as Server-Sent Events. Each of these responsibilities maps naturally to Rust’s async primitives.

This article architects a complete agent gateway: an Axum-based HTTP layer that streams SSE events, a turn orchestrator that loops through LLM calls and tool dispatches as async pipeline stages, and a Redis integration layer for durable state and horizontal fan-out.

The code in this article is a simplified reference architecture. A complete, runnable implementation requires wiring together the module declarations, providing a tool registry, and handling edge cases described in the inline warnings. Treat the snippets as a structural guide rather than a copy-paste-and-run project.

Architecture Overview of the Agent Turn Pipeline

Request Lifecycle: From HTTP to Agent Turn to SSE Stream

The request lifecycle follows a linear pipeline. A client opens an HTTP connection to the /v1/agent/stream endpoint. The Axum handler validates the request, constructs a bounded mpsc channel, and spawns a Tokio task running the turn orchestrator. That orchestrator enters a loop: it sends the current conversation context to an LLM provider via a streaming HTTP call, parses the response chunk by chunk, and forwards token events through the channel. When the LLM response contains a tool-call directive, the orchestrator spawns a concurrent task for tool execution, feeds the result back into the conversation context, and initiates another turn. The mpsc receiver side of the channel is mapped into an SSE Event stream that Axum writes to the client connection incrementally.

Component Map

Axum handles HTTP routing, request parsing, and SSE response serialization. Tokio provides the async runtime, task spawning for concurrent tool calls, and channel primitives for backpressure. These two form the request-handling core, while Redis stores inter-turn conversation state with TTL-based expiration and provides pub/sub for fan-out scenarios where multiple clients observe the same agent session. An LLM provider abstraction built on reqwest rounds out the stack, consuming upstream SSE streams from OpenAI-compatible APIs and translating them into typed gateway events.

Design Decisions and Trade-offs

The gateway uses SSE instead of WebSockets because agent streaming is inherently unidirectional: the server pushes token chunks to the client. SSE operates over standard HTTP and avoids the complexity of WebSocket frame management. SSE still holds long-lived connections, so configure load balancer idle timeouts accordingly. For bidirectional interaction, the client simply issues a new HTTP request for each user turn.

Horizontal scaling requires shared state across gateway instances. When multiple instances sit behind a load balancer, conversation state must be accessible from any instance, and fan-out events must propagate across process boundaries. Redis provides both. In-process tokio::broadcast channels are faster (sub-microsecond dispatch vs. the ~0.1-0.3 ms Redis localhost round-trip) but confine state to a single process.

Backpressure relies on bounded mpsc channels. When the client stalls and stops reading SSE events, the channel buffer fills, and the orchestrator’s send operations begin to exert backpressure naturally. If the client disconnects entirely, the receiver drops, causing the sender to fail, which signals the orchestrator to cancel its work.

Project Scaffolding and Dependencies

Prerequisites: Rust 1.75 or later (required for axum 0.7 compatibility), Redis 6 or later, and optionally k6 for load testing. You will need the LLM_API_KEY environment variable set, and optionally LLM_BASE_URL and REDIS_URL.

[package]name="agent-gateway"version="0.1.0"edition="2021"[dependencies]axum={version="0.7",features=["macros"]}tokio={version="1",features=["full"]}redis={version="0.25",features=["aio","tokio-comp"]}reqwest={version="0.12",features=["stream","json"]}serde={version="1",features=["derive"]}serde_json="1"bytes="1"futures="0.3"tokio-stream={version="0.1",features=["sync"]}tower={version="0.4",features=["timeout","limit","load-shed"]}tower-http={version="0.5",features=["trace"]}tracing="0.1"tracing-subscriber={version="0.3",features=["env-filter"]}tokio-util={version="0.7",features=["rt","sync"]}uuid={version="1",features=["v4"]}

Directory Structure

src/├── main.rs├── gateway/│   ├── mod.rs│   └── router.rs├── agent/│   ├── mod.rs│   └── turn_orchestrator.rs├── streaming/│   ├── mod.rs│   ├── sse_handler.rs│   └── events.rs└── state/├── mod.rs└── redis_store.rs

Each parent module (gateway/mod.rs, agent/mod.rs, streaming/mod.rs, state/mod.rs) must declare its child modules with pub mod statements, and main.rs must declare mod gateway; mod agent; mod streaming; mod state; for cross-module paths to resolve.

Building the Axum SSE Streaming Layer

Setting Up the Axum Router and Shared State

useaxum::{Router,extract::State};useaxum::error_handling::HandleErrorLayer;useaxum::http::StatusCode;useredis::aio::MultiplexedConnection;usestd::sync::Arc;usestd::fmt;usetower::{BoxError,ServiceBuilder};usetower_http::trace::TraceLayer;usetracing_subscriber::EnvFilter;#[derive(Clone)]pubstructApiKey(String);implApiKey{pubfnnew(key:String)->Self{Self(key)}pub(crate)fnapi_key(&self)->&str{&self.0}}implfmt::DebugforApiKey{fnfmt(&self, f:&mutfmt::Formatter)->fmt::Result{write!(f,"[REDACTED]")}}#[derive(Clone)]pubstructAppState{pub redis_conn:MultiplexedConnection,pub llm_client:reqwest::Client,pub llm_base_url:String,pub llm_api_key:ApiKey,}#[tokio::main]asyncfnmain(){tracing_subscriber::fmt().with_env_filter(EnvFilter::from_default_env()).init();let redis_url =std::env::var("REDIS_URL").unwrap_or_else(|_|"redis://127.0.0.1:6379".into());let redis_client =redis::Client::open(redis_url.as_str()).expect("Invalid Redis URL");let redis_conn = redis_client.get_multiplexed_async_connection().await.expect("Failed to connect to Redis");let state =AppState{redis_conn,llm_client:reqwest::Client::builder().pool_max_idle_per_host(20).build().expect("Failed to build HTTP client"),llm_base_url:std::env::var("LLM_BASE_URL").unwrap_or_else(|_|"https://api.openai.com".into()),llm_api_key:ApiKey::new(std::env::var("LLM_API_KEY").expect("LLM_API_KEY must be set"),),};let app =Router::new().route("/v1/agent/stream",axum::routing::post(crate::streaming::sse_handler::handle_stream),).route("/health",axum::routing::get(||async{"ok"})).layer(axum::extract::DefaultBodyLimit::max(1_048_576)).layer(ServiceBuilder::new().layer(HandleErrorLayer::new(|err:BoxError|asyncmove{if err.is::<tower::load_shed::error::Overloaded>(){(StatusCode::SERVICE_UNAVAILABLE,"server at capacity".to_string(),)}else{(StatusCode::INTERNAL_SERVER_ERROR,format!("internal error: {err}"),)}})).layer(TraceLayer::new_for_http()).layer(tower::timeout::TimeoutLayer::new(std::time::Duration::from_secs(120),)).layer(tower::limit::ConcurrencyLimitLayer::new(256)),).with_state(state);let listener =tokio::net::TcpListener::bind("0.0.0.0:3000").await.expect("Failed to bind");tracing::info!("Agent gateway listening on :3000");axum::serve(listener, app).await.expect("Server error");}

MultiplexedConnection is Clone and internally handles concurrent command multiplexing over a single TCP connection. There is no need to wrap it in Arc<Mutex<>>. Simply clone the connection for each operation. For details on when to switch to a connection pool, see the Redis section below.

Security note: The /v1/agent/stream endpoint shown here has no authentication. In production, add an authentication middleware layer (e.g., API key validation or JWT verification before exposing this endpoint

Implementing the SSE Endpoint

useaxum::{extract::State,response::sse::{Event,KeepAlive,Sse},Json,};usefutures::stream::Stream;usestd::convert::Infallible;usetokio::sync::mpsc;usetokio_stream::wrappers::ReceiverStream;usetokio_stream::StreamExt;usecrate::agent::turn_orchestrator::orchestrate_turn;usecrate::streaming::events::{AgentEvent,AgentRequest};usecrate::AppState;pubasyncfnhandle_stream(State(state):State<AppState>,Json(request):Json<AgentRequest>,)->Result<Sse<implStream<Item=Result<Event,Infallible>>>>,(axum::http::StatusCode,String)>{if request.session_id.is_empty()||!request.session_id.chars().all(|c| c.is_alphanumeric()|| c =='-'){returnErr((axum::http::StatusCode::BAD_REQUEST,"invalid session_id: must be alphanumeric or hyphens only".into(),));}let(tx, rx)=mpsc::channel::<AgentEvent>(64);tokio::spawn(asyncmove{ifletErr(e)=orchestrate_turn(state, request, tx.clone()).await{let _ = tx.send(AgentEvent::Error{message: e.to_string(),}).await;}});let stream =ReceiverStream::new(rx).map(|event|{let data =serde_json::to_string(&event).unwrap_or_else(|e|{tracing::error!("Event serialization failed: {e}");"{"type":"error","payload":{"message":"serialization failed"}}".into()});let event_type = event.event_type();Ok(Event::default().event(event_type).data(data))});Ok(Sse::new(stream).keep_alive(KeepAlive::new().interval(std::time::Duration::from_secs(15)).text("ping"),))}

The handler constructs a bounded mpsc channel with a capacity of 64 events. This bound is deliberate: it limits memory consumption per connection and creates natural backpressure against the orchestrator. When the client disconnects, the ReceiverStream drops, which drops the receiver half of the channel. Subsequent tx.send() calls in the orchestrator return an error, signaling the spawned task to terminate.

The KeepAlive configuration sends a keepalive frame every 15 seconds, preventing intermediate proxies and load balancers from closing idle connections during pauses between agent turns. .text(“ping”) sends a data: ping SSE event, not an SSE comment line (: ping). If your client uses a browser Eventnt (ignored by Event

Structuring SSE Event Types for Agent Turns

useserde::{Deserialize,Serialize};#[derive(Debug, Clone, Serialize, Deserialize)]pubstructAgentRequest{pub session_id:String,pub message:String,#[serde(default)]pub model:Option<String>,}#[derive(Debug, Clone, Serialize, Deserialize)]#[serde(tag = "type", content = "payload")]pubenumAgentEvent{TurnStart{ turn_id:String, turn_number:u32},TokenChunk{ content:String},ToolCall{ tool_name:String, arguments:serde_json::Value},ToolResult{ tool_name:String, result:serde_json::Value},TurnEnd{ turn_id:String, finish_reason:String},Error{ message:String},}implAgentEvent{pubfnevent_type(&self)->&'staticstr{matchself{AgentEvent::TurnStart{..}=>"turn_start",AgentEvent::TokenChunk{..}=>"token_chunk",AgentEvent::ToolCall{..}=>"tool_call",AgentEvent::ToolResult{..}=>"tool_result",AgentEvent::TurnEnd{..}=>"turn_end",AgentEvent::Error{..}=>"error",}}}

The AgentEvent enum uses Serde’s internally tagged representation. For example, TokenChunk { content: "Hello".into() } serializes to {"type": "token_chunk", "payload": {"content": "Hello"}}, and TurnStart { turn_id: "abc".into(), turn_number: 1 } serializes to {"type": "turn_start", "payload": {"turn_id": "abc", "turn_number": 1}}. Front-end clients can pattern-match on the type field of each SSE event to route data into the appropriate UI handler. The enum is exhaustive over the lifecycle of an agent turn: start, token streaming, tool interactions, completion, and error.

Designing the Asynchronous Agent Turn Pipeline

The Turn Orchestrator

usetokio::sync::mpsc;usetokio_util::sync::CancellationToken;useuuid::Uuid;usecrate::streaming::events::{AgentEvent,AgentRequest};usecrate::state::redis_store;usecrate::AppState;pubasyncfnorchestrate_turn(state:AppState,request:AgentRequest,tx:mpsc::Sender<AgentEvent>,)->Result<(),Box<dynstd::error::Error+Send+Sync>>{letmut turn_number:u32=0;let max_turns:u32=10;letmut messages =redis_store::load_context(&state,&request.session_id).await?;messages.push(serde_json::json!({"role":"user","content": request.message}));loop{turn_number +=1;if turn_number > max_turns {tx.send(AgentEvent::Error{message:"Max turns exceeded".into(),}).await?;break;}let turn_id =Uuid::new_v4().to_string();tx.send(AgentEvent::TurnStart{turn_id: turn_id.clone(),turn_number,}).await?;let(tool_calls, finish_reason)=crate::streaming::llm_stream::stream_llm_response(&state,&messages,&tx,).await?;if tool_calls.is_empty(){tx.send(AgentEvent::TurnEnd{turn_id,finish_reason: finish_reason.unwrap_or_else(||"stop".into()),}).await?;break;}letmut tool_handles:Vec<(String,CancellationToken,tokio::task::JoinHandle<serde_json::Value>,)>=Vec::new();for tc in&tool_calls {let tool_name = tc.tool_name.clone();let arguments = tc.arguments.clone();let tx_clone = tx.clone();let token =CancellationToken::new();let token_child = token.clone();let handle =tokio::spawn(asyncmove{tokio::select!{result =execute_tool(&tool_name,&arguments)=>{let _ = tx_clone.send(AgentEvent::ToolResult{tool_name: tool_name.clone(),result: result.clone(),}).await;result}_ = token_child.cancelled()=>{tracing::warn!(tool =%tool_name,"Tool task cancelled");serde_json::json!({"status":"cancelled"})}}});tool_handles.push((tc.tool_name.clone(), token, handle));}for(name, cancel_token, handle)in tool_handles {matchtokio::time::timeout(std::time::Duration::from_secs(30),handle,).await{Ok(Ok(result))=>{let content =match&result {serde_json::Value::String(s)=> s.clone(),other => other.to_string(),};messages.push(serde_json::json!({"role":"tool","name": name,"content": content}));}Ok(Err(join_err))=>{tracing::error!("Tool task panicked: {join_err}");tx.send(AgentEvent::Error{message:"Tool call failed (task panic)".into(),}).await?;}Err(_elapsed)=>{cancel_token.cancel();tx.send(AgentEvent::Error{message:format!("Tool call '{name}' timed out"),}).await?;}}}tx.send(AgentEvent::TurnEnd{turn_id,finish_reason:"tool_calls".into(),}).await?;}redis_store::save_context(&state,&request.session_id,&messages).await?;Ok(())}asyncfnexecute_tool(_name:&str,_args:&serde_json::Value,)->serde_json::Value{serde_json::json!({"status":"not_implemented"})}

The orchestrator enforces a maximum turn count to prevent runaway loops when an LLM repeatedly requests tool calls. Each tool call spawns as a separate Tokio task, enabling concurrent execution. The 30-second timeout applies when awaiting the JoinHandle. When a timeout fires, the CancellationToken is cancelled, signaling the spawned task to stop execution and release resources. The match arms separate panicked tasks (JoinError) from timed-out tasks (Elapsed) so that a panic does not silently masquerade as a timeout. Tool result content is extracted carefully: serde_json::Value::String variants are unwrapped directly to avoid double-encoding (where to_string() would produce ""foo"" instead of "foo").

Streaming from the LLM Provider

usebytes::{Buf,BytesMut};usefutures::StreamExt;usetokio::sync::mpsc;usestd::collections::HashMap;usecrate::streaming::events::AgentEvent;usecrate::AppState;pub(crate)structToolCallDirective{pub tool_name:String,pub arguments:serde_json::Value,}pub(crate)asyncfnstream_llm_response(state:&AppState,messages:&[serde_json::Value],tx:&mpsc::Sender<AgentEvent>,)->Result<(Vec<ToolCallDirective>,Option<String>),Box<dynstd::error::Error+Send+Sync>,>{let response = state.llm_client.post(format!("{}/v1/chat/completions", state.llm_base_url)).bearer_auth(state.llm_api_key.api_key()).timeout(std::time::Duration::from_secs(90)).json(&serde_json::json!({"model":"gpt-4o","messages": messages,"stream":true})).send().await?.error_for_status()?;letmut byte_stream = response.bytes_stream();letmut buf =BytesMut::with_capacity(4096);letmut finish_reason:Option<String>=None;letmut tool_call_buffers:HashMap<usize,(String,String)>=HashMap::new();'outer:whileletSome(chunk)= byte_stream.next().await{buf.extend_from_slice(&chunk?);loop{letSome(pos)= buf.iter().position(|&b| b == b'')else{break;};let line_bytes = buf.split_to(pos +1);let line =matchstd::str::from_utf8(&line_bytes){Ok(s)=> s.trim().to_owned(),Err(_)=>{tracing::warn!("Invalid UTF-8 in LLM stream line; skipping");continue;}};if line =="data: [DONE]"{break'outer;}ifletSome(json_str)= line.strip_prefix("data: "){ifletOk(parsed)=serde_json::from_str::<serde_json::Value>(json_str){ifletSome(delta)=parsed["choices"][0]["delta"]["content"].as_str(){if!delta.is_empty(){if tx.send(AgentEvent::TokenChunk{content: delta.to_string(),}).await.is_err(){break'outer;}}}ifletSome(reason)=parsed["choices"][0]["finish_reason"].as_str(){finish_reason =Some(reason.to_string());}ifletSome(tc)=parsed["choices"][0]["delta"]["tool_calls"].as_array(){for call in tc {let index =call["index"].as_u64().unwrap_or(0)asusize;let entry = tool_call_buffers.entry(index).or_insert_with(||{(String::new(),String::new())});ifletSome(name)=call["function"]["name"].as_str(){entry.0.push_str(name);}ifletSome(args_fragment)=call["function"]["arguments"].as_str(){entry.1.push_str(args_fragment);}}}}}}}let tool_calls =finalize_tool_calls(&tool_call_buffers, tx).await?;Ok((tool_calls, finish_reason))}asyncfnfinalize_tool_calls(buffers:&HashMap<usize,(String,String)>,tx:&mpsc::Sender<AgentEvent>,)->Result<Vec<ToolCallDirective>,Box<dynstd::error::Error+Send+Sync>,>{letmut indices:Vec<usize>= buffers.keys().copied().collect();indices.sort_unstable();letmut tool_calls =Vec::new();for index in indices {let(name, args_str)=&buffers[&index];if name.is_empty(){continue;}let arguments =matchserde_json::from_str::<serde_json::Value>(args_str){Ok(v)=> v,Err(e)=>{tracing::error!(tool_name =%name,raw_args =%args_str,error =%e,"Failed to parse tool-call arguments JSON");let _ = tx.send(AgentEvent::Error{message:format!("Malformed arguments for tool '{name}': {e}"),}).await;continue;}};let _ = tx.send(AgentEvent::ToolCall{tool_name: name.clone(),arguments: arguments.clone(),}).await;tool_calls.push(ToolCallDirective{tool_name: name.clone(),arguments,});}Ok(tool_calls)}

This function uses reqwest::Response::bytes_stream() to consume the upstream SSE response without buffering the entire body. Each incoming byte chunk is appended to a BytesMut buffer, and complete lines are extracted using split_to for efficient zero-copy line extraction. The data: [DONE] sentinel terminates parsing. Token chunks are forwarded through the bounded mpsc sender immediately, which means the client receives tokens with minimal gateway-added latency. The bounded channel’s send operation will await if the buffer is full, naturally throttling the parse rate to match the client’s consumption speed. When the receiver drops (client disconnect), the send returns an error and the loop exits early, stopping upstream consumption.

The per-request .timeout(Duration::from_secs(90)) on the reqwest call ensures that a stalled upstream LLM connection terminates independently of the Axum handler-level timeout, preventing a hung connection from holding a Tokio task and channel for the full 120-second handler timeout.

Important: OpenAI streams tool-call function.arguments as incremental string deltas across multiple chunks. No single chunk contains a complete parseable JSON arguments string. The code above accumulates these fragments in a per-index buffer (tool_call_buffers) and parses them only after the stream completes. Tool calls are finalized in sorted index order to ensure deterministic message ordering. If argument JSON is malformed (e.g., the stream was cut mid-object), an error event is emitted and the tool call is skipped rather than propagating corrupt data. If you send tool definitions in the request, also include a "tools" field in the request JSON body (omitted here for brevity).

Tool-Call Dispatch as Async Tasks

Tool calls spawn as independent Tokio tasks using tokio::spawn, which places them onto the work-stealing thread pool. This means multiple tools execute concurrently. The timeout applies when awaiting the JoinHandle, not at the spawn point, allowing the task to run freely while the orchestrator enforces a deadline when collecting results. When a timeout fires, the CancellationToken is cancelled, signaling the task to stop and release its resources. The orchestrator’s match arms distinguish between panicked tasks and timed-out tasks to avoid silently swallowing errors.

Redis as the State and Fan-Out Layer

Turn State Persistence

useredis::AsyncCommands;usecrate::AppState;pubasyncfnsave_context(state:&AppState,session_id:&str,messages:&[serde_json::Value],)->Result<(),Box<dynstd::error::Error+Send+Sync>>{if session_id.is_empty()||!session_id.chars().all(|c| c.is_alphanumeric()|| c =='-'){returnErr("invalid session_id: must be alphanumeric or hyphens only".into(),);}letmut conn = state.redis_conn.clone();let serialized =serde_json::to_string(messages)?;conn.set_ex::<_, _,()>(format!("session:{session_id}:context"),serialized,3600,).await?;Ok(())}pubasyncfnload_context(state:&AppState,session_id:&str,)->Result<Vec<serde_json::Value>,Box<dynstd::error::Error+Send+Sync>>{if session_id.is_empty()||!session_id.chars().all(|c| c.is_alphanumeric()|| c =='-'){returnErr("invalid session_id: must be alphanumeric or hyphens only".into(),);}letmut conn = state.redis_conn.clone();let result:Option<String>= conn.get(format!("session:{session_id}:context")).await?;match result {Some(data)=>Ok(serde_json::from_str(&data)?),None=>Ok(Vec::new()),}}

Storing conversation context in Redis with a one-hour TTL ensures that state survives gateway restarts and is accessible from any instance behind a load balancer. The MultiplexedConnection allows multiple async commands to share a single TCP connection to Redis. It is Clone by design and handles internal multiplexing, so no Mutex is needed. Call .clone() to obtain a handle for each operation. A single multiplexed connection saturates around 200-300 concurrent sessions in typical workloads; beyond that, replace it with a pool of 8-16 connections using bb8 or deadpool-redis.

Pub/Sub for Multi-Client Fan-Out

When multiple clients need to observe the same agent session (for example, a monitoring dashboard alongside the primary chat interface), Redis pub/sub provides cross-process event distribution. Each gateway instance subscribes to a channel keyed by session ID. The orchestrator publishes each AgentEvent to this channel in addition to sending it through the local mpsc channel. Subscriber instances map incoming pub/sub messages into SSE streams for their connected clients.

Pub/sub requires a separate redis::aio::PubSub connection; it cannot share the multiplexed command connection. Create a dedicated connection for subscriptions.

Use this pattern only when fan-out crosses process boundaries. For single-instance deployments or when all observers connect to the same gateway instance, tokio::broadcast channels avoid the Redis round-trip entirely (sub-microsecond dispatch vs. ~0.1-0.3 ms for Redis on localhost).

Performance Hardening and Production Concerns

Backpressure and Client Disconnect Handling

The bounded mpsc channel with a capacity of 64 events is the primary backpressure mechanism. When the channel is full, the orchestrator blocks on send().await, which slows LLM response consumption and prevents unbounded memory growth. When the client drops the connection, Axum drops the response body future, which drops the ReceiverStream, which drops the mpsc::Receiver. The next send() in the orchestrator returns an error, and the task logs and terminates. Between the client disconnect and the send error, up to 64 events worth of delay may occur depending on how many events the channel has buffered. For more granular shutdown coordination, share a CancellationToken between the handler and the spawned task.

Connection Pooling and Re

The ConcurrencyLimitLayer configured at 256 limits the number of simultaneously active agent sessions. Beyond this threshold, the gateway rejects incoming requests with an HTTP 503 response via the HandleErrorLayer, preventing resource exhaustion. Redis connection pool sizing should match the expected concurrency; a single multiplexed connection handles roughly 200-300 concurrent sessions before saturating, but a pool of 8 to 16 connections is appropriate for hundreds of concurrent sessions. The server constructs the reqwest client once and reuses it across requests, maintaining a connection pool to the LLM provider with HTTP keep-alive to amortize TLS handshake costs.

Observability

Wrap each agent turn in a tracing::instrument span that captures the session ID, turn number, and model name. Key metrics to export include time-to-first-token (the interval between the handler receiving the request and emitting the first TokenChunk), tokens per second (measuring throughput through the gateway), and turn count per request (tracking how many LLM round-trips each user message triggers). These metrics surface degradation in the LLM provider, tool execution latency, or Redis contention.

Benchmarking the Gateway

Benchmark Setup

Build a benchmark suite using k6 for HTTP-level load testing and a custom Rust harness for measuring internal pipeline latencies. Benchmark scenarios include ramping concurrent SSE connections from 10 to 1,000, sustained streaming throughput over 60-second windows, and P99 latency measurement under load. A mock LLM server that returns deterministic streaming responses eliminates provider variability from measurements.

Results Comparison

The following are representative benchmark figures from a 4-core, 8 GB machine using a mock LLM server. Baselines were Node.js 20 with express-sse and Python 3.12 with FastAPI + uvicorn, all using the same mock LLM server. Metrics are gateway-internal (listener receipt to first SSE byte sent); end-to-end latency including network hops will be higher. Absolute numbers will vary with hardware, OS, and kernel tuning.

The Rust gateway’s memory profile remains nearly flat as connection count increases because each SSE connection holds only a small channel buffer and a lightweight Tokio task. The Node.js and Python baselines allocate per-connection buffers and carry runtime overhead that scales linearly. Readers should reproduce these benchmarks on their target hardware, as numbers will vary with core count, network configuration, and kernel tuning.

Deployment Considerations

Configure the gateway following the 12-factor pattern: extract LLM API keys, Redis URLs (via REDIS_URL), bind addresses, and concurrency limits from environment variables. Use gcr.io/distroless/static as the base image for containerized deployments. A statically linked Rust binary built with musl produces a container image under 15 MB. A bare scratch image omits CA certificates, causing TLS failures when connecting to external APIs like OpenAI. Reserve scratch only for non-TLS internal services.

Health check (/health) and readiness endpoints should verify Redis connectivity before reporting the instance as ready. TLS termination is best handled at the load balancer level in Kubernetes deployments; if end-to-end encryption within the cluster is required, Axum supports rustlstion

What Comes Next

Natural extensions include a WebSocket upgrade path for bidirectional agent communication, multi-model routing that selects providers based on task type or cost constraints, and authentication middleware using tower layers for API key validation or JWT verification. The architecture choices (bounded channels for backpressure, SSE over WebSockets for unidirectional streaming, Redis for cross-instance coordination) each reflect specific production constraints rather than abstract preferences. Adapt the gateway to a production agent workflow by wiring in your tool registry, adding authentication, and tuning concurrency limits for your deployment target.

Sharing our passion for building incredible internet things.