Skip to content

Commit 12d8ac9

Browse files
ihrprihrpr
authored
Merge pull request #347 from modelcontextprotocol/ihrpr/backwards_compatibility
StreamableHttp transport - backwards compatibility examples
2 parents 4e3ddf4 + 29ae6b8 commit 12d8ac9

File tree

3 files changed

+677
-0
lines changed

3 files changed

+677
-0
lines changed

‎src/examples/client/streamableHttpWithSseFallbackClient.ts‎

Lines changed: 192 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,192 @@
1+
import { Client } from '../../client/index.js';
2+
import { StreamableHTTPClientTransport } from '../../client/streamableHttp.js';
3+
import { SSEClientTransport } from '../../client/sse.js';
4+
import {
5+
ListToolsRequest,
6+
ListToolsResultSchema,
7+
CallToolRequest,
8+
CallToolResultSchema,
9+
LoggingMessageNotificationSchema,
10+
} from '../../types.js';
11+
12+
/**
13+
* Simplified Backwards Compatible MCP Client
14+
*
15+
* This client demonstrates backward compatibility with both:
16+
* 1. Modern servers using Streamable HTTP transport (protocol version 2025-03-26)
17+
* 2. Older servers using HTTP+SSE transport (protocol version 2024-11-05)
18+
*
19+
* Following the MCP specification for backwards compatibility:
20+
* - Attempts to POST an initialize request to the server URL first (modern transport)
21+
* - If that fails with 4xx status, falls back to GET request for SSE stream (older transport)
22+
*/
23+
24+
// Command line args processing
25+
const args = process.argv.slice(2);
26+
const serverUrl = args[0] || 'http://localhost:3000/mcp';
27+
28+
async function main(): Promise<void> {
29+
console.log('MCP Backwards Compatible Client');
30+
console.log('===============================');
31+
console.log(`Connecting to server at: ${serverUrl}`);
32+
33+
let client: Client;
34+
let transport: StreamableHTTPClientTransport | SSEClientTransport;
35+
36+
try {
37+
// Try connecting with automatic transport detection
38+
const connection = await connectWithBackwardsCompatibility(serverUrl);
39+
client = connection.client;
40+
transport = connection.transport;
41+
42+
// Set up notification handler
43+
client.setNotificationHandler(LoggingMessageNotificationSchema, (notification) => {
44+
console.log(`Notification: ${notification.params.level} - ${notification.params.data}`);
45+
});
46+
47+
// DEMO WORKFLOW:
48+
// 1. List available tools
49+
console.log('\n=== Listing Available Tools ===');
50+
await listTools(client);
51+
52+
// 2. Call the notification tool
53+
console.log('\n=== Starting Notification Stream ===');
54+
await startNotificationTool(client);
55+
56+
// 3. Wait for all notifications (5 seconds)
57+
console.log('\n=== Waiting for all notifications ===');
58+
await new Promise(resolve => setTimeout(resolve, 5000));
59+
60+
// 4. Disconnect
61+
console.log('\n=== Disconnecting ===');
62+
await transport.close();
63+
console.log('Disconnected from MCP server');
64+
65+
} catch (error) {
66+
console.error('Error running client:', error);
67+
process.exit(1);
68+
}
69+
}
70+
71+
/**
72+
* Connect to an MCP server with backwards compatibility
73+
* Following the spec for client backward compatibility
74+
*/
75+
async function connectWithBackwardsCompatibility(url: string): Promise<{
76+
client: Client,
77+
transport: StreamableHTTPClientTransport | SSEClientTransport,
78+
transportType: 'streamable-http' | 'sse'
79+
}> {
80+
console.log('1. Trying Streamable HTTP transport first...');
81+
82+
// Step 1: Try Streamable HTTP transport first
83+
const client = new Client({
84+
name: 'backwards-compatible-client',
85+
version: '1.0.0'
86+
});
87+
88+
client.onerror = (error) => {
89+
console.error('Client error:', error);
90+
};
91+
const baseUrl = new URL(url);
92+
93+
try {
94+
// Create modern transport
95+
const streamableTransport = new StreamableHTTPClientTransport(baseUrl);
96+
await client.connect(streamableTransport);
97+
98+
console.log('Successfully connected using modern Streamable HTTP transport.');
99+
return {
100+
client,
101+
transport: streamableTransport,
102+
transportType: 'streamable-http'
103+
};
104+
} catch (error) {
105+
// Step 2: If transport fails, try the older SSE transport
106+
console.log(`StreamableHttp transport connection failed: ${error}`);
107+
console.log('2. Falling back to deprecated HTTP+SSE transport...');
108+
109+
try {
110+
// Create SSE transport pointing to /sse endpoint
111+
const sseTransport = new SSEClientTransport(baseUrl);
112+
const sseClient = new Client({
113+
name: 'backwards-compatible-client',
114+
version: '1.0.0'
115+
});
116+
await sseClient.connect(sseTransport);
117+
118+
console.log('Successfully connected using deprecated HTTP+SSE transport.');
119+
return {
120+
client: sseClient,
121+
transport: sseTransport,
122+
transportType: 'sse'
123+
};
124+
} catch (sseError) {
125+
console.error(`Failed to connect with either transport method:\n1. Streamable HTTP error: ${error}\n2. SSE error: ${sseError}`);
126+
throw new Error('Could not connect to server with any available transport');
127+
}
128+
}
129+
}
130+
131+
/**
132+
* List available tools on the server
133+
*/
134+
async function listTools(client: Client): Promise<void> {
135+
try {
136+
const toolsRequest: ListToolsRequest = {
137+
method: 'tools/list',
138+
params: {}
139+
};
140+
const toolsResult = await client.request(toolsRequest, ListToolsResultSchema);
141+
142+
console.log('Available tools:');
143+
if (toolsResult.tools.length === 0) {
144+
console.log(' No tools available');
145+
} else {
146+
for (const tool of toolsResult.tools) {
147+
console.log(` - ${tool.name}: ${tool.description}`);
148+
}
149+
}
150+
} catch (error) {
151+
console.log(`Tools not supported by this server: ${error}`);
152+
}
153+
}
154+
155+
/**
156+
* Start a notification stream by calling the notification tool
157+
*/
158+
async function startNotificationTool(client: Client): Promise<void> {
159+
try {
160+
// Call the notification tool using reasonable defaults
161+
const request: CallToolRequest = {
162+
method: 'tools/call',
163+
params: {
164+
name: 'start-notification-stream',
165+
arguments: {
166+
interval: 1000, // 1 second between notifications
167+
count: 5 // Send 5 notifications
168+
}
169+
}
170+
};
171+
172+
console.log('Calling notification tool...');
173+
const result = await client.request(request, CallToolResultSchema);
174+
175+
console.log('Tool result:');
176+
result.content.forEach(item => {
177+
if (item.type === 'text') {
178+
console.log(` ${item.text}`);
179+
} else {
180+
console.log(` ${item.type} content:`, item);
181+
}
182+
});
183+
} catch (error) {
184+
console.log(`Error calling notification tool: ${error}`);
185+
}
186+
}
187+
188+
// Start the client
189+
main().catch((error: unknown) => {
190+
console.error('Error running MCP client:', error);
191+
process.exit(1);
192+
});

‎src/examples/server/simpleSseServer.ts‎

Lines changed: 169 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,169 @@
1+
import express, { Request, Response } from 'express';
2+
import { McpServer } from '../../server/mcp.js';
3+
import { SSEServerTransport } from '../../server/sse.js';
4+
import { z } from 'zod';
5+
import { CallToolResult } from '../../types.js';
6+
7+
/**
8+
* This example server demonstrates the deprecated HTTP+SSE transport
9+
* (protocol version 2024-11-05). It mainly used for testing backward compatible clients.
10+
*
11+
* The server exposes two endpoints:
12+
* - /sse: For establishing the SSE stream (GET)
13+
* - /messages: For receiving client messages (POST)
14+
*
15+
*/
16+
17+
// Create an MCP server instance
18+
const server = new McpServer({
19+
name: 'simple-sse-server',
20+
version: '1.0.0',
21+
}, { capabilities: { logging: {} } });
22+
23+
server.tool(
24+
'start-notification-stream',
25+
'Starts sending periodic notifications',
26+
{
27+
interval: z.number().describe('Interval in milliseconds between notifications').default(1000),
28+
count: z.number().describe('Number of notifications to send').default(10),
29+
},
30+
async ({ interval, count }, { sendNotification }): Promise<CallToolResult> => {
31+
const sleep = (ms: number) => new Promise(resolve => setTimeout(resolve, ms));
32+
let counter = 0;
33+
34+
// Send the initial notification
35+
await sendNotification({
36+
method: "notifications/message",
37+
params: {
38+
level: "info",
39+
data: `Starting notification stream with ${count} messages every ${interval}ms`
40+
}
41+
});
42+
43+
// Send periodic notifications
44+
while (counter < count) {
45+
counter++;
46+
await sleep(interval);
47+
48+
try {
49+
await sendNotification({
50+
method: "notifications/message",
51+
params: {
52+
level: "info",
53+
data: `Notification #${counter} at ${new Date().toISOString()}`
54+
}
55+
});
56+
}
57+
catch (error) {
58+
console.error("Error sending notification:", error);
59+
}
60+
}
61+
62+
return {
63+
content: [
64+
{
65+
type: 'text',
66+
text: `Completed sending ${count} notifications every ${interval}ms`,
67+
}
68+
],
69+
};
70+
}
71+
);
72+
73+
const app = express();
74+
app.use(express.json());
75+
76+
// Store transports by session ID
77+
const transports: Record<string, SSEServerTransport> = {};
78+
79+
// SSE endpoint for establishing the stream
80+
app.get('/mcp', async (req: Request, res: Response) => {
81+
console.log('Received GET request to /sse (establishing SSE stream)');
82+
83+
try {
84+
// Create a new SSE transport for the client
85+
// The endpoint for POST messages is '/messages'
86+
const transport = new SSEServerTransport('/messages', res);
87+
88+
// Store the transport by session ID
89+
const sessionId = transport.sessionId;
90+
transports[sessionId] = transport;
91+
92+
// Set up onclose handler to clean up transport when closed
93+
transport.onclose = () => {
94+
console.log(`SSE transport closed for session ${sessionId}`);
95+
delete transports[sessionId];
96+
};
97+
98+
// Connect the transport to the MCP server
99+
await server.connect(transport);
100+
101+
// Start the SSE transport to begin streaming
102+
// This sends an initial 'endpoint' event with the session ID in the URL
103+
await transport.start();
104+
105+
console.log(`Established SSE stream with session ID: ${sessionId}`);
106+
} catch (error) {
107+
console.error('Error establishing SSE stream:', error);
108+
if (!res.headersSent) {
109+
res.status(500).send('Error establishing SSE stream');
110+
}
111+
}
112+
});
113+
114+
// Messages endpoint for receiving client JSON-RPC requests
115+
app.post('/messages', async (req: Request, res: Response) => {
116+
console.log('Received POST request to /messages');
117+
118+
// Extract session ID from URL query parameter
119+
// In the SSE protocol, this is added by the client based on the endpoint event
120+
const sessionId = req.query.sessionId as string | undefined;
121+
122+
if (!sessionId) {
123+
console.error('No session ID provided in request URL');
124+
res.status(400).send('Missing sessionId parameter');
125+
return;
126+
}
127+
128+
const transport = transports[sessionId];
129+
if (!transport) {
130+
console.error(`No active transport found for session ID: ${sessionId}`);
131+
res.status(404).send('Session not found');
132+
return;
133+
}
134+
135+
try {
136+
// Handle the POST message with the transport
137+
await transport.handlePostMessage(req, res, req.body);
138+
} catch (error) {
139+
console.error('Error handling request:', error);
140+
if (!res.headersSent) {
141+
res.status(500).send('Error handling request');
142+
}
143+
}
144+
});
145+
146+
// Start the server
147+
const PORT = 3000;
148+
app.listen(PORT, () => {
149+
console.log(`Simple SSE Server (deprecated protocol version 2024-11-05) listening on port ${PORT}`);
150+
});
151+
152+
// Handle server shutdown
153+
process.on('SIGINT', async () => {
154+
console.log('Shutting down server...');
155+
156+
// Close all active transports to properly clean up resources
157+
for (const sessionId in transports) {
158+
try {
159+
console.log(`Closing transport for session ${sessionId}`);
160+
await transports[sessionId].close();
161+
delete transports[sessionId];
162+
} catch (error) {
163+
console.error(`Error closing transport for session ${sessionId}:`, error);
164+
}
165+
}
166+
await server.close();
167+
console.log('Server shutdown complete');
168+
process.exit(0);
169+
});

0 commit comments

Comments
 (0)