example-remote-server/docs/streamable-http-design.md

Design Document: Streamable HTTP Transport Implementation

Current Implementation

Dual Transport Architecture

The example remote server implements both transport methods:

Legacy SSE Transport:

1. SSE Endpoint: /sse - Creates SSE connection using SSEServerTransport
2. Message Endpoint: /message - Receives POST requests and forwards them via Redis

Modern Streamable HTTP Transport:

1. Unified Endpoint: /mcp - Handles GET, POST, DELETE with StreamableHTTPServerTransport
2. Stateful Sessions: Requires initialization and session ID tracking
3. SSE Response Format: Returns results via Server-Sent Events streams

Shared Infrastructure:

1. Redis Integration: Messages published/subscribed through Redis channels using session IDs
2. Auth: Uses requireBearerAuth middleware with mode-dependent auth providers
3. Session Management: Session ownership tracked via Redis for multi-user isolation

Key Files:

• /src/index.ts:156-162 - SSE and Streamable HTTP endpoints with auth
• /src/handlers/sse.ts - SSE connection handler with Redis integration
• /src/handlers/shttp.ts - Streamable HTTP handler
• /src/services/mcp.ts - MCP server implementation with tools, resources, prompts

Streamable HTTP Transport Specification (2025-03-26)

The new Streamable HTTP transport replaces the old HTTP+SSE approach with a single endpoint that supports:

1. Single Endpoint: One URL that handles GET, POST, and DELETE methods
2. POST Requests: Send JSON-RPC messages, can return either JSON responses or SSE streams
3. GET Requests: Open SSE streams for server-to-client messages
4. Session Management: Optional session IDs in Mcp-Session-Id headers
5. Resumability: Optional event storage with Last-Event-ID support
6. Auth Integration: Same authentication patterns as SSE

Key Specification Requirements:

• Accept header must include both application/json and text/event-stream
• Session ID management via Mcp-Session-Id headers
• 202 Accepted for notifications/responses only
• SSE streams or JSON responses for requests
• Security: Origin validation, localhost binding, proper auth

TypeScript SDK Implementation

The SDK provides StreamableHTTPServerTransport with:

1. Two Modes:

   • Stateful: Session ID generator provided, maintains sessions in memory
   • Stateless: Session ID generator undefined, no session state

2. Key Features:

   • Built-in session validation
   • Event store support for resumability
   • Automatic response correlation
   • Auth info threading via req.auth

3. Integration Patterns:

   • Stateful: Store transports by session ID, reuse across requests
   • Stateless: New transport per request, immediate cleanup
   • Auth: Same bearer auth middleware as SSE

Implementation Plan

1. New Streamable HTTP Endpoint

Add /mcp endpoint that handles GET, POST, DELETE methods:

// In src/index.ts
app.get("/mcp", cors(corsOptions), bearerAuth, authContext, handleStreamableHTTP);  
app.post("/mcp", cors(corsOptions), bearerAuth, authContext, handleStreamableHTTP);
app.delete("/mcp", cors(corsOptions), bearerAuth, authContext, handleStreamableHTTP);

2. Handler Implementation

Create new handler in /src/handlers/mcp.ts:

export async function handleStreamableHTTP(req: Request, res: Response) {
  // Use same Redis-based architecture as SSE transport
  // but with StreamableHTTPServerTransport instead of SSEServerTransport
}

3. Transport Integration Strategy (Horizontally Scalable)

Redis-based Session Management (Required for Horizontal Scaling)

• Store session state in Redis, not in-memory
• Any server instance can handle any request for any session
• Session lifecycle independent of SSE connection lifecycle
• Message buffering in Redis when SSE connection is down
• Session TTL of 5 minutes to prevent Redis bloat

4. Redis Integration

Maintain current Redis architecture:

• Use session ID as Redis channel key
• Same message publishing/subscribing pattern
• Same MCP server creation logic
• Transport acts as bridge to Redis like current SSE implementation

5. Auth Integration

Use identical auth setup as SSE:

• Same bearerAuth middleware
• Same authContext middleware
• Same EverythingAuthProvider
• Auth info flows through req.auth to transport

6. Backwards Compatibility

Keep existing /sse and /message endpoints:

• Maintain current SSE transport for existing clients
• Add new /mcp endpoint alongside
• Both transports share same Redis infrastructure
• Same auth provider serves both

Key Differences from Current SSE Implementation

1. Single Endpoint: /mcp handles all HTTP methods vs separate /sse + /message
2. Transport Class: StreamableHTTPServerTransport vs SSEServerTransport
3. Session Headers: Mcp-Session-Id headers vs URL session ID
4. Request Handling: Transport handles HTTP details vs manual SSE headers
5. Response Correlation: Built into transport vs manual request tracking

Benefits of This Approach

1. Spec Compliance: Follows 2025-03-26 MCP specification exactly
2. Minimal Changes: Reuses existing Redis infrastructure and auth
3. Feature Parity: Same functionality as current SSE transport
4. Future Proof: Can add resumability with event store later
5. Clean Integration: Same auth patterns and middleware stack

Implementation Steps

1. Add Dependencies: StreamableHTTPServerTransport from SDK
2. Create Redis Session Management: Implement SessionManager and MessageDelivery classes
3. Create Handler: New streamable HTTP handler function with Redis integration
4. Add Routes: New /mcp endpoint with all HTTP methods
5. Session Management: Redis-based session storage with TTL
6. Message Buffering: Redis-based message buffering for disconnected clients
7. Testing: Verify auth, Redis integration, horizontal scaling, and MCP protocol compliance
8. Documentation: Update README with new endpoint usage

Additional Implementation Considerations

• Redis Connection Management: Ensure Redis connections are properly pooled and cleaned up
• Error Handling: Robust error handling for Redis operations and session timeouts
• Monitoring: Add logging for session creation, cleanup, and message buffering metrics
• Performance: Consider Redis memory usage and implement appropriate limits on message buffer size
• Security: Ensure session IDs are cryptographically secure and validate all session operations

Technical Details

Session Management Architecture (Redis-based)

Redis Data Structures for Horizontal Scaling:

// Redis keys for session management
const SESSION_METADATA_KEY = (sessionId: string) => `session:${sessionId}:metadata`;
const SESSION_MESSAGES_KEY = (sessionId: string) => `session:${sessionId}:messages`;
const SESSION_CONNECTION_KEY = (sessionId: string) => `session:${sessionId}:connection`;

// Session metadata structure
interface SessionMetadata {
  sessionId: string;
  clientId: string;
  createdAt: number;
  lastActivity: number;
}

// Session lifecycle management
class SessionManager {
  private static SESSION_TTL = 5 * 60; // 5 minutes in seconds
  
  static async createSession(sessionId: string, clientId: string): Promise<void> {
    const metadata: SessionMetadata = {
      sessionId,
      clientId,
      createdAt: Date.now(),
      lastActivity: Date.now()
    };
    
    // Store session metadata with TTL
    await redisClient.set(
      SESSION_METADATA_KEY(sessionId), 
      JSON.stringify(metadata),
      { EX: this.SESSION_TTL }
    );
    
    // Initialize empty message buffer
    await redisClient.del(SESSION_MESSAGES_KEY(sessionId));
    
    // Mark connection as disconnected initially
    await redisClient.set(SESSION_CONNECTION_KEY(sessionId), 'disconnected', { EX: this.SESSION_TTL });
  }
  
  static async refreshSession(sessionId: string): Promise<boolean> {
    const metadata = await this.getSessionMetadata(sessionId);
    if (!metadata) return false;
    
    // Update last activity and refresh TTL
    metadata.lastActivity = Date.now();
    await redisClient.set(
      SESSION_METADATA_KEY(sessionId), 
      JSON.stringify(metadata),
      { EX: this.SESSION_TTL }
    );
    
    // Refresh other keys too
    await redisClient.expire(SESSION_MESSAGES_KEY(sessionId), this.SESSION_TTL);
    await redisClient.expire(SESSION_CONNECTION_KEY(sessionId), this.SESSION_TTL);
    
    return true;
  }
  
  static async deleteSession(sessionId: string): Promise<void> {
    await redisClient.del(SESSION_METADATA_KEY(sessionId));
    await redisClient.del(SESSION_MESSAGES_KEY(sessionId));
    await redisClient.del(SESSION_CONNECTION_KEY(sessionId));
  }
  
  static async getSessionMetadata(sessionId: string): Promise<SessionMetadata | null> {
    const data = await redisClient.get(SESSION_METADATA_KEY(sessionId));
    return data ? JSON.parse(data) : null;
  }
  
  // Mark SSE connection as connected/disconnected
  static async setConnectionState(sessionId: string, connected: boolean): Promise<void> {
    await redisClient.set(
      SESSION_CONNECTION_KEY(sessionId), 
      connected ? 'connected' : 'disconnected',
      { EX: this.SESSION_TTL }
    );
  }
  
  static async isConnected(sessionId: string): Promise<boolean> {
    const state = await redisClient.get(SESSION_CONNECTION_KEY(sessionId));
    return state === 'connected';
  }
}

Redis Integration Pattern with Message Buffering

The implementation extends the current Redis pattern to support message buffering:

// Message delivery with buffering support
class MessageDelivery {
  static async deliverMessage(sessionId: string, message: JSONRPCMessage): Promise<void> {
    const isConnected = await SessionManager.isConnected(sessionId);
    
    if (isConnected) {
      // Direct delivery via existing Redis pub/sub
      const redisChannel = `mcp:${sessionId}`;
      await redisClient.publish(redisChannel, JSON.stringify(message));
    } else {
      // Buffer the message for later delivery
      await redisClient.lpush(
        SESSION_MESSAGES_KEY(sessionId), 
        JSON.stringify(message)
      );
      // Set TTL on the messages list
      await redisClient.expire(SESSION_MESSAGES_KEY(sessionId), SessionManager.SESSION_TTL);
    }
  }
  
  static async deliverBufferedMessages(sessionId: string, transport: StreamableHTTPServerTransport): Promise<void> {
    // Get all buffered messages
    const bufferedMessages = await redisClient.lrange(SESSION_MESSAGES_KEY(sessionId), 0, -1);
    
    // Deliver buffered messages in order (reverse because lpush)
    for (let i = bufferedMessages.length - 1; i >= 0; i--) {
      const message = JSON.parse(bufferedMessages[i]);
      await transport.send(message);
    }
    
    // Clear the buffer after delivery
    await redisClient.del(SESSION_MESSAGES_KEY(sessionId));
  }
}

// Enhanced Redis subscription for SSE connections
const setupRedisSubscription = async (sessionId: string, transport: StreamableHTTPServerTransport) => {
  const redisChannel = `mcp:${sessionId}`;
  
  const redisCleanup = await redisClient.createSubscription(
    redisChannel,
    async (message) => {
      const jsonMessage = JSON.parse(message);
      try {
        await transport.send(jsonMessage);
      } catch (error) {
        console.error(`Failed to send message on transport for session ${sessionId}:`, error);
        // Mark connection as disconnected so future messages get buffered
        await SessionManager.setConnectionState(sessionId, false);
      }
    },
    async (error) => {
      console.error('Redis subscription error:', error);
      await SessionManager.setConnectionState(sessionId, false);
    }
  );
  
  return redisCleanup;
};

Handler Implementation Flow

The new streamable HTTP handler integrates with the Redis-based session management:

export async function handleStreamableHTTP(req: Request, res: Response) {
  const method = req.method;
  const sessionId = req.headers['mcp-session-id'] as string | undefined;
  
  if (method === 'POST') {
    // Handle POST requests (initialization or message sending)
    
    if (isInitializeRequest(req.body) && !sessionId) {
      // New session initialization
      const newSessionId = randomUUID();
      const authInfo = req.auth;
      
      // Create session in Redis
      await SessionManager.createSession(newSessionId, authInfo?.clientId || 'unknown');
      
      // Create transport with Redis-based session management
      const transport = new StreamableHTTPServerTransport({
        sessionIdGenerator: () => newSessionId,
        // Custom implementation - don't store transport in memory
      });
      
      const { server: mcpServer, cleanup: mcpCleanup } = createMcpServer();
      
      // Set up Redis subscription for this session but don't store transport globally
      // Instead, rely on Redis for all message routing
      
      await mcpServer.connect(transport);
      await transport.handleRequest(req, res, req.body);
      
    } else if (sessionId) {
      // Existing session - validate and handle request
      const sessionValid = await SessionManager.refreshSession(sessionId);
      if (!sessionValid) {
        res.writeHead(404).end(JSON.stringify({
          jsonrpc: "2.0",
          error: { code: -32001, message: "Session not found" },
          id: null
        }));
        return;
      }
      
      // Create ephemeral transport for this request
      const transport = new StreamableHTTPServerTransport({
        sessionIdGenerator: () => sessionId, // Use existing session ID
      });
      
      const { server: mcpServer, cleanup: mcpCleanup } = createMcpServer();
      await mcpServer.connect(transport);
      await transport.handleRequest(req, res, req.body);
      
      // Clean up after request completes
      res.on('close', mcpCleanup);
    }
    
  } else if (method === 'GET') {
    // Handle SSE stream requests
    
    if (!sessionId) {
      res.writeHead(400).end('Session ID required');
      return;
    }
    
    const sessionValid = await SessionManager.refreshSession(sessionId);
    if (!sessionValid) {
      res.writeHead(404).end('Session not found');
      return;
    }
    
    // Create transport for SSE stream
    const transport = new StreamableHTTPServerTransport({
      sessionIdGenerator: () => sessionId,
    });
    
    // Mark connection as active
    await SessionManager.setConnectionState(sessionId, true);
    
    // Deliver any buffered messages first
    await MessageDelivery.deliverBufferedMessages(sessionId, transport);
    
    // Set up Redis subscription for live messages
    const redisCleanup = await setupRedisSubscription(sessionId, transport);
    
    // Handle connection cleanup
    res.on('close', async () => {
      await SessionManager.setConnectionState(sessionId, false);
      redisCleanup();
    });
    
    await transport.handleRequest(req, res);
    
  } else if (method === 'DELETE') {
    // Handle session deletion
    
    if (!sessionId) {
      res.writeHead(400).end('Session ID required');
      return;
    }
    
    // Delete session from Redis
    await SessionManager.deleteSession(sessionId);
    res.writeHead(200).end();
  }
}

Auth Information Flow

Auth information flows through the middleware stack:

// Auth middleware adds req.auth
const authInfo: AuthInfo = req.auth;

// Transport receives auth info
await transport.handleRequest(req, res);

// Auth info is available in MCP server handlers
server.tool('example', 'description', schema, async (params, { authInfo }) => {
  // authInfo contains token, clientId, scopes, etc.
});

Conclusion

This design provides horizontally scalable streamable HTTP support by using Redis for all session state management and message buffering. Key advantages:

1. Horizontal Scalability: Any server instance can handle any request for any session
2. Resilient Connection Handling: SSE disconnects don't end sessions; messages are buffered
3. Automatic Cleanup: 5-minute session TTL prevents Redis bloat
4. Backwards Compatibility: Existing /sse and /message endpoints remain unchanged
5. Spec Compliance: Follows 2025-03-26 MCP specification exactly

The implementation is more complex than a single-instance approach but essential for production deployment in a horizontally scaled environment. The Redis-based architecture ensures sessions persist across server instances and SSE connection interruptions.