diff --git a/src/examples/README.md b/src/examples/README.md index 6e53fdec..937159c3 100644 --- a/src/examples/README.md +++ b/src/examples/README.md @@ -6,6 +6,34 @@ This directory contains example implementations of MCP clients and servers using Multi node with stete management example will be added soon after we add support. +### Server with JSON response mode (`server/jsonResponseStreamableHttp.ts`) + +A simple MCP server that uses the Streamable HTTP transport with JSON response mode enabled, implemented with Express. The server provides a simple `greet` tool that returns a greeting for a name. + +#### Running the server + +```bash +npx tsx src/examples/server/jsonResponseStreamableHttp.ts +``` + +The server will start on port 3000. You can test the initialization and tool calling: + +```bash +# Initialize the server and get the session ID from headers +SESSION_ID=$(curl -X POST -H "Content-Type: application/json" -H "Accept: application/json" \ + -d '{"jsonrpc":"2.0","method":"initialize","params":{"capabilities":{}},"id":"1"}' \ + -i http://localhost:3000/mcp 2>&1 | grep -i "mcp-session-id" | cut -d' ' -f2 | tr -d '\r') +echo "Session ID: $SESSION_ID" + +# Call the greet tool using the saved session ID +curl -X POST -H "Content-Type: application/json" -H "Accept: application/json" \ + -H "mcp-session-id: $SESSION_ID" \ + -d '{"jsonrpc":"2.0","method":"mcp.call_tool","params":{"name":"greet","arguments":{"name":"World"}},"id":"2"}' \ + http://localhost:3000/mcp +``` + +Note that in this example, we're using plain JSON response mode by setting `Accept: application/json` header. + ### Server (`server/simpleStreamableHttp.ts`) A simple MCP server that uses the Streamable HTTP transport, implemented with Express. The server provides: diff --git a/src/examples/client/simpleStreamableHttp.ts b/src/examples/client/simpleStreamableHttp.ts index 9bf43ce8..589c1f41 100644 --- a/src/examples/client/simpleStreamableHttp.ts +++ b/src/examples/client/simpleStreamableHttp.ts @@ -15,9 +15,9 @@ import { async function main(): Promise { // Create a new client with streamable HTTP transport - const client = new Client({ - name: 'example-client', - version: '1.0.0' + const client = new Client({ + name: 'example-client', + version: '1.0.0' }); const transport = new StreamableHTTPClientTransport( new URL('http://localhost:3000/mcp') @@ -27,7 +27,6 @@ async function main(): Promise { await client.connect(transport); console.log('Connected to MCP server'); - // List available tools const toolsRequest: ListToolsRequest = { method: 'tools/list', @@ -48,32 +47,44 @@ async function main(): Promise { console.log('Greeting result:', greetResult.content[0].text); // List available prompts - const promptsRequest: ListPromptsRequest = { - method: 'prompts/list', - params: {} - }; - const promptsResult = await client.request(promptsRequest, ListPromptsResultSchema); - console.log('Available prompts:', promptsResult.prompts); + try { + const promptsRequest: ListPromptsRequest = { + method: 'prompts/list', + params: {} + }; + const promptsResult = await client.request(promptsRequest, ListPromptsResultSchema); + console.log('Available prompts:', promptsResult.prompts); + } catch (error) { + console.log(`Prompts not supported by this server (${error})`); + } // Get a prompt - const promptRequest: GetPromptRequest = { - method: 'prompts/get', - params: { - name: 'greeting-template', - arguments: { name: 'MCP User' } - } - }; - const promptResult = await client.request(promptRequest, GetPromptResultSchema); - console.log('Prompt template:', promptResult.messages[0].content.text); + try { + const promptRequest: GetPromptRequest = { + method: 'prompts/get', + params: { + name: 'greeting-template', + arguments: { name: 'MCP User' } + } + }; + const promptResult = await client.request(promptRequest, GetPromptResultSchema); + console.log('Prompt template:', promptResult.messages[0].content.text); + } catch (error) { + console.log(`Prompt retrieval not supported by this server (${error})`); + } // List available resources - const resourcesRequest: ListResourcesRequest = { - method: 'resources/list', - params: {} - }; - const resourcesResult = await client.request(resourcesRequest, ListResourcesResultSchema); - console.log('Available resources:', resourcesResult.resources); - + try { + const resourcesRequest: ListResourcesRequest = { + method: 'resources/list', + params: {} + }; + const resourcesResult = await client.request(resourcesRequest, ListResourcesResultSchema); + console.log('Available resources:', resourcesResult.resources); + } catch (error) { + console.log(`Resources not supported by this server (${error})`); + } + // Close the connection await client.close(); } diff --git a/src/examples/server/jsonResponseStreamableHttp.ts b/src/examples/server/jsonResponseStreamableHttp.ts new file mode 100644 index 00000000..2e83b54a --- /dev/null +++ b/src/examples/server/jsonResponseStreamableHttp.ts @@ -0,0 +1,119 @@ +import express, { Request, Response } from 'express'; +import { randomUUID } from 'node:crypto'; +import { McpServer } from '../../server/mcp.js'; +import { StreamableHTTPServerTransport } from '../../server/streamableHttp.js'; +import { z } from 'zod'; +import { CallToolResult } from '../../types.js'; + +// Create an MCP server with implementation details +const server = new McpServer({ + name: 'json-response-streamable-http-server', + version: '1.0.0', +}); + +// Register a simple tool that returns a greeting +server.tool( + 'greet', + 'A simple greeting tool', + { + name: z.string().describe('Name to greet'), + }, + async ({ name }): Promise => { + return { + content: [ + { + type: 'text', + text: `Hello, ${name}!`, + }, + ], + }; + } +); + +const app = express(); +app.use(express.json()); + +// Map to store transports by session ID +const transports: { [sessionId: string]: StreamableHTTPServerTransport } = {}; + +app.post('/mcp', async (req: Request, res: Response) => { + console.log('Received MCP request:', req.body); + try { + // Check for existing session ID + const sessionId = req.headers['mcp-session-id'] as string | undefined; + let transport: StreamableHTTPServerTransport; + + if (sessionId && transports[sessionId]) { + // Reuse existing transport + transport = transports[sessionId]; + } else if (!sessionId && isInitializeRequest(req.body)) { + // New initialization request - use JSON response mode + transport = new StreamableHTTPServerTransport({ + sessionIdGenerator: () => randomUUID(), + enableJsonResponse: true, // Enable JSON response mode + }); + + // Connect the transport to the MCP server BEFORE handling the request + await server.connect(transport); + + // After handling the request, if we get a session ID back, store the transport + await transport.handleRequest(req, res, req.body); + + // Store the transport by session ID for future requests + if (transport.sessionId) { + transports[transport.sessionId] = transport; + } + return; // Already handled + } else { + // Invalid request - no session ID or not initialization request + res.status(400).json({ + jsonrpc: '2.0', + error: { + code: -32000, + message: 'Bad Request: No valid session ID provided', + }, + id: null, + }); + return; + } + + // Handle the request with existing transport - no need to reconnect + await transport.handleRequest(req, res, req.body); + } catch (error) { + console.error('Error handling MCP request:', error); + if (!res.headersSent) { + res.status(500).json({ + jsonrpc: '2.0', + error: { + code: -32603, + message: 'Internal server error', + }, + id: null, + }); + } + } +}); + +// Helper function to detect initialize requests +function isInitializeRequest(body: unknown): boolean { + if (Array.isArray(body)) { + return body.some(msg => typeof msg === 'object' && msg !== null && 'method' in msg && msg.method === 'initialize'); + } + return typeof body === 'object' && body !== null && 'method' in body && body.method === 'initialize'; +} + +// Start the server +const PORT = 3000; +app.listen(PORT, () => { + console.log(`MCP Streamable HTTP Server with JSON responses listening on port ${PORT}`); + console.log(`Server is running. Press Ctrl+C to stop.`); + console.log(`Initialize with: curl -X POST -H "Content-Type: application/json" -H "Accept: application/json" -d '{"jsonrpc":"2.0","method":"initialize","params":{"capabilities":{}},"id":"1"}' http://localhost:${PORT}/mcp`); + console.log(`Then call tool with: curl -X POST -H "Content-Type: application/json" -H "Accept: application/json" -H "mcp-session-id: YOUR_SESSION_ID" -d '{"jsonrpc":"2.0","method":"mcp.call_tool","params":{"name":"greet","arguments":{"name":"World"}},"id":"2"}' http://localhost:${PORT}/mcp`); +}); + +// Handle server shutdown +process.on('SIGINT', async () => { + console.log('Shutting down server...'); + await server.close(); + process.exit(0); +}); \ No newline at end of file diff --git a/src/server/.streamableHttp.test.ts.swp b/src/server/.streamableHttp.test.ts.swp new file mode 100644 index 00000000..4ab20fa3 Binary files /dev/null and b/src/server/.streamableHttp.test.ts.swp differ diff --git a/src/server/streamableHttp.test.ts b/src/server/streamableHttp.test.ts index bdbfb6ba..74868979 100644 --- a/src/server/streamableHttp.test.ts +++ b/src/server/streamableHttp.test.ts @@ -1008,8 +1008,8 @@ describe("StreamableHTTPServerTransport", () => { await transport.handleRequest(req, response); // Verify that the request is tracked in the SSE map - expect(transport["_sseResponseMapping"].size).toBe(2); - expect(transport["_sseResponseMapping"].has("cleanup-test")).toBe(true); + expect(transport["_responseMapping"].size).toBe(2); + expect(transport["_responseMapping"].has("cleanup-test")).toBe(true); // Send a response await transport.send({ @@ -1019,8 +1019,8 @@ describe("StreamableHTTPServerTransport", () => { }); // Verify that the mapping was cleaned up - expect(transport["_sseResponseMapping"].size).toBe(1); - expect(transport["_sseResponseMapping"].has("cleanup-test")).toBe(false); + expect(transport["_responseMapping"].size).toBe(1); + expect(transport["_responseMapping"].has("cleanup-test")).toBe(false); }); it("should clean up connection tracking when client disconnects", async () => { @@ -1052,17 +1052,17 @@ describe("StreamableHTTPServerTransport", () => { await transport.handleRequest(req, response); // Both requests should be mapped to the same response - expect(transport["_sseResponseMapping"].size).toBe(3); - expect(transport["_sseResponseMapping"].get("req1")).toBe(response); - expect(transport["_sseResponseMapping"].get("req2")).toBe(response); + expect(transport["_responseMapping"].size).toBe(3); + expect(transport["_responseMapping"].get("req1")).toBe(response); + expect(transport["_responseMapping"].get("req2")).toBe(response); // Simulate client disconnect by triggering the stored callback if (closeCallback) closeCallback(); // All entries using this response should be removed - expect(transport["_sseResponseMapping"].size).toBe(1); - expect(transport["_sseResponseMapping"].has("req1")).toBe(false); - expect(transport["_sseResponseMapping"].has("req2")).toBe(false); + expect(transport["_responseMapping"].size).toBe(1); + expect(transport["_responseMapping"].has("req1")).toBe(false); + expect(transport["_responseMapping"].has("req2")).toBe(false); }); }); @@ -1214,6 +1214,148 @@ describe("StreamableHTTPServerTransport", () => { }); }); + describe("JSON Response Mode", () => { + let jsonResponseTransport: StreamableHTTPServerTransport; + let mockResponse: jest.Mocked; + + beforeEach(async () => { + jsonResponseTransport = new StreamableHTTPServerTransport({ + sessionIdGenerator: () => randomUUID(), + enableJsonResponse: true, + }); + + // Initialize the transport + const initMessage: JSONRPCMessage = { + jsonrpc: "2.0", + method: "initialize", + params: { + clientInfo: { name: "test-client", version: "1.0" }, + protocolVersion: "2025-03-26" + }, + id: "init-1", + }; + + const initReq = createMockRequest({ + method: "POST", + headers: { + "content-type": "application/json", + "accept": "application/json, text/event-stream", + }, + body: JSON.stringify(initMessage), + }); + + mockResponse = createMockResponse(); + await jsonResponseTransport.handleRequest(initReq, mockResponse); + mockResponse = createMockResponse(); // Reset for tests + }); + + it("should return JSON response for a single request", async () => { + const requestMessage: JSONRPCMessage = { + jsonrpc: "2.0", + method: "test", + params: {}, + id: "test-req-id", + }; + + const req = createMockRequest({ + method: "POST", + headers: { + "content-type": "application/json", + "accept": "application/json, text/event-stream", + "mcp-session-id": jsonResponseTransport.sessionId, + }, + body: JSON.stringify(requestMessage), + }); + + // Mock immediate response + jsonResponseTransport.onmessage = (message) => { + if ('method' in message && 'id' in message) { + const responseMessage: JSONRPCMessage = { + jsonrpc: "2.0", + result: { value: "test-result" }, + id: message.id, + }; + setTimeout(() => { + void jsonResponseTransport.send(responseMessage); + }, 10); + } + }; + + await jsonResponseTransport.handleRequest(req, mockResponse); + + // Wait for all promises to resolve + await new Promise(resolve => setTimeout(resolve, 50)); + + // Should respond with application/json header + expect(mockResponse.writeHead).toHaveBeenCalledWith( + 200, + expect.objectContaining({ + "Content-Type": "application/json", + }) + ); + + // Should return the response as JSON + const expectedResponse = { + jsonrpc: "2.0", + result: { value: "test-result" }, + id: "test-req-id", + }; + + expect(mockResponse.end).toHaveBeenCalledWith(JSON.stringify(expectedResponse)); + }); + + it("should return JSON response for batch requests", async () => { + const batchMessages: JSONRPCMessage[] = [ + { jsonrpc: "2.0", method: "test1", params: {}, id: "req1" }, + { jsonrpc: "2.0", method: "test2", params: {}, id: "req2" }, + ]; + + const req = createMockRequest({ + method: "POST", + headers: { + "content-type": "application/json", + "accept": "application/json, text/event-stream", + "mcp-session-id": jsonResponseTransport.sessionId, + }, + body: JSON.stringify(batchMessages), + }); + + // Mock sequential responses - send them in specific order + jsonResponseTransport.onmessage = (message) => { + if ('method' in message && 'id' in message) { + const responseMessage: JSONRPCMessage = { + jsonrpc: "2.0", + result: { value: `result-for-${message.id}` }, + id: message.id, + }; + // Send responses in order - req1 first, then req2 + if (message.id === 'req1') { + void jsonResponseTransport.send(responseMessage); + } else { + // Add a tiny delay to req2 to ensure it's processed after req1 + setTimeout(() => void jsonResponseTransport.send(responseMessage), 5); + } + } + }; + + await jsonResponseTransport.handleRequest(req, mockResponse); + + // Wait for all promises to resolve - give it enough time + await new Promise(resolve => setTimeout(resolve, 100)); + + // Should respond with application/json header + expect(mockResponse.writeHead).toHaveBeenCalledWith( + 200, + expect.objectContaining({ + "Content-Type": "application/json", + }) + ); + + expect(mockResponse.end).toHaveBeenCalled(); + expect(mockResponse.end.mock.calls[0][0]).toContain("result-for-req2"); + }); + }); + describe("Handling Pre-Parsed Body", () => { beforeEach(async () => { // Initialize the transport diff --git a/src/server/streamableHttp.ts b/src/server/streamableHttp.ts index b0fcce6d..3fd96307 100644 --- a/src/server/streamableHttp.ts +++ b/src/server/streamableHttp.ts @@ -18,8 +18,12 @@ export interface StreamableHTTPServerTransportOptions { */ sessionIdGenerator: () => string | undefined; - - + /** + * If true, the server will return JSON responses instead of starting an SSE stream. + * This can be useful for simple request/response scenarios without streaming. + * Default is false (SSE streams are preferred). + */ + enableJsonResponse?: boolean; } /** @@ -60,8 +64,11 @@ export class StreamableHTTPServerTransport implements Transport { // when sessionId is not set (undefined), it means the transport is in stateless mode private sessionIdGenerator: () => string | undefined; private _started: boolean = false; - private _sseResponseMapping: Map = new Map(); + private _responseMapping: Map = new Map(); + private _requestResponseMap: Map = new Map(); private _initialized: boolean = false; + private _enableJsonResponse: boolean = false; + sessionId?: string | undefined; onclose?: () => void; @@ -70,6 +77,7 @@ export class StreamableHTTPServerTransport implements Transport { constructor(options: StreamableHTTPServerTransportOptions) { this.sessionIdGenerator = options.sessionIdGenerator; + this._enableJsonResponse = options.enableJsonResponse ?? false; } /** @@ -221,33 +229,37 @@ export class StreamableHTTPServerTransport implements Transport { this.onmessage?.(message); } } else if (hasRequests) { - const headers: Record = { - "Content-Type": "text/event-stream", - "Cache-Control": "no-cache", - Connection: "keep-alive", - }; + // The default behavior is to use SSE streaming + // but in some cases server will return JSON responses + if (!this._enableJsonResponse) { + const headers: Record = { + "Content-Type": "text/event-stream", + "Cache-Control": "no-cache", + Connection: "keep-alive", + }; - // After initialization, always include the session ID if we have one - if (this.sessionId !== undefined) { - headers["mcp-session-id"] = this.sessionId; + // After initialization, always include the session ID if we have one + if (this.sessionId !== undefined) { + headers["mcp-session-id"] = this.sessionId; + } + + res.writeHead(200, headers); } - - res.writeHead(200, headers); - // Store the response for this request to send messages back through this connection // We need to track by request ID to maintain the connection for (const message of messages) { if ('method' in message && 'id' in message) { - this._sseResponseMapping.set(message.id, res); + this._responseMapping.set(message.id, res); } } // Set up close handler for client disconnects res.on("close", () => { // Remove all entries that reference this response - for (const [id, storedRes] of this._sseResponseMapping.entries()) { + for (const [id, storedRes] of this._responseMapping.entries()) { if (storedRes === res) { - this._sseResponseMapping.delete(id); + this._responseMapping.delete(id); + this._requestResponseMap.delete(id); } } }); @@ -350,10 +362,14 @@ export class StreamableHTTPServerTransport implements Transport { async close(): Promise { // Close all SSE connections - this._sseResponseMapping.forEach((response) => { + this._responseMapping.forEach((response) => { response.end(); }); - this._sseResponseMapping.clear(); + this._responseMapping.clear(); + + // Clear any pending responses + this._requestResponseMap.clear(); + this.onclose?.(); } @@ -367,24 +383,69 @@ export class StreamableHTTPServerTransport implements Transport { throw new Error("No request ID provided for the message"); } - const sseResponse = this._sseResponseMapping.get(requestId); - if (!sseResponse) { - throw new Error(`No SSE connection established for request ID: ${String(requestId)}`); + // Get the response for this request + const response = this._responseMapping.get(requestId); + if (!response) { + throw new Error(`No connection established for request ID: ${String(requestId)}`); } - // Send the message as an SSE event - sseResponse.write( - `event: message\ndata: ${JSON.stringify(message)}\n\n`, - ); - // After all JSON-RPC responses have been sent, the server SHOULD close the SSE stream. + + if (!this._enableJsonResponse) { + response.write(`event: message\ndata: ${JSON.stringify(message)}\n\n`); + } if ('result' in message || 'error' in message) { - this._sseResponseMapping.delete(requestId); - // Only close the connection if it's not needed by other requests - const canCloseConnection = ![...this._sseResponseMapping.entries()].some(([id, res]) => res === sseResponse && id !== requestId); - if (canCloseConnection) { - sseResponse?.end(); + this._requestResponseMap.set(requestId, message); + + // Get all request IDs that share the same request response object + const relatedIds = Array.from(this._responseMapping.entries()) + .filter(([_, res]) => res === response) + .map(([id]) => id); + + // Check if we have responses for all requests using this connection + const allResponsesReady = relatedIds.every(id => this._requestResponseMap.has(id)); + + if (allResponsesReady) { + if (this._enableJsonResponse) { + // If we are in SSE mode, we don't need to do anything else + + // All responses ready, send as JSON + const headers: Record = { + 'Content-Type': 'application/json', + }; + if (this.sessionId !== undefined) { + headers['mcp-session-id'] = this.sessionId; + } + + // In tests, relatedIds might be coming in non-deterministic order + // We need to sort numerically for numeric IDs, alphabetically for string IDs + const responses = relatedIds + .sort((a, b) => { + // If both IDs are numbers, sort numerically + if (typeof a === 'number' && typeof b === 'number') { + return a - b; + } + + // Otherwise sort by string lexical order + return String(a).localeCompare(String(b)); + }) + .map(id => this._requestResponseMap.get(id)!); + + response.writeHead(200, headers); + if (responses.length === 1) { + response.end(JSON.stringify(responses[0])); + } else { + response.end(JSON.stringify(responses)); + } + } else { + response.end(); + } + // Clean up + for (const id of relatedIds) { + this._requestResponseMap.delete(id); + this._responseMapping.delete(id); + } } } } +} -} \ No newline at end of file