Merge branch 'main' into patch-1

This commit is contained in:
Simon Knott 2025-05-06 15:57:25 +02:00 • committed by GitHub
commit fbcf010f31
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
34 changed files with 5605 additions and 1909 deletions

285
README.md
View file

@ -12,7 +12,7 @@
- [Prompts](#prompts)
- [Running Your Server](#running-your-server)
- [stdio](#stdio)
- [HTTP with SSE](#http-with-sse)
- [Streamable HTTP](#streamable-http)
- [Testing and Debugging](#testing-and-debugging)
- [Examples](#examples)
- [Echo Server](#echo-server)
@ -22,6 +22,7 @@
- [Writing MCP Clients](#writing-mcp-clients)
- [Server Capabilities](#server-capabilities)
- [Proxy OAuth Server](#proxy-authorization-requests-upstream)
- [Backwards Compatibility](#backwards-compatibility)
## Overview
@ -29,7 +30,7 @@ The Model Context Protocol allows applications to provide context for LLMs in a
- Build MCP clients that can connect to any MCP server
- Create MCP servers that expose resources, prompts and tools
- Use standard transports like stdio and SSE
- Use standard transports like stdio and Streamable HTTP
- Handle all MCP protocol messages and lifecycle events
## Installation
@ -207,50 +208,179 @@ const transport = new StdioServerTransport();
await server.connect(transport);
```
### HTTP with SSE
### Streamable HTTP
For remote servers, start a web server with a Server-Sent Events (SSE) endpoint, and a separate endpoint for the client to send its messages to:
For remote servers, set up a Streamable HTTP transport that handles both client requests and server-to-client notifications.
#### With Session Management
In some cases, servers need to be stateful. This is achieved by [session management](https://modelcontextprotocol.io/specification/2025-03-26/basic/transports#session-management).
```typescript
import express, { Request, Response } from "express";
import express from "express";
import { randomUUID } from "node:crypto";
import { McpServer } from "@modelcontextprotocol/sdk/server/mcp.js";
import { SSEServerTransport } from "@modelcontextprotocol/sdk/server/sse.js";
import { StreamableHTTPServerTransport } from "@modelcontextprotocol/sdk/server/streamableHttp.js";
import { isInitializeRequest } from "@modelcontextprotocol/sdk/types.js"
const server = new McpServer({
name: "example-server",
version: "1.0.0"
});
// ... set up server resources, tools, and prompts ...
const app = express();
app.use(express.json());
// to support multiple simultaneous connections we have a lookup object from
// sessionId to transport
const transports: {[sessionId: string]: SSEServerTransport} = {};
// Map to store transports by session ID
const transports: { [sessionId: string]: StreamableHTTPServerTransport } = {};
app.get("/sse", async (_: Request, res: Response) => {
const transport = new SSEServerTransport('/messages', res);
transports[transport.sessionId] = transport;
res.on("close", () => {
delete transports[transport.sessionId];
});
await server.connect(transport);
// Handle POST requests for client-to-server communication
app.post('/mcp', async (req, res) => {
// 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
transport = new StreamableHTTPServerTransport({
sessionIdGenerator: () => randomUUID(),
onsessioninitialized: (sessionId) => {
// Store the transport by session ID
transports[sessionId] = transport;
}
});
// Clean up transport when closed
transport.onclose = () => {
if (transport.sessionId) {
delete transports[transport.sessionId];
}
};
const server = new McpServer({
name: "example-server",
version: "1.0.0"
});
// ... set up server resources, tools, and prompts ...
// Connect to the MCP server
await server.connect(transport);
} else {
// Invalid 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
await transport.handleRequest(req, res, req.body);
});
app.post("/messages", async (req: Request, res: Response) => {
const sessionId = req.query.sessionId as string;
// Reusable handler for GET and DELETE requests
const handleSessionRequest = async (req: express.Request, res: express.Response) => {
const sessionId = req.headers['mcp-session-id'] as string | undefined;
if (!sessionId || !transports[sessionId]) {
res.status(400).send('Invalid or missing session ID');
return;
}
const transport = transports[sessionId];
if (transport) {
await transport.handlePostMessage(req, res);
} else {
res.status(400).send('No transport found for sessionId');
await transport.handleRequest(req, res);
};
// Handle GET requests for server-to-client notifications via SSE
app.get('/mcp', handleSessionRequest);
// Handle DELETE requests for session termination
app.delete('/mcp', handleSessionRequest);
app.listen(3000);
```
#### Without Session Management (Stateless)
For simpler use cases where session management isn't needed:
```typescript
const app = express();
app.use(express.json());
app.post('/mcp', async (req: Request, res: Response) => {
// In stateless mode, create a new instance of transport and server for each request
// to ensure complete isolation. A single instance would cause request ID collisions
// when multiple clients connect concurrently.
try {
const server = getServer();
const transport: StreamableHTTPServerTransport = new StreamableHTTPServerTransport({
sessionIdGenerator: undefined,
});
res.on('close', () => {
console.log('Request closed');
transport.close();
server.close();
});
await server.connect(transport);
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,
});
}
}
});
app.listen(3001);
app.get('/mcp', async (req: Request, res: Response) => {
console.log('Received GET MCP request');
res.writeHead(405).end(JSON.stringify({
jsonrpc: "2.0",
error: {
code: -32000,
message: "Method not allowed."
},
id: null
}));
});
app.delete('/mcp', async (req: Request, res: Response) => {
console.log('Received DELETE MCP request');
res.writeHead(405).end(JSON.stringify({
jsonrpc: "2.0",
error: {
code: -32000,
message: "Method not allowed."
},
id: null
}));
});
// Start the server
const PORT = 3000;
app.listen(PORT, () => {
console.log(`MCP Stateless Streamable HTTP Server listening on port ${PORT}`);
});
```
This stateless approach is useful for:
- Simple API wrappers
- RESTful scenarios where each request is independent
- Horizontally scaled deployments without shared session state
### Testing and Debugging
To test your server, you can use the [MCP Inspector](https://github.com/modelcontextprotocol/inspector). See its README for more information.
@ -596,6 +726,105 @@ This setup allows you to:
- Provide custom documentation URLs
- Maintain control over the OAuth flow while delegating to an external provider
### Backwards Compatibility
Clients and servers with StreamableHttp tranport can maintain [backwards compatibility](https://modelcontextprotocol.io/specification/2025-03-26/basic/transports#backwards-compatibility) with the deprecated HTTP+SSE transport (from protocol version 2024-11-05) as follows
#### Client-Side Compatibility
For clients that need to work with both Streamable HTTP and older SSE servers:
```typescript
import { Client } from "@modelcontextprotocol/sdk/client/index.js";
import { StreamableHTTPClientTransport } from "@modelcontextprotocol/sdk/client/streamableHttp.js";
import { SSEClientTransport } from "@modelcontextprotocol/sdk/client/sse.js";
let client: Client|undefined = undefined
const baseUrl = new URL(url);
try {
client = new Client({
name: 'streamable-http-client',
version: '1.0.0'
});
const transport = new StreamableHTTPClientTransport(
new URL(baseUrl)
);
await client.connect(transport);
console.log("Connected using Streamable HTTP transport");
} catch (error) {
// If that fails with a 4xx error, try the older SSE transport
console.log("Streamable HTTP connection failed, falling back to SSE transport");
client = new Client({
name: 'sse-client',
version: '1.0.0'
});
const sseTransport = new SSEClientTransport(baseUrl);
await client.connect(sseTransport);
console.log("Connected using SSE transport");
}
```
#### Server-Side Compatibility
For servers that need to support both Streamable HTTP and older clients:
```typescript
import express from "express";
import { McpServer } from "@modelcontextprotocol/sdk/server/mcp.js";
import { StreamableHTTPServerTransport } from "@modelcontextprotocol/sdk/server/streamableHttp.js";
import { SSEServerTransport } from "@modelcontextprotocol/sdk/server/sse.js";
const server = new McpServer({
name: "backwards-compatible-server",
version: "1.0.0"
});
// ... set up server resources, tools, and prompts ...
const app = express();
app.use(express.json());
// Store transports for each session type
const transports = {
streamable: {} as Record<string, StreamableHTTPServerTransport>,
sse: {} as Record<string, SSEServerTransport>
};
// Modern Streamable HTTP endpoint
app.all('/mcp', async (req, res) => {
// Handle Streamable HTTP transport for modern clients
// Implementation as shown in the "With Session Management" example
// ...
});
// Legacy SSE endpoint for older clients
app.get('/sse', async (req, res) => {
// Create SSE transport for legacy clients
const transport = new SSEServerTransport('/messages', res);
transports.sse[transport.sessionId] = transport;
res.on("close", () => {
delete transports.sse[transport.sessionId];
});
await server.connect(transport);
});
// Legacy message endpoint for older clients
app.post('/messages', async (req, res) => {
const sessionId = req.query.sessionId as string;
const transport = transports.sse[sessionId];
if (transport) {
await transport.handlePostMessage(req, res, req.body);
} else {
res.status(400).send('No transport found for sessionId');
}
});
app.listen(3000);
```
**Note**: The SSE transport is now deprecated in favor of Streamable HTTP. New implementations should use Streamable HTTP, and existing SSE implementations should plan to migrate.
## Documentation
- [Model Context Protocol documentation](https://modelcontextprotocol.io)

View file

@ -1,6 +1,6 @@
{
"name": "@modelcontextprotocol/sdk",
"version": "1.10.0",
"version": "1.11.0",
"description": "Model Context Protocol implementation for TypeScript",
"license": "MIT",
"author": "Anthropic, PBC (https://anthropic.com)",

View file

@ -39,7 +39,7 @@ describe("OAuth Authorization", () => {
const [url, options] = calls[0];
expect(url.toString()).toBe("https://auth.example.com/.well-known/oauth-authorization-server");
expect(options.headers).toEqual({
"MCP-Protocol-Version": "2024-11-05"
"MCP-Protocol-Version": "2025-03-26"
});
});
@ -478,4 +478,4 @@ describe("OAuth Authorization", () => {
).rejects.toThrow("Dynamic client registration failed");
});
});
});
});

View file

@ -165,6 +165,194 @@ test("should reject unsupported protocol version", async () => {
expect(clientTransport.close).toHaveBeenCalled();
});
test("should connect new client to old, supported server version", async () => {
const OLD_VERSION = SUPPORTED_PROTOCOL_VERSIONS[1];
const server = new Server(
{
name: "test server",
version: "1.0",
},
{
capabilities: {
resources: {},
tools: {},
},
},
);
server.setRequestHandler(InitializeRequestSchema, (_request) => ({
protocolVersion: OLD_VERSION,
capabilities: {
resources: {},
tools: {},
},
serverInfo: {
name: "old server",
version: "1.0",
},
}));
server.setRequestHandler(ListResourcesRequestSchema, () => ({
resources: [],
}));
server.setRequestHandler(ListToolsRequestSchema, () => ({
tools: [],
}));
const [clientTransport, serverTransport] =
InMemoryTransport.createLinkedPair();
const client = new Client(
{
name: "new client",
version: "1.0",
protocolVersion: LATEST_PROTOCOL_VERSION,
},
{
capabilities: {
sampling: {},
},
enforceStrictCapabilities: true,
},
);
await Promise.all([
client.connect(clientTransport),
server.connect(serverTransport),
]);
expect(client.getServerVersion()).toEqual({
name: "old server",
version: "1.0",
});
});
test("should negotiate version when client is old, and newer server supports its version", async () => {
const OLD_VERSION = SUPPORTED_PROTOCOL_VERSIONS[1];
const server = new Server(
{
name: "new server",
version: "1.0",
},
{
capabilities: {
resources: {},
tools: {},
},
},
);
server.setRequestHandler(InitializeRequestSchema, (_request) => ({
protocolVersion: LATEST_PROTOCOL_VERSION,
capabilities: {
resources: {},
tools: {},
},
serverInfo: {
name: "new server",
version: "1.0",
},
}));
server.setRequestHandler(ListResourcesRequestSchema, () => ({
resources: [],
}));
server.setRequestHandler(ListToolsRequestSchema, () => ({
tools: [],
}));
const [clientTransport, serverTransport] =
InMemoryTransport.createLinkedPair();
const client = new Client(
{
name: "old client",
version: "1.0",
protocolVersion: OLD_VERSION,
},
{
capabilities: {
sampling: {},
},
enforceStrictCapabilities: true,
},
);
await Promise.all([
client.connect(clientTransport),
server.connect(serverTransport),
]);
expect(client.getServerVersion()).toEqual({
name: "new server",
version: "1.0",
});
});
test("should throw when client is old, and server doesn't support its version", async () => {
const OLD_VERSION = SUPPORTED_PROTOCOL_VERSIONS[1];
const FUTURE_VERSION = "FUTURE_VERSION";
const server = new Server(
{
name: "new server",
version: "1.0",
},
{
capabilities: {
resources: {},
tools: {},
},
},
);
server.setRequestHandler(InitializeRequestSchema, (_request) => ({
protocolVersion: FUTURE_VERSION,
capabilities: {
resources: {},
tools: {},
},
serverInfo: {
name: "new server",
version: "1.0",
},
}));
server.setRequestHandler(ListResourcesRequestSchema, () => ({
resources: [],
}));
server.setRequestHandler(ListToolsRequestSchema, () => ({
tools: [],
}));
const [clientTransport, serverTransport] =
InMemoryTransport.createLinkedPair();
const client = new Client(
{
name: "old client",
version: "1.0",
protocolVersion: OLD_VERSION,
},
{
capabilities: {
sampling: {},
},
enforceStrictCapabilities: true,
},
);
await Promise.all([
expect(client.connect(clientTransport)).rejects.toThrow(
"Server's protocol version is not supported: FUTURE_VERSION"
),
server.connect(serverTransport),
]);
});
test("should respect server capabilities", async () => {
const server = new Server(
{

View file

@ -126,7 +126,11 @@ export class Client<
override async connect(transport: Transport, options?: RequestOptions): Promise<void> {
await super.connect(transport);
// When transport sessionId is already set this means we are trying to reconnect.
// In this case we don't need to initialize again.
if (transport.sessionId !== undefined) {
return;
}
try {
const result = await this.request(
{

View file

@ -1,7 +1,7 @@
import { ChildProcess, IOType } from "node:child_process";
import spawn from "cross-spawn";
import process from "node:process";
import { Stream } from "node:stream";
import { Stream, PassThrough } from "node:stream";
import { ReadBuffer, serializeMessage } from "../shared/stdio.js";
import { Transport } from "../shared/transport.js";
import { JSONRPCMessage } from "../types.js";
@ -93,6 +93,7 @@ export class StdioClientTransport implements Transport {
private _abortController: AbortController = new AbortController();
private _readBuffer: ReadBuffer = new ReadBuffer();
private _serverParams: StdioServerParameters;
private _stderrStream: PassThrough | null = null;
onclose?: () => void;
onerror?: (error: Error) => void;
@ -100,6 +101,9 @@ export class StdioClientTransport implements Transport {
constructor(server: StdioServerParameters) {
this._serverParams = server;
if (server.stderr === "pipe" || server.stderr === "overlapped") {
this._stderrStream = new PassThrough();
}
}
/**
@ -158,15 +162,25 @@ export class StdioClientTransport implements Transport {
this._process.stdout?.on("error", (error) => {
this.onerror?.(error);
});
if (this._stderrStream && this._process.stderr) {
this._process.stderr.pipe(this._stderrStream);
}
});
}
/**
* The stderr stream of the child process, if `StdioServerParameters.stderr` was set to "pipe" or "overlapped".
*
* This is only available after the process has been started.
* If stderr piping was requested, a PassThrough stream is returned _immediately_, allowing callers to
* attach listeners before the start method is invoked. This prevents loss of any early
* error output emitted by the child process.
*/
get stderr(): Stream | null {
if (this._stderrStream) {
return this._stderrStream;
}
return this._process?.stderr ?? null;
}

View file

@ -1,12 +1,24 @@
import { StreamableHTTPClientTransport } from "./streamableHttp.js";
import { StreamableHTTPClientTransport, StreamableHTTPReconnectionOptions } from "./streamableHttp.js";
import { OAuthClientProvider, UnauthorizedError } from "./auth.js";
import { JSONRPCMessage } from "../types.js";
describe("StreamableHTTPClientTransport", () => {
let transport: StreamableHTTPClientTransport;
let mockAuthProvider: jest.Mocked<OAuthClientProvider>;
beforeEach(() => {
transport = new StreamableHTTPClientTransport(new URL("http://localhost:1234/mcp"));
mockAuthProvider = {
get redirectUrl() { return "http://localhost/callback"; },
get clientMetadata() { return { redirect_uris: ["http://localhost/callback"] }; },
clientInformation: jest.fn(() => ({ client_id: "test-client-id", client_secret: "test-client-secret" })),
tokens: jest.fn(),
saveTokens: jest.fn(),
redirectToAuthorization: jest.fn(),
saveCodeVerifier: jest.fn(),
codeVerifier: jest.fn(),
};
transport = new StreamableHTTPClientTransport(new URL("http://localhost:1234/mcp"), { authProvider: mockAuthProvider });
jest.spyOn(global, "fetch");
});
@ -101,6 +113,77 @@ describe("StreamableHTTPClientTransport", () => {
expect(lastCall[1].headers.get("mcp-session-id")).toBe("test-session-id");
});
it("should terminate session with DELETE request", async () => {
// First, simulate getting a session ID
const message: JSONRPCMessage = {
jsonrpc: "2.0",
method: "initialize",
params: {
clientInfo: { name: "test-client", version: "1.0" },
protocolVersion: "2025-03-26"
},
id: "init-id"
};
(global.fetch as jest.Mock).mockResolvedValueOnce({
ok: true,
status: 200,
headers: new Headers({ "content-type": "text/event-stream", "mcp-session-id": "test-session-id" }),
});
await transport.send(message);
expect(transport.sessionId).toBe("test-session-id");
// Now terminate the session
(global.fetch as jest.Mock).mockResolvedValueOnce({
ok: true,
status: 200,
headers: new Headers()
});
await transport.terminateSession();
// Verify the DELETE request was sent with the session ID
const calls = (global.fetch as jest.Mock).mock.calls;
const lastCall = calls[calls.length - 1];
expect(lastCall[1].method).toBe("DELETE");
expect(lastCall[1].headers.get("mcp-session-id")).toBe("test-session-id");
// The session ID should be cleared after successful termination
expect(transport.sessionId).toBeUndefined();
});
it("should handle 405 response when server doesn't support session termination", async () => {
// First, simulate getting a session ID
const message: JSONRPCMessage = {
jsonrpc: "2.0",
method: "initialize",
params: {
clientInfo: { name: "test-client", version: "1.0" },
protocolVersion: "2025-03-26"
},
id: "init-id"
};
(global.fetch as jest.Mock).mockResolvedValueOnce({
ok: true,
status: 200,
headers: new Headers({ "content-type": "text/event-stream", "mcp-session-id": "test-session-id" }),
});
await transport.send(message);
// Now terminate the session, but server responds with 405
(global.fetch as jest.Mock).mockResolvedValueOnce({
ok: false,
status: 405,
statusText: "Method Not Allowed",
headers: new Headers()
});
await expect(transport.terminateSession()).resolves.not.toThrow();
});
it("should handle 404 response when session expires", async () => {
const message: JSONRPCMessage = {
jsonrpc: "2.0",
@ -164,8 +247,7 @@ describe("StreamableHTTPClientTransport", () => {
// We expect the 405 error to be caught and handled gracefully
// This should not throw an error that breaks the transport
await transport.start();
await expect(transport.openSseStream()).rejects.toThrow("Failed to open SSE stream: Method Not Allowed");
await expect(transport["_startOrAuthSse"]({})).resolves.not.toThrow("Failed to open SSE stream: Method Not Allowed");
// Check that GET was attempted
expect(global.fetch).toHaveBeenCalledWith(
expect.anything(),
@ -209,7 +291,7 @@ describe("StreamableHTTPClientTransport", () => {
transport.onmessage = messageSpy;
await transport.start();
await transport.openSseStream();
await transport["_startOrAuthSse"]({});
// Give time for the SSE event to be processed
await new Promise(resolve => setTimeout(resolve, 50));
@ -276,45 +358,62 @@ describe("StreamableHTTPClientTransport", () => {
})).toBe(true);
});
it("should include last-event-id header when resuming a broken connection", async () => {
// First make a successful connection that provides an event ID
const encoder = new TextEncoder();
const stream = new ReadableStream({
start(controller) {
const event = "id: event-123\nevent: message\ndata: {\"jsonrpc\": \"2.0\", \"method\": \"serverNotification\", \"params\": {}}\n\n";
controller.enqueue(encoder.encode(event));
controller.close();
it("should support custom reconnection options", () => {
// Create a transport with custom reconnection options
transport = new StreamableHTTPClientTransport(new URL("http://localhost:1234/mcp"), {
reconnectionOptions: {
initialReconnectionDelay: 500,
maxReconnectionDelay: 10000,
reconnectionDelayGrowFactor: 2,
maxRetries: 5,
}
});
(global.fetch as jest.Mock).mockResolvedValueOnce({
// Verify options were set correctly (checking implementation details)
// Access private properties for testing
const transportInstance = transport as unknown as {
_reconnectionOptions: StreamableHTTPReconnectionOptions;
};
expect(transportInstance._reconnectionOptions.initialReconnectionDelay).toBe(500);
expect(transportInstance._reconnectionOptions.maxRetries).toBe(5);
});
it("should pass lastEventId when reconnecting", async () => {
// Create a fresh transport
transport = new StreamableHTTPClientTransport(new URL("http://localhost:1234/mcp"));
// Mock fetch to verify headers sent
const fetchSpy = global.fetch as jest.Mock;
fetchSpy.mockReset();
fetchSpy.mockResolvedValue({
ok: true,
status: 200,
headers: new Headers({ "content-type": "text/event-stream" }),
body: stream
body: new ReadableStream()
});
// Call the reconnect method directly with a lastEventId
await transport.start();
await transport.openSseStream();
await new Promise(resolve => setTimeout(resolve, 50));
// Type assertion to access private method
const transportWithPrivateMethods = transport as unknown as {
_startOrAuthSse: (options: { resumptionToken?: string }) => Promise<void>
};
await transportWithPrivateMethods._startOrAuthSse({ resumptionToken: "test-event-id" });
// Now simulate attempting to reconnect
(global.fetch as jest.Mock).mockResolvedValueOnce({
ok: true,
status: 200,
headers: new Headers({ "content-type": "text/event-stream" }),
body: null
});
await transport.openSseStream();
// Check that Last-Event-ID was included
const calls = (global.fetch as jest.Mock).mock.calls;
const lastCall = calls[calls.length - 1];
expect(lastCall[1].headers.get("last-event-id")).toBe("event-123");
// Verify fetch was called with the lastEventId header
expect(fetchSpy).toHaveBeenCalled();
const fetchCall = fetchSpy.mock.calls[0];
const headers = fetchCall[1].headers;
expect(headers.get("last-event-id")).toBe("test-event-id");
});
it("should throw error when invalid content-type is received", async () => {
// Clear any previous state from other tests
jest.clearAllMocks();
// Create a fresh transport instance
transport = new StreamableHTTPClientTransport(new URL("http://localhost:1234/mcp"));
const message: JSONRPCMessage = {
jsonrpc: "2.0",
method: "test",
@ -324,7 +423,7 @@ describe("StreamableHTTPClientTransport", () => {
const stream = new ReadableStream({
start(controller) {
controller.enqueue("invalid text response");
controller.enqueue(new TextEncoder().encode("invalid text response"));
controller.close();
}
});
@ -366,7 +465,7 @@ describe("StreamableHTTPClientTransport", () => {
await transport.start();
await transport.openSseStream();
await transport["_startOrAuthSse"]({});
expect((actualReqInit.headers as Headers).get("x-custom-header")).toBe("CustomValue");
requestInit.headers["X-Custom-Header"] = "SecondCustomValue";
@ -376,4 +475,61 @@ describe("StreamableHTTPClientTransport", () => {
expect(global.fetch).toHaveBeenCalledTimes(2);
});
it("should have exponential backoff with configurable maxRetries", () => {
// This test verifies the maxRetries and backoff calculation directly
// Create transport with specific options for testing
transport = new StreamableHTTPClientTransport(new URL("http://localhost:1234/mcp"), {
reconnectionOptions: {
initialReconnectionDelay: 100,
maxReconnectionDelay: 5000,
reconnectionDelayGrowFactor: 2,
maxRetries: 3,
}
});
// Get access to the internal implementation
const getDelay = transport["_getNextReconnectionDelay"].bind(transport);
// First retry - should use initial delay
expect(getDelay(0)).toBe(100);
// Second retry - should double (2^1 * 100 = 200)
expect(getDelay(1)).toBe(200);
// Third retry - should double again (2^2 * 100 = 400)
expect(getDelay(2)).toBe(400);
// Fourth retry - should double again (2^3 * 100 = 800)
expect(getDelay(3)).toBe(800);
// Tenth retry - should be capped at maxReconnectionDelay
expect(getDelay(10)).toBe(5000);
});
it("attempts auth flow on 401 during POST request", async () => {
const message: JSONRPCMessage = {
jsonrpc: "2.0",
method: "test",
params: {},
id: "test-id"
};
(global.fetch as jest.Mock)
.mockResolvedValueOnce({
ok: false,
status: 401,
statusText: "Unauthorized",
headers: new Headers()
})
.mockResolvedValue({
ok: false,
status: 404
});
await expect(transport.send(message)).rejects.toThrow(UnauthorizedError);
expect(mockAuthProvider.redirectToAuthorization.mock.calls).toHaveLength(1);
});
});

View file

@ -1,8 +1,16 @@
import { Transport } from "../shared/transport.js";
import { JSONRPCMessage, JSONRPCMessageSchema } from "../types.js";
import { isInitializedNotification, isJSONRPCRequest, isJSONRPCResponse, JSONRPCMessage, JSONRPCMessageSchema } from "../types.js";
import { auth, AuthResult, OAuthClientProvider, UnauthorizedError } from "./auth.js";
import { EventSourceParserStream } from "eventsource-parser/stream";
// Default reconnection options for StreamableHTTP connections
const DEFAULT_STREAMABLE_HTTP_RECONNECTION_OPTIONS: StreamableHTTPReconnectionOptions = {
initialReconnectionDelay: 1000,
maxReconnectionDelay: 30000,
reconnectionDelayGrowFactor: 1.5,
maxRetries: 2,
};
export class StreamableHTTPError extends Error {
constructor(
public readonly code: number | undefined,
@ -12,6 +20,60 @@ export class StreamableHTTPError extends Error {
}
}
/**
* Options for starting or authenticating an SSE connection
*/
interface StartSSEOptions {
/**
* The resumption token used to continue long-running requests that were interrupted.
*
* This allows clients to reconnect and continue from where they left off.
*/
resumptionToken?: string;
/**
* A callback that is invoked when the resumption token changes.
*
* This allows clients to persist the latest token for potential reconnection.
*/
onresumptiontoken?: (token: string) => void;
/**
* Override Message ID to associate with the replay message
* so that response can be associate with the new resumed request.
*/
replayMessageId?: string | number;
}
/**
* Configuration options for reconnection behavior of the StreamableHTTPClientTransport.
*/
export interface StreamableHTTPReconnectionOptions {
/**
* Maximum backoff time between reconnection attempts in milliseconds.
* Default is 30000 (30 seconds).
*/
maxReconnectionDelay: number;
/**
* Initial backoff time between reconnection attempts in milliseconds.
* Default is 1000 (1 second).
*/
initialReconnectionDelay: number;
/**
* The factor by which the reconnection delay increases after each attempt.
* Default is 1.5.
*/
reconnectionDelayGrowFactor: number;
/**
* Maximum number of reconnection attempts before giving up.
* Default is 2.
*/
maxRetries: number;
}
/**
* Configuration options for the `StreamableHTTPClientTransport`.
*/
@ -36,6 +98,17 @@ export type StreamableHTTPClientTransportOptions = {
* Customizes HTTP requests to the server.
*/
requestInit?: RequestInit;
/**
* Options to configure the reconnection behavior.
*/
reconnectionOptions?: StreamableHTTPReconnectionOptions;
/**
* Session ID for the connection. This is used to identify the session on the server.
* When not provided and connecting to a server that supports session IDs, the server will generate a new session ID.
*/
sessionId?: string;
};
/**
@ -49,7 +122,7 @@ export class StreamableHTTPClientTransport implements Transport {
private _requestInit?: RequestInit;
private _authProvider?: OAuthClientProvider;
private _sessionId?: string;
private _lastEventId?: string;
private _reconnectionOptions: StreamableHTTPReconnectionOptions;
onclose?: () => void;
onerror?: (error: Error) => void;
@ -62,6 +135,8 @@ export class StreamableHTTPClientTransport implements Transport {
this._url = url;
this._requestInit = opts?.requestInit;
this._authProvider = opts?.authProvider;
this._sessionId = opts?.sessionId;
this._reconnectionOptions = opts?.reconnectionOptions ?? DEFAULT_STREAMABLE_HTTP_RECONNECTION_OPTIONS;
}
private async _authThenStart(): Promise<void> {
@ -81,7 +156,7 @@ export class StreamableHTTPClientTransport implements Transport {
throw new UnauthorizedError();
}
return await this._startOrAuthStandaloneSSE();
return await this._startOrAuthSse({ resumptionToken: undefined });
}
private async _commonHeaders(): Promise<Headers> {
@ -102,16 +177,18 @@ export class StreamableHTTPClientTransport implements Transport {
);
}
private async _startOrAuthStandaloneSSE(): Promise<void> {
private async _startOrAuthSse(options: StartSSEOptions): Promise<void> {
const { resumptionToken } = options;
try {
// Try to open an initial SSE stream with GET to listen for server messages
// This is optional according to the spec - server may not support it
const headers = await this._commonHeaders();
headers.set("Accept", "text/event-stream");
// Include Last-Event-ID header for resumable streams
if (this._lastEventId) {
headers.set("last-event-id", this._lastEventId);
// Include Last-Event-ID header for resumable streams if provided
if (resumptionToken) {
headers.set("last-event-id", resumptionToken);
}
const response = await fetch(this._url, {
@ -126,50 +203,139 @@ export class StreamableHTTPClientTransport implements Transport {
return await this._authThenStart();
}
// 405 indicates that the server does not offer an SSE stream at GET endpoint
// This is an expected case that should not trigger an error
if (response.status === 405) {
return;
}
throw new StreamableHTTPError(
response.status,
`Failed to open SSE stream: ${response.statusText}`,
);
}
// Successful connection, handle the SSE stream as a standalone listener
this._handleSseStream(response.body);
this._handleSseStream(response.body, options);
} catch (error) {
this.onerror?.(error as Error);
throw error;
}
}
private _handleSseStream(stream: ReadableStream<Uint8Array> | null): void {
if (!stream) {
/**
* Calculates the next reconnection delay using backoff algorithm
*
* @param attempt Current reconnection attempt count for the specific stream
* @returns Time to wait in milliseconds before next reconnection attempt
*/
private _getNextReconnectionDelay(attempt: number): number {
// Access default values directly, ensuring they're never undefined
const initialDelay = this._reconnectionOptions.initialReconnectionDelay;
const growFactor = this._reconnectionOptions.reconnectionDelayGrowFactor;
const maxDelay = this._reconnectionOptions.maxReconnectionDelay;
// Cap at maximum delay
return Math.min(initialDelay * Math.pow(growFactor, attempt), maxDelay);
}
/**
* Schedule a reconnection attempt with exponential backoff
*
* @param lastEventId The ID of the last received event for resumability
* @param attemptCount Current reconnection attempt count for this specific stream
*/
private _scheduleReconnection(options: StartSSEOptions, attemptCount = 0): void {
// Use provided options or default options
const maxRetries = this._reconnectionOptions.maxRetries;
// Check if we've exceeded maximum retry attempts
if (maxRetries > 0 && attemptCount >= maxRetries) {
this.onerror?.(new Error(`Maximum reconnection attempts (${maxRetries}) exceeded.`));
return;
}
const processStream = async () => {
// Create a pipeline: binary stream -> text decoder -> SSE parser
const eventStream = stream
.pipeThrough(new TextDecoderStream())
.pipeThrough(new EventSourceParserStream());
// Calculate next delay based on current attempt count
const delay = this._getNextReconnectionDelay(attemptCount);
for await (const event of eventStream) {
// Update last event ID if provided
if (event.id) {
this._lastEventId = event.id;
// Schedule the reconnection
setTimeout(() => {
// Use the last event ID to resume where we left off
this._startOrAuthSse(options).catch(error => {
this.onerror?.(new Error(`Failed to reconnect SSE stream: ${error instanceof Error ? error.message : String(error)}`));
// Schedule another attempt if this one failed, incrementing the attempt counter
this._scheduleReconnection(options, attemptCount + 1);
});
}, delay);
}
private _handleSseStream(stream: ReadableStream<Uint8Array> | null, options: StartSSEOptions): void {
if (!stream) {
return;
}
const { onresumptiontoken, replayMessageId } = options;
let lastEventId: string | undefined;
const processStream = async () => {
// this is the closest we can get to trying to catch network errors
// if something happens reader will throw
try {
// Create a pipeline: binary stream -> text decoder -> SSE parser
const reader = stream
.pipeThrough(new TextDecoderStream())
.pipeThrough(new EventSourceParserStream())
.getReader();
while (true) {
const { value: event, done } = await reader.read();
if (done) {
break;
}
// Update last event ID if provided
if (event.id) {
lastEventId = event.id;
onresumptiontoken?.(event.id);
}
if (!event.event || event.event === "message") {
try {
const message = JSONRPCMessageSchema.parse(JSON.parse(event.data));
if (replayMessageId !== undefined && isJSONRPCResponse(message)) {
message.id = replayMessageId;
}
this.onmessage?.(message);
} catch (error) {
this.onerror?.(error as Error);
}
}
}
// Handle message events (default event type is undefined per docs)
// or explicit 'message' event type
if (!event.event || event.event === "message") {
try {
const message = JSONRPCMessageSchema.parse(JSON.parse(event.data));
this.onmessage?.(message);
} catch (error) {
this.onerror?.(error as Error);
} catch (error) {
// Handle stream errors - likely a network disconnect
this.onerror?.(new Error(`SSE stream disconnected: ${error}`));
// Attempt to reconnect if the stream disconnects unexpectedly and we aren't closing
if (this._abortController && !this._abortController.signal.aborted) {
// Use the exponential backoff reconnection strategy
if (lastEventId !== undefined) {
try {
this._scheduleReconnection({
resumptionToken: lastEventId,
onresumptiontoken,
replayMessageId
}, 0);
}
catch (error) {
this.onerror?.(new Error(`Failed to reconnect: ${error instanceof Error ? error.message : String(error)}`));
}
}
}
}
};
processStream().catch(err => this.onerror?.(err));
processStream();
}
async start() {
@ -203,8 +369,16 @@ export class StreamableHTTPClientTransport implements Transport {
this.onclose?.();
}
async send(message: JSONRPCMessage | JSONRPCMessage[]): Promise<void> {
async send(message: JSONRPCMessage | JSONRPCMessage[], options?: { resumptionToken?: string, onresumptiontoken?: (token: string) => void }): Promise<void> {
try {
const { resumptionToken, onresumptiontoken } = options || {};
if (resumptionToken) {
// If we have at last event ID, we need to reconnect the SSE stream
this._startOrAuthSse({ resumptionToken, replayMessageId: isJSONRPCRequest(message) ? message.id : undefined }).catch(err => this.onerror?.(err));
return;
}
const headers = await this._commonHeaders();
headers.set("content-type", "application/json");
headers.set("accept", "application/json, text/event-stream");
@ -244,6 +418,12 @@ export class StreamableHTTPClientTransport implements Transport {
// If the response is 202 Accepted, there's no body to process
if (response.status === 202) {
// if the accepted notification is initialized, we start the SSE stream
// if it's supported by the server
if (isInitializedNotification(message)) {
// Start without a lastEventId since this is a fresh connection
this._startOrAuthSse({ resumptionToken: undefined }).catch(err => this.onerror?.(err));
}
return;
}
@ -257,7 +437,10 @@ export class StreamableHTTPClientTransport implements Transport {
if (hasRequests) {
if (contentType?.includes("text/event-stream")) {
this._handleSseStream(response.body);
// Handle SSE stream responses for requests
// We use the same handler as standalone streams, which now supports
// reconnection with the last event ID
this._handleSseStream(response.body, { onresumptiontoken });
} else if (contentType?.includes("application/json")) {
// For non-streaming servers, we might get direct JSON responses
const data = await response.json();
@ -281,19 +464,51 @@ export class StreamableHTTPClientTransport implements Transport {
}
}
get sessionId(): string | undefined {
return this._sessionId;
}
/**
* Opens SSE stream to receive messages from the server.
*
* This allows the server to push messages to the client without requiring the client
* to first send a request via HTTP POST. Some servers may not support this feature.
* If authentication is required but fails, this method will throw an UnauthorizedError.
* Terminates the current session by sending a DELETE request to the server.
*
* Clients that no longer need a particular session
* (e.g., because the user is leaving the client application) SHOULD send an
* HTTP DELETE to the MCP endpoint with the Mcp-Session-Id header to explicitly
* terminate the session.
*
* The server MAY respond with HTTP 405 Method Not Allowed, indicating that
* the server does not allow clients to terminate sessions.
*/
async openSseStream(): Promise<void> {
if (!this._abortController) {
throw new Error(
"StreamableHTTPClientTransport not started! Call connect() before openSseStream().",
);
async terminateSession(): Promise<void> {
if (!this._sessionId) {
return; // No session to terminate
}
try {
const headers = await this._commonHeaders();
const init = {
...this._requestInit,
method: "DELETE",
headers,
signal: this._abortController?.signal,
};
const response = await fetch(this._url, init);
// We specifically handle 405 as a valid response according to the spec,
// meaning the server does not support explicit session termination
if (!response.ok && response.status !== 405) {
throw new StreamableHTTPError(
response.status,
`Failed to terminate session: ${response.statusText}`
);
}
this._sessionId = undefined;
} catch (error) {
this.onerror?.(error as Error);
throw error;
}
await this._startOrAuthStandaloneSSE();
}
}

View file

@ -2,129 +2,296 @@
This directory contains example implementations of MCP clients and servers using the TypeScript SDK.
## Streamable HTTP - single node deployment with basic session state management
## Table of Contents
Multi node with stete management example will be added soon after we add support.
- [Client Implementations](#client-implementations)
- [Streamable HTTP Client](#streamable-http-client)
- [Backwards Compatible Client](#backwards-compatible-client)
- [Server Implementations](#server-implementations)
- [Single Node Deployment](#single-node-deployment)
- [Streamable HTTP Transport](#streamable-http-transport)
- [Deprecated SSE Transport](#deprecated-sse-transport)
- [Backwards Compatible Server](#streamable-http-backwards-compatible-server-with-sse)
- [Multi-Node Deployment](#multi-node-deployment)
- [Backwards Compatibility](#testing-streamable-http-backwards-compatibility-with-sse)
### Server with JSON response mode (`server/jsonResponseStreamableHttp.ts`)
## Client Implementations
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.
### Streamable HTTP Client
#### Running the server
A full-featured interactive client that connects to a Streamable HTTP server, demonstrating how to:
```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" \
-H "Accept: text/event-stream" \
-d '{
"jsonrpc": "2.0",
"method": "initialize",
"params": {
"capabilities": {},
"protocolVersion": "2025-03-26",
"clientInfo": {
"name": "test",
"version": "1.0.0"
}
},
"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 "Accept: text/event-stream" \
-H "mcp-session-id: $SESSION_ID" \
-d '{
"jsonrpc": "2.0",
"method": "tools/call",
"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:
- A simple `greet` tool that returns a greeting for a name
- A `greeting-template` prompt that generates a greeting template
- A static `greeting-resource` resource
#### Running the server
```bash
npx tsx src/examples/server/simpleStreamableHttp.ts
```
The server will start on port 3000. You can test the initialization and tool listing:
```bash
# First initialize the server and save the session ID to a variable
SESSION_ID=$(curl -X POST \
-H "Content-Type: application/json" \
-H "Accept: application/json" \
-H "Accept: text/event-stream" \
-d '{
"jsonrpc": "2.0",
"method": "initialize",
"params": {
"capabilities": {},
"protocolVersion": "2025-03-26",
"clientInfo": {
"name": "test",
"version": "1.0.0"
}
},
"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
# Then list tools using the saved session ID
curl -X POST -H "Content-Type: application/json" -H "Accept: application/json, text/event-stream" \
-H "mcp-session-id: $SESSION_ID" \
-d '{"jsonrpc":"2.0","method":"tools/list","params":{},"id":"2"}' \
http://localhost:3000/mcp
```
### Client (`client/simpleStreamableHttp.ts`)
A client that connects to the server, initializes it, and demonstrates how to:
- List available tools and call the `greet` tool
- List available prompts and get the `greeting-template` prompt
- Establish and manage a connection to an MCP server
- List and call tools with arguments
- Handle notifications through the SSE stream
- List and get prompts with arguments
- List available resources
#### Running the client
- Handle session termination and reconnection
- Support for resumability with Last-Event-ID tracking
```bash
npx tsx src/examples/client/simpleStreamableHttp.ts
```
Make sure the server is running before starting the client.
### Backwards Compatible Client
## Notes
A client that implements backwards compatibility according to the [MCP specification](https://modelcontextprotocol.io/specification/2025-03-26/basic/transports#backwards-compatibility), allowing it to work with both new and legacy servers. This client demonstrates:
- These examples demonstrate the basic usage of the Streamable HTTP transport
- The server manages sessions between the calls
- The client handles both direct HTTP responses and SSE streaming responses
- The client first POSTs an initialize request to the server URL:
- If successful, it uses the Streamable HTTP transport
- If it fails with a 4xx status, it attempts a GET request to establish an SSE stream
```bash
npx tsx src/examples/client/streamableHttpWithSseFallbackClient.ts
```
## Server Implementations
### Single Node Deployment
These examples demonstrate how to set up an MCP server on a single node with different transport options.
#### Streamable HTTP Transport
##### Simple Streamable HTTP Server
A server that implements the Streamable HTTP transport (protocol version 2025-03-26).
- Basic server setup with Express and the Streamable HTTP transport
- Session management with an in-memory event store for resumability
- Tool implementation with the `greet` and `multi-greet` tools
- Prompt implementation with the `greeting-template` prompt
- Static resource exposure
- Support for notifications via SSE stream established by GET requests
- Session termination via DELETE requests
```bash
npx tsx src/examples/server/simpleStreamableHttp.ts
```
##### JSON Response Mode Server
A server that uses Streamable HTTP transport with JSON response mode enabled (no SSE).
- Streamable HTTP with JSON response mode, which returns responses directly in the response body
- Limited support for notifications (since SSE is disabled)
- Proper response handling according to the MCP specification for servers that don't support SSE
- Returning appropriate HTTP status codes for unsupported methods
```bash
npx tsx src/examples/server/jsonResponseStreamableHttp.ts
```
##### Streamable HTTP with server notifications
A server that demonstrates server notifications using Streamable HTTP.
- Resource list change notifications with dynamically added resources
- Automatic resource creation on a timed interval
```bash
npx tsx src/examples/server/standaloneSseWithGetStreamableHttp.ts
```
#### Deprecated SSE Transport
A server that implements the deprecated HTTP+SSE transport (protocol version 2024-11-05). This example only used for testing backwards compatibility for clients.
- Two separate endpoints: `/mcp` for the SSE stream (GET) and `/messages` for client messages (POST)
- Tool implementation with a `start-notification-stream` tool that demonstrates sending periodic notifications
```bash
npx tsx src/examples/server/simpleSseServer.ts
```
#### Streamable Http Backwards Compatible Server with SSE
A server that supports both Streamable HTTP and SSE transports, adhering to the [MCP specification for backwards compatibility](https://modelcontextprotocol.io/specification/2025-03-26/basic/transports#backwards-compatibility).
- Single MCP server instance with multiple transport options
- Support for Streamable HTTP requests at `/mcp` endpoint (GET/POST/DELETE)
- Support for deprecated SSE transport with `/sse` (GET) and `/messages` (POST)
- Session type tracking to avoid mixing transport types
- Notifications and tool execution across both transport types
```bash
npx tsx src/examples/server/sseAndStreamableHttpCompatibleServer.ts
```
### Multi-Node Deployment
When deploying MCP servers in a horizontally scaled environment (multiple server instances), there are a few different options that can be useful for different use cases:
- **Stateless mode** - No need to maintain state between calls to MCP servers. Useful for simple API wrapper servers.
- **Persistent storage mode** - No local state needed, but session data is stored in a database. Example: an MCP server for online ordering where the shopping cart is stored in a database.
- **Local state with message routing** - Local state is needed, and all requests for a session must be routed to the correct node. This can be done with a message queue and pub/sub system.
#### Stateless Mode
The Streamable HTTP transport can be configured to operate without tracking sessions. This is perfect for simple API proxies or when each request is completely independent.
##### Implementation
To enable stateless mode, configure the `StreamableHTTPServerTransport` with:
```typescript
sessionIdGenerator: undefined
```
This disables session management entirely, and the server won't generate or expect session IDs.
- No session ID headers are sent or expected
- Any server node can process any request
- No state is preserved between requests
- Perfect for RESTful or stateless API scenarios
- Simplest deployment model with minimal infrastructure requirements
```
┌─────────────────────────────────────────────┐
│ Client │
└─────────────────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────┐
│ Load Balancer │
└─────────────────────────────────────────────┘
│ │
▼ ▼
┌─────────────────┐ ┌─────────────────────┐
│ MCP Server #1 │ │ MCP Server #2 │
│ (Node.js) │ │ (Node.js) │
└─────────────────┘ └─────────────────────┘
```
#### Persistent Storage Mode
For cases where you need session continuity but don't need to maintain in-memory state on specific nodes, you can use a database to persist session data while still allowing any node to handle requests.
##### Implementation
Configure the transport with session management, but retrieve and store all state in an external persistent storage:
```typescript
sessionIdGenerator: () => randomUUID(),
eventStore: databaseEventStore
```
All session state is stored in the database, and any node can serve any client by retrieving the state when needed.
- Maintains sessions with unique IDs
- Stores all session data in an external database
- Provides resumability through the database-backed EventStore
- Any node can handle any request for the same session
- No node-specific memory state means no need for message routing
- Good for applications where state can be fully externalized
- Somewhat higher latency due to database access for each request
```
┌─────────────────────────────────────────────┐
│ Client │
└─────────────────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────┐
│ Load Balancer │
└─────────────────────────────────────────────┘
│ │
▼ ▼
┌─────────────────┐ ┌─────────────────────┐
│ MCP Server #1 │ │ MCP Server #2 │
│ (Node.js) │ │ (Node.js) │
└─────────────────┘ └─────────────────────┘
│ │
│ │
▼ ▼
┌─────────────────────────────────────────────┐
│ Database (PostgreSQL) │
│ │
│ • Session state │
│ • Event storage for resumability │
└─────────────────────────────────────────────┘
```
#### Streamable HTTP with Distributed Message Routing
For scenarios where local in-memory state must be maintained on specific nodes (such as Computer Use or complex session state), the Streamable HTTP transport can be combined with a pub/sub system to route messages to the correct node handling each session.
1. **Bidirectional Message Queue Integration**:
- All nodes both publish to and subscribe from the message queue
- Each node registers the sessions it's actively handling
- Messages are routed based on session ownership
2. **Request Handling Flow**:
- When a client connects to Node A with an existing `mcp-session-id`
- If Node A doesn't own this session, it:
- Establishes and maintains the SSE connection with the client
- Publishes the request to the message queue with the session ID
- Node B (which owns the session) receives the request from the queue
- Node B processes the request with its local session state
- Node B publishes responses/notifications back to the queue
- Node A subscribes to the response channel and forwards to the client
3. **Channel Identification**:
- Each message channel combines both `mcp-session-id` and `stream-id`
- This ensures responses are correctly routed back to the originating connection
```
┌─────────────────────────────────────────────┐
│ Client │
└─────────────────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────┐
│ Load Balancer │
└─────────────────────────────────────────────┘
│ │
▼ ▼
┌─────────────────┐ ┌─────────────────────┐
│ MCP Server #1 │◄───►│ MCP Server #2 │
│ (Has Session A) │ │ (Has Session B) │
└─────────────────┘ └─────────────────────┘
▲│ ▲│
│▼ │▼
┌─────────────────────────────────────────────┐
│ Message Queue / Pub-Sub │
│ │
│ • Session ownership registry │
│ • Bidirectional message routing │
│ • Request/response forwarding │
└─────────────────────────────────────────────┘
```
- Maintains session affinity for stateful operations without client redirection
- Enables horizontal scaling while preserving complex in-memory state
- Provides fault tolerance through the message queue as intermediary
## Backwards Compatibility
### Testing Streamable HTTP Backwards Compatibility with SSE
To test the backwards compatibility features:
1. Start one of the server implementations:
```bash
# Legacy SSE server (protocol version 2024-11-05)
npx tsx src/examples/server/simpleSseServer.ts
# Streamable HTTP server (protocol version 2025-03-26)
npx tsx src/examples/server/simpleStreamableHttp.ts
# Backwards compatible server (supports both protocols)
npx tsx src/examples/server/sseAndStreamableHttpCompatibleServer.ts
```
2. Then run the backwards compatible client:
```bash
npx tsx src/examples/client/streamableHttpWithSseFallbackClient.ts
```
This demonstrates how the MCP ecosystem ensures interoperability between clients and servers regardless of which protocol version they were built for.

View file

@ -0,0 +1,160 @@
import { Client } from '../../client/index.js';
import { StreamableHTTPClientTransport } from '../../client/streamableHttp.js';
import {
CallToolRequest,
CallToolResultSchema,
LoggingMessageNotificationSchema,
CallToolResult,
} from '../../types.js';
/**
* Multiple Clients MCP Example
*
* This client demonstrates how to:
* 1. Create multiple MCP clients in parallel
* 2. Each client calls a single tool
* 3. Track notifications from each client independently
*/
// Command line args processing
const args = process.argv.slice(2);
const serverUrl = args[0] || 'http://localhost:3000/mcp';
interface ClientConfig {
id: string;
name: string;
toolName: string;
toolArguments: Record<string, string | number | boolean>;
}
async function createAndRunClient(config: ClientConfig): Promise<{ id: string; result: CallToolResult }> {
console.log(`[${config.id}] Creating client: ${config.name}`);
const client = new Client({
name: config.name,
version: '1.0.0'
});
const transport = new StreamableHTTPClientTransport(new URL(serverUrl));
// Set up client-specific error handler
client.onerror = (error) => {
console.error(`[${config.id}] Client error:`, error);
};
// Set up client-specific notification handler
client.setNotificationHandler(LoggingMessageNotificationSchema, (notification) => {
console.log(`[${config.id}] Notification: ${notification.params.data}`);
});
try {
// Connect to the server
await client.connect(transport);
console.log(`[${config.id}] Connected to MCP server`);
// Call the specified tool
console.log(`[${config.id}] Calling tool: ${config.toolName}`);
const toolRequest: CallToolRequest = {
method: 'tools/call',
params: {
name: config.toolName,
arguments: {
...config.toolArguments,
// Add client ID to arguments for identification in notifications
caller: config.id
}
}
};
const result = await client.request(toolRequest, CallToolResultSchema);
console.log(`[${config.id}] Tool call completed`);
// Keep the connection open for a bit to receive notifications
await new Promise(resolve => setTimeout(resolve, 5000));
// Disconnect
await transport.close();
console.log(`[${config.id}] Disconnected from MCP server`);
return { id: config.id, result };
} catch (error) {
console.error(`[${config.id}] Error:`, error);
throw error;
}
}
async function main(): Promise<void> {
console.log('MCP Multiple Clients Example');
console.log('============================');
console.log(`Server URL: ${serverUrl}`);
console.log('');
try {
// Define client configurations
const clientConfigs: ClientConfig[] = [
{
id: 'client1',
name: 'basic-client-1',
toolName: 'start-notification-stream',
toolArguments: {
interval: 3, // 1 second between notifications
count: 5 // Send 5 notifications
}
},
{
id: 'client2',
name: 'basic-client-2',
toolName: 'start-notification-stream',
toolArguments: {
interval: 2, // 2 seconds between notifications
count: 3 // Send 3 notifications
}
},
{
id: 'client3',
name: 'basic-client-3',
toolName: 'start-notification-stream',
toolArguments: {
interval: 1, // 0.5 second between notifications
count: 8 // Send 8 notifications
}
}
];
// Start all clients in parallel
console.log(`Starting ${clientConfigs.length} clients in parallel...`);
console.log('');
const clientPromises = clientConfigs.map(config => createAndRunClient(config));
const results = await Promise.all(clientPromises);
// Display results from all clients
console.log('\n=== Final Results ===');
results.forEach(({ id, result }) => {
console.log(`\n[${id}] Tool result:`);
if (Array.isArray(result.content)) {
result.content.forEach((item: { type: string; text?: string }) => {
if (item.type === 'text' && item.text) {
console.log(` ${item.text}`);
} else {
console.log(` ${item.type} content:`, item);
}
});
} else {
console.log(` Unexpected result format:`, result);
}
});
console.log('\n=== All clients completed successfully ===');
} catch (error) {
console.error('Error running multiple clients:', error);
process.exit(1);
}
}
// Start the example
main().catch((error: unknown) => {
console.error('Error running MCP multiple clients example:', error);
process.exit(1);
});

View file

@ -0,0 +1,197 @@
import { Client } from '../../client/index.js';
import { StreamableHTTPClientTransport } from '../../client/streamableHttp.js';
import {
ListToolsRequest,
ListToolsResultSchema,
CallToolResultSchema,
LoggingMessageNotificationSchema,
CallToolResult,
} from '../../types.js';
/**
* Parallel Tool Calls MCP Client
*
* This client demonstrates how to:
* 1. Start multiple tool calls in parallel
* 2. Track notifications from each tool call using a caller parameter
*/
// Command line args processing
const args = process.argv.slice(2);
const serverUrl = args[0] || 'http://localhost:3000/mcp';
async function main(): Promise<void> {
console.log('MCP Parallel Tool Calls Client');
console.log('==============================');
console.log(`Connecting to server at: ${serverUrl}`);
let client: Client;
let transport: StreamableHTTPClientTransport;
try {
// Create client with streamable HTTP transport
client = new Client({
name: 'parallel-tool-calls-client',
version: '1.0.0'
});
client.onerror = (error) => {
console.error('Client error:', error);
};
// Connect to the server
transport = new StreamableHTTPClientTransport(new URL(serverUrl));
await client.connect(transport);
console.log('Successfully connected to MCP server');
// Set up notification handler with caller identification
client.setNotificationHandler(LoggingMessageNotificationSchema, (notification) => {
console.log(`Notification: ${notification.params.data}`);
});
console.log("List tools")
const toolsRequest = await listTools(client);
console.log("Tools: ", toolsRequest)
// 2. Start multiple notification tools in parallel
console.log('\n=== Starting Multiple Notification Streams in Parallel ===');
const toolResults = await startParallelNotificationTools(client);
// Log the results from each tool call
for (const [caller, result] of Object.entries(toolResults)) {
console.log(`\n=== Tool result for ${caller} ===`);
result.content.forEach((item: { type: string; text?: string; }) => {
if (item.type === 'text') {
console.log(` ${item.text}`);
} else {
console.log(` ${item.type} content:`, item);
}
});
}
// 3. Wait for all notifications (10 seconds)
console.log('\n=== Waiting for all notifications ===');
await new Promise(resolve => setTimeout(resolve, 10000));
// 4. Disconnect
console.log('\n=== Disconnecting ===');
await transport.close();
console.log('Disconnected from MCP server');
} catch (error) {
console.error('Error running client:', error);
process.exit(1);
}
}
/**
* List available tools on the server
*/
async function listTools(client: Client): Promise<void> {
try {
const toolsRequest: ListToolsRequest = {
method: 'tools/list',
params: {}
};
const toolsResult = await client.request(toolsRequest, ListToolsResultSchema);
console.log('Available tools:');
if (toolsResult.tools.length === 0) {
console.log(' No tools available');
} else {
for (const tool of toolsResult.tools) {
console.log(` - ${tool.name}: ${tool.description}`);
}
}
} catch (error) {
console.log(`Tools not supported by this server: ${error}`);
}
}
/**
* Start multiple notification tools in parallel with different configurations
* Each tool call includes a caller parameter to identify its notifications
*/
async function startParallelNotificationTools(client: Client): Promise<Record<string, CallToolResult>> {
try {
// Define multiple tool calls with different configurations
const toolCalls = [
{
caller: 'fast-notifier',
request: {
method: 'tools/call',
params: {
name: 'start-notification-stream',
arguments: {
interval: 2, // 0.5 second between notifications
count: 10, // Send 10 notifications
caller: 'fast-notifier' // Identify this tool call
}
}
}
},
{
caller: 'slow-notifier',
request: {
method: 'tools/call',
params: {
name: 'start-notification-stream',
arguments: {
interval: 5, // 2 seconds between notifications
count: 5, // Send 5 notifications
caller: 'slow-notifier' // Identify this tool call
}
}
}
},
{
caller: 'burst-notifier',
request: {
method: 'tools/call',
params: {
name: 'start-notification-stream',
arguments: {
interval: 1, // 0.1 second between notifications
count: 3, // Send just 3 notifications
caller: 'burst-notifier' // Identify this tool call
}
}
}
}
];
console.log(`Starting ${toolCalls.length} notification tools in parallel...`);
// Start all tool calls in parallel
const toolPromises = toolCalls.map(({ caller, request }) => {
console.log(`Starting tool call for ${caller}...`);
return client.request(request, CallToolResultSchema)
.then(result => ({ caller, result }))
.catch(error => {
console.error(`Error in tool call for ${caller}:`, error);
throw error;
});
});
// Wait for all tool calls to complete
const results = await Promise.all(toolPromises);
// Organize results by caller
const resultsByTool: Record<string, CallToolResult> = {};
results.forEach(({ caller, result }) => {
resultsByTool[caller] = result;
});
return resultsByTool;
} catch (error) {
console.error(`Error starting parallel notification tools:`, error);
throw error;
}
}
// Start the client
main().catch((error: unknown) => {
console.error('Error running MCP client:', error);
process.exit(1);
});

View file

@ -1,5 +1,6 @@
import { Client } from '../../client/index.js';
import { StreamableHTTPClientTransport } from '../../client/streamableHttp.js';
import { createInterface } from 'node:readline';
import {
ListToolsRequest,
ListToolsResultSchema,
@ -11,107 +12,491 @@ import {
GetPromptResultSchema,
ListResourcesRequest,
ListResourcesResultSchema,
LoggingMessageNotificationSchema
LoggingMessageNotificationSchema,
ResourceListChangedNotificationSchema,
} from '../../types.js';
// Create readline interface for user input
const readline = createInterface({
input: process.stdin,
output: process.stdout
});
// Track received notifications for debugging resumability
let notificationCount = 0;
// Global client and transport for interactive commands
let client: Client | null = null;
let transport: StreamableHTTPClientTransport | null = null;
let serverUrl = 'http://localhost:3000/mcp';
let notificationsToolLastEventId: string | undefined = undefined;
let sessionId: string | undefined = undefined;
async function main(): Promise<void> {
// Create a new client with streamable HTTP transport
const client = new Client({
name: 'example-client',
version: '1.0.0'
});
console.log('MCP Interactive Client');
console.log('=====================');
const transport = new StreamableHTTPClientTransport(
new URL('http://localhost:3000/mcp')
);
// Connect to server immediately with default settings
await connect();
// Connect the client using the transport and initialize the server
await client.connect(transport);
client.setNotificationHandler(LoggingMessageNotificationSchema, (notification) => {
console.log(`Notification received: ${notification.params.level} - ${notification.params.data}`);
});
// Print help and start the command loop
printHelp();
commandLoop();
}
function printHelp(): void {
console.log('\nAvailable commands:');
console.log(' connect [url] - Connect to MCP server (default: http://localhost:3000/mcp)');
console.log(' disconnect - Disconnect from server');
console.log(' terminate-session - Terminate the current session');
console.log(' reconnect - Reconnect to the server');
console.log(' list-tools - List available tools');
console.log(' call-tool <name> [args] - Call a tool with optional JSON arguments');
console.log(' greet [name] - Call the greet tool');
console.log(' multi-greet [name] - Call the multi-greet tool with notifications');
console.log(' start-notifications [interval] [count] - Start periodic notifications');
console.log(' list-prompts - List available prompts');
console.log(' get-prompt [name] [args] - Get a prompt with optional JSON arguments');
console.log(' list-resources - List available resources');
console.log(' help - Show this help');
console.log(' quit - Exit the program');
}
console.log('Connected to MCP server');
// List available tools
const toolsRequest: ListToolsRequest = {
method: 'tools/list',
params: {}
};
const toolsResult = await client.request(toolsRequest, ListToolsResultSchema);
console.log('Available tools:', toolsResult.tools);
function commandLoop(): void {
readline.question('\n> ', async (input) => {
const args = input.trim().split(/\s+/);
const command = args[0]?.toLowerCase();
// Call the 'greet' tool
const greetRequest: CallToolRequest = {
method: 'tools/call',
params: {
name: 'greet',
arguments: { name: 'MCP User' }
try {
switch (command) {
case 'connect':
await connect(args[1]);
break;
case 'disconnect':
await disconnect();
break;
case 'terminate-session':
await terminateSession();
break;
case 'reconnect':
await reconnect();
break;
case 'list-tools':
await listTools();
break;
case 'call-tool':
if (args.length < 2) {
console.log('Usage: call-tool <name> [args]');
} else {
const toolName = args[1];
let toolArgs = {};
if (args.length > 2) {
try {
toolArgs = JSON.parse(args.slice(2).join(' '));
} catch {
console.log('Invalid JSON arguments. Using empty args.');
}
}
await callTool(toolName, toolArgs);
}
break;
case 'greet':
await callGreetTool(args[1] || 'MCP User');
break;
case 'multi-greet':
await callMultiGreetTool(args[1] || 'MCP User');
break;
case 'start-notifications': {
const interval = args[1] ? parseInt(args[1], 10) : 2000;
const count = args[2] ? parseInt(args[2], 10) : 10;
await startNotifications(interval, count);
break;
}
case 'list-prompts':
await listPrompts();
break;
case 'get-prompt':
if (args.length < 2) {
console.log('Usage: get-prompt <name> [args]');
} else {
const promptName = args[1];
let promptArgs = {};
if (args.length > 2) {
try {
promptArgs = JSON.parse(args.slice(2).join(' '));
} catch {
console.log('Invalid JSON arguments. Using empty args.');
}
}
await getPrompt(promptName, promptArgs);
}
break;
case 'list-resources':
await listResources();
break;
case 'help':
printHelp();
break;
case 'quit':
case 'exit':
await cleanup();
return;
default:
if (command) {
console.log(`Unknown command: ${command}`);
}
break;
}
} catch (error) {
console.error(`Error executing command: ${error}`);
}
};
const greetResult = await client.request(greetRequest, CallToolResultSchema);
console.log('Greeting result:', greetResult.content[0].text);
// Call the new 'multi-greet' tool
console.log('\nCalling multi-greet tool (with notifications)...');
const multiGreetRequest: CallToolRequest = {
method: 'tools/call',
params: {
name: 'multi-greet',
arguments: { name: 'MCP User' }
}
};
const multiGreetResult = await client.request(multiGreetRequest, CallToolResultSchema);
console.log('Multi-greet results:');
multiGreetResult.content.forEach(item => {
if (item.type === 'text') {
console.log(`- ${item.text}`);
}
// Continue the command loop
commandLoop();
});
}
async function connect(url?: string): Promise<void> {
if (client) {
console.log('Already connected. Disconnect first.');
return;
}
if (url) {
serverUrl = url;
}
console.log(`Connecting to ${serverUrl}...`);
try {
// Create a new client
client = new Client({
name: 'example-client',
version: '1.0.0'
});
client.onerror = (error) => {
console.error('\x1b[31mClient error:', error, '\x1b[0m');
}
transport = new StreamableHTTPClientTransport(
new URL(serverUrl),
{
sessionId: sessionId
}
);
// Set up notification handlers
client.setNotificationHandler(LoggingMessageNotificationSchema, (notification) => {
notificationCount++;
console.log(`\nNotification #${notificationCount}: ${notification.params.level} - ${notification.params.data}`);
// Re-display the prompt
process.stdout.write('> ');
});
client.setNotificationHandler(ResourceListChangedNotificationSchema, async (_) => {
console.log(`\nResource list changed notification received!`);
try {
if (!client) {
console.log('Client disconnected, cannot fetch resources');
return;
}
const resourcesResult = await client.request({
method: 'resources/list',
params: {}
}, ListResourcesResultSchema);
console.log('Available resources count:', resourcesResult.resources.length);
} catch {
console.log('Failed to list resources after change notification');
}
// Re-display the prompt
process.stdout.write('> ');
});
// Connect the client
await client.connect(transport);
sessionId = transport.sessionId
console.log('Transport created with session ID:', sessionId);
console.log('Connected to MCP server');
} catch (error) {
console.error('Failed to connect:', error);
client = null;
transport = null;
}
}
async function disconnect(): Promise<void> {
if (!client || !transport) {
console.log('Not connected.');
return;
}
try {
await transport.close();
console.log('Disconnected from MCP server');
client = null;
transport = null;
} catch (error) {
console.error('Error disconnecting:', error);
}
}
async function terminateSession(): Promise<void> {
if (!client || !transport) {
console.log('Not connected.');
return;
}
try {
console.log('Terminating session with ID:', transport.sessionId);
await transport.terminateSession();
console.log('Session terminated successfully');
// Check if sessionId was cleared after termination
if (!transport.sessionId) {
console.log('Session ID has been cleared');
sessionId = undefined;
// Also close the transport and clear client objects
await transport.close();
console.log('Transport closed after session termination');
client = null;
transport = null;
} else {
console.log('Server responded with 405 Method Not Allowed (session termination not supported)');
console.log('Session ID is still active:', transport.sessionId);
}
} catch (error) {
console.error('Error terminating session:', error);
}
}
async function reconnect(): Promise<void> {
if (client) {
await disconnect();
}
await connect();
}
async function listTools(): Promise<void> {
if (!client) {
console.log('Not connected to server.');
return;
}
try {
const toolsRequest: ListToolsRequest = {
method: 'tools/list',
params: {}
};
const toolsResult = await client.request(toolsRequest, ListToolsResultSchema);
console.log('Available tools:');
if (toolsResult.tools.length === 0) {
console.log(' No tools available');
} else {
for (const tool of toolsResult.tools) {
console.log(` - ${tool.name}: ${tool.description}`);
}
}
} catch (error) {
console.log(`Tools not supported by this server (${error})`);
}
}
async function callTool(name: string, args: Record<string, unknown>): Promise<void> {
if (!client) {
console.log('Not connected to server.');
return;
}
try {
const request: CallToolRequest = {
method: 'tools/call',
params: {
name,
arguments: args
}
};
console.log(`Calling tool '${name}' with args:`, args);
const onLastEventIdUpdate = (event: string) => {
notificationsToolLastEventId = event;
};
const result = await client.request(request, CallToolResultSchema, {
resumptionToken: notificationsToolLastEventId, onresumptiontoken: onLastEventIdUpdate
});
console.log('Tool result:');
result.content.forEach(item => {
if (item.type === 'text') {
console.log(` ${item.text}`);
} else {
console.log(` ${item.type} content:`, item);
}
});
} catch (error) {
console.log(`Error calling tool ${name}: ${error}`);
}
}
async function callGreetTool(name: string): Promise<void> {
await callTool('greet', { name });
}
async function callMultiGreetTool(name: string): Promise<void> {
console.log('Calling multi-greet tool with notifications...');
await callTool('multi-greet', { name });
}
async function startNotifications(interval: number, count: number): Promise<void> {
console.log(`Starting notification stream: interval=${interval}ms, count=${count || 'unlimited'}`);
await callTool('start-notification-stream', { interval, count });
}
async function listPrompts(): Promise<void> {
if (!client) {
console.log('Not connected to server.');
return;
}
// List available prompts
try {
const promptsRequest: ListPromptsRequest = {
method: 'prompts/list',
params: {}
};
const promptsResult = await client.request(promptsRequest, ListPromptsResultSchema);
console.log('Available prompts:', promptsResult.prompts);
console.log('Available prompts:');
if (promptsResult.prompts.length === 0) {
console.log(' No prompts available');
} else {
for (const prompt of promptsResult.prompts) {
console.log(` - ${prompt.name}: ${prompt.description}`);
}
}
} catch (error) {
console.log(`Prompts not supported by this server (${error})`);
}
}
async function getPrompt(name: string, args: Record<string, unknown>): Promise<void> {
if (!client) {
console.log('Not connected to server.');
return;
}
// Get a prompt
try {
const promptRequest: GetPromptRequest = {
method: 'prompts/get',
params: {
name: 'greeting-template',
arguments: { name: 'MCP User' }
name,
arguments: args as Record<string, string>
}
};
const promptResult = await client.request(promptRequest, GetPromptResultSchema);
console.log('Prompt template:', promptResult.messages[0].content.text);
console.log('Prompt template:');
promptResult.messages.forEach((msg, index) => {
console.log(` [${index + 1}] ${msg.role}: ${msg.content.text}`);
});
} catch (error) {
console.log(`Prompt retrieval not supported by this server (${error})`);
console.log(`Error getting prompt ${name}: ${error}`);
}
}
async function listResources(): Promise<void> {
if (!client) {
console.log('Not connected to server.');
return;
}
// List available resources
try {
const resourcesRequest: ListResourcesRequest = {
method: 'resources/list',
params: {}
};
const resourcesResult = await client.request(resourcesRequest, ListResourcesResultSchema);
console.log('Available resources:', resourcesResult.resources);
console.log('Available resources:');
if (resourcesResult.resources.length === 0) {
console.log(' No resources available');
} else {
for (const resource of resourcesResult.resources) {
console.log(` - ${resource.name}: ${resource.uri}`);
}
}
} catch (error) {
console.log(`Resources not supported by this server (${error})`);
}
// Close the connection
await client.close();
}
async function cleanup(): Promise<void> {
if (client && transport) {
try {
// First try to terminate the session gracefully
if (transport.sessionId) {
try {
console.log('Terminating session before exit...');
await transport.terminateSession();
console.log('Session terminated successfully');
} catch (error) {
console.error('Error terminating session:', error);
}
}
// Then close the transport
await transport.close();
} catch (error) {
console.error('Error closing transport:', error);
}
}
process.stdin.setRawMode(false);
readline.close();
console.log('\nGoodbye!');
process.exit(0);
}
// Set up raw mode for keyboard input to capture Escape key
process.stdin.setRawMode(true);
process.stdin.on('data', async (data) => {
// Check for Escape key (27)
if (data.length === 1 && data[0] === 27) {
console.log('\nESC key pressed. Disconnecting from server...');
// Abort current operation and disconnect from server
if (client && transport) {
await disconnect();
console.log('Disconnected. Press Enter to continue.');
} else {
console.log('Not connected to server.');
}
// Re-display the prompt
process.stdout.write('> ');
}
});
// Handle Ctrl+C
process.on('SIGINT', async () => {
console.log('\nReceived SIGINT. Cleaning up...');
await cleanup();
});
// Start the interactive client
main().catch((error: unknown) => {
console.error('Error running MCP client:', error);
process.exit(1);

View file

@ -0,0 +1,192 @@
import { Client } from '../../client/index.js';
import { StreamableHTTPClientTransport } from '../../client/streamableHttp.js';
import { SSEClientTransport } from '../../client/sse.js';
import {
ListToolsRequest,
ListToolsResultSchema,
CallToolRequest,
CallToolResultSchema,
LoggingMessageNotificationSchema,
} from '../../types.js';
/**
* Simplified Backwards Compatible MCP Client
*
* This client demonstrates backward compatibility with both:
* 1. Modern servers using Streamable HTTP transport (protocol version 2025-03-26)
* 2. Older servers using HTTP+SSE transport (protocol version 2024-11-05)
*
* Following the MCP specification for backwards compatibility:
* - Attempts to POST an initialize request to the server URL first (modern transport)
* - If that fails with 4xx status, falls back to GET request for SSE stream (older transport)
*/
// Command line args processing
const args = process.argv.slice(2);
const serverUrl = args[0] || 'http://localhost:3000/mcp';
async function main(): Promise<void> {
console.log('MCP Backwards Compatible Client');
console.log('===============================');
console.log(`Connecting to server at: ${serverUrl}`);
let client: Client;
let transport: StreamableHTTPClientTransport | SSEClientTransport;
try {
// Try connecting with automatic transport detection
const connection = await connectWithBackwardsCompatibility(serverUrl);
client = connection.client;
transport = connection.transport;
// Set up notification handler
client.setNotificationHandler(LoggingMessageNotificationSchema, (notification) => {
console.log(`Notification: ${notification.params.level} - ${notification.params.data}`);
});
// DEMO WORKFLOW:
// 1. List available tools
console.log('\n=== Listing Available Tools ===');
await listTools(client);
// 2. Call the notification tool
console.log('\n=== Starting Notification Stream ===');
await startNotificationTool(client);
// 3. Wait for all notifications (5 seconds)
console.log('\n=== Waiting for all notifications ===');
await new Promise(resolve => setTimeout(resolve, 5000));
// 4. Disconnect
console.log('\n=== Disconnecting ===');
await transport.close();
console.log('Disconnected from MCP server');
} catch (error) {
console.error('Error running client:', error);
process.exit(1);
}
}
/**
* Connect to an MCP server with backwards compatibility
* Following the spec for client backward compatibility
*/
async function connectWithBackwardsCompatibility(url: string): Promise<{
client: Client,
transport: StreamableHTTPClientTransport | SSEClientTransport,
transportType: 'streamable-http' | 'sse'
}> {
console.log('1. Trying Streamable HTTP transport first...');
// Step 1: Try Streamable HTTP transport first
const client = new Client({
name: 'backwards-compatible-client',
version: '1.0.0'
});
client.onerror = (error) => {
console.error('Client error:', error);
};
const baseUrl = new URL(url);
try {
// Create modern transport
const streamableTransport = new StreamableHTTPClientTransport(baseUrl);
await client.connect(streamableTransport);
console.log('Successfully connected using modern Streamable HTTP transport.');
return {
client,
transport: streamableTransport,
transportType: 'streamable-http'
};
} catch (error) {
// Step 2: If transport fails, try the older SSE transport
console.log(`StreamableHttp transport connection failed: ${error}`);
console.log('2. Falling back to deprecated HTTP+SSE transport...');
try {
// Create SSE transport pointing to /sse endpoint
const sseTransport = new SSEClientTransport(baseUrl);
const sseClient = new Client({
name: 'backwards-compatible-client',
version: '1.0.0'
});
await sseClient.connect(sseTransport);
console.log('Successfully connected using deprecated HTTP+SSE transport.');
return {
client: sseClient,
transport: sseTransport,
transportType: 'sse'
};
} catch (sseError) {
console.error(`Failed to connect with either transport method:\n1. Streamable HTTP error: ${error}\n2. SSE error: ${sseError}`);
throw new Error('Could not connect to server with any available transport');
}
}
}
/**
* List available tools on the server
*/
async function listTools(client: Client): Promise<void> {
try {
const toolsRequest: ListToolsRequest = {
method: 'tools/list',
params: {}
};
const toolsResult = await client.request(toolsRequest, ListToolsResultSchema);
console.log('Available tools:');
if (toolsResult.tools.length === 0) {
console.log(' No tools available');
} else {
for (const tool of toolsResult.tools) {
console.log(` - ${tool.name}: ${tool.description}`);
}
}
} catch (error) {
console.log(`Tools not supported by this server: ${error}`);
}
}
/**
* Start a notification stream by calling the notification tool
*/
async function startNotificationTool(client: Client): Promise<void> {
try {
// Call the notification tool using reasonable defaults
const request: CallToolRequest = {
method: 'tools/call',
params: {
name: 'start-notification-stream',
arguments: {
interval: 1000, // 1 second between notifications
count: 5 // Send 5 notifications
}
}
};
console.log('Calling notification tool...');
const result = await client.request(request, CallToolResultSchema);
console.log('Tool result:');
result.content.forEach(item => {
if (item.type === 'text') {
console.log(` ${item.text}`);
} else {
console.log(` ${item.type} content:`, item);
}
});
} catch (error) {
console.log(`Error calling notification tool: ${error}`);
}
}
// Start the client
main().catch((error: unknown) => {
console.error('Error running MCP client:', error);
process.exit(1);
});

View file

@ -3,76 +3,80 @@ 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';
import { CallToolResult, isInitializeRequest } from '../../types.js';
// Create an MCP server with implementation details
const server = new McpServer({
name: 'json-response-streamable-http-server',
version: '1.0.0',
}, {
capabilities: {
logging: {},
}
});
const getServer = () => {
const server = new McpServer({
name: 'json-response-streamable-http-server',
version: '1.0.0',
}, {
capabilities: {
logging: {},
}
});
// 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<CallToolResult> => {
return {
content: [
{
type: 'text',
text: `Hello, ${name}!`,
},
],
};
}
);
// 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<CallToolResult> => {
return {
content: [
{
type: 'text',
text: `Hello, ${name}!`,
},
],
};
}
);
// Register a tool that sends multiple greetings with notifications
server.tool(
'multi-greet',
'A tool that sends different greetings with delays between them',
{
name: z.string().describe('Name to greet'),
},
async ({ name }, { sendNotification }): Promise<CallToolResult> => {
const sleep = (ms: number) => new Promise(resolve => setTimeout(resolve, ms));
// Register a tool that sends multiple greetings with notifications
server.tool(
'multi-greet',
'A tool that sends different greetings with delays between them',
{
name: z.string().describe('Name to greet'),
},
async ({ name }, { sendNotification }): Promise<CallToolResult> => {
const sleep = (ms: number) => new Promise(resolve => setTimeout(resolve, ms));
await sendNotification({
method: "notifications/message",
params: { level: "debug", data: `Starting multi-greet for ${name}` }
});
await sendNotification({
method: "notifications/message",
params: { level: "debug", data: `Starting multi-greet for ${name}` }
});
await sleep(1000); // Wait 1 second before first greeting
await sleep(1000); // Wait 1 second before first greeting
await sendNotification({
method: "notifications/message",
params: { level: "info", data: `Sending first greeting to ${name}` }
});
await sendNotification({
method: "notifications/message",
params: { level: "info", data: `Sending first greeting to ${name}` }
});
await sleep(1000); // Wait another second before second greeting
await sleep(1000); // Wait another second before second greeting
await sendNotification({
method: "notifications/message",
params: { level: "info", data: `Sending second greeting to ${name}` }
});
await sendNotification({
method: "notifications/message",
params: { level: "info", data: `Sending second greeting to ${name}` }
});
return {
content: [
{
type: 'text',
text: `Good morning, ${name}!`,
}
],
};
}
);
return {
content: [
{
type: 'text',
text: `Good morning, ${name}!`,
}
],
};
}
);
return server;
}
const app = express();
app.use(express.json());
@ -95,18 +99,18 @@ app.post('/mcp', async (req: Request, res: Response) => {
transport = new StreamableHTTPServerTransport({
sessionIdGenerator: () => randomUUID(),
enableJsonResponse: true, // Enable JSON response mode
onsessioninitialized: (sessionId) => {
// Store the transport by session ID when session is initialized
// This avoids race conditions where requests might come in before the session is stored
console.log(`Session initialized with ID: ${sessionId}`);
transports[sessionId] = transport;
}
});
// Connect the transport to the MCP server BEFORE handling the request
const server = getServer();
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
@ -138,45 +142,21 @@ app.post('/mcp', async (req: Request, res: Response) => {
}
});
// 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';
}
// Handle GET requests for SSE streams according to spec
app.get('/mcp', async (req: Request, res: Response) => {
// Since this is a very simple example, we don't support GET requests for this server
// The spec requires returning 405 Method Not Allowed in this case
res.status(405).set('Allow', 'POST').send('Method Not Allowed');
});
// Start the server
const PORT = 3000;
app.listen(PORT, () => {
console.log(`MCP Streamable HTTP Server listening on port ${PORT}`);
console.log(`Initialize session with the command below id you are using curl for testing:
-----------------------------
SESSION_ID=$(curl -X POST \
-H "Content-Type: application/json" \
-H "Accept: application/json" \
-H "Accept: text/event-stream" \
-d '{
"jsonrpc": "2.0",
"method": "initialize",
"params": {
"capabilities": {},
"protocolVersion": "2025-03-26",
"clientInfo": {
"name": "test",
"version": "1.0.0"
}
},
"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"
-----------------------------`);
});
// Handle server shutdown
process.on('SIGINT', async () => {
console.log('Shutting down server...');
await server.close();
process.exit(0);
});

View file

@ -0,0 +1,168 @@
import express, { Request, Response } from 'express';
import { McpServer } from '../../server/mcp.js';
import { SSEServerTransport } from '../../server/sse.js';
import { z } from 'zod';
import { CallToolResult } from '../../types.js';
/**
* This example server demonstrates the deprecated HTTP+SSE transport
* (protocol version 2024-11-05). It mainly used for testing backward compatible clients.
*
* The server exposes two endpoints:
* - /mcp: For establishing the SSE stream (GET)
* - /messages: For receiving client messages (POST)
*
*/
// Create an MCP server instance
const getServer = () => {
const server = new McpServer({
name: 'simple-sse-server',
version: '1.0.0',
}, { capabilities: { logging: {} } });
server.tool(
'start-notification-stream',
'Starts sending periodic notifications',
{
interval: z.number().describe('Interval in milliseconds between notifications').default(1000),
count: z.number().describe('Number of notifications to send').default(10),
},
async ({ interval, count }, { sendNotification }): Promise<CallToolResult> => {
const sleep = (ms: number) => new Promise(resolve => setTimeout(resolve, ms));
let counter = 0;
// Send the initial notification
await sendNotification({
method: "notifications/message",
params: {
level: "info",
data: `Starting notification stream with ${count} messages every ${interval}ms`
}
});
// Send periodic notifications
while (counter < count) {
counter++;
await sleep(interval);
try {
await sendNotification({
method: "notifications/message",
params: {
level: "info",
data: `Notification #${counter} at ${new Date().toISOString()}`
}
});
}
catch (error) {
console.error("Error sending notification:", error);
}
}
return {
content: [
{
type: 'text',
text: `Completed sending ${count} notifications every ${interval}ms`,
}
],
};
}
);
return server;
};
const app = express();
app.use(express.json());
// Store transports by session ID
const transports: Record<string, SSEServerTransport> = {};
// SSE endpoint for establishing the stream
app.get('/mcp', async (req: Request, res: Response) => {
console.log('Received GET request to /sse (establishing SSE stream)');
try {
// Create a new SSE transport for the client
// The endpoint for POST messages is '/messages'
const transport = new SSEServerTransport('/messages', res);
// Store the transport by session ID
const sessionId = transport.sessionId;
transports[sessionId] = transport;
// Set up onclose handler to clean up transport when closed
transport.onclose = () => {
console.log(`SSE transport closed for session ${sessionId}`);
delete transports[sessionId];
};
// Connect the transport to the MCP server
const server = getServer();
await server.connect(transport);
console.log(`Established SSE stream with session ID: ${sessionId}`);
} catch (error) {
console.error('Error establishing SSE stream:', error);
if (!res.headersSent) {
res.status(500).send('Error establishing SSE stream');
}
}
});
// Messages endpoint for receiving client JSON-RPC requests
app.post('/messages', async (req: Request, res: Response) => {
console.log('Received POST request to /messages');
// Extract session ID from URL query parameter
// In the SSE protocol, this is added by the client based on the endpoint event
const sessionId = req.query.sessionId as string | undefined;
if (!sessionId) {
console.error('No session ID provided in request URL');
res.status(400).send('Missing sessionId parameter');
return;
}
const transport = transports[sessionId];
if (!transport) {
console.error(`No active transport found for session ID: ${sessionId}`);
res.status(404).send('Session not found');
return;
}
try {
// Handle the POST message with the transport
await transport.handlePostMessage(req, res, req.body);
} catch (error) {
console.error('Error handling request:', error);
if (!res.headersSent) {
res.status(500).send('Error handling request');
}
}
});
// Start the server
const PORT = 3000;
app.listen(PORT, () => {
console.log(`Simple SSE Server (deprecated protocol version 2024-11-05) listening on port ${PORT}`);
});
// Handle server shutdown
process.on('SIGINT', async () => {
console.log('Shutting down server...');
// Close all active transports to properly clean up resources
for (const sessionId in transports) {
try {
console.log(`Closing transport for session ${sessionId}`);
await transports[sessionId].close();
delete transports[sessionId];
} catch (error) {
console.error(`Error closing transport for session ${sessionId}:`, error);
}
}
console.log('Server shutdown complete');
process.exit(0);
});

View file

@ -0,0 +1,162 @@
import express, { Request, Response } from 'express';
import { McpServer } from '../../server/mcp.js';
import { StreamableHTTPServerTransport } from '../../server/streamableHttp.js';
import { z } from 'zod';
import { CallToolResult, GetPromptResult, ReadResourceResult } from '../../types.js';
const getServer = () => {
// Create an MCP server with implementation details
const server = new McpServer({
name: 'stateless-streamable-http-server',
version: '1.0.0',
}, { capabilities: { logging: {} } });
// Register a simple prompt
server.prompt(
'greeting-template',
'A simple greeting prompt template',
{
name: z.string().describe('Name to include in greeting'),
},
async ({ name }): Promise<GetPromptResult> => {
return {
messages: [
{
role: 'user',
content: {
type: 'text',
text: `Please greet ${name} in a friendly manner.`,
},
},
],
};
}
);
// Register a tool specifically for testing resumability
server.tool(
'start-notification-stream',
'Starts sending periodic notifications for testing resumability',
{
interval: z.number().describe('Interval in milliseconds between notifications').default(100),
count: z.number().describe('Number of notifications to send (0 for 100)').default(10),
},
async ({ interval, count }, { sendNotification }): Promise<CallToolResult> => {
const sleep = (ms: number) => new Promise(resolve => setTimeout(resolve, ms));
let counter = 0;
while (count === 0 || counter < count) {
counter++;
try {
await sendNotification({
method: "notifications/message",
params: {
level: "info",
data: `Periodic notification #${counter} at ${new Date().toISOString()}`
}
});
}
catch (error) {
console.error("Error sending notification:", error);
}
// Wait for the specified interval
await sleep(interval);
}
return {
content: [
{
type: 'text',
text: `Started sending periodic notifications every ${interval}ms`,
}
],
};
}
);
// Create a simple resource at a fixed URI
server.resource(
'greeting-resource',
'https://example.com/greetings/default',
{ mimeType: 'text/plain' },
async (): Promise<ReadResourceResult> => {
return {
contents: [
{
uri: 'https://example.com/greetings/default',
text: 'Hello, world!',
},
],
};
}
);
return server;
}
const app = express();
app.use(express.json());
app.post('/mcp', async (req: Request, res: Response) => {
const server = getServer();
try {
const transport: StreamableHTTPServerTransport = new StreamableHTTPServerTransport({
sessionIdGenerator: undefined,
});
await server.connect(transport);
await transport.handleRequest(req, res, req.body);
res.on('close', () => {
console.log('Request closed');
transport.close();
server.close();
});
} 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,
});
}
}
});
app.get('/mcp', async (req: Request, res: Response) => {
console.log('Received GET MCP request');
res.writeHead(405).end(JSON.stringify({
jsonrpc: "2.0",
error: {
code: -32000,
message: "Method not allowed."
},
id: null
}));
});
app.delete('/mcp', async (req: Request, res: Response) => {
console.log('Received DELETE MCP request');
res.writeHead(405).end(JSON.stringify({
jsonrpc: "2.0",
error: {
code: -32000,
message: "Method not allowed."
},
id: null
}));
});
// Start the server
const PORT = 3000;
app.listen(PORT, () => {
console.log(`MCP Stateless Streamable HTTP Server listening on port ${PORT}`);
});
// Handle server shutdown
process.on('SIGINT', async () => {
console.log('Shutting down server...');
process.exit(0);
});

View file

@ -1,113 +1,163 @@
import express, { Request, Response } from 'express';
import { randomUUID } from 'node:crypto';
import { z } from 'zod';
import { McpServer } from '../../server/mcp.js';
import { StreamableHTTPServerTransport } from '../../server/streamableHttp.js';
import { z } from 'zod';
import { CallToolResult, GetPromptResult, ReadResourceResult } from '../../types.js';
import { CallToolResult, GetPromptResult, isInitializeRequest, ReadResourceResult } from '../../types.js';
import { InMemoryEventStore } from '../shared/inMemoryEventStore.js';
// Create an MCP server with implementation details
const server = new McpServer({
name: 'simple-streamable-http-server',
version: '1.0.0',
}, { capabilities: { logging: {} } });
const getServer = () => {
const server = new McpServer({
name: 'simple-streamable-http-server',
version: '1.0.0',
}, { capabilities: { logging: {} } });
// 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<CallToolResult> => {
return {
content: [
{
type: 'text',
text: `Hello, ${name}!`,
},
],
};
}
);
// Register a tool that sends multiple greetings with notifications
server.tool(
'multi-greet',
'A tool that sends different greetings with delays between them',
{
name: z.string().describe('Name to greet'),
},
async ({ name }, { sendNotification }): Promise<CallToolResult> => {
const sleep = (ms: number) => new Promise(resolve => setTimeout(resolve, ms));
await sendNotification({
method: "notifications/message",
params: { level: "debug", data: `Starting multi-greet for ${name}` }
});
await sleep(1000); // Wait 1 second before first greeting
await sendNotification({
method: "notifications/message",
params: { level: "info", data: `Sending first greeting to ${name}` }
});
await sleep(1000); // Wait another second before second greeting
await sendNotification({
method: "notifications/message",
params: { level: "info", data: `Sending second greeting to ${name}` }
});
return {
content: [
{
type: 'text',
text: `Good morning, ${name}!`,
}
],
};
}
);
// Register a simple prompt
server.prompt(
'greeting-template',
'A simple greeting prompt template',
{
name: z.string().describe('Name to include in greeting'),
},
async ({ name }): Promise<GetPromptResult> => {
return {
messages: [
{
role: 'user',
content: {
// 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<CallToolResult> => {
return {
content: [
{
type: 'text',
text: `Please greet ${name} in a friendly manner.`,
text: `Hello, ${name}!`,
},
},
],
};
}
);
],
};
}
);
// Create a simple resource at a fixed URI
server.resource(
'greeting-resource',
'https://example.com/greetings/default',
{ mimeType: 'text/plain' },
async (): Promise<ReadResourceResult> => {
return {
contents: [
{
uri: 'https://example.com/greetings/default',
text: 'Hello, world!',
},
],
};
}
);
// Register a tool that sends multiple greetings with notifications (with annotations)
server.tool(
'multi-greet',
'A tool that sends different greetings with delays between them',
{
name: z.string().describe('Name to greet'),
},
{
title: 'Multiple Greeting Tool',
readOnlyHint: true,
openWorldHint: false
},
async ({ name }, { sendNotification }): Promise<CallToolResult> => {
const sleep = (ms: number) => new Promise(resolve => setTimeout(resolve, ms));
await sendNotification({
method: "notifications/message",
params: { level: "debug", data: `Starting multi-greet for ${name}` }
});
await sleep(1000); // Wait 1 second before first greeting
await sendNotification({
method: "notifications/message",
params: { level: "info", data: `Sending first greeting to ${name}` }
});
await sleep(1000); // Wait another second before second greeting
await sendNotification({
method: "notifications/message",
params: { level: "info", data: `Sending second greeting to ${name}` }
});
return {
content: [
{
type: 'text',
text: `Good morning, ${name}!`,
}
],
};
}
);
// Register a simple prompt
server.prompt(
'greeting-template',
'A simple greeting prompt template',
{
name: z.string().describe('Name to include in greeting'),
},
async ({ name }): Promise<GetPromptResult> => {
return {
messages: [
{
role: 'user',
content: {
type: 'text',
text: `Please greet ${name} in a friendly manner.`,
},
},
],
};
}
);
// Register a tool specifically for testing resumability
server.tool(
'start-notification-stream',
'Starts sending periodic notifications for testing resumability',
{
interval: z.number().describe('Interval in milliseconds between notifications').default(100),
count: z.number().describe('Number of notifications to send (0 for 100)').default(50),
},
async ({ interval, count }, { sendNotification }): Promise<CallToolResult> => {
const sleep = (ms: number) => new Promise(resolve => setTimeout(resolve, ms));
let counter = 0;
while (count === 0 || counter < count) {
counter++;
try {
await sendNotification({
method: "notifications/message",
params: {
level: "info",
data: `Periodic notification #${counter} at ${new Date().toISOString()}`
}
});
}
catch (error) {
console.error("Error sending notification:", error);
}
// Wait for the specified interval
await sleep(interval);
}
return {
content: [
{
type: 'text',
text: `Started sending periodic notifications every ${interval}ms`,
}
],
};
}
);
// Create a simple resource at a fixed URI
server.resource(
'greeting-resource',
'https://example.com/greetings/default',
{ mimeType: 'text/plain' },
async (): Promise<ReadResourceResult> => {
return {
contents: [
{
uri: 'https://example.com/greetings/default',
text: 'Hello, world!',
},
],
};
}
);
return server;
};
const app = express();
app.use(express.json());
@ -127,21 +177,33 @@ app.post('/mcp', async (req: Request, res: Response) => {
transport = transports[sessionId];
} else if (!sessionId && isInitializeRequest(req.body)) {
// New initialization request
const eventStore = new InMemoryEventStore();
transport = new StreamableHTTPServerTransport({
sessionIdGenerator: () => randomUUID(),
eventStore, // Enable resumability
onsessioninitialized: (sessionId) => {
// Store the transport by session ID when session is initialized
// This avoids race conditions where requests might come in before the session is stored
console.log(`Session initialized with ID: ${sessionId}`);
transports[sessionId] = transport;
}
});
// Set up onclose handler to clean up transport when closed
transport.onclose = () => {
const sid = transport.sessionId;
if (sid && transports[sid]) {
console.log(`Transport closed for session ${sid}, removing from transports map`);
delete transports[sid];
}
};
// Connect the transport to the MCP server BEFORE handling the request
// so responses can flow back through the same transport
const server = getServer();
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
@ -174,45 +236,67 @@ app.post('/mcp', async (req: Request, res: Response) => {
}
});
// 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');
// Handle GET requests for SSE streams (using built-in support from StreamableHTTP)
app.get('/mcp', async (req: Request, res: Response) => {
const sessionId = req.headers['mcp-session-id'] as string | undefined;
if (!sessionId || !transports[sessionId]) {
res.status(400).send('Invalid or missing session ID');
return;
}
return typeof body === 'object' && body !== null && 'method' in body && body.method === 'initialize';
}
// Check for Last-Event-ID header for resumability
const lastEventId = req.headers['last-event-id'] as string | undefined;
if (lastEventId) {
console.log(`Client reconnecting with Last-Event-ID: ${lastEventId}`);
} else {
console.log(`Establishing new SSE stream for session ${sessionId}`);
}
const transport = transports[sessionId];
await transport.handleRequest(req, res);
});
// Handle DELETE requests for session termination (according to MCP spec)
app.delete('/mcp', async (req: Request, res: Response) => {
const sessionId = req.headers['mcp-session-id'] as string | undefined;
if (!sessionId || !transports[sessionId]) {
res.status(400).send('Invalid or missing session ID');
return;
}
console.log(`Received session termination request for session ${sessionId}`);
try {
const transport = transports[sessionId];
await transport.handleRequest(req, res);
} catch (error) {
console.error('Error handling session termination:', error);
if (!res.headersSent) {
res.status(500).send('Error processing session termination');
}
}
});
// Start the server
const PORT = 3000;
app.listen(PORT, () => {
console.log(`MCP Streamable HTTP Server listening on port ${PORT}`);
console.log(`Initialize session with the command below id you are using curl for testing:
-----------------------------
SESSION_ID=$(curl -X POST \
-H "Content-Type: application/json" \
-H "Accept: application/json" \
-H "Accept: text/event-stream" \
-d '{
"jsonrpc": "2.0",
"method": "initialize",
"params": {
"capabilities": {},
"protocolVersion": "2025-03-26",
"clientInfo": {
"name": "test",
"version": "1.0.0"
}
},
"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"
-----------------------------`);
});
// Handle server shutdown
process.on('SIGINT', async () => {
console.log('Shutting down server...');
await server.close();
// Close all active transports to properly clean up resources
for (const sessionId in transports) {
try {
console.log(`Closing transport for session ${sessionId}`);
await transports[sessionId].close();
delete transports[sessionId];
} catch (error) {
console.error(`Error closing transport for session ${sessionId}:`, error);
}
}
console.log('Server shutdown complete');
process.exit(0);
});
});

View file

@ -0,0 +1,246 @@
import express, { Request, Response } from 'express';
import { randomUUID } from "node:crypto";
import { McpServer } from '../../server/mcp.js';
import { StreamableHTTPServerTransport } from '../../server/streamableHttp.js';
import { SSEServerTransport } from '../../server/sse.js';
import { z } from 'zod';
import { CallToolResult, isInitializeRequest } from '../../types.js';
import { InMemoryEventStore } from '../shared/inMemoryEventStore.js';
/**
* This example server demonstrates backwards compatibility with both:
* 1. The deprecated HTTP+SSE transport (protocol version 2024-11-05)
* 2. The Streamable HTTP transport (protocol version 2025-03-26)
*
* It maintains a single MCP server instance but exposes two transport options:
* - /mcp: The new Streamable HTTP endpoint (supports GET/POST/DELETE)
* - /sse: The deprecated SSE endpoint for older clients (GET to establish stream)
* - /messages: The deprecated POST endpoint for older clients (POST to send messages)
*/
const getServer = () => {
const server = new McpServer({
name: 'backwards-compatible-server',
version: '1.0.0',
}, { capabilities: { logging: {} } });
// Register a simple tool that sends notifications over time
server.tool(
'start-notification-stream',
'Starts sending periodic notifications for testing resumability',
{
interval: z.number().describe('Interval in milliseconds between notifications').default(100),
count: z.number().describe('Number of notifications to send (0 for 100)').default(50),
},
async ({ interval, count }, { sendNotification }): Promise<CallToolResult> => {
const sleep = (ms: number) => new Promise(resolve => setTimeout(resolve, ms));
let counter = 0;
while (count === 0 || counter < count) {
counter++;
try {
await sendNotification({
method: "notifications/message",
params: {
level: "info",
data: `Periodic notification #${counter} at ${new Date().toISOString()}`
}
});
}
catch (error) {
console.error("Error sending notification:", error);
}
// Wait for the specified interval
await sleep(interval);
}
return {
content: [
{
type: 'text',
text: `Started sending periodic notifications every ${interval}ms`,
}
],
};
}
);
return server;
};
// Create Express application
const app = express();
app.use(express.json());
// Store transports by session ID
const transports: Record<string, StreamableHTTPServerTransport | SSEServerTransport> = {};
//=============================================================================
// STREAMABLE HTTP TRANSPORT (PROTOCOL VERSION 2025-03-26)
//=============================================================================
// Handle all MCP Streamable HTTP requests (GET, POST, DELETE) on a single endpoint
app.all('/mcp', async (req: Request, res: Response) => {
console.log(`Received ${req.method} request to /mcp`);
try {
// Check for existing session ID
const sessionId = req.headers['mcp-session-id'] as string | undefined;
let transport: StreamableHTTPServerTransport;
if (sessionId && transports[sessionId]) {
// Check if the transport is of the correct type
const existingTransport = transports[sessionId];
if (existingTransport instanceof StreamableHTTPServerTransport) {
// Reuse existing transport
transport = existingTransport;
} else {
// Transport exists but is not a StreamableHTTPServerTransport (could be SSEServerTransport)
res.status(400).json({
jsonrpc: '2.0',
error: {
code: -32000,
message: 'Bad Request: Session exists but uses a different transport protocol',
},
id: null,
});
return;
}
} else if (!sessionId && req.method === 'POST' && isInitializeRequest(req.body)) {
const eventStore = new InMemoryEventStore();
transport = new StreamableHTTPServerTransport({
sessionIdGenerator: () => randomUUID(),
eventStore, // Enable resumability
onsessioninitialized: (sessionId) => {
// Store the transport by session ID when session is initialized
console.log(`StreamableHTTP session initialized with ID: ${sessionId}`);
transports[sessionId] = transport;
}
});
// Set up onclose handler to clean up transport when closed
transport.onclose = () => {
const sid = transport.sessionId;
if (sid && transports[sid]) {
console.log(`Transport closed for session ${sid}, removing from transports map`);
delete transports[sid];
}
};
// Connect the transport to the MCP server
const server = getServer();
await server.connect(transport);
} 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 the transport
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,
});
}
}
});
//=============================================================================
// DEPRECATED HTTP+SSE TRANSPORT (PROTOCOL VERSION 2024-11-05)
//=============================================================================
app.get('/sse', async (req: Request, res: Response) => {
console.log('Received GET request to /sse (deprecated SSE transport)');
const transport = new SSEServerTransport('/messages', res);
transports[transport.sessionId] = transport;
res.on("close", () => {
delete transports[transport.sessionId];
});
const server = getServer();
await server.connect(transport);
});
app.post("/messages", async (req: Request, res: Response) => {
const sessionId = req.query.sessionId as string;
let transport: SSEServerTransport;
const existingTransport = transports[sessionId];
if (existingTransport instanceof SSEServerTransport) {
// Reuse existing transport
transport = existingTransport;
} else {
// Transport exists but is not a SSEServerTransport (could be StreamableHTTPServerTransport)
res.status(400).json({
jsonrpc: '2.0',
error: {
code: -32000,
message: 'Bad Request: Session exists but uses a different transport protocol',
},
id: null,
});
return;
}
if (transport) {
await transport.handlePostMessage(req, res, req.body);
} else {
res.status(400).send('No transport found for sessionId');
}
});
// Start the server
const PORT = 3000;
app.listen(PORT, () => {
console.log(`Backwards compatible MCP server listening on port ${PORT}`);
console.log(`
==============================================
SUPPORTED TRANSPORT OPTIONS:
1. Streamable Http(Protocol version: 2025-03-26)
Endpoint: /mcp
Methods: GET, POST, DELETE
Usage:
- Initialize with POST to /mcp
- Establish SSE stream with GET to /mcp
- Send requests with POST to /mcp
- Terminate session with DELETE to /mcp
2. Http + SSE (Protocol version: 2024-11-05)
Endpoints: /sse (GET) and /messages (POST)
Usage:
- Establish SSE stream with GET to /sse
- Send requests with POST to /messages?sessionId=<id>
==============================================
`);
});
// Handle server shutdown
process.on('SIGINT', async () => {
console.log('Shutting down server...');
// Close all active transports to properly clean up resources
for (const sessionId in transports) {
try {
console.log(`Closing transport for session ${sessionId}`);
await transports[sessionId].close();
delete transports[sessionId];
} catch (error) {
console.error(`Error closing transport for session ${sessionId}:`, error);
}
}
console.log('Server shutdown complete');
process.exit(0);
});

View file

@ -0,0 +1,125 @@
import express, { Request, Response } from 'express';
import { randomUUID } from 'node:crypto';
import { McpServer } from '../../server/mcp.js';
import { StreamableHTTPServerTransport } from '../../server/streamableHttp.js';
import { isInitializeRequest, ReadResourceResult } from '../../types.js';
// Create an MCP server with implementation details
const server = new McpServer({
name: 'resource-list-changed-notification-server',
version: '1.0.0',
});
// Store transports by session ID to send notifications
const transports: { [sessionId: string]: StreamableHTTPServerTransport } = {};
const addResource = (name: string, content: string) => {
const uri = `https://mcp-example.com/dynamic/${encodeURIComponent(name)}`;
server.resource(
name,
uri,
{ mimeType: 'text/plain', description: `Dynamic resource: ${name}` },
async (): Promise<ReadResourceResult> => {
return {
contents: [{ uri, text: content }],
};
}
);
};
addResource('example-resource', 'Initial content for example-resource');
const resourceChangeInterval = setInterval(() => {
const name = randomUUID();
addResource(name, `Content for ${name}`);
}, 5000); // Change resources every 5 seconds for testing
const app = express();
app.use(express.json());
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
transport = new StreamableHTTPServerTransport({
sessionIdGenerator: () => randomUUID(),
onsessioninitialized: (sessionId) => {
// Store the transport by session ID when session is initialized
// This avoids race conditions where requests might come in before the session is stored
console.log(`Session initialized with ID: ${sessionId}`);
transports[sessionId] = transport;
}
});
// Connect the transport to the MCP server
await server.connect(transport);
// Handle the request - the onsessioninitialized callback will store the transport
await transport.handleRequest(req, res, req.body);
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
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,
});
}
}
});
// Handle GET requests for SSE streams (now using built-in support from StreamableHTTP)
app.get('/mcp', async (req: Request, res: Response) => {
const sessionId = req.headers['mcp-session-id'] as string | undefined;
if (!sessionId || !transports[sessionId]) {
res.status(400).send('Invalid or missing session ID');
return;
}
console.log(`Establishing SSE stream for session ${sessionId}`);
const transport = transports[sessionId];
await transport.handleRequest(req, res);
});
// Start the server
const PORT = 3000;
app.listen(PORT, () => {
console.log(`Server listening on port ${PORT}`);
});
// Handle server shutdown
process.on('SIGINT', async () => {
console.log('Shutting down server...');
clearInterval(resourceChangeInterval);
await server.close();
process.exit(0);
});

View file

@ -0,0 +1,77 @@
import { JSONRPCMessage } from '../../types.js';
import { EventStore } from '../../server/streamableHttp.js';
/**
* Simple in-memory implementation of the EventStore interface for resumability
* This is primarily intended for examples and testing, not for production use
* where a persistent storage solution would be more appropriate.
*/
export class InMemoryEventStore implements EventStore {
private events: Map<string, { streamId: string, message: JSONRPCMessage }> = new Map();
/**
* Generates a unique event ID for a given stream ID
*/
private generateEventId(streamId: string): string {
return `${streamId}_${Date.now()}_${Math.random().toString(36).substring(2, 10)}`;
}
/**
* Extracts the stream ID from an event ID
*/
private getStreamIdFromEventId(eventId: string): string {
const parts = eventId.split('_');
return parts.length > 0 ? parts[0] : '';
}
/**
* Stores an event with a generated event ID
* Implements EventStore.storeEvent
*/
async storeEvent(streamId: string, message: JSONRPCMessage): Promise<string> {
const eventId = this.generateEventId(streamId);
this.events.set(eventId, { streamId, message });
return eventId;
}
/**
* Replays events that occurred after a specific event ID
* Implements EventStore.replayEventsAfter
*/
async replayEventsAfter(lastEventId: string,
{ send }: { send: (eventId: string, message: JSONRPCMessage) => Promise<void> }
): Promise<string> {
if (!lastEventId || !this.events.has(lastEventId)) {
return '';
}
// Extract the stream ID from the event ID
const streamId = this.getStreamIdFromEventId(lastEventId);
if (!streamId) {
return '';
}
let foundLastEvent = false;
// Sort events by eventId for chronological ordering
const sortedEvents = [...this.events.entries()].sort((a, b) => a[0].localeCompare(b[0]));
for (const [eventId, { streamId: eventStreamId, message }] of sortedEvents) {
// Only include events from the same stream
if (eventStreamId !== streamId) {
continue;
}
// Start sending events after we find the lastEventId
if (eventId === lastEventId) {
foundLastEvent = true;
continue;
}
if (foundLastEvent) {
await send(eventId, message);
}
}
return streamId;
}
}

Some files were not shown because too many files have changed in this diff Show more