Add comprehensive structured logging to streamable HTTP

- Remove sensitive data (auth tokens, cookies) from request logging
- Add detailed logging throughout streamable HTTP request lifecycle:
  - Session creation/reuse
  - Session ownership validation
  - Transport initialization
  - Message flow through Redis channels
  - Session shutdown and cleanup
- Add logging for inactivity timeouts and control messages
- Use appropriate log levels (info for lifecycle, debug for details, error for failures)

All logs now include contextual information like sessionId, userId, and channel names
for easier debugging and monitoring.

🤖 Generated with [Claude Code](https://claude.ai/code)

Co-Authored-By: Claude <[email protected]>

Author: jerome3o-anthropic

src/handlers/shttp.ts

@@ -29,6 +29,10 @@ export async function handleStreamableHTTP(req: Request, res: Response) {
   });
 
   const onsessionclosed = async (sessionId: string) => {
+    logger.info('Session closed callback triggered', {
+      sessionId,
+      userId: getUserIdFromAuth(req.auth)
+    });
     await shutdownSession(sessionId);
   }
 
@@ -39,6 +43,10 @@ export async function handleStreamableHTTP(req: Request, res: Response) {
 
     // if no userid, return 401, we shouldn't get here ideally
     if (!userId) {
+      logger.warning('Request without user ID', {
+        sessionId,
+        hasAuth: !!req.auth
+      });
       res.status(401)
       return;
     }
@@ -48,14 +56,29 @@ export async function handleStreamableHTTP(req: Request, res: Response) {
     // incorrect session for the authed user, return 401
     if (sessionId) {
       if (!(await isSessionOwnedBy(sessionId, userId))) {
+        logger.warning('Session ownership mismatch', {
+          sessionId,
+          userId,
+          requestMethod: req.method
+        });
         res.status(401)
         return;
       }
       // Reuse existing transport for owned session
+      logger.info('Reusing existing session', {
+        sessionId,
+        userId,
+        isGetRequest
+      });
       shttpTransport = await getShttpTransport(sessionId, onsessionclosed, isGetRequest);
     } else if (isInitializeRequest(req.body)) {
       // New initialization request - use JSON response mode
       const onsessioninitialized = async (sessionId: string) => {
+        logger.info('Initializing new session', {
+          sessionId,
+          userId
+        });
+        
         const { server, cleanup: mcpCleanup } = createMcpServer();
 
         const serverRedisTransport = new ServerRedisTransport(sessionId);
@@ -64,6 +87,11 @@ export async function handleStreamableHTTP(req: Request, res: Response) {
 
         // Set session ownership
         await setSessionOwner(sessionId, userId);
+        
+        logger.info('Session initialized successfully', {
+          sessionId,
+          userId
+        });
       }
 
       const sessionId = randomUUID();
@@ -75,6 +103,12 @@ export async function handleStreamableHTTP(req: Request, res: Response) {
       shttpTransport.onclose = await redisRelayToMcpServer(sessionId, shttpTransport);
     } else {
       // Invalid request - no session ID and not initialization request
+      logger.warning('Invalid request: no session ID and not initialization', {
+        hasSessionId: !!sessionId,
+        isInitRequest: false,
+        userId,
+        method: req.method
+      });
       res.status(400)
       return;
     }

src/index.ts

@@ -39,12 +39,27 @@ const baseSecurityHeaders = (req: express.Request, res: express.Response, next:
 const loggingMiddleware = (req: express.Request, res: express.Response, next: express.NextFunction) => {
   const startTime = Date.now();
 
-  // Log request
+  // Sanitize headers to remove sensitive information
+  const sanitizedHeaders = { ...req.headers };
+  delete sanitizedHeaders.authorization;
+  delete sanitizedHeaders.cookie;
+  delete sanitizedHeaders['x-api-key'];
+  
+  // Log request (without sensitive data)
   logger.info('Request received', {
     method: req.method,
     url: req.url,
-    headers: req.headers,
-    body: req.method === 'GET' && req.url.includes('/mcp') ? req.body : undefined
+    // Only log specific safe headers
+    headers: {
+      'content-type': sanitizedHeaders['content-type'],
+      'user-agent': sanitizedHeaders['user-agent'],
+      'mcp-protocol-version': sanitizedHeaders['mcp-protocol-version'],
+      'mcp-session-id': sanitizedHeaders['mcp-session-id'],
+      'accept': sanitizedHeaders['accept'],
+      'x-cloud-trace-context': sanitizedHeaders['x-cloud-trace-context']
+    },
+    // Don't log request body as it may contain sensitive data
+    bodySize: req.headers['content-length']
   });
 
   // Log response when finished

src/services/redisTransport.ts

@@ -24,6 +24,14 @@ type RedisMessage =
 
 function sendToMcpServer(sessionId: string, message: JSONRPCMessage, extra?: { authInfo?: AuthInfo; }, options?: TransportSendOptions): Promise<void> {
   const toServerChannel = getToServerChannel(sessionId);
+  
+  logger.debug('Sending message to MCP server via Redis', {
+    sessionId,
+    channel: toServerChannel,
+    method: (message as any).method,
+    id: (message as any).id
+  });
+  
   const redisMessage: RedisMessage = { type: 'mcp', message, extra, options };
   return redisClient.publish(toServerChannel, JSON.stringify(redisMessage));
 }
@@ -51,6 +59,7 @@ function sendControlMessage(sessionId: string, action: 'SHUTDOWN' | 'PING' | 'ST
 }
 
 export async function shutdownSession(sessionId: string): Promise<void> {
+  logger.info('Sending shutdown control message', { sessionId });
   return sendControlMessage(sessionId, 'SHUTDOWN');
 }
 
@@ -61,6 +70,7 @@ export async function isLive(sessionId: string): Promise<boolean> {
 }
 
 export async function setSessionOwner(sessionId: string, userId: string): Promise<void> {
+  logger.debug('Setting session owner', { sessionId, userId });
   await redisClient.set(`session:${sessionId}:owner`, userId);
 }
 
@@ -76,30 +86,54 @@ export async function validateSessionOwnership(sessionId: string, userId: string
 export async function isSessionOwnedBy(sessionId: string, userId: string): Promise<boolean> {
   const isLiveSession = await isLive(sessionId);
   if (!isLiveSession) {
+    logger.debug('Session not live', { sessionId });
     return false;
   }
-  return await validateSessionOwnership(sessionId, userId);
+  const isOwned = await validateSessionOwnership(sessionId, userId);
+  logger.debug('Session ownership check', { sessionId, userId, isOwned });
+  return isOwned;
 }
 
 
 export async function redisRelayToMcpServer(sessionId: string, transport: Transport, isGetRequest: boolean = false): Promise<() => Promise<void>> {
+  logger.debug('Setting up Redis relay to MCP server', {
+    sessionId,
+    isGetRequest
+  });
+  
   let redisCleanup: (() => Promise<void>) | undefined = undefined;
   const cleanup = async () => {
     // TODO: solve race conditions where we call cleanup while the subscription is being created / before it is created
     if (redisCleanup) {
+      logger.debug('Cleaning up Redis relay', { sessionId });
       await redisCleanup();
     }
   }
 
   const subscribe = async (requestId: string) => {
     const toClientChannel = getToClientChannel(sessionId, requestId);
+    
+    logger.debug('Subscribing to client channel', {
+      sessionId,
+      requestId,
+      channel: toClientChannel
+    });
 
     redisCleanup = await redisClient.createSubscription(toClientChannel, async (redisMessageJson) => {
       const redisMessage = JSON.parse(redisMessageJson) as RedisMessage;
       if (redisMessage.type === 'mcp') {
+        logger.debug('Relaying message from Redis to client', {
+          sessionId,
+          requestId,
+          method: (redisMessage.message as any).method
+        });
         await transport.send(redisMessage.message, redisMessage.options);
       }
     }, (error) => {
+      logger.error('Error in Redis relay subscription', error, {
+        sessionId,
+        channel: toClientChannel
+      });
       transport.onerror?.(error);
     });
   }
@@ -111,6 +145,11 @@ export async function redisRelayToMcpServer(sessionId: string, transport: Transp
       transport.onmessage = async (message, extra) => {
         // First, set up response subscription if needed
         if ("id" in message) {
+          logger.debug('Setting up response subscription', {
+            sessionId,
+            messageId: message.id,
+            method: (message as any).method
+          });
           await subscribe(message.id.toString());
         }
         // Now send the message to the MCP server
@@ -171,40 +210,78 @@ export class ServerRedisTransport implements Transport {
   }
 
   async start(): Promise<void> {
+    logger.info('Starting ServerRedisTransport', {
+      sessionId: this._sessionId,
+      inactivityTimeoutMs: this.INACTIVITY_TIMEOUT_MS
+    });
+    
     // Start inactivity timer
     this.resetInactivityTimer();
 
     // Subscribe to MCP messages from clients
     const serverChannel = getToServerChannel(this._sessionId);
+    logger.debug('Subscribing to server channel', {
+      sessionId: this._sessionId,
+      channel: serverChannel
+    });
+    
     this.serverCleanup = await redisClient.createSubscription(
       serverChannel,
       (messageJson) => {
         const redisMessage = JSON.parse(messageJson) as RedisMessage;
         if (redisMessage.type === 'mcp') {
           // Reset inactivity timer on each message from client
           this.resetInactivityTimer();
+          
+          logger.debug('Received MCP message from client', {
+            sessionId: this._sessionId,
+            method: (redisMessage.message as any).method,
+            id: (redisMessage.message as any).id
+          });
+          
           this.onmessage?.(redisMessage.message, redisMessage.extra);
         }
       },
       (error) => {
+        logger.error('Error in server channel subscription', error, {
+          sessionId: this._sessionId,
+          channel: serverChannel
+        });
         this.onerror?.(error);
       }
     );
 
     // Subscribe to control messages for shutdown
     const controlChannel = getControlChannel(this._sessionId);
+    logger.debug('Subscribing to control channel', {
+      sessionId: this._sessionId,
+      channel: controlChannel
+    });
+    
     this.controlCleanup = await redisClient.createSubscription(
       controlChannel,
       (messageJson) => {
         const redisMessage = JSON.parse(messageJson) as RedisMessage;
         if (redisMessage.type === 'control') {
+          logger.info('Received control message', {
+            sessionId: this._sessionId,
+            action: redisMessage.action
+          });
+          
           if (redisMessage.action === 'SHUTDOWN') {
+            logger.info('Shutting down transport due to control message', {
+              sessionId: this._sessionId
+            });
             this.shouldShutdown = true;
             this.close();
           }
         }
       },
       (error) => {
+        logger.error('Error in control channel subscription', error, {
+          sessionId: this._sessionId,
+          channel: controlChannel
+        });
         this.onerror?.(error);
       }
     );
@@ -215,12 +292,25 @@ export class ServerRedisTransport implements Transport {
     const relatedRequestId = options?.relatedRequestId?.toString() ?? ("id" in message ? message.id?.toString() : notificationStreamId);
     const channel = getToClientChannel(this._sessionId, relatedRequestId)
 
+    logger.debug('Sending message to client', {
+      sessionId: this._sessionId,
+      channel,
+      method: (message as any).method,
+      id: (message as any).id,
+      relatedRequestId
+    });
+
     const redisMessage: RedisMessage = { type: 'mcp', message, options };
     const messageStr = JSON.stringify(redisMessage);
     await redisClient.publish(channel, messageStr);
   }
 
   async close(): Promise<void> {
+    logger.info('Closing ServerRedisTransport', {
+      sessionId: this._sessionId,
+      wasShutdown: this.shouldShutdown
+    });
+    
     // Clear inactivity timer
     this.clearInactivityTimer();
 
@@ -241,6 +331,11 @@ export class ServerRedisTransport implements Transport {
 }
 
 export async function getShttpTransport(sessionId: string, onsessionclosed: (sessionId: string) => void | Promise<void>, isGetRequest: boolean = false): Promise<StreamableHTTPServerTransport> {
+  logger.debug('Getting StreamableHTTPServerTransport for existing session', {
+    sessionId,
+    isGetRequest
+  });
+  
   // Giving undefined here and setting the sessionId means the 
   // transport wont try to create a new session.
   const shttpTransport = new StreamableHTTPServerTransport({