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

Merge pull request #304 from modelcontextprotocol/get-sse

Browse source 
StreamableHTTP -  listening for messages from the server trough GET established SSE
This commit is contained in:
ihrpr 2025-04-11 14:33:19 +01:00 • committed by GitHub
commit 56b0427957
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
9 changed files with 517 additions and 144 deletions
Download patch file Download diff file Expand all files Collapse all files

11
src/client/streamableHttp.test.ts
View file

@ -164,8 +164,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["_startOrAuthStandaloneSSE"]()).resolves.not.toThrow("Failed to open SSE stream: Method Not Allowed");
// Check that GET was attempted
expect(global.fetch).toHaveBeenCalledWith(
expect.anything(),
@ -209,7 +208,7 @@ describe("StreamableHTTPClientTransport", () => {
transport.onmessage = messageSpy;
await transport.start();
await transport.openSseStream();
await transport["_startOrAuthStandaloneSSE"]();
// Give time for the SSE event to be processed
await new Promise(resolve => setTimeout(resolve, 50));
@ -295,7 +294,7 @@ describe("StreamableHTTPClientTransport", () => {
});
await transport.start();
await transport.openSseStream();
await transport["_startOrAuthStandaloneSSE"]();
await new Promise(resolve => setTimeout(resolve, 50));
// Now simulate attempting to reconnect
@ -306,7 +305,7 @@ describe("StreamableHTTPClientTransport", () => {
body: null
});
await transport.openSseStream();
await transport["_startOrAuthStandaloneSSE"]();
// Check that Last-Event-ID was included
const calls = (global.fetch as jest.Mock).mock.calls;
@ -366,7 +365,7 @@ describe("StreamableHTTPClientTransport", () => {
await transport.start();
await transport.openSseStream();
await transport["_startOrAuthStandaloneSSE"]();
expect((actualReqInit.headers as Headers).get("x-custom-header")).toBe("CustomValue");
requestInit.headers["X-Custom-Header"] = "SecondCustomValue";

31
src/client/streamableHttp.ts
View file

@ -1,5 +1,5 @@
import { Transport } from "../shared/transport.js";
import { JSONRPCMessage, JSONRPCMessageSchema } from "../types.js";
import { isJSONRPCNotification, JSONRPCMessage, JSONRPCMessageSchema } from "../types.js";
import { auth, AuthResult, OAuthClientProvider, UnauthorizedError } from "./auth.js";
import { EventSourceParserStream } from "eventsource-parser/stream";
@ -126,12 +126,17 @@ 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);
} catch (error) {
@ -244,6 +249,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 (isJSONRPCNotification(message) && message.method === "notifications/initialized") {
// We don't need to handle 405 here anymore as it's handled in _startOrAuthStandaloneSSE
this._startOrAuthStandaloneSSE().catch(err => this.onerror?.(err));
}
return;
}
@ -280,20 +291,4 @@ export class StreamableHTTPClientTransport implements Transport {
throw error;
}
}
/**
* 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.
*/
async openSseStream(): Promise<void> {
if (!this._abortController) {
throw new Error(
"StreamableHTTPClientTransport not started! Call connect() before openSseStream().",
);
}
await this._startOrAuthStandaloneSSE();
}
}

157
src/examples/README.md
View file

@ -2,67 +2,21 @@
This directory contains example implementations of MCP clients and servers using the TypeScript SDK.
## Table of Contents
- [Streamable HTTP Servers - Single Node Deployment](#streamable-http---single-node-deployment-with-basic-session-state-management)
- [Simple Server with Streamable HTTP](#simple-server-with-streamable-http-transport-serversimplestreamablehttpts)
- [Server Supporting SSE via GET](#server-supporting-with-sse-via-get-serverstandalonessewithgetstreamablehttpts)
- [Server with JSON Response Mode](#server-with-json-response-mode-serverjsonresponsestreamablehttpts)
- [Client Example - Streamable HTTP](#client-clientsimplestreamablehttpts)
- [Useful bash commands for testing](#useful-commands-for-testing)
## Streamable HTTP - single node deployment with basic session state management
Multi node with stete management example will be added soon after we add support.
Multi node with state management example will be added soon after we add support.
### Server with JSON response mode (`server/jsonResponseStreamableHttp.ts`)
A simple MCP server that uses the Streamable HTTP transport with JSON response mode enabled, implemented with Express. The server provides a simple `greet` tool that returns a greeting for a name.
#### Running the server
```bash
npx tsx src/examples/server/jsonResponseStreamableHttp.ts
```
The server will start on port 3000. You can test the initialization and tool calling:
```bash
# Initialize the server and get the session ID from headers
SESSION_ID=$(curl -X POST \
-H "Content-Type: application/json" \
-H "Accept: application/json" \
-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`)
### Simple server with Streamable HTTP transport (`server/simpleStreamableHttp.ts`)
A simple MCP server that uses the Streamable HTTP transport, implemented with Express. The server provides:
@ -78,6 +32,53 @@ npx tsx src/examples/server/simpleStreamableHttp.ts
The server will start on port 3000. You can test the initialization and tool listing:
### Server supporting SSE via GET (`server/standaloneSseWithGetStreamableHttp.ts`)
An MCP server that demonstrates how to support SSE notifications via GET requests using the Streamable HTTP transport with Express. The server dynamically adds resources at regular intervals and supports notifications for resource list changes (server notifications are available through the standalone SSE connection established by GET request).
#### Running the server
```bash
npx tsx src/examples/server/standaloneSseWithGetStreamableHttp.ts
```
The server will start on port 3000 and automatically create new resources every 5 seconds.
### Server with JSON response mode (`server/jsonResponseStreamableHttp.ts`)
A simple MCP server that uses the Streamable HTTP transport with JSON response mode enabled, implemented with Express. The server provides a simple `greet` tool that returns a greeting for a name.
_NOTE: This demonstrates a server that does not use SSE at all. Note that this limits its support for MCP features; for example, it cannot provide logging and progress notifications for tool execution._
#### Running the server
```bash
npx tsx src/examples/server/jsonResponseStreamableHttp.ts
```
### 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
- List available resources
#### Running the client
```bash
npx tsx src/examples/client/simpleStreamableHttp.ts
```
Make sure the server is running before starting the client.
### Useful commands for testing
#### Initialize
Streamable HTTP transport requires to do the initialization first.
```bash
# First initialize the server and save the session ID to a variable
SESSION_ID=$(curl -X POST \
@ -100,6 +101,11 @@ SESSION_ID=$(curl -X POST \
-i http://localhost:3000/mcp 2>&1 | grep -i "mcp-session-id" | cut -d' ' -f2 | tr -d '\r')
echo "Session ID: $SESSION_ID
```
Once a session is established, we can send POST requests:
#### List tools
```bash
# 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" \
@ -107,24 +113,25 @@ curl -X POST -H "Content-Type: application/json" -H "Accept: application/json, t
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
- List available resources
#### Running the client
#### Call tool
```bash
npx tsx src/examples/client/simpleStreamableHttp.ts
# 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
```
Make sure the server is running before starting the client.
## Notes
- 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

118
src/examples/client/simpleStreamableHttp.ts
View file

@ -11,7 +11,8 @@ import {
GetPromptResultSchema,
ListResourcesRequest,
ListResourcesResultSchema,
LoggingMessageNotificationSchema
LoggingMessageNotificationSchema,
ResourceListChangedNotificationSchema,
} from '../../types.js';
async function main(): Promise<void> {
@ -25,50 +26,31 @@ async function main(): Promise<void> {
new URL('http://localhost:3000/mcp')
);
// Connect the client using the transport and initialize the server
// Connect the client using the transport and initialize the server
await client.connect(transport);
console.log('Connected to MCP server');
// Set up notification handlers for server-initiated messages
client.setNotificationHandler(LoggingMessageNotificationSchema, (notification) => {
console.log(`Notification received: ${notification.params.level} - ${notification.params.data}`);
});
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);
// Call the 'greet' tool
const greetRequest: CallToolRequest = {
method: 'tools/call',
params: {
name: 'greet',
arguments: { name: 'MCP User' }
}
};
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}`);
}
client.setNotificationHandler(ResourceListChangedNotificationSchema, async (_) => {
console.log(`Resource list changed notification received!`);
const resourcesRequest: ListResourcesRequest = {
method: 'resources/list',
params: {}
};
const resourcesResult = await client.request(resourcesRequest, ListResourcesResultSchema);
console.log('Available resources count:', resourcesResult.resources.length);
});
// List and call tools
await listTools(client);
await callGreetTool(client);
await callMultiGreetTool(client);
// List available prompts
try {
const promptsRequest: ListPromptsRequest = {
@ -107,9 +89,63 @@ async function main(): Promise<void> {
} catch (error) {
console.log(`Resources not supported by this server (${error})`);
}
// Keep the connection open to receive notifications
console.log('\nKeeping connection open to receive notifications. Press Ctrl+C to exit.');
}
// Close the connection
await client.close();
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:', toolsResult.tools);
if (toolsResult.tools.length === 0) {
console.log('No tools available from the server');
}
} catch (error) {
console.log(`Tools not supported by this server (${error})`);
return
}
}
async function callGreetTool(client: Client): Promise<void> {
try {
const greetRequest: CallToolRequest = {
method: 'tools/call',
params: {
name: 'greet',
arguments: { name: 'MCP User' }
}
};
const greetResult = await client.request(greetRequest, CallToolResultSchema);
console.log('Greeting result:', greetResult.content[0].text);
} catch (error) {
console.log(`Error calling greet tool: ${error}`);
}
}
async function callMultiGreetTool(client: Client): Promise<void> {
try {
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}`);
}
});
} catch (error) {
console.log(`Error calling multi-greet tool: ${error}`);
}
}
main().catch((error: unknown) => {

7
src/examples/server/jsonResponseStreamableHttp.ts
View file

@ -138,6 +138,13 @@ app.post('/mcp', async (req: Request, res: Response) => {
}
});
// 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');
});
// Helper function to detect initialize requests
function isInitializeRequest(body: unknown): boolean {
if (Array.isArray(body)) {

13
src/examples/server/simpleStreamableHttp.ts
View file

@ -174,6 +174,19 @@ app.post('/mcp', async (req: Request, res: Response) => {
}
});
// 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;
}
console.log(`Establishing SSE stream for session ${sessionId}`);
const transport = transports[sessionId];
await transport.handleRequest(req, res);
});
// Helper function to detect initialize requests
function isInitializeRequest(body: unknown): boolean {
if (Array.isArray(body)) {

130
src/examples/server/standaloneSseWithGetStreamableHttp.ts Normal file
View file

@ -0,0 +1,130 @@
import express, { Request, Response } from 'express';
import { randomUUID } from 'node:crypto';
import { McpServer } from '../../server/mcp.js';
import { StreamableHTTPServerTransport } from '../../server/streamableHttp.js';
import { 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(),
});
// Connect the transport to the MCP server
await server.connect(transport);
await transport.handleRequest(req, res, req.body);
// Store the transport by session ID for future requests
if (transport.sessionId) {
transports[transport.sessionId] = transport;
}
return; // Already handled
} else {
// Invalid request - no session ID or not initialization request
res.status(400).json({
jsonrpc: '2.0',
error: {
code: -32000,
message: 'Bad Request: No valid session ID provided',
},
id: null,
});
return;
}
// Handle the request with existing transport
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);
});
// Helper function to detect initialize requests
function isInitializeRequest(body: unknown): boolean {
if (Array.isArray(body)) {
return body.some(msg => typeof msg === 'object' && msg !== null && 'method' in msg && msg.method === 'initialize');
}
return typeof body === 'object' && body !== null && 'method' in body && body.method === 'initialize';
}
// Start the server
const PORT = 3000;
app.listen(PORT, () => {
console.log(`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);
});

114
src/server/streamableHttp.test.ts
View file

@ -32,6 +32,7 @@ function createMockResponse(): jest.Mocked<ServerResponse> {
emit: jest.fn().mockReturnThis(),
getHeader: jest.fn(),
setHeader: jest.fn(),
flushHeaders: jest.fn(),
} as unknown as jest.Mocked<ServerResponse>;
return response;
}
@ -564,22 +565,123 @@ describe("StreamableHTTPServerTransport", () => {
mockResponse.writeHead.mockClear();
});
it("should reject GET requests for SSE with 405 Method Not Allowed", async () => {
it("should accept GET requests for SSE with proper Accept header", async () => {
const req = createMockRequest({
method: "GET",
headers: {
"accept": "application/json, text/event-stream",
"accept": "text/event-stream",
"mcp-session-id": transport.sessionId,
},
});
await transport.handleRequest(req, mockResponse);
expect(mockResponse.writeHead).toHaveBeenCalledWith(405, expect.objectContaining({
"Allow": "POST, DELETE"
expect(mockResponse.writeHead).toHaveBeenCalledWith(200, expect.objectContaining({
"Content-Type": "text/event-stream",
"Cache-Control": "no-cache, no-transform",
"Connection": "keep-alive",
"mcp-session-id": transport.sessionId,
}));
expect(mockResponse.end).toHaveBeenCalledWith(expect.stringContaining('"jsonrpc":"2.0"'));
expect(mockResponse.end).toHaveBeenCalledWith(expect.stringContaining('Method not allowed'));
});
it("should reject GET requests without Accept: text/event-stream header", async () => {
const req = createMockRequest({
method: "GET",
headers: {
"accept": "application/json",
"mcp-session-id": transport.sessionId,
},
});
await transport.handleRequest(req, mockResponse);
expect(mockResponse.writeHead).toHaveBeenCalledWith(406);
expect(mockResponse.end).toHaveBeenCalledWith(expect.stringContaining('"message":"Not Acceptable: Client must accept text/event-stream"'));
});
it("should send server-initiated requests to GET SSE stream", async () => {
// Open a standalone SSE stream with GET
const req = createMockRequest({
method: "GET",
headers: {
"accept": "text/event-stream",
"mcp-session-id": transport.sessionId,
},
});
const sseResponse = createMockResponse();
await transport.handleRequest(req, sseResponse);
// Send a notification without a related request ID
const notification: JSONRPCMessage = {
jsonrpc: "2.0",
method: "notifications/resources/updated",
params: { uri: "someuri" }
};
await transport.send(notification);
// Verify notification was sent on SSE stream
expect(sseResponse.write).toHaveBeenCalledWith(
expect.stringContaining(`event: message\ndata: ${JSON.stringify(notification)}\n\n`)
);
});
it("should not close GET SSE stream after sending server requests or notifications", async () => {
// Open a standalone SSE stream
const req = createMockRequest({
method: "GET",
headers: {
"accept": "text/event-stream",
"mcp-session-id": transport.sessionId,
},
});
const sseResponse = createMockResponse();
await transport.handleRequest(req, sseResponse);
// Send multiple notifications
const notification1: JSONRPCMessage = { jsonrpc: "2.0", method: "event1", params: {} };
const notification2: JSONRPCMessage = { jsonrpc: "2.0", method: "event2", params: {} };
await transport.send(notification1);
await transport.send(notification2);
// Stream should remain open
expect(sseResponse.end).not.toHaveBeenCalled();
});
it("should reject second GET SSE stream for the same session", async () => {
// Open first SSE stream - should succeed
const req1 = createMockRequest({
method: "GET",
headers: {
"accept": "text/event-stream",
"mcp-session-id": transport.sessionId,
},
});
const sseResponse1 = createMockResponse();
await transport.handleRequest(req1, sseResponse1);
// Try to open a second SSE stream - should be rejected
const req2 = createMockRequest({
method: "GET",
headers: {
"accept": "text/event-stream",
"mcp-session-id": transport.sessionId,
},
});
const sseResponse2 = createMockResponse();
await transport.handleRequest(req2, sseResponse2);
// First stream should be good
expect(sseResponse1.writeHead).toHaveBeenCalledWith(200, expect.anything());
// Second stream should get 409 Conflict
expect(sseResponse2.writeHead).toHaveBeenCalledWith(409);
expect(sseResponse2.end).toHaveBeenCalledWith(expect.stringContaining('"message":"Conflict: Only one SSE stream is allowed per session"'));
});
it("should reject POST requests without proper Accept header", async () => {

94
src/server/streamableHttp.ts
View file

@ -68,6 +68,7 @@ export class StreamableHTTPServerTransport implements Transport {
private _requestResponseMap: Map<RequestId, JSONRPCMessage> = new Map();
private _initialized: boolean = false;
private _enableJsonResponse: boolean = false;
private _standaloneSSE: ServerResponse | undefined;
sessionId?: string | undefined;
@ -97,6 +98,8 @@ export class StreamableHTTPServerTransport implements Transport {
async handleRequest(req: IncomingMessage, res: ServerResponse, parsedBody?: unknown): Promise<void> {
if (req.method === "POST") {
await this.handlePostRequest(req, res, parsedBody);
} else if (req.method === "GET") {
await this.handleGetRequest(req, res);
} else if (req.method === "DELETE") {
await this.handleDeleteRequest(req, res);
} else {
@ -105,12 +108,78 @@ export class StreamableHTTPServerTransport implements Transport {
}
/**
* Handles unsupported requests (GET, PUT, PATCH, etc.)
* For now we support only POST and DELETE requests. Support for GET for SSE connections will be added later.
* Handles GET requests for SSE stream
*/
private async handleGetRequest(req: IncomingMessage, res: ServerResponse): Promise<void> {
// The client MUST include an Accept header, listing text/event-stream as a supported content type.
const acceptHeader = req.headers.accept;
if (!acceptHeader?.includes("text/event-stream")) {
res.writeHead(406).end(JSON.stringify({
jsonrpc: "2.0",
error: {
code: -32000,
message: "Not Acceptable: Client must accept text/event-stream"
},
id: null
}));
return;
}
// If an Mcp-Session-Id is returned by the server during initialization,
// clients using the Streamable HTTP transport MUST include it
// in the Mcp-Session-Id header on all of their subsequent HTTP requests.
if (!this.validateSession(req, res)) {
return;
}
// The server MUST either return Content-Type: text/event-stream in response to this HTTP GET,
// or else return HTTP 405 Method Not Allowed
const headers: Record<string, string> = {
"Content-Type": "text/event-stream",
"Cache-Control": "no-cache, no-transform",
Connection: "keep-alive",
};
// After initialization, always include the session ID if we have one
if (this.sessionId !== undefined) {
headers["mcp-session-id"] = this.sessionId;
}
// The server MAY include a Last-Event-ID header in the response to this HTTP GET.
// Resumability will be supported in the future
// Check if there's already an active standalone SSE stream for this session
if (this._standaloneSSE !== undefined) {
// Only one GET SSE stream is allowed per session
res.writeHead(409).end(JSON.stringify({
jsonrpc: "2.0",
error: {
code: -32000,
message: "Conflict: Only one SSE stream is allowed per session"
},
id: null
}));
return;
}
// We need to send headers immediately as message will arrive much later,
// otherwise the client will just wait for the first message
res.writeHead(200, headers).flushHeaders();
// Assing the response to the standalone SSE stream
this._standaloneSSE = res;
// Set up close handler for client disconnects
res.on("close", () => {
this._standaloneSSE = undefined;
});
}
/**
* Handles unsupported requests (PUT, PATCH, etc.)
*/
private async handleUnsupportedRequest(res: ServerResponse): Promise<void> {
res.writeHead(405, {
"Allow": "POST, DELETE"
"Allow": "GET, POST, DELETE"
}).end(JSON.stringify({
jsonrpc: "2.0",
error: {
@ -379,8 +448,24 @@ export class StreamableHTTPServerTransport implements Transport {
// If the message is a response, use the request ID from the message
requestId = message.id;
}
// Check if this message should be sent on the standalone SSE stream (no request ID)
// Ignore notifications from tools (which have relatedRequestId set)
// Those will be sent via dedicated response SSE streams
if (requestId === undefined) {
throw new Error("No request ID provided for the message");
// For standalone SSE streams, we can only send requests and notifications
if ('result' in message || 'error' in message) {
throw new Error("Cannot send a response on a standalone SSE stream unless resuming a previous client request");
}
if (this._standaloneSSE === undefined) {
// The spec says the server MAY send messages on the stream, so it's ok to discard if no stream
return;
}
// Send the message to the standalone SSE stream
this._standaloneSSE.write(`event: message\ndata: ${JSON.stringify(message)}\n\n`);
return;
}
// Get the response for this request
@ -389,7 +474,6 @@ export class StreamableHTTPServerTransport implements Transport {
throw new Error(`No connection established for request ID: ${String(requestId)}`);
}
if (!this._enableJsonResponse) {
response.write(`event: message\ndata: ${JSON.stringify(message)}\n\n`);
}