# ADR-038: Multi-Worker Session Affinity for SSE and Streamable HTTP > MCP Gateway supports horizontal scaling with multiple worker processes (e.g., gunicorn -w 4). - Skill: `tools-only/adr-038-multi-worker-session-affinity-for-sse-and` (Agent Skill, multi-file: 3 files) - Install (CLI): `npx skillmds@latest add tools-only/adr-038-multi-worker-session-affinity-for-sse-and` - Raw SKILL.md: https://api.skillmd.com/api/skills/tools-only/adr-038-multi-worker-session-affinity-for-sse-and/raw - Safety review: pending (external: skill-scanner PASS, skillspector PASS) - Works with: Claude Code, Claude.ai, OpenAI Codex - Category: Coding & Dev Tools - Author: tools-only (https://skillmd.com/u/tools-only) - Updated: 2026-09-29 - Page: https://skillmd.com/skills/tools-only/adr-038-multi-worker-session-affinity-for-sse-and --- # ADR-038: Multi-Worker Session Affinity for SSE and Streamable HTTP - *Status:* Accepted - *Date:* 2025-01-31 - *Deciders:* Platform Team ## Context MCP Gateway supports horizontal scaling with multiple worker processes (e.g., `gunicorn -w 4`). When clients connect via SSE or Streamable HTTP, the gateway maintains pooled sessions to backend MCP servers for efficiency (see ADR-032). **The Problem:** In a multi-worker deployment, a client's requests may hit different workers. If each worker creates its own upstream MCP session, we lose: 1. **Connection efficiency** - Multiple sessions to the same backend instead of one 2. **Session state** - Some backends maintain state per session 3. **Resource waste** - Each session consumes memory and connections **The Goal:** Route all requests from the same client to the same upstream MCP session, regardless of which worker receives the HTTP request. ## Decision Implement **unified session affinity** using Redis Pub/Sub for cross-worker coordination: 1. **SSE Transport**: Uses Redis Pub/Sub for JSON-RPC message routing via `broadcast()` → `respond()` pattern 2. **Streamable HTTP Transport**: Uses Redis Pub/Sub for full HTTP request forwarding via `forward_streamable_http_to_owner()` pattern Both transports share the core session pool (`MCPSessionPool`), ownership registration (`register_session_mapping()`), and **Redis Pub/Sub communication mechanism**. The only difference is the payload format (JSON-RPC messages vs full HTTP requests). ## Architecture Overview ``` ┌─────────────────────────────────────────────────────────────────────────────────┐ │ MULTI-WORKER SESSION AFFINITY │ ├─────────────────────────────────────────────────────────────────────────────────┤ │ │ │ ┌─────────────┐ ┌─────────────┐ ┌─────────────┐ │ │ │ WORKER_A │ │ WORKER_B │ │ WORKER_C │ │ │ │ │ │ │ │ │ │ │ │ ┌─────────┐ │ │ ┌─────────┐ │ │ ┌─────────┐ │ │ │ │ │ Session │ │ │ │ Session │ │ │ │ Session │ │ MCPSessionPool │ │ │ │ Pool │ │ │ │ Pool │ │ │ │ Pool │ │ (per worker) │ │ │ └─────────┘ │ │ └─────────┘ │ │ └─────────┘ │ │ │ └──────┬──────┘ └──────┬──────┘ └──────┬──────┘ │ │ │ │ │ │ │ └──────────────────┼──────────────────┘ │ │ │ │ │ ┌──────▼──────┐ │ │ │ Redis │ Ownership Registry: │ │ │ │ mcpgw:pool_owner:{session_id} → WORKER_ID │ │ │ - Pub/Sub │ Session Mapping: │ │ │ - Keys │ mcpgw:session_mapping:{...} → pool_key │ │ └─────────────┘ │ │ │ └─────────────────────────────────────────────────────────────────────────────────┘ ``` ## SSE Transport Flow SSE uses a **persistent connection** from client to gateway. Only one worker owns the SSE stream, so all responses MUST go through that worker. ### Key Components | Component | Location | Purpose | |-----------|----------|---------| | `SessionRegistry` | `mcpgateway/cache/session_registry.py` | Manages SSE sessions and message routing | | `SSETransport` | `mcpgateway/transports/sse_transport.py` | Handles SSE stream I/O | | `broadcast()` | `session_registry.py:961` | Routes messages to session owner via Redis/DB | | `respond()` | `session_registry.py:1119` | Listens for messages and processes them | | `generate_response()` | `session_registry.py:1863` | Executes requests via internal `/rpc` call | | `_register_session_mapping()` | `session_registry.py:900` | Registers session ownership in pool | ### Sequence Diagram ```mermaid sequenceDiagram participant Client participant LB as Load Balancer participant WB as WORKER_B participant Redis participant WA as WORKER_A (SSE Owner) participant Backend as Backend MCP Server Note over Client,WA: 1. SSE Connection Established Client->>LB: GET /sse LB->>WA: Route to WORKER_A WA->>WA: Create SSETransport WA->>WA: Start respond() loop WA-->>Client: SSE Stream Connected Note over Client,Backend: 2. Tool Call (may hit different worker) Client->>LB: POST /message {tools/call} LB->>WB: Route to WORKER_B (different worker!) WB->>WB: broadcast(session_id, message) WB->>WB: _register_session_mapping() WB->>Redis: SETNX mcpgw:pool_owner:{session_id} Note over WB,Redis: Fails if WORKER_A already owns WB->>Redis: Publish to session channel Note over WA: respond() receives message Redis-->>WA: Message received WA->>WA: generate_response() WA->>WA: POST /rpc (internal) WA->>WA: tool_service.invoke_tool() WA->>WA: MCPSessionPool.acquire() WA->>Backend: Execute tool Backend-->>WA: Result WA->>WA: transport.send_message() WA-->>Client: SSE: {result} ``` ### Detailed Flow ``` ┌─────────────────────────────────────────────────────────────────────────────────┐ │ SSE SESSION AFFINITY FLOW │ ├─────────────────────────────────────────────────────────────────────────────────┤ │ │ │ CLIENT │ │ │ │ │ │ GET /sse │ │ ▼ │ │ WORKER_A (receives SSE connection) │ │ │ │ │ ├─► session_registry.add_session(session_id, transport) │ │ │ │ │ └─► asyncio.create_task(session_registry.respond(...)) │ │ │ │ │ └─► Listens on Redis channel: {session_id} │ │ │ │ ───────────────────────────────────────────────────────────────────────────── │ │ │ │ CLIENT │ │ │ │ │ │ POST /message {"method": "tools/call", "params": {"name": "my_tool"}} │ │ ▼ │ │ WORKER_B (receives HTTP request - different worker!) │ │ │ │ │ ├─► session_registry.broadcast(session_id, message) │ │ │ │ │ │ │ ├─► _register_session_mapping(session_id, message, user_email) │ │ │ │ │ │ │ │ │ ├─► tool_lookup_cache.get(tool_name) │ │ │ │ │ └─► Returns: {gateway: {url, id, transport}} │ │ │ │ │ │ │ │ │ └─► pool.register_session_mapping(session_id, url, ...) │ │ │ │ │ │ │ │ │ ├─► Redis SETNX mcpgw:pool_owner:{session_id} = WORKER_B │ │ │ │ │ └─► Returns False (WORKER_A already owns) │ │ │ │ │ │ │ │ │ └─► Ownership NOT changed (atomic SETNX) │ │ │ │ │ │ │ └─► Redis PUBLISH {session_id} {message} │ │ │ │ │ └─► Return HTTP 202 Accepted │ │ │ │ ───────────────────────────────────────────────────────────────────────────── │ │ │ │ WORKER_A (respond() loop receives message) │ │ │ │ │ ├─► pubsub.get_message() returns the broadcast message │ │ │ │ │ └─► generate_response(message, transport, ...) │ │ │ │ │ ├─► POST http://127.0.0.1:{port}/rpc (internal call) │ │ │ │ │ │ │ └─► /rpc handler │ │ │ │ │ │ │ ├─► Check x-forwarded-internally header │ │ │ │ │ │ │ └─► tool_service.invoke_tool() │ │ │ │ │ │ │ └─► MCPSessionPool.acquire() │ │ │ │ │ │ │ └─► Use pooled session to backend │ │ │ │ │ └─► transport.send_message(response) │ │ │ │ │ └─► SSE Stream ──────────────────────────────────────► CLIENT │ │ │ └─────────────────────────────────────────────────────────────────────────────────┘ ``` ## Streamable HTTP Transport Flow Streamable HTTP uses **independent HTTP request/response cycles**. Any worker can respond to the client, so we use **Redis Pub/Sub forwarding** to route requests to the session owner, which then executes them via internal `/rpc` routing. ### Why Redis Pub/Sub Instead of HTTP Forwarding? An earlier implementation attempted HTTP-based forwarding using the worker's hostname extracted from `WORKER_ID` (format: `hostname:pid`). This approach **failed for single-host multi-worker deployments** because: - All workers share the same hostname (e.g., gunicorn with 4 workers on one server) - Worker IDs: `myserver:1234`, `myserver:5678`, `myserver:9012` - Extracting hostname → `myserver` for all workers - HTTP forwarding to `http://myserver:4444/mcp` routes to load balancer, not specific worker PID - Result: Random worker receives request, session not found, error **Redis Pub/Sub solution** works universally: - Each worker listens on unique channel: `mcpgw:pool_http:{worker_id}` - Direct worker-to-worker messaging without hostname routing - Works identically for single-host and multi-host deployments ### Why Not Use SDK's Session Manager? The MCP SDK's `StreamableHTTPSessionManager` stores sessions in an in-memory `_server_instances` dictionary. This is problematic because: 1. **Sessions are per-worker** - Not shared across workers 2. **Sessions get cleaned up** - SDK clears `_server_instances` between requests 3. **RedisEventStore only handles events** - It stores events for resumability, not session routing Our solution: **bypass the SDK for request routing** and use `/rpc` endpoint directly, which leverages `MCPSessionPool` for upstream connections. ### Key Components | Component | Location | Purpose | |-----------|----------|---------| | `handle_streamable_http()` | `streamablehttp_transport.py:1252` | ASGI handler with affinity routing | | `forward_streamable_http_to_owner()` | `mcp_session_pool.py:1651` | Redis Pub/Sub forwarding to owner worker | | `_execute_forwarded_http_request()` | `mcp_session_pool.py:1554` | Executes forwarded HTTP requests on owner | | `get_streamable_http_session_owner()` | `mcp_session_pool.py:1545` | Checks Redis for session ownership | | `start_rpc_listener()` | `mcp_session_pool.py:1438` | Listens on both RPC and HTTP Redis channels | | `/rpc` endpoint | `main.py:5259` | Unified request handler for all methods | | `register_session_mapping()` | `mcp_session_pool.py:545` | Registers session ownership atomically | | `WORKER_ID` | `mcp_session_pool.py:61` | Unique identifier: `{hostname}:{pid}` | ### Sequence Diagram ```mermaid sequenceDiagram participant Client participant LB as Load Balancer participant WB as WORKER_B participant Redis participant WA as WORKER_A (Session Owner) participant Backend as Backend MCP Server Note over Client,Backend: 1. Initialize - Establishes Ownership Client->>LB: POST /mcp {initialize} LB->>WA: Route to WORKER_A WA->>WA: SDK creates session, returns mcp-session-id WA->>Redis: SETNX mcpgw:pool_owner:{session_id} = host_a:1 WA-->>Client: HTTP Response {mcp-session-id: ABC} Note over Client,Backend: 2. Subsequent Request - Same Worker (Owner) Client->>LB: POST /mcp {tools/call} LB->>WA: Route to WORKER_A WA->>Redis: GET mcpgw:pool_owner:ABC → host_a:1 (us!) WA->>WA: Route to /rpc (bypass SDK sessions) WA->>WA: tool_service.invoke_tool() WA->>WA: MCPSessionPool.acquire() WA->>Backend: Execute tool Backend-->>WA: Result WA-->>Client: HTTP Response {result} Note over Client,Backend: 3. Subsequent Request - Different Worker Client->>LB: POST /mcp {tools/call, mcp-session-id: ABC} LB->>WB: Route to WORKER_B (different worker!) WB->>Redis: GET mcpgw:pool_owner:ABC Redis-->>WB: host_a:1 (not us!) WB->>WB: forward_streamable_http_to_owner() WB->>Redis: Subscribe to response channel WB->>Redis: PUBLISH to mcpgw:pool_http:host_a:1 (hex-encoded request) Note over WA: start_rpc_listener() receives message Redis-->>WA: HTTP forward message WA->>WA: _execute_forwarded_http_request() WA->>WA: POST /rpc (internal, x-forwarded-internally: true) WA->>WA: tool_service.invoke_tool() WA->>Backend: Execute tool Backend-->>WA: Result WA->>Redis: PUBLISH response (hex-encoded) Redis-->>WB: Response received WB-->>Client: HTTP Response {result} ``` ### Detailed Flow ``` ┌─────────────────────────────────────────────────────────────────────────────────┐ │ STREAMABLE HTTP SESSION AFFINITY FLOW │ ├─────────────────────────────────────────────────────────────────────────────────┤ │ │ │ CLIENT │ │ │ │ │ │ POST /mcp │ │ │ Headers: mcp-session-id: ABC123 │ │ │ Body: {"method": "tools/call", "params": {"name": "my_tool"}} │ │ ▼ │ │ WORKER_B (receives HTTP request) - hostname:pid = "host_b:1" │ │ │ │ │ └─► handle_streamable_http(scope, receive, send) │ │ │ │ │ ├─► Extract mcp_session_id from headers │ │ │ (checks both "mcp-session-id" and "x-mcp-session-id") │ │ │ │ │ ├─► pool.get_streamable_http_session_owner(mcp_session_id) │ │ │ │ │ │ │ └─► Redis GET mcpgw:pool_owner:{session_id} │ │ │ └─► Returns: "host_a:1" (WORKER_A owns it) │ │ │ │ │ ├─► owner != WORKER_ID → Forward HTTP request │ │ │ │ │ └─► pool.forward_streamable_http_to_owner(...) │ │ │ │ │ ├─► Generate unique response channel UUID │ │ │ response_channel = mcpgw:pool_http_response:{uuid} │ │ │ │ │ ├─► Serialize HTTP request for Redis: │ │ │ { │ │ │ "type": "http_forward", │ │ │ "response_channel": response_channel, │ │ │ "method": "POST", │ │ │ "path": "/mcp", │ │ │ "headers": {...}, │ │ │ "body": "..." (hex-encoded), │ │ │ "original_worker": "host_b:1" │ │ │ } │ │ │ │ │ ├─► Subscribe to response channel (prevent race) │ │ │ │ │ ├─► Redis PUBLISH to mcpgw:pool_http:host_a:1 │ │ │ │ │ ├─► Wait for response on response_channel (with timeout) │ │ │ │ │ : (see WORKER_A processing below) │ │ : │ │ ▼ │ │ Response received from Redis │ │ │ │ │ ├─► Decode hex body back to bytes │ │ │ │ │ └─► Forward response to client ──────────────────────► CLIENT │ │ │ │ ───────────────────────────────────────────────────────────────────────────── │ │ │ │ WORKER_A (start_rpc_listener receives Redis message) - hostname:pid = "host_a:1" │ │ │ │ │ └─► start_rpc_listener() loop │ │ │ │ │ ├─► Redis message received on mcpgw:pool_http:host_a:1 │ │ │ type = "http_forward" │ │ │ │ │ └─► _execute_forwarded_http_request(request, redis) │ │ │ │ │ ├─► Deserialize HTTP request from Redis message │ │ │ │ │ ├─► Decode hex body back to bytes │ │ │ │ │ ├─► Add x-forwarded-internally: true header │ │ │ │ │ ├─► HTTP POST http://127.0.0.1:4444/mcp │ │ │ Headers: │ │ │ x-forwarded-internally: true │ │ │ x-original-worker: host_b:1 │ │ │ mcp-session-id: ABC123 │ │ │ (+ all original headers) │ │ │ │ │ └─► /rpc handler (main.py) │ │ │ │ │ ├─► Sees x-forwarded-internally: true │ │ │ └─► Skip session affinity check (no loops) │ │ │ │ │ └─► tool_service.invoke_tool() │ │ │ │ │ ├─► Normalize session ID from headers │ │ │ (both mcp-session-id and x-mcp-session-id) │ │ │ │ │ └─► MCPSessionPool.acquire() │ │ │ │ │ ├─► Reuse pooled connection to backend │ │ │ │ │ └─► Execute tool on upstream server │ │ │ │ │ └─► Response returned to _execute...() │ │ │ │ │ ├─► Hex-encode response body │ │ │ │ │ └─► Redis PUBLISH to response_channel │ │ │ │ ───────────────────────────────────────────────────────────────────────────── │ │ │ │ ALTERNATE: WORKER_A is the owner AND receives request directly │ │ │ │ │ └─► handle_streamable_http(scope, receive, send) │ │ │ │ │ ├─► is_internally_forwarded = False │ │ │ │ │ ├─► pool.get_streamable_http_session_owner(mcp_session_id) │ │ │ └─► Returns: "host_a:1" (that's us!) │ │ │ │ │ ├─► owner == WORKER_ID → We own it, but DON'T use SDK │ │ │ (SDK's _server_instances is empty between requests) │ │ │ │ │ └─► Route to /rpc (same path as forwarded requests) │ │ │ │ │ └─► tool_service.invoke_tool() → MCPSessionPool → Backend │ │ │ └─────────────────────────────────────────────────────────────────────────────────┘ ``` ### Why Route to /rpc Instead of SDK? The MCP SDK's `StreamableHTTPSessionManager` has a fundamental issue: ```python # SDK stores sessions in memory self._server_instances: dict[str, ServerInstance] = {} ``` **Problem:** This dictionary is cleared between requests, causing "Session not found" errors. **Solution:** Route ALL requests with session IDs to `/rpc`, which: 1. Uses `MCPSessionPool` for upstream connections (Redis-backed) 2. Doesn't depend on SDK's in-memory session storage 3. Works identically for forwarded and local requests ## Comparison: SSE vs Streamable HTTP | Aspect | SSE | Streamable HTTP | |--------|-----|-----------------| | **Client Connection** | Persistent SSE stream | Independent HTTP requests | | **Response Path** | Via SSE stream on owner worker | Via HTTP on any worker (forwarded if needed) | | **Routing Mechanism** | `broadcast()` → Redis Pub/Sub → `respond()` | Redis Pub/Sub → `_execute_forwarded_http_request()` | | **Message Format** | JSON-RPC message only | Full HTTP request (method, headers, body) | | **Message Storage** | Redis/DB (persistent until consumed) | Redis Pub/Sub only (synchronous) | | **Latency** | Higher (message queue + polling) | Lower (direct pub/sub) | | **SDK Dependency** | Uses SDK for SSE streaming | Bypasses SDK for session routing | | **Hex Encoding** | Not needed (JSON only) | Required (binary HTTP bodies) | ### Why Same Mechanism, Different Payloads? **Both transports now use Redis Pub/Sub** for consistency and reliability: **SSE Transport:** - Forwards JSON-RPC messages via `mcpgw:pool_rpc:{worker_id}` - Owner worker's `respond()` loop processes messages - Responses sent via persistent SSE stream back to client **Streamable HTTP Transport:** - Forwards full HTTP requests via `mcpgw:pool_http:{worker_id}` - Owner worker's `_execute_forwarded_http_request()` executes locally - Responses returned via Redis Pub/Sub to requesting worker - Requesting worker sends HTTP response to client **Benefits of unified approach:** - Works in ALL deployment scenarios (single-host, multi-host, containers) - No dependency on hostname-based routing - Consistent debugging and monitoring - Same timeout and error handling logic ## Redis Keys | Key Pattern | Purpose | TTL | |-------------|---------|-----| | `mcpgw:pool_owner:{session_id}` | Worker ID that owns the session (e.g., `host_a:1`) | Configurable (default 5min) | | `mcpgw:session_mapping:{session_id}:{url}:{transport}:{gateway_id}` | Pool key for session | Configurable | | `mcpgw:pool_rpc:{worker_id}` | Pub/sub channel for SSE JSON-RPC forwards | N/A (pub/sub) | | `mcpgw:pool_rpc_response:{uuid}` | Pub/sub channel for SSE responses | N/A (pub/sub) | | `mcpgw:pool_http:{worker_id}` | Pub/sub channel for Streamable HTTP forwards | N/A (pub/sub) | | `mcpgw:pool_http_response:{uuid}` | Pub/sub channel for HTTP responses | N/A (pub/sub) | | `mcpgw:eventstore:{stream_id}:events` | Sorted set for event storage (resumability) | Configurable | | `mcpgw:eventstore:{stream_id}:meta` | Hash for stream metadata | Configurable | | `mcpgw:eventstore:event_index` | Hash mapping event_id to stream | Configurable | **Note:** Both SSE and Streamable HTTP use Redis Pub/Sub for worker-to-worker communication. The difference is payload format: JSON-RPC messages for SSE, full HTTP requests for Streamable HTTP. ## Configuration Session affinity requires Redis for cross-worker coordination. The following environment variables must be configured: ```bash # REQUIRED: Enable Redis client (session affinity depends on Redis) CACHE_TYPE=redis REDIS_URL=redis://localhost:6379/0 # REQUIRED: Enable stateful sessions for Streamable HTTP transport USE_STATEFUL_SESSIONS=true # REQUIRED: Enable session affinity for multi-worker deployments MCPGATEWAY_SESSION_AFFINITY_ENABLED=true # OPTIONAL: Session ownership TTL (seconds, default: 300) MCPGATEWAY_SESSION_AFFINITY_TTL=300 # OPTIONAL: Forwarded request timeout (seconds, default: 30) MCPGATEWAY_POOL_RPC_FORWARD_TIMEOUT=30 ``` **Important Notes**: - `CACHE_TYPE=redis` is **required** - the Redis client will not initialize without it - Session affinity works independently of session pooling - the MCP session pool will be initialized automatically when `MCPGATEWAY_SESSION_AFFINITY_ENABLED=true`, even if `MCP_SESSION_POOL_ENABLED=false` - Redis must be accessible at the configured `REDIS_URL` before starting the application ## Atomic Ownership with SETNX Ownership is registered atomically using Redis `SETNX` (SET if Not eXists): ```python # In register_session_mapping() owner_key = f"mcpgw:pool_owner:{mcp_session_id}" was_set = await redis.setnx(owner_key, WORKER_ID) if was_set: # We claimed it - set TTL await redis.expire(owner_key, ttl) else: # Another worker already owns it existing_owner = await redis.get(owner_key) ``` This ensures: 1. **First worker wins** - Only one worker can claim ownership 2. **No race conditions** - SETNX is atomic 3. **Subsequent calls respect ownership** - SETNX fails, existing owner returned ## Worker Identification Each worker has a unique `WORKER_ID` using the format `{hostname}:{pid}`: ```python # In mcp_session_pool.py import socket import os WORKER_ID = f"{socket.gethostname()}:{os.getpid()}" # Example: "gateway-container-abc:1" ``` **Why hostname:pid?** - In Docker, each container has PID 1 for the main process - Using PID alone would cause collisions (`1` vs `1`) - Hostname is unique per container (container ID or configured hostname) - Combined format ensures uniqueness across containers and processes **Important Note on Single-Host Deployments:** While `hostname:pid` provides unique worker identification, it **cannot be used for HTTP routing** in single-host multi-worker scenarios. All workers on the same host share the same hostname, making hostname-based HTTP forwarding fail. This is why Redis Pub/Sub is used instead - each worker listens on a unique Redis channel based on the full `WORKER_ID` (including PID), enabling direct worker-to-worker communication without hostname routing. ## Deployment Compatibility | Deployment Type | Hostname Uniqueness | Worker ID Example | Compatibility | |----------------|---------------------|-------------------|---------------| | Kubernetes StatefulSet | ✅ Unique per pod | `mcpgateway-0:1`, `mcpgateway-1:1` | ✅ Fully compatible | | Docker Compose (multi-container) | ✅ Unique per container | `gateway-1:1`, `gateway-2:1` | ✅ Fully compatible | | Kubernetes Deployment (replicas) | ✅ Unique per pod | `mcpgateway-abc123:1`, `mcpgateway-def456:1` | ✅ Fully compatible | | **Gunicorn multi-worker (single host)** | ❌ Shared hostname | `myserver:1234`, `myserver:5678` | ✅ Fixed with Redis Pub/Sub | | **Docker single container + gunicorn** | ❌ Shared hostname | `container:1`, `container:2` | ✅ Fixed with Redis Pub/Sub | | **Traditional server + gunicorn** | ❌ Shared hostname | `prod-server:1234`, `prod-server:5678` | ✅ Fixed with Redis Pub/Sub | **Key Insight:** Redis Pub/Sub channels use the full `WORKER_ID` (including PID), so even when hostnames are identical, each worker listens on a unique channel. This makes the solution work universally across all deployment scenarios. ## Binary Data Encoding Since Redis Pub/Sub transports messages as strings (JSON), binary HTTP request/response bodies must be encoded: ```python # In forward_streamable_http_to_owner() - encoding request forward_data = { "type": "http_forward", "body": body.hex() if body else "", # bytes → hex string ... } # In _execute_forwarded_http_request() - decoding request body = bytes.fromhex(body_hex) if body_hex else b"" # In _execute_forwarded_http_request() - encoding response response_data = { "body": response.content.hex(), # bytes → hex string ... } # In forward_streamable_http_to_owner() - decoding response response_data["body"] = bytes.fromhex(body_hex) if body_hex else b"" ``` **Performance Impact:** - Hex encoding doubles payload size (each byte → 2 hex chars) - Negligible for typical JSON-RPC requests (<10KB) - Redis default message limit is ~512MB (sufficient for MCP use cases) - Encoding/decoding is fast (built-in Python operations) ## Loop Prevention When forwarding requests, we prevent infinite loops with the `x-forwarded-internally` header: ```python # In _execute_forwarded_http_request() when making internal call internal_headers["x-forwarded-internally"] = "true" internal_headers["x-original-worker"] = request.get("original_worker", "unknown") # In handle_streamable_http() when receiving is_internally_forwarded = headers.get("x-forwarded-internally") == "true" if is_internally_forwarded: # Route to /rpc, don't check affinity again # In /rpc handler (main.py) if is_internally_forwarded and mcp_session_id: # Execute locally - don't forward again ``` This prevents the following loop: 1. WORKER_B receives request for session owned by WORKER_A 2. WORKER_B forwards via Redis to WORKER_A 3. WORKER_A's `_execute_forwarded_http_request()` adds `x-forwarded-internally: true` 4. WORKER_A makes internal HTTP call to `127.0.0.1:4444/mcp` 5. Load balancer routes to... WORKER_A (with header set) 6. WORKER_A sees header, skips affinity check, executes locally ✓ ## Startup Initialization The RPC/HTTP listener must be started during application startup: ```python # In main.py lifespan if settings.mcpgateway_session_affinity_enabled: pool = get_mcp_session_pool() pool._rpc_listener_task = asyncio.create_task(pool.start_rpc_listener()) ``` The `start_rpc_listener()` subscribes to both channels: - `mcpgw:pool_rpc:{WORKER_ID}` - for SSE JSON-RPC forwards - `mcpgw:pool_http:{WORKER_ID}` - for Streamable HTTP request forwards Messages are routed based on the `type` field: - `"rpc_forward"` → `_execute_forwarded_request()` (SSE) - `"http_forward"` → `_execute_forwarded_http_request()` (Streamable HTTP) ## Consequences ### Positive - Enables horizontal scaling with session affinity - Reuses upstream MCP sessions efficiently via `MCPSessionPool` - Works transparently for both SSE and Streamable HTTP - Atomic ownership prevents race conditions (SETNX) - Bypasses SDK session issues for reliable operation - **Universal deployment compatibility** - works for single-host and multi-host deployments - **Consistent communication mechanism** - both transports use Redis Pub/Sub - **No hostname routing dependency** - eliminates single-host multi-worker failures ### Negative - Requires Redis for multi-worker deployments - Adds latency for cross-worker requests (~1-2ms Redis roundtrip) - More complex debugging (requests may span workers via Redis) - SDK is partially bypassed, requiring maintenance of parallel path - Binary data requires hex encoding (2x payload size in Redis) ### Neutral - Unified Redis Pub/Sub approach for both transports (different payload formats) - Streamable HTTP routing could be simplified further by eliminating SDK dependency entirely ## Future Improvements The current implementation works around SDK limitations. A cleaner approach would be: 1. **Remove SDK session manager** for Streamable HTTP entirely 2. **Parse JSON-RPC directly** in `handle_streamable_http()` 3. **Route all requests to `/rpc`** from the start 4. **Handle SSE streaming** separately without SDK session state This would eliminate the SDK dependency for session routing while keeping it only for protocol compliance (JSON-RPC formatting, SSE streaming). ## Troubleshooting ### Redis Client Returns None **Symptoms**: Logs show `[REDIS_DEBUG] Redis client is None, cannot register session ownership` **Cause**: Redis client is not initialized because `CACHE_TYPE` is not set to `redis`. **Solution**: ```bash # Add to .env CACHE_TYPE=redis REDIS_URL=redis://localhost:6379/0 ``` Verify Redis is running: ```bash redis-cli ping # Should return "PONG" ``` ### MCP Session Pool Not Initialized **Symptoms**: Error message `MCP session pool not initialized. Call init_mcp_session_pool() first.` **Cause**: Session pool initialization was skipped because both `MCP_SESSION_POOL_ENABLED=false` and `MCPGATEWAY_SESSION_AFFINITY_ENABLED=false`. **Solution**: The session pool is automatically initialized when session affinity is enabled. Ensure: ```bash MCPGATEWAY_SESSION_AFFINITY_ENABLED=true ``` The pool will initialize even if `MCP_SESSION_POOL_ENABLED=false` (affinity needs the ownership registry but not the full pooling functionality). ### Session Affinity Not Working (No "Owner from Redis" in Logs) **Symptoms**: Requests fail with "No valid session ID provided" or session state is lost between requests. **Root Cause Checklist**: 1. ✅ `CACHE_TYPE=redis` - Redis client must be enabled 2. ✅ `USE_STATEFUL_SESSIONS=true` - Required for Streamable HTTP stateful sessions 3. ✅ `MCPGATEWAY_SESSION_AFFINITY_ENABLED=true` - Enables session affinity 4. ✅ Redis is accessible at `REDIS_URL` 5. ✅ Server restarted after changing `.env` **Debug Logging**: Look for these log messages during startup: ``` Redis client initialized: parser=AsyncHiredisParser, pool_size=10... MCP session pool initialized Multi-worker session affinity RPC listener started ``` During requests: ``` [HTTP_AFFINITY_SDK] Worker hostname:pid | Session abc12345... | Registered ownership after SDK handling [HTTP_AFFINITY_CHECK] Worker hostname:pid | Session abc12345... | Owner from Redis: hostname:pid ``` ## References - [ADR-032: MCP Session Pool](./032-mcp-session-pool.md) - [Redis SETNX Documentation](https://redis.io/commands/setnx/) - [MCP Protocol Specification](https://modelcontextprotocol.io/)