diff --git a/src/handlers/shttp.ts b/src/handlers/shttp.ts index 53bd80a..bc62990 100644 --- a/src/handlers/shttp.ts +++ b/src/handlers/shttp.ts @@ -17,43 +17,25 @@ declare module "express-serve-static-core" { } export async function handleStreamableHTTP(req: Request, res: Response) { - console.log('=== handleStreamableHTTP START ==='); - console.log('Received MCP request:', JSON.stringify(req.body, null, 2)); - console.log('Request headers:', JSON.stringify(req.headers, null, 2)); - let shttpTransport: StreamableHTTPServerTransport | undefined = undefined; let cleanup: (() => Promise) | undefined = undefined; try { // Check for existing session ID const sessionId = req.headers['mcp-session-id'] as string | undefined; - console.log('Session ID from headers:', sessionId); if (sessionId && await isLive(sessionId)) { - console.log('Session is live, reusing existing transport for session:', sessionId); // Reuse existing transport ({ shttpTransport, cleanup } = await getShttpTransport(sessionId)); - - console.log('Created transport for session:', sessionId); - console.log('Retrieved transport from Redis for session:', sessionId); } else if (!sessionId && isInitializeRequest(req.body)) { - console.log('New initialization request detected, creating new session'); // New initialization request - use JSON response mode const sessionId = randomUUID(); - console.log('Generated new session ID:', sessionId); const server = createMcpServer(); - console.log('Created MCP server instance'); await startServerListeningToRedis(server, sessionId) - console.log('Started server listening to Redis for session:', sessionId); ({ shttpTransport, cleanup } = await getFirstShttpTransport(sessionId)); - console.log('Retrieved first transport for session:', sessionId); - console.log('Transport object:', shttpTransport.constructor.name, 'sessionId:', shttpTransport.sessionId); } else { - console.log('Invalid request - no session ID and not an initialization request'); - console.log('Session ID present:', !!sessionId); - console.log('Is initialize request:', isInitializeRequest(req.body)); // Invalid request - no session ID or not initialization request res.status(400).json({ jsonrpc: '2.0', @@ -63,26 +45,15 @@ export async function handleStreamableHTTP(req: Request, res: Response) { }, id: null, }); - console.log('Sent 400 Bad Request response'); - console.log('=== handleStreamableHTTP END (bad request) ==='); return; } // Handle the request with existing transport - no need to reconnect - console.log('Handling request with existing transport for session:', sessionId); - console.log('Transport object:', shttpTransport.constructor.name, 'sessionId:', shttpTransport.sessionId); - console.log('Request body:', JSON.stringify(req.body, null, 2)); await shttpTransport.handleRequest(req, res, req.body); - console.log('Request handled successfully'); } catch (error) { - console.error('=== ERROR in handleStreamableHTTP ==='); - console.error('Error type:', error instanceof Error ? error.constructor.name : typeof error); - console.error('Error message:', error instanceof Error ? error.message : String(error)); - console.error('Error stack:', error instanceof Error ? error.stack : 'No stack trace'); console.error('Error handling MCP request:', error); if (!res.headersSent) { - console.log('Sending 500 Internal Server Error response'); res.status(500).json({ jsonrpc: '2.0', error: { @@ -91,19 +62,14 @@ export async function handleStreamableHTTP(req: Request, res: Response) { }, id: null, }); - } else { - console.log('Response headers already sent, cannot send error response'); } } finally { // Set up cleanup when response is complete res.on('finish', async () => { - console.log('HTTP response finished, closing transport'); await shttpTransport?.close(); if (cleanup) { await cleanup(); } - console.log('Transport closed after response'); }); - console.log('=== handleStreamableHTTP END ==='); } } diff --git a/src/index.ts b/src/index.ts index db99324..5f7aa32 100644 --- a/src/index.ts +++ b/src/index.ts @@ -36,12 +36,7 @@ const baseSecurityHeaders = (req: express.Request, res: express.Response, next: // simple logging middleware const logger = (req: express.Request, res: express.Response, next: express.NextFunction) => { - console.log(`${req.method} ${req.url}`); next(); - // Log the response status code - res.on('finish', () => { - console.log(`Response status code: ${res.statusCode}`); - }); }; diff --git a/src/redis.ts b/src/redis.ts index e45e85e..cc57f00 100644 --- a/src/redis.ts +++ b/src/redis.ts @@ -104,34 +104,24 @@ export class RedisClientImpl implements RedisClient { onMessage: (message: string) => void, onError: (error: Error) => void, ): Promise<() => Promise> { - console.log(`[RedisClient] Creating subscription for channel: ${channel}`); const subscriber = this.redis.duplicate(); subscriber.on("error", (error) => { - console.error(`[RedisClient] Subscriber error for channel ${channel}:`, error); onError(error); }); - console.log(`[RedisClient] Connecting subscriber for channel: ${channel}`); await subscriber.connect(); - console.log(`[RedisClient] Connected subscriber for channel: ${channel}`); - console.log(`[RedisClient] Subscribing to channel: ${channel}`); await subscriber.subscribe(channel, (message) => { - console.log(`[RedisClient] Received message on channel ${channel}:`, message.substring(0, 100)); onMessage(message); }); - console.log(`[RedisClient] Successfully subscribed to channel: ${channel}`); return async () => { - console.log(`[RedisClient] Cleaning up subscription for channel: ${channel}`); await subscriber.disconnect(); }; } async publish(channel: string, message: string): Promise { - console.log(`[RedisClient] Publishing to channel: ${channel}, message length: ${message.length}`); await this.redis.publish(channel, message); - console.log(`[RedisClient] Successfully published to channel: ${channel}`); } async exists(key: string): Promise { diff --git a/src/services/redisTransport.ts b/src/services/redisTransport.ts index 647fa15..5fff564 100644 --- a/src/services/redisTransport.ts +++ b/src/services/redisTransport.ts @@ -25,8 +25,6 @@ type RedisMessage = function sendToMcpServer(sessionId: string, message: JSONRPCMessage, extra?: { authInfo?: AuthInfo; }, options?: TransportSendOptions): Promise { const toServerChannel = getToServerChannel(sessionId); const redisMessage: RedisMessage = { type: 'mcp', message, extra, options }; - console.log(`[sendToMcpServer] Publishing to Redis channel: ${toServerChannel}`); - console.log(`[sendToMcpServer] Message:`, JSON.stringify(message)); return redisClient.publish(toServerChannel, JSON.stringify(redisMessage)); } @@ -64,10 +62,8 @@ export async function isLive(sessionId: string): Promise { export function redisRelayToMcpServer(sessionId: string, transport: Transport): () => Promise { - console.log(`[redisRelayToMcpServer] Setting up relay for session: ${sessionId}`); let redisCleanup: (() => Promise) | undefined = undefined; const cleanup = async () => { - console.log(`[redisRelayToMcpServer] Cleaning up relay for session: ${sessionId}`); // TODO: solve race conditions where we call cleanup while the subscription is being created / before it is created if (redisCleanup) { await redisCleanup(); @@ -75,42 +71,28 @@ export function redisRelayToMcpServer(sessionId: string, transport: Transport): } const messagePromise = new Promise((resolve) => { - console.log(`[redisRelayToMcpServer] Setting up transport.onmessage handler for session: ${sessionId}`); transport.onmessage = async (message, extra) => { - console.log(`[redisRelayToMcpServer] Received message from transport for session ${sessionId}:`, JSON.stringify(message)); - // First, set up response subscription if needed if ("id" in message) { - console.log(`[redisRelayToMcpServer] Setting up response subscription for message ID: ${message.id}`); const toClientChannel = getToClientChannel(sessionId, message.id.toString()); - console.log(`[redisRelayToMcpServer] Subscribing to response channel: ${toClientChannel}`); redisCleanup = await redisClient.createSubscription(toClientChannel, async (redisMessageJson) => { - console.log(`[redisRelayToMcpServer] Received response from Redis for session ${sessionId}:`, redisMessageJson.substring(0, 200)); const redisMessage = JSON.parse(redisMessageJson) as RedisMessage; if (redisMessage.type === 'mcp') { - console.log(`[redisRelayToMcpServer] Sending response back to transport for session ${sessionId}`); await transport.send(redisMessage.message, redisMessage.options); - console.log(`[redisRelayToMcpServer] Response sent to transport for session ${sessionId}`); } }, (error) => { - console.error(`[redisRelayToMcpServer] Error in response subscription for session ${sessionId}:`, error); transport.onerror?.(error); }); - console.log(`[redisRelayToMcpServer] Response subscription established for session ${sessionId}`); - } else { - console.log(`[redisRelayToMcpServer] Message is notification (no ID) for session ${sessionId}, skipping response subscription`); } // Now send the message to the MCP server await sendToMcpServer(sessionId, message, extra); - console.log(`[redisRelayToMcpServer] Sent message to MCP server for session ${sessionId}`); resolve(message); } }); messagePromise.catch((error) => { - console.error(`[redisRelayToMcpServer] Error setting up relay for session ${sessionId}:`, error); transport.onerror?.(error); cleanup(); }); @@ -140,20 +122,15 @@ export class ServerRedisTransport implements Transport { // Subscribe to MCP messages from clients const serverChannel = getToServerChannel(this._sessionId); - console.log(`[ServerRedisTransport.${this.counter}] Subscribing to server channel: ${serverChannel}`); this.serverCleanup = await redisClient.createSubscription( serverChannel, (messageJson) => { - console.log(`[ServerRedisTransport.${this.counter}] Received message from Redis:`, messageJson.substring(0, 200)); const redisMessage = JSON.parse(messageJson) as RedisMessage; if (redisMessage.type === 'mcp') { - console.log(`[ServerRedisTransport.${this.counter}] Processing MCP message:`, JSON.stringify(redisMessage.message)); this.onmessage?.(redisMessage.message, redisMessage.extra); - console.log(`[ServerRedisTransport.${this.counter}] MCP message processed`); } }, (error) => { - console.error(`[ServerRedisTransport.${this.counter}] Server channel error:`, error); this.onerror?.(error); } ); @@ -181,14 +158,10 @@ export class ServerRedisTransport implements Transport { async send(message: JSONRPCMessage, options?: TransportSendOptions): Promise { const relatedRequestId = options?.relatedRequestId?.toString() ?? "id" in message ? message.id : notificationStreamId; const channel = getToClientChannel(this._sessionId, relatedRequestId) - console.log(`[ServerRedisTransport.${this.counter}] Sending message to channel: ${channel}`); - console.log(`[ServerRedisTransport.${this.counter}] Message:`, JSON.stringify(message)); - console.log(`[ServerRedisTransport.${this.counter}] Options:`, JSON.stringify(options)); const redisMessage: RedisMessage = { type: 'mcp', message, options }; const messageStr = JSON.stringify(redisMessage); await redisClient.publish(channel, messageStr); - console.log(`[ServerRedisTransport.${this.counter}] Message published to Redis`); } async close(): Promise { @@ -226,19 +199,14 @@ export async function startServerListeningToRedis(serverWrapper: { server: Serve } export async function getFirstShttpTransport(sessionId: string): Promise<{shttpTransport: StreamableHTTPServerTransport, cleanup: () => Promise}> { - console.log(`[getFirstShttpTransport] Creating transport for session: ${sessionId}`); const shttpTransport = new StreamableHTTPServerTransport({ sessionIdGenerator: () => sessionId, enableJsonResponse: true, // Enable JSON response mode }); - console.log(`[getFirstShttpTransport] Created transport, sessionId: ${shttpTransport.sessionId}`); - // Use the new request-id based relay approach - console.log(`[getFirstShttpTransport] Setting up Redis relay for session: ${sessionId}`); const cleanup = redisRelayToMcpServer(sessionId, shttpTransport); - console.log(`[getFirstShttpTransport] Transport setup complete for session: ${sessionId}`); return { shttpTransport, cleanup }; }