Remove debugging console logs for cleaner output
- Removed verbose logging from Redis transport operations - Cleaned up HTTP handler debug statements - Simplified Redis client logging - Removed request/response logging middleware - Maintains error logging for troubleshooting 🤖 Generated with [Claude Code](https://claude.ai/code) Co-Authored-By: Claude <noreply@anthropic.com>
This commit is contained in:
Jerome
parent
de8c4a4ed2
commit
f660459275
4 changed files with 0 additions and 81 deletions
|
|
@ -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<void>) | 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 ===');
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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}`);
|
||||
});
|
||||
};
|
||||
|
||||
|
||||
|
|
|
|||
10
src/redis.ts
10
src/redis.ts
|
|
@ -104,34 +104,24 @@ export class RedisClientImpl implements RedisClient {
|
|||
onMessage: (message: string) => void,
|
||||
onError: (error: Error) => void,
|
||||
): Promise<() => Promise<void>> {
|
||||
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<void> {
|
||||
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<boolean> {
|
||||
|
|
|
|||
|
|
@ -25,8 +25,6 @@ type RedisMessage =
|
|||
function sendToMcpServer(sessionId: string, message: JSONRPCMessage, extra?: { authInfo?: AuthInfo; }, options?: TransportSendOptions): Promise<void> {
|
||||
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<boolean> {
|
|||
|
||||
|
||||
export function redisRelayToMcpServer(sessionId: string, transport: Transport): () => Promise<void> {
|
||||
console.log(`[redisRelayToMcpServer] Setting up relay for session: ${sessionId}`);
|
||||
let redisCleanup: (() => Promise<void>) | 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<JSONRPCMessage>((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<void> {
|
||||
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<void> {
|
||||
|
|
@ -226,19 +199,14 @@ export async function startServerListeningToRedis(serverWrapper: { server: Serve
|
|||
}
|
||||
|
||||
export async function getFirstShttpTransport(sessionId: string): Promise<{shttpTransport: StreamableHTTPServerTransport, cleanup: () => Promise<void>}> {
|
||||
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 };
|
||||
}
|
||||
|
||||
|
|
|
|||
Loading…
Reference in a new issue