From d67ef14ad6603ec829138a9f29cb74ae18c15936 Mon Sep 17 00:00:00 2001 From: Jerome Date: Thu, 10 Jul 2025 11:05:50 +0100 Subject: [PATCH] Add comprehensive structured logging to streamable HTTP MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 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 --- src/handlers/shttp.ts | 34 ++++++++++++ src/index.ts | 21 ++++++-- src/services/redisTransport.ts | 97 +++++++++++++++++++++++++++++++++- 3 files changed, 148 insertions(+), 4 deletions(-) diff --git a/src/handlers/shttp.ts b/src/handlers/shttp.ts index 3ad47f4..37fee5e 100644 --- a/src/handlers/shttp.ts +++ b/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; } diff --git a/src/index.ts b/src/index.ts index 2f0b858..83eb92f 100644 --- a/src/index.ts +++ b/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 diff --git a/src/services/redisTransport.ts b/src/services/redisTransport.ts index 9b6004d..d48f78e 100644 --- a/src/services/redisTransport.ts +++ b/src/services/redisTransport.ts @@ -24,6 +24,14 @@ type RedisMessage = function sendToMcpServer(sessionId: string, message: JSONRPCMessage, extra?: { authInfo?: AuthInfo; }, options?: TransportSendOptions): Promise { 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 { + logger.info('Sending shutdown control message', { sessionId }); return sendControlMessage(sessionId, 'SHUTDOWN'); } @@ -61,6 +70,7 @@ export async function isLive(sessionId: string): Promise { } export async function setSessionOwner(sessionId: string, userId: string): Promise { + 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 { 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> { + logger.debug('Setting up Redis relay to MCP server', { + sessionId, + isGetRequest + }); + let redisCleanup: (() => Promise) | 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,11 +210,21 @@ export class ServerRedisTransport implements Transport { } async start(): Promise { + 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) => { @@ -183,28 +232,56 @@ export class ServerRedisTransport implements Transport { 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 { + 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, isGetRequest: boolean = false): Promise { + 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({