diff --git a/src/handlers/common.ts b/src/handlers/common.ts index 02b52b0..3a95c18 100644 --- a/src/handlers/common.ts +++ b/src/handlers/common.ts @@ -1,6 +1,7 @@ import { NextFunction, Request, Response } from "express"; import { withContext } from "../context.js"; import { readMcpInstallation } from "../services/auth.js"; +import { logger } from "../utils/logger.js"; import { JSONRPCError, JSONRPCNotification, JSONRPCRequest, JSONRPCResponse } from "@modelcontextprotocol/sdk/types.js"; @@ -11,18 +12,23 @@ export function logMcpMessage( // check if message has a method field if ("method" in message) { if (message.method === "tools/call") { - console.info( - `[session ${sessionId}] Processing ${message.method}, for tool ${message.params?.name}`, - ); + logger.info('Processing MCP method', { + sessionId, + method: message.method, + toolName: message.params?.name + }); } else { - console.info( - `[session ${sessionId}] Processing ${message.method} method`, - ); - } + logger.info('Processing MCP method', { + sessionId, + method: message.method + }); + } } else if ("error" in message) { - console.warn( - `[session ${sessionId}] Received error message: ${message.error.message}, ${message.error.code}`, - ) + logger.warning('Received error message', { + sessionId, + errorMessage: message.error.message, + errorCode: message.error.code + }); } } diff --git a/src/handlers/shttp.ts b/src/handlers/shttp.ts index d52bd18..3ad47f4 100644 --- a/src/handlers/shttp.ts +++ b/src/handlers/shttp.ts @@ -5,6 +5,7 @@ import { getShttpTransport, isSessionOwnedBy, redisRelayToMcpServer, ServerRedis import { isInitializeRequest } from "@modelcontextprotocol/sdk/types.js"; import { randomUUID } from "crypto"; import { createMcpServer } from "../services/mcp.js"; +import { logger } from "../utils/logger.js"; declare module "express-serve-static-core" { @@ -80,7 +81,11 @@ export async function handleStreamableHTTP(req: Request, res: Response) { // Handle the request with existing transport - no need to reconnect await shttpTransport.handleRequest(req, res, req.body); } catch (error) { - console.error('Error handling MCP request:', error); + logger.error('Error handling MCP request', error as Error, { + sessionId: req.headers['mcp-session-id'] as string | undefined, + method: req.method, + userId: getUserIdFromAuth(req.auth) + }); if (!res.headersSent) { res.status(500) diff --git a/src/handlers/sse.ts b/src/handlers/sse.ts index 8896603..fb9a654 100644 --- a/src/handlers/sse.ts +++ b/src/handlers/sse.ts @@ -6,6 +6,7 @@ import getRawBody from "raw-body"; import { redisClient } from "../redis.js"; import { createMcpServer } from "../services/mcp.js"; import { logMcpMessage } from "./common.js"; +import { logger } from "../utils/logger.js"; const MAXIMUM_MESSAGE_SIZE = "4mb"; @@ -25,7 +26,9 @@ function redisChannelForSession(sessionId: string): string { export async function handleSSEConnection(req: Request, res: Response) { const { server: mcpServer, cleanup: mcpCleanup } = createMcpServer(); const transport = new SSEServerTransport("/message", res); - console.info(`[session ${transport.sessionId}] Received MCP SSE connection`); + logger.info('Received MCP SSE connection', { + sessionId: transport.sessionId + }); const redisCleanup = await redisClient.createSubscription( redisChannelForSession(transport.sessionId), @@ -36,24 +39,21 @@ export async function handleSSEConnection(req: Request, res: Response) { const message = JSON.parse(json); logMcpMessage(message, transport.sessionId); transport.handleMessage(message).catch((error) => { - console.error( - `[session ${transport.sessionId}] Error handling message:`, - error, - ); + logger.error('Error handling message', error as Error, { + sessionId: transport.sessionId + }); }); }, (error) => { - console.error( - `[session ${transport.sessionId}] Disconnecting due to error in Redis subscriber:`, - error, - ); + logger.error('Disconnecting due to error in Redis subscriber', error as Error, { + sessionId: transport.sessionId + }); transport .close() .catch((error) => - console.error( - `[session ${transport.sessionId}] Error closing transport:`, - error, - ), + logger.error('Error closing transport', error as Error, { + sessionId: transport.sessionId + }), ); }, ); @@ -61,17 +61,19 @@ export async function handleSSEConnection(req: Request, res: Response) { const cleanup = () => { void mcpCleanup(); redisCleanup().catch((error) => - console.error( - `[session ${transport.sessionId}] Error disconnecting Redis subscriber:`, - error, - ), + logger.error('Error disconnecting Redis subscriber', error as Error, { + sessionId: transport.sessionId + }), ); } // Clean up Redis subscription when the connection closes mcpServer.onclose = cleanup - console.info(`[session ${transport.sessionId}] Listening on Redis channel`); + logger.info('Listening on Redis channel', { + sessionId: transport.sessionId, + channel: redisChannel + }); await mcpServer.connect(transport); } @@ -94,7 +96,10 @@ export async function handleMessage(req: Request, res: Response) { }); } catch (error) { res.status(400).json(error); - console.error("Bad POST request:", error); + logger.error('Bad POST request', error as Error, { + sessionId, + contentType: req.headers['content-type'] + }); return; } await redisClient.publish(redisChannelForSession(sessionId), body); diff --git a/src/index.ts b/src/index.ts index ebeb7c6..2f0b858 100644 --- a/src/index.ts +++ b/src/index.ts @@ -9,6 +9,7 @@ import { handleFakeAuthorize, handleFakeAuthorizeRedirect } from "./handlers/fak import { handleStreamableHTTP } from "./handlers/shttp.js"; import { handleMessage, handleSSEConnection } from "./handlers/sse.js"; import { redisClient } from "./redis.js"; +import { logger } from "./utils/logger.js"; const app = express(); @@ -34,16 +35,29 @@ const baseSecurityHeaders = (req: express.Request, res: express.Response, next: next(); }; -// simple logging middleware -const logger = (req: express.Request, res: express.Response, next: express.NextFunction) => { - // if GET and /mcp, log the body returned - console.log(`${new Date().toISOString()} ${req.method} ${req.url}`); - if (req.method === 'GET' && req.url.includes('/mcp')) { - console.log(' Body:', JSON.stringify(req.body, null, 2)); - } - res.on('finish', () => { - console.log(' Response:', res.statusCode); +// Structured logging middleware +const loggingMiddleware = (req: express.Request, res: express.Response, next: express.NextFunction) => { + const startTime = Date.now(); + + // Log request + logger.info('Request received', { + method: req.method, + url: req.url, + headers: req.headers, + body: req.method === 'GET' && req.url.includes('/mcp') ? req.body : undefined }); + + // Log response when finished + res.on('finish', () => { + const duration = Date.now() - startTime; + logger.info('Request completed', { + method: req.method, + url: req.url, + statusCode: res.statusCode, + duration: `${duration}ms` + }); + }); + next(); }; @@ -73,7 +87,12 @@ const corsOptions = { app.use(express.json()); -app.use(logger); + +// Add structured logging context middleware first +app.use(logger.middleware()); + +// Then add the logging middleware +app.use(loggingMiddleware); // Apply base security headers to all routes app.use(baseSecurityHeaders); @@ -126,10 +145,14 @@ app.get("/fakeupstreamauth/callback", cors(corsOptions), handleFakeAuthorizeRedi try { await redisClient.connect(); } catch (error) { - console.error("Could not connect to Redis:", error); + logger.error("Could not connect to Redis", error as Error); process.exit(1); } app.listen(PORT, () => { - console.log(`Server running on http://localhost:${PORT}`); + logger.info('Server started', { + port: PORT, + url: `http://localhost:${PORT}`, + environment: process.env.NODE_ENV || 'development' + }); }); diff --git a/src/redis.ts b/src/redis.ts index 1316592..1826bff 100644 --- a/src/redis.ts +++ b/src/redis.ts @@ -1,4 +1,5 @@ import { createClient, SetOptions } from "@redis/client"; +import { logger } from "./utils/logger.js"; /** * Describes the Redis primitives we use in this application, to be able to mock @@ -46,7 +47,7 @@ export class RedisClientImpl implements RedisClient { constructor() { this.redis.on("error", (error) => - console.error("Redis client error:", error), + logger.error("Redis client error", error as Error), ); } diff --git a/src/services/auth.ts b/src/services/auth.ts index 425de79..02dae0f 100644 --- a/src/services/auth.ts +++ b/src/services/auth.ts @@ -3,6 +3,7 @@ import crypto from "crypto"; import { redisClient } from "../redis.js"; import { McpInstallation, PendingAuthorization, TokenExchange } from "../types.js"; import { OAuthClientInformationFull, OAuthTokens } from "@modelcontextprotocol/sdk/shared/auth.js"; +import { logger } from "../utils/logger.js"; export function generatePKCEChallenge(verifier: string): string { const buffer = Buffer.from(verifier); @@ -243,7 +244,9 @@ export async function exchangeToken( const tokenExchange: TokenExchange = JSON.parse(decoded); if (tokenExchange.alreadyUsed) { - console.error("Duplicate use of authorization code detected; revoking tokens"); + logger.error('Duplicate use of authorization code detected; revoking tokens', undefined, { + authorizationCode: authorizationCode.substring(0, 8) + '...' + }); await revokeMcpInstallation(tokenExchange.mcpAccessToken); throw new Error("Duplicate use of authorization code detected; tokens revoked"); } @@ -257,7 +260,9 @@ export async function exchangeToken( if (rereadData !== data) { // Data concurrently changed while we were updating it. This necessarily means a duplicate use. - console.error("2Duplicate use of authorization code detected; revoking tokens"); + logger.error('Duplicate use of authorization code detected (concurrent update); revoking tokens', undefined, { + authorizationCode: authorizationCode.substring(0, 8) + '...' + }); await revokeMcpInstallation(tokenExchange.mcpAccessToken); throw new Error("Duplicate use of authorization code detected; tokens revoked"); } diff --git a/src/services/redisTransport.ts b/src/services/redisTransport.ts index 4f665e6..9b6004d 100644 --- a/src/services/redisTransport.ts +++ b/src/services/redisTransport.ts @@ -3,6 +3,7 @@ import { redisClient } from "../redis.js"; import { Transport, TransportSendOptions } from "@modelcontextprotocol/sdk/shared/transport.js"; import { AuthInfo } from "@modelcontextprotocol/sdk/server/auth/types.js"; import { JSONRPCMessage, MessageExtraInfo } from "@modelcontextprotocol/sdk/types.js"; +import { logger } from "../utils/logger.js"; let redisTransportCounter = 0; const notificationStreamId = "__GET_stream"; @@ -154,7 +155,10 @@ export class ServerRedisTransport implements Transport { // Set new timeout this.inactivityTimeout = setTimeout(() => { - console.log(`Session ${this._sessionId} timed out due to inactivity`); + logger.info('Session timed out due to inactivity', { + sessionId: this._sessionId, + timeoutMs: this.INACTIVITY_TIMEOUT_MS + }); void shutdownSession(this._sessionId); }, this.INACTIVITY_TIMEOUT_MS); } diff --git a/src/utils/logger.ts b/src/utils/logger.ts new file mode 100644 index 0000000..88acc73 --- /dev/null +++ b/src/utils/logger.ts @@ -0,0 +1,172 @@ +import { AsyncLocalStorage } from 'async_hooks'; +import { Request } from 'express'; + +// Severity levels as per Google Cloud Logging +export enum LogSeverity { + DEFAULT = 'DEFAULT', + DEBUG = 'DEBUG', + INFO = 'INFO', + NOTICE = 'NOTICE', + WARNING = 'WARNING', + ERROR = 'ERROR', + CRITICAL = 'CRITICAL', + ALERT = 'ALERT', + EMERGENCY = 'EMERGENCY' +} + +interface LogContext { + trace?: string; + spanId?: string; + [key: string]: any; +} + +interface StructuredLogEntry { + severity: LogSeverity; + message: string; + timestamp: string; + 'logging.googleapis.com/trace'?: string; + 'logging.googleapis.com/spanId'?: string; + [key: string]: any; +} + +class StructuredLogger { + private asyncLocalStorage = new AsyncLocalStorage(); + private projectId: string | undefined; + + constructor() { + // Get project ID from environment or metadata server + this.projectId = process.env.GOOGLE_CLOUD_PROJECT || process.env.GCP_PROJECT; + } + + /** + * Run a function with a specific logging context + */ + runWithContext(context: LogContext, fn: () => T): T { + return this.asyncLocalStorage.run(context, fn); + } + + /** + * Extract trace context from Cloud Run request + */ + extractTraceContext(req: Request): LogContext { + const context: LogContext = {}; + + const traceHeader = req.header('X-Cloud-Trace-Context'); + if (traceHeader && this.projectId) { + const [trace, spanId] = traceHeader.split('/'); + context.trace = `projects/${this.projectId}/traces/${trace}`; + if (spanId) { + context.spanId = spanId.split(';')[0]; // Remove any trace flags + } + } + + // Add other useful request context + context.requestId = req.header('X-Request-Id'); + context.userAgent = req.header('User-Agent'); + context.method = req.method; + context.path = req.path; + + return context; + } + + /** + * Create Express middleware for request context + */ + middleware() { + return (req: Request, res: any, next: any) => { + const context = this.extractTraceContext(req); + this.runWithContext(context, () => { + next(); + }); + }; + } + + /** + * Log a structured message + */ + private log(severity: LogSeverity, message: string, metadata?: Record) { + const context = this.asyncLocalStorage.getStore() || {}; + + const entry: StructuredLogEntry = { + severity, + message, + timestamp: new Date().toISOString(), + ...metadata + }; + + // Add trace context if available + if (context.trace) { + entry['logging.googleapis.com/trace'] = context.trace; + } + if (context.spanId) { + entry['logging.googleapis.com/spanId'] = context.spanId; + } + + // Add any other context fields + Object.keys(context).forEach(key => { + if (key !== 'trace' && key !== 'spanId') { + entry[`context.${key}`] = context[key]; + } + }); + + // Output as JSON for Cloud Logging + console.log(JSON.stringify(entry)); + } + + // Convenience methods for different severity levels + debug(message: string, metadata?: Record) { + this.log(LogSeverity.DEBUG, message, metadata); + } + + info(message: string, metadata?: Record) { + this.log(LogSeverity.INFO, message, metadata); + } + + notice(message: string, metadata?: Record) { + this.log(LogSeverity.NOTICE, message, metadata); + } + + warning(message: string, metadata?: Record) { + this.log(LogSeverity.WARNING, message, metadata); + } + + error(message: string, error?: Error, metadata?: Record) { + const errorMetadata = { + ...metadata, + error: error ? { + name: error.name, + message: error.message, + stack: error.stack + } : undefined + }; + this.log(LogSeverity.ERROR, message, errorMetadata); + } + + critical(message: string, metadata?: Record) { + this.log(LogSeverity.CRITICAL, message, metadata); + } + + alert(message: string, metadata?: Record) { + this.log(LogSeverity.ALERT, message, metadata); + } + + emergency(message: string, metadata?: Record) { + this.log(LogSeverity.EMERGENCY, message, metadata); + } + + /** + * Add additional context to the current async context + */ + addContext(context: LogContext) { + const currentContext = this.asyncLocalStorage.getStore(); + if (currentContext) { + Object.assign(currentContext, context); + } + } +} + +// Export singleton instance +export const logger = new StructuredLogger(); + +// Re-export for convenience +export { LogContext, StructuredLogEntry }; \ No newline at end of file