modelcontextprotocol/example-remote-server
Watch
1
0
Fork
You've already forked example-remote-server
 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Add comprehensive structured logging to streamable HTTP

Browse source 
- 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 <noreply@anthropic.com>
This commit is contained in:
Jerome 2025-07-10 11:05:50 +01:00
commit d67ef14ad6
No known key found for this signature in database
1 changed files with 148 additions and 4 deletions
Download patch file Download diff file Expand all files Collapse all files

97
src/services/redisTransport.ts
View file

@ -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,11 +210,21 @@ 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) => {
@ -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<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({