feat: fetch transport (#1209)

Co-authored-by: Felix Weinberger <3823880+felixweinberger@users.noreply.github.com>

package-lock.json

@@ -9,6 +9,7 @@
             "version": "1.24.3",
             "license": "MIT",
             "dependencies": {
+                "@hono/node-server": "^1.19.7",
                 "ajv": "^8.17.1",
                 "ajv-formats": "^3.0.1",
                 "content-type": "^1.0.5",
@@ -663,6 +664,18 @@
                 "node": "^18.18.0 || ^20.9.0 || >=21.1.0"
             }
         },
+        "node_modules/@hono/node-server": {
+            "version": "1.19.7",
+            "resolved": "https://registry.npmjs.org/@hono/node-server/-/node-server-1.19.7.tgz",
+            "integrity": "sha512-vUcD0uauS7EU2caukW8z5lJKtoGMokxNbJtBiwHgpqxEXokaHCBkQUmCHhjFB1VUTWdqj25QoMkMKzgjq+uhrw==",
+            "license": "MIT",
+            "engines": {
+                "node": ">=18.14.1"
+            },
+            "peerDependencies": {
+                "hono": "^4"
+            }
+        },
         "node_modules/@humanfs/core": {
             "version": "0.19.1",
             "resolved": "https://registry.npmjs.org/@humanfs/core/-/core-0.19.1.tgz",
@@ -3027,6 +3040,16 @@
                 "node": ">= 0.4"
             }
         },
+        "node_modules/hono": {
+            "version": "4.10.8",
+            "resolved": "https://registry.npmjs.org/hono/-/hono-4.10.8.tgz",
+            "integrity": "sha512-DDT0A0r6wzhe8zCGoYOmMeuGu3dyTAE40HHjwUsWFTEy5WxK1x2WDSsBPlEXgPbRIFY6miDualuUDbasPogIww==",
+            "license": "MIT",
+            "peer": true,
+            "engines": {
+                "node": ">=16.9.0"
+            }
+        },
         "node_modules/http-errors": {
             "version": "2.0.1",
             "resolved": "https://registry.npmjs.org/http-errors/-/http-errors-2.0.1.tgz",

package.json

@@ -86,6 +86,7 @@
         "client": "tsx scripts/cli.ts client"
     },
     "dependencies": {
+        "@hono/node-server": "^1.19.7",
         "ajv": "^8.17.1",
         "ajv-formats": "^3.0.1",
         "content-type": "^1.0.5",

src/examples/server/honoWebStandardStreamableHttp.ts

@@ -0,0 +1,74 @@
+/**
+ * Example MCP server using Hono with WebStandardStreamableHTTPServerTransport
+ *
+ * This example demonstrates using the Web Standard transport directly with Hono,
+ * which works on any runtime: Node.js, Cloudflare Workers, Deno, Bun, etc.
+ *
+ * Run with: npx tsx src/examples/server/honoWebStandardStreamableHttp.ts
+ */
+
+import { Hono } from 'hono';
+import { cors } from 'hono/cors';
+import { serve } from '@hono/node-server';
+import * as z from 'zod/v4';
+import { McpServer } from '../../server/mcp.js';
+import { WebStandardStreamableHTTPServerTransport } from '../../server/webStandardStreamableHttp.js';
+import { CallToolResult } from '../../types.js';
+
+// Create the MCP server
+const server = new McpServer({
+    name: 'hono-webstandard-mcp-server',
+    version: '1.0.0'
+});
+
+// Register a simple greeting tool
+server.registerTool(
+    'greet',
+    {
+        title: 'Greeting Tool',
+        description: 'A simple greeting tool',
+        inputSchema: { name: z.string().describe('Name to greet') }
+    },
+    async ({ name }): Promise<CallToolResult> => {
+        return {
+            content: [{ type: 'text', text: `Hello, ${name}! (from Hono + WebStandard transport)` }]
+        };
+    }
+);
+
+// Create a stateless transport (no options = no session management)
+const transport = new WebStandardStreamableHTTPServerTransport();
+
+// Create the Hono app
+const app = new Hono();
+
+// Enable CORS for all origins
+app.use(
+    '*',
+    cors({
+        origin: '*',
+        allowMethods: ['GET', 'POST', 'DELETE', 'OPTIONS'],
+        allowHeaders: ['Content-Type', 'mcp-session-id', 'Last-Event-ID', 'mcp-protocol-version'],
+        exposeHeaders: ['mcp-session-id', 'mcp-protocol-version']
+    })
+);
+
+// Health check endpoint
+app.get('/health', c => c.json({ status: 'ok' }));
+
+// MCP endpoint
+app.all('/mcp', c => transport.handleRequest(c.req.raw));
+
+// Start the server
+const PORT = process.env.MCP_PORT ? parseInt(process.env.MCP_PORT, 10) : 3000;
+
+server.connect(transport).then(() => {
+    console.log(`Starting Hono MCP server on port ${PORT}`);
+    console.log(`Health check: http://localhost:${PORT}/health`);
+    console.log(`MCP endpoint: http://localhost:${PORT}/mcp`);
+
+    serve({
+        fetch: app.fetch,
+        port: PORT
+    });
+});

src/server/webStandardStreamableHttp.ts

@@ -0,0 +1,997 @@
+/**
+ * Web Standards Streamable HTTP Server Transport
+ *
+ * This is the core transport implementation using Web Standard APIs (Request, Response, ReadableStream).
+ * It can run on any runtime that supports Web Standards: Node.js 18+, Cloudflare Workers, Deno, Bun, etc.
+ *
+ * For Node.js Express/HTTP compatibility, use `StreamableHTTPServerTransport` which wraps this transport.
+ */
+
+import { Transport } from '../shared/transport.js';
+import { AuthInfo } from './auth/types.js';
+import {
+    MessageExtraInfo,
+    RequestInfo,
+    isInitializeRequest,
+    isJSONRPCErrorResponse,
+    isJSONRPCRequest,
+    isJSONRPCResultResponse,
+    JSONRPCMessage,
+    JSONRPCMessageSchema,
+    RequestId,
+    SUPPORTED_PROTOCOL_VERSIONS,
+    DEFAULT_NEGOTIATED_PROTOCOL_VERSION
+} from '../types.js';
+
+export type StreamId = string;
+export type EventId = string;
+
+/**
+ * Interface for resumability support via event storage
+ */
+export interface EventStore {
+    /**
+     * Stores an event for later retrieval
+     * @param streamId ID of the stream the event belongs to
+     * @param message The JSON-RPC message to store
+     * @returns The generated event ID for the stored event
+     */
+    storeEvent(streamId: StreamId, message: JSONRPCMessage): Promise<EventId>;
+
+    /**
+     * Get the stream ID associated with a given event ID.
+     * @param eventId The event ID to look up
+     * @returns The stream ID, or undefined if not found
+     *
+     * Optional: If not provided, the SDK will use the streamId returned by
+     * replayEventsAfter for stream mapping.
+     */
+    getStreamIdForEventId?(eventId: EventId): Promise<StreamId | undefined>;
+
+    replayEventsAfter(
+        lastEventId: EventId,
+        {
+            send
+        }: {
+            send: (eventId: EventId, message: JSONRPCMessage) => Promise<void>;
+        }
+    ): Promise<StreamId>;
+}
+
+/**
+ * Internal stream mapping for managing SSE connections
+ */
+interface StreamMapping {
+    /** Stream controller for pushing SSE data - only used with ReadableStream approach */
+    controller?: ReadableStreamDefaultController<Uint8Array>;
+    /** Text encoder for SSE formatting */
+    encoder?: TextEncoder;
+    /** Promise resolver for JSON response mode */
+    resolveJson?: (response: Response) => void;
+    /** Cleanup function to close stream and remove mapping */
+    cleanup: () => void;
+}
+
+/**
+ * Configuration options for WebStandardStreamableHTTPServerTransport
+ */
+export interface WebStandardStreamableHTTPServerTransportOptions {
+    /**
+     * Function that generates a session ID for the transport.
+     * The session ID SHOULD be globally unique and cryptographically secure (e.g., a securely generated UUID, a JWT, or a cryptographic hash)
+     *
+     * If not provided, session management is disabled (stateless mode).
+     */
+    sessionIdGenerator?: () => string;
+
+    /**
+     * A callback for session initialization events
+     * This is called when the server initializes a new session.
+     * Useful in cases when you need to register multiple mcp sessions
+     * and need to keep track of them.
+     * @param sessionId The generated session ID
+     */
+    onsessioninitialized?: (sessionId: string) => void | Promise<void>;
+
+    /**
+     * A callback for session close events
+     * This is called when the server closes a session due to a DELETE request.
+     * Useful in cases when you need to clean up resources associated with the session.
+     * Note that this is different from the transport closing, if you are handling
+     * HTTP requests from multiple nodes you might want to close each
+     * WebStandardStreamableHTTPServerTransport after a request is completed while still keeping the
+     * session open/running.
+     * @param sessionId The session ID that was closed
+     */
+    onsessionclosed?: (sessionId: string) => void | Promise<void>;
+
+    /**
+     * If true, the server will return JSON responses instead of starting an SSE stream.
+     * This can be useful for simple request/response scenarios without streaming.
+     * Default is false (SSE streams are preferred).
+     */
+    enableJsonResponse?: boolean;
+
+    /**
+     * Event store for resumability support
+     * If provided, resumability will be enabled, allowing clients to reconnect and resume messages
+     */
+    eventStore?: EventStore;
+
+    /**
+     * List of allowed host header values for DNS rebinding protection.
+     * If not specified, host validation is disabled.
+     * @deprecated Use external middleware for host validation instead.
+     */
+    allowedHosts?: string[];
+
+    /**
+     * List of allowed origin header values for DNS rebinding protection.
+     * If not specified, origin validation is disabled.
+     * @deprecated Use external middleware for origin validation instead.
+     */
+    allowedOrigins?: string[];
+
+    /**
+     * Enable DNS rebinding protection (requires allowedHosts and/or allowedOrigins to be configured).
+     * Default is false for backwards compatibility.
+     * @deprecated Use external middleware for DNS rebinding protection instead.
+     */
+    enableDnsRebindingProtection?: boolean;
+
+    /**
+     * Retry interval in milliseconds to suggest to clients in SSE retry field.
+     * When set, the server will send a retry field in SSE priming events to control
+     * client reconnection timing for polling behavior.
+     */
+    retryInterval?: number;
+}
+
+/**
+ * Options for handling a request
+ */
+export interface HandleRequestOptions {
+    /**
+     * Pre-parsed request body. If provided, the transport will use this instead of parsing req.json().
+     * Useful when using body-parser middleware that has already parsed the body.
+     */
+    parsedBody?: unknown;
+
+    /**
+     * Authentication info from middleware. If provided, will be passed to message handlers.
+     */
+    authInfo?: AuthInfo;
+}
+
+/**
+ * Server transport for Web Standards Streamable HTTP: this implements the MCP Streamable HTTP transport specification
+ * using Web Standard APIs (Request, Response, ReadableStream).
+ *
+ * This transport works on any runtime that supports Web Standards: Node.js 18+, Cloudflare Workers, Deno, Bun, etc.
+ *
+ * Usage example:
+ *
+ * ```typescript
+ * // Stateful mode - server sets the session ID
+ * const statefulTransport = new WebStandardStreamableHTTPServerTransport({
+ *   sessionIdGenerator: () => crypto.randomUUID(),
+ * });
+ *
+ * // Stateless mode - explicitly set session ID to undefined
+ * const statelessTransport = new WebStandardStreamableHTTPServerTransport({
+ *   sessionIdGenerator: undefined,
+ * });
+ *
+ * // Hono.js usage
+ * app.all('/mcp', async (c) => {
+ *   return transport.handleRequest(c.req.raw);
+ * });
+ *
+ * // Cloudflare Workers usage
+ * export default {
+ *   async fetch(request: Request): Promise<Response> {
+ *     return transport.handleRequest(request);
+ *   }
+ * };
+ * ```
+ *
+ * In stateful mode:
+ * - Session ID is generated and included in response headers
+ * - Session ID is always included in initialization responses
+ * - Requests with invalid session IDs are rejected with 404 Not Found
+ * - Non-initialization requests without a session ID are rejected with 400 Bad Request
+ * - State is maintained in-memory (connections, message history)
+ *
+ * In stateless mode:
+ * - No Session ID is included in any responses
+ * - No session validation is performed
+ */
+export class WebStandardStreamableHTTPServerTransport implements Transport {
+    // when sessionId is not set (undefined), it means the transport is in stateless mode
+    private sessionIdGenerator: (() => string) | undefined;
+    private _started: boolean = false;
+    private _streamMapping: Map<string, StreamMapping> = new Map();
+    private _requestToStreamMapping: Map<RequestId, string> = new Map();
+    private _requestResponseMap: Map<RequestId, JSONRPCMessage> = new Map();
+    private _initialized: boolean = false;
+    private _enableJsonResponse: boolean = false;
+    private _standaloneSseStreamId: string = '_GET_stream';
+    private _eventStore?: EventStore;
+    private _onsessioninitialized?: (sessionId: string) => void | Promise<void>;
+    private _onsessionclosed?: (sessionId: string) => void | Promise<void>;
+    private _allowedHosts?: string[];
+    private _allowedOrigins?: string[];
+    private _enableDnsRebindingProtection: boolean;
+    private _retryInterval?: number;
+
+    sessionId?: string;
+    onclose?: () => void;
+    onerror?: (error: Error) => void;
+    onmessage?: (message: JSONRPCMessage, extra?: MessageExtraInfo) => void;
+
+    constructor(options: WebStandardStreamableHTTPServerTransportOptions = {}) {
+        this.sessionIdGenerator = options.sessionIdGenerator;
+        this._enableJsonResponse = options.enableJsonResponse ?? false;
+        this._eventStore = options.eventStore;
+        this._onsessioninitialized = options.onsessioninitialized;
+        this._onsessionclosed = options.onsessionclosed;
+        this._allowedHosts = options.allowedHosts;
+        this._allowedOrigins = options.allowedOrigins;
+        this._enableDnsRebindingProtection = options.enableDnsRebindingProtection ?? false;
+        this._retryInterval = options.retryInterval;
+    }
+
+    /**
+     * Starts the transport. This is required by the Transport interface but is a no-op
+     * for the Streamable HTTP transport as connections are managed per-request.
+     */
+    async start(): Promise<void> {
+        if (this._started) {
+            throw new Error('Transport already started');
+        }
+        this._started = true;
+    }
+
+    /**
+     * Helper to create a JSON error response
+     */
+    private createJsonErrorResponse(
+        status: number,
+        code: number,
+        message: string,
+        options?: { headers?: Record<string, string>; data?: string }
+    ): Response {
+        const error: { code: number; message: string; data?: string } = { code, message };
+        if (options?.data !== undefined) {
+            error.data = options.data;
+        }
+        return new Response(
+            JSON.stringify({
+                jsonrpc: '2.0',
+                error,
+                id: null
+            }),
+            {
+                status,
+                headers: {
+                    'Content-Type': 'application/json',
+                    ...options?.headers
+                }
+            }
+        );
+    }
+
+    /**
+     * Validates request headers for DNS rebinding protection.
+     * @returns Error response if validation fails, undefined if validation passes.
+     */
+    private validateRequestHeaders(req: Request): Response | undefined {
+        // Skip validation if protection is not enabled
+        if (!this._enableDnsRebindingProtection) {
+            return undefined;
+        }
+
+        // Validate Host header if allowedHosts is configured
+        if (this._allowedHosts && this._allowedHosts.length > 0) {
+            const hostHeader = req.headers.get('host');
+            if (!hostHeader || !this._allowedHosts.includes(hostHeader)) {
+                const error = `Invalid Host header: ${hostHeader}`;
+                this.onerror?.(new Error(error));
+                return this.createJsonErrorResponse(403, -32000, error);
+            }
+        }
+
+        // Validate Origin header if allowedOrigins is configured
+        if (this._allowedOrigins && this._allowedOrigins.length > 0) {
+            const originHeader = req.headers.get('origin');
+            if (originHeader && !this._allowedOrigins.includes(originHeader)) {
+                const error = `Invalid Origin header: ${originHeader}`;
+                this.onerror?.(new Error(error));
+                return this.createJsonErrorResponse(403, -32000, error);
+            }
+        }
+
+        return undefined;
+    }
+
+    /**
+     * Handles an incoming HTTP request, whether GET, POST, or DELETE
+     * Returns a Response object (Web Standard)
+     */
+    async handleRequest(req: Request, options?: HandleRequestOptions): Promise<Response> {
+        // Validate request headers for DNS rebinding protection
+        const validationError = this.validateRequestHeaders(req);
+        if (validationError) {
+            return validationError;
+        }
+
+        switch (req.method) {
+            case 'POST':
+                return this.handlePostRequest(req, options);
+            case 'GET':
+                return this.handleGetRequest(req);
+            case 'DELETE':
+                return this.handleDeleteRequest(req);
+            default:
+                return this.handleUnsupportedRequest();
+        }
+    }
+
+    /**
+     * Writes a priming event to establish resumption capability.
+     * Only sends if eventStore is configured (opt-in for resumability) and
+     * the client's protocol version supports empty SSE data (>= 2025-11-25).
+     */
+    private async writePrimingEvent(
+        controller: ReadableStreamDefaultController<Uint8Array>,
+        encoder: TextEncoder,
+        streamId: string,
+        protocolVersion: string
+    ): Promise<void> {
+        if (!this._eventStore) {
+            return;
+        }
+
+        // Priming events have empty data which older clients cannot handle.
+        // Only send priming events to clients with protocol version >= 2025-11-25
+        // which includes the fix for handling empty SSE data.
+        if (protocolVersion < '2025-11-25') {
+            return;
+        }
+
+        const primingEventId = await this._eventStore.storeEvent(streamId, {} as JSONRPCMessage);
+
+        let primingEvent = `id: ${primingEventId}\ndata: \n\n`;
+        if (this._retryInterval !== undefined) {
+            primingEvent = `id: ${primingEventId}\nretry: ${this._retryInterval}\ndata: \n\n`;
+        }
+        controller.enqueue(encoder.encode(primingEvent));
+    }
+
+    /**
+     * Handles GET requests for SSE stream
+     */
+    private async handleGetRequest(req: Request): Promise<Response> {
+        // The client MUST include an Accept header, listing text/event-stream as a supported content type.
+        const acceptHeader = req.headers.get('accept');
+        if (!acceptHeader?.includes('text/event-stream')) {
+            return this.createJsonErrorResponse(406, -32000, 'Not Acceptable: Client must accept text/event-stream');
+        }
+
+        // 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.
+        const sessionError = this.validateSession(req);
+        if (sessionError) {
+            return sessionError;
+        }
+        const protocolError = this.validateProtocolVersion(req);
+        if (protocolError) {
+            return protocolError;
+        }
+
+        // Handle resumability: check for Last-Event-ID header
+        if (this._eventStore) {
+            const lastEventId = req.headers.get('last-event-id');
+            if (lastEventId) {
+                return this.replayEvents(lastEventId);
+            }
+        }
+
+        // Check if there's already an active standalone SSE stream for this session
+        if (this._streamMapping.get(this._standaloneSseStreamId) !== undefined) {
+            // Only one GET SSE stream is allowed per session
+            return this.createJsonErrorResponse(409, -32000, 'Conflict: Only one SSE stream is allowed per session');
+        }
+
+        const encoder = new TextEncoder();
+        let streamController: ReadableStreamDefaultController<Uint8Array>;
+
+        // Create a ReadableStream with a controller we can use to push SSE events
+        const readable = new ReadableStream<Uint8Array>({
+            start: controller => {
+                streamController = controller;
+            },
+            cancel: () => {
+                // Stream was cancelled by client
+                this._streamMapping.delete(this._standaloneSseStreamId);
+            }
+        });
+
+        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;
+        }
+
+        // Store the stream mapping with the controller for pushing data
+        this._streamMapping.set(this._standaloneSseStreamId, {
+            controller: streamController!,
+            encoder,
+            cleanup: () => {
+                this._streamMapping.delete(this._standaloneSseStreamId);
+                try {
+                    streamController!.close();
+                } catch {
+                    // Controller might already be closed
+                }
+            }
+        });
+
+        return new Response(readable, { headers });
+    }
+
+    /**
+     * Replays events that would have been sent after the specified event ID
+     * Only used when resumability is enabled
+     */
+    private async replayEvents(lastEventId: string): Promise<Response> {
+        if (!this._eventStore) {
+            return this.createJsonErrorResponse(400, -32000, 'Event store not configured');
+        }
+
+        try {
+            // If getStreamIdForEventId is available, use it for conflict checking
+            let streamId: string | undefined;
+            if (this._eventStore.getStreamIdForEventId) {
+                streamId = await this._eventStore.getStreamIdForEventId(lastEventId);
+
+                if (!streamId) {
+                    return this.createJsonErrorResponse(400, -32000, 'Invalid event ID format');
+                }
+
+                // Check conflict with the SAME streamId we'll use for mapping
+                if (this._streamMapping.get(streamId) !== undefined) {
+                    return this.createJsonErrorResponse(409, -32000, 'Conflict: Stream already has an active connection');
+                }
+            }
+
+            const headers: Record<string, string> = {
+                'Content-Type': 'text/event-stream',
+                'Cache-Control': 'no-cache, no-transform',
+                Connection: 'keep-alive'
+            };
+
+            if (this.sessionId !== undefined) {
+                headers['mcp-session-id'] = this.sessionId;
+            }
+
+            // Create a ReadableStream with controller for SSE
+            const encoder = new TextEncoder();
+            let streamController: ReadableStreamDefaultController<Uint8Array>;
+
+            const readable = new ReadableStream<Uint8Array>({
+                start: controller => {
+                    streamController = controller;
+                },
+                cancel: () => {
+                    // Stream was cancelled by client
+                    // Cleanup will be handled by the mapping
+                }
+            });
+
+            // Replay events - returns the streamId for backwards compatibility
+            const replayedStreamId = await this._eventStore.replayEventsAfter(lastEventId, {
+                send: async (eventId: string, message: JSONRPCMessage) => {
+                    const success = this.writeSSEEvent(streamController!, encoder, message, eventId);
+                    if (!success) {
+                        this.onerror?.(new Error('Failed replay events'));
+                        try {
+                            streamController!.close();
+                        } catch {
+                            // Controller might already be closed
+                        }
+                    }
+                }
+            });
+
+            this._streamMapping.set(replayedStreamId, {
+                controller: streamController!,
+                encoder,
+                cleanup: () => {
+                    this._streamMapping.delete(replayedStreamId);
+                    try {
+                        streamController!.close();
+                    } catch {
+                        // Controller might already be closed
+                    }
+                }
+            });
+
+            return new Response(readable, { headers });
+        } catch (error) {
+            this.onerror?.(error as Error);
+            return this.createJsonErrorResponse(500, -32000, 'Error replaying events');
+        }
+    }
+
+    /**
+     * Writes an event to an SSE stream via controller with proper formatting
+     */
+    private writeSSEEvent(
+        controller: ReadableStreamDefaultController<Uint8Array>,
+        encoder: TextEncoder,
+        message: JSONRPCMessage,
+        eventId?: string
+    ): boolean {
+        try {
+            let eventData = `event: message\n`;
+            // Include event ID if provided - this is important for resumability
+            if (eventId) {
+                eventData += `id: ${eventId}\n`;
+            }
+            eventData += `data: ${JSON.stringify(message)}\n\n`;
+            controller.enqueue(encoder.encode(eventData));
+            return true;
+        } catch {
+            return false;
+        }
+    }
+
+    /**
+     * Handles unsupported requests (PUT, PATCH, etc.)
+     */
+    private handleUnsupportedRequest(): Response {
+        return new Response(
+            JSON.stringify({
+                jsonrpc: '2.0',
+                error: {
+                    code: -32000,
+                    message: 'Method not allowed.'
+                },
+                id: null
+            }),
+            {
+                status: 405,
+                headers: {
+                    Allow: 'GET, POST, DELETE',
+                    'Content-Type': 'application/json'
+                }
+            }
+        );
+    }
+
+    /**
+     * Handles POST requests containing JSON-RPC messages
+     */
+    private async handlePostRequest(req: Request, options?: HandleRequestOptions): Promise<Response> {
+        try {
+            // Validate the Accept header
+            const acceptHeader = req.headers.get('accept');
+            // The client MUST include an Accept header, listing both application/json and text/event-stream as supported content types.
+            if (!acceptHeader?.includes('application/json') || !acceptHeader.includes('text/event-stream')) {
+                return this.createJsonErrorResponse(
+                    406,
+                    -32000,
+                    'Not Acceptable: Client must accept both application/json and text/event-stream'
+                );
+            }
+
+            const ct = req.headers.get('content-type');
+            if (!ct || !ct.includes('application/json')) {
+                return this.createJsonErrorResponse(415, -32000, 'Unsupported Media Type: Content-Type must be application/json');
+            }
+
+            // Build request info from headers
+            const requestInfo: RequestInfo = {
+                headers: Object.fromEntries(req.headers.entries())
+            };
+
+            let rawMessage;
+            if (options?.parsedBody !== undefined) {
+                rawMessage = options.parsedBody;
+            } else {
+                try {
+                    rawMessage = await req.json();
+                } catch {
+                    return this.createJsonErrorResponse(400, -32700, 'Parse error: Invalid JSON');
+                }
+            }
+
+            let messages: JSONRPCMessage[];
+
+            // handle batch and single messages
+            try {
+                if (Array.isArray(rawMessage)) {
+                    messages = rawMessage.map(msg => JSONRPCMessageSchema.parse(msg));
+                } else {
+                    messages = [JSONRPCMessageSchema.parse(rawMessage)];
+                }
+            } catch {
+                return this.createJsonErrorResponse(400, -32700, 'Parse error: Invalid JSON-RPC message');
+            }
+
+            // Check if this is an initialization request
+            // https://spec.modelcontextprotocol.io/specification/2025-03-26/basic/lifecycle/
+            const isInitializationRequest = messages.some(isInitializeRequest);
+            if (isInitializationRequest) {
+                // If it's a server with session management and the session ID is already set we should reject the request
+                // to avoid re-initialization.
+                if (this._initialized && this.sessionId !== undefined) {
+                    return this.createJsonErrorResponse(400, -32600, 'Invalid Request: Server already initialized');
+                }
+                if (messages.length > 1) {
+                    return this.createJsonErrorResponse(400, -32600, 'Invalid Request: Only one initialization request is allowed');
+                }
+                this.sessionId = this.sessionIdGenerator?.();
+                this._initialized = true;
+
+                // If we have a session ID and an onsessioninitialized handler, call it immediately
+                // This is needed in cases where the server needs to keep track of multiple sessions
+                if (this.sessionId && this._onsessioninitialized) {
+                    await Promise.resolve(this._onsessioninitialized(this.sessionId));
+                }
+            }
+            if (!isInitializationRequest) {
+                // 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.
+                const sessionError = this.validateSession(req);
+                if (sessionError) {
+                    return sessionError;
+                }
+                // Mcp-Protocol-Version header is required for all requests after initialization.
+                const protocolError = this.validateProtocolVersion(req);
+                if (protocolError) {
+                    return protocolError;
+                }
+            }
+
+            // check if it contains requests
+            const hasRequests = messages.some(isJSONRPCRequest);
+
+            if (!hasRequests) {
+                // if it only contains notifications or responses, return 202
+                for (const message of messages) {
+                    this.onmessage?.(message, { authInfo: options?.authInfo, requestInfo });
+                }
+                return new Response(null, { status: 202 });
+            }
+
+            // The default behavior is to use SSE streaming
+            // but in some cases server will return JSON responses
+            const streamId = crypto.randomUUID();
+
+            // Extract protocol version for priming event decision.
+            // For initialize requests, get from request params.
+            // For other requests, get from header (already validated).
+            const initRequest = messages.find(m => isInitializeRequest(m));
+            const clientProtocolVersion = initRequest
+                ? initRequest.params.protocolVersion
+                : (req.headers.get('mcp-protocol-version') ?? DEFAULT_NEGOTIATED_PROTOCOL_VERSION);
+
+            if (this._enableJsonResponse) {
+                // For JSON response mode, return a Promise that resolves when all responses are ready
+                return new Promise<Response>(resolve => {
+                    this._streamMapping.set(streamId, {
+                        resolveJson: resolve,
+                        cleanup: () => {
+                            this._streamMapping.delete(streamId);
+                        }
+                    });
+
+                    for (const message of messages) {
+                        if (isJSONRPCRequest(message)) {
+                            this._requestToStreamMapping.set(message.id, streamId);
+                        }
+                    }
+
+                    for (const message of messages) {
+                        this.onmessage?.(message, { authInfo: options?.authInfo, requestInfo });
+                    }
+                });
+            }
+
+            // SSE streaming mode - use ReadableStream with controller for more reliable data pushing
+            const encoder = new TextEncoder();
+            let streamController: ReadableStreamDefaultController<Uint8Array>;
+
+            const readable = new ReadableStream<Uint8Array>({
+                start: controller => {
+                    streamController = controller;
+                },
+                cancel: () => {
+                    // Stream was cancelled by client
+                    this._streamMapping.delete(streamId);
+                }
+            });
+
+            const headers: Record<string, string> = {
+                'Content-Type': 'text/event-stream',
+                'Cache-Control': 'no-cache',
+                Connection: 'keep-alive'
+            };
+
+            // After initialization, always include the session ID if we have one
+            if (this.sessionId !== undefined) {
+                headers['mcp-session-id'] = this.sessionId;
+            }
+
+            // Store the response for this request to send messages back through this connection
+            // We need to track by request ID to maintain the connection
+            for (const message of messages) {
+                if (isJSONRPCRequest(message)) {
+                    this._streamMapping.set(streamId, {
+                        controller: streamController!,
+                        encoder,
+                        cleanup: () => {
+                            this._streamMapping.delete(streamId);
+                            try {
+                                streamController!.close();
+                            } catch {
+                                // Controller might already be closed
+                            }
+                        }
+                    });
+                    this._requestToStreamMapping.set(message.id, streamId);
+                }
+            }
+
+            // Write priming event if event store is configured (after mapping is set up)
+            await this.writePrimingEvent(streamController!, encoder, streamId, clientProtocolVersion);
+
+            // handle each message
+            for (const message of messages) {
+                // Build closeSSEStream callback for requests when eventStore is configured
+                // AND client supports resumability (protocol version >= 2025-11-25).
+                // Old clients can't resume if the stream is closed early because they
+                // didn't receive a priming event with an event ID.
+                let closeSSEStream: (() => void) | undefined;
+                let closeStandaloneSSEStream: (() => void) | undefined;
+                if (isJSONRPCRequest(message) && this._eventStore && clientProtocolVersion >= '2025-11-25') {
+                    closeSSEStream = () => {
+                        this.closeSSEStream(message.id);
+                    };
+                    closeStandaloneSSEStream = () => {
+                        this.closeStandaloneSSEStream();
+                    };
+                }
+
+                this.onmessage?.(message, { authInfo: options?.authInfo, requestInfo, closeSSEStream, closeStandaloneSSEStream });
+            }
+            // The server SHOULD NOT close the SSE stream before sending all JSON-RPC responses
+            // This will be handled by the send() method when responses are ready
+
+            return new Response(readable, { status: 200, headers });
+        } catch (error) {
+            // return JSON-RPC formatted error
+            this.onerror?.(error as Error);
+            return this.createJsonErrorResponse(400, -32700, 'Parse error', { data: String(error) });
+        }
+    }
+
+    /**
+     * Handles DELETE requests to terminate sessions
+     */
+    private async handleDeleteRequest(req: Request): Promise<Response> {
+        const sessionError = this.validateSession(req);
+        if (sessionError) {
+            return sessionError;
+        }
+        const protocolError = this.validateProtocolVersion(req);
+        if (protocolError) {
+            return protocolError;
+        }
+
+        await Promise.resolve(this._onsessionclosed?.(this.sessionId!));
+        await this.close();
+        return new Response(null, { status: 200 });
+    }
+
+    /**
+     * Validates session ID for non-initialization requests.
+     * Returns Response error if invalid, undefined otherwise
+     */
+    private validateSession(req: Request): Response | undefined {
+        if (this.sessionIdGenerator === undefined) {
+            // If the sessionIdGenerator ID is not set, the session management is disabled
+            // and we don't need to validate the session ID
+            return undefined;
+        }
+        if (!this._initialized) {
+            // If the server has not been initialized yet, reject all requests
+            return this.createJsonErrorResponse(400, -32000, 'Bad Request: Server not initialized');
+        }
+
+        const sessionId = req.headers.get('mcp-session-id');
+
+        if (!sessionId) {
+            // Non-initialization requests without a session ID should return 400 Bad Request
+            return this.createJsonErrorResponse(400, -32000, 'Bad Request: Mcp-Session-Id header is required');
+        }
+
+        if (sessionId !== this.sessionId) {
+            // Reject requests with invalid session ID with 404 Not Found
+            return this.createJsonErrorResponse(404, -32001, 'Session not found');
+        }
+
+        return undefined;
+    }
+
+    /**
+     * Validates the MCP-Protocol-Version header on incoming requests.
+     *
+     * For initialization: Version negotiation handles unknown versions gracefully
+     * (server responds with its supported version).
+     *
+     * For subsequent requests with MCP-Protocol-Version header:
+     * - Accept if in supported list
+     * - 400 if unsupported
+     *
+     * For HTTP requests without the MCP-Protocol-Version header:
+     * - Accept and default to the version negotiated at initialization
+     */
+    private validateProtocolVersion(req: Request): Response | undefined {
+        const protocolVersion = req.headers.get('mcp-protocol-version');
+
+        if (protocolVersion !== null && !SUPPORTED_PROTOCOL_VERSIONS.includes(protocolVersion)) {
+            return this.createJsonErrorResponse(
+                400,
+                -32000,
+                `Bad Request: Unsupported protocol version: ${protocolVersion} (supported versions: ${SUPPORTED_PROTOCOL_VERSIONS.join(', ')})`
+            );
+        }
+        return undefined;
+    }
+
+    async close(): Promise<void> {
+        // Close all SSE connections
+        this._streamMapping.forEach(({ cleanup }) => {
+            cleanup();
+        });
+        this._streamMapping.clear();
+
+        // Clear any pending responses
+        this._requestResponseMap.clear();
+        this.onclose?.();
+    }
+
+    /**
+     * Close an SSE stream for a specific request, triggering client reconnection.
+     * Use this to implement polling behavior during long-running operations -
+     * client will reconnect after the retry interval specified in the priming event.
+     */
+    closeSSEStream(requestId: RequestId): void {
+        const streamId = this._requestToStreamMapping.get(requestId);
+        if (!streamId) return;
+
+        const stream = this._streamMapping.get(streamId);
+        if (stream) {
+            stream.cleanup();
+        }
+    }
+
+    /**
+     * Close the standalone GET SSE stream, triggering client reconnection.
+     * Use this to implement polling behavior for server-initiated notifications.
+     */
+    closeStandaloneSSEStream(): void {
+        const stream = this._streamMapping.get(this._standaloneSseStreamId);
+        if (stream) {
+            stream.cleanup();
+        }
+    }
+
+    async send(message: JSONRPCMessage, options?: { relatedRequestId?: RequestId }): Promise<void> {
+        let requestId = options?.relatedRequestId;
+        if (isJSONRPCResultResponse(message) || isJSONRPCErrorResponse(message)) {
+            // 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) {
+            // For standalone SSE streams, we can only send requests and notifications
+            if (isJSONRPCResultResponse(message) || isJSONRPCErrorResponse(message)) {
+                throw new Error('Cannot send a response on a standalone SSE stream unless resuming a previous client request');
+            }
+
+            // Generate and store event ID if event store is provided
+            // Store even if stream is disconnected so events can be replayed on reconnect
+            let eventId: string | undefined;
+            if (this._eventStore) {
+                // Stores the event and gets the generated event ID
+                eventId = await this._eventStore.storeEvent(this._standaloneSseStreamId, message);
+            }
+
+            const standaloneSse = this._streamMapping.get(this._standaloneSseStreamId);
+            if (standaloneSse === undefined) {
+                // Stream is disconnected - event is stored for replay, nothing more to do
+                return;
+            }
+
+            // Send the message to the standalone SSE stream
+            if (standaloneSse.controller && standaloneSse.encoder) {
+                this.writeSSEEvent(standaloneSse.controller, standaloneSse.encoder, message, eventId);
+            }
+            return;
+        }
+
+        // Get the response for this request
+        const streamId = this._requestToStreamMapping.get(requestId);
+        if (!streamId) {
+            throw new Error(`No connection established for request ID: ${String(requestId)}`);
+        }
+
+        const stream = this._streamMapping.get(streamId);
+
+        if (!this._enableJsonResponse && stream?.controller && stream?.encoder) {
+            // For SSE responses, generate event ID if event store is provided
+            let eventId: string | undefined;
+
+            if (this._eventStore) {
+                eventId = await this._eventStore.storeEvent(streamId, message);
+            }
+            // Write the event to the response stream
+            this.writeSSEEvent(stream.controller, stream.encoder, message, eventId);
+        }
+
+        if (isJSONRPCResultResponse(message) || isJSONRPCErrorResponse(message)) {
+            this._requestResponseMap.set(requestId, message);
+            const relatedIds = Array.from(this._requestToStreamMapping.entries())
+                .filter(([_, sid]) => sid === streamId)
+                .map(([id]) => id);
+
+            // Check if we have responses for all requests using this connection
+            const allResponsesReady = relatedIds.every(id => this._requestResponseMap.has(id));
+
+            if (allResponsesReady) {
+                if (!stream) {
+                    throw new Error(`No connection established for request ID: ${String(requestId)}`);
+                }
+                if (this._enableJsonResponse && stream.resolveJson) {
+                    // All responses ready, send as JSON
+                    const headers: Record<string, string> = {
+                        'Content-Type': 'application/json'
+                    };
+                    if (this.sessionId !== undefined) {
+                        headers['mcp-session-id'] = this.sessionId;
+                    }
+
+                    const responses = relatedIds.map(id => this._requestResponseMap.get(id)!);
+
+                    if (responses.length === 1) {
+                        stream.resolveJson(new Response(JSON.stringify(responses[0]), { status: 200, headers }));
+                    } else {
+                        stream.resolveJson(new Response(JSON.stringify(responses), { status: 200, headers }));
+                    }
+                } else {
+                    // End the SSE stream
+                    stream.cleanup();
+                }
+                // Clean up
+                for (const id of relatedIds) {
+                    this._requestResponseMap.delete(id);
+                    this._requestToStreamMapping.delete(id);
+                }
+            }
+        }
+    }
+}

test/server/streamableHttp.test.ts

@@ -119,7 +119,12 @@ async function sendPostRequest(
     });
 }
 
-function expectErrorResponse(data: unknown, expectedCode: number, expectedMessagePattern: RegExp): void {
+function expectErrorResponse(
+    data: unknown,
+    expectedCode: number,
+    expectedMessagePattern: RegExp,
+    options?: { expectData?: boolean }
+): void {
     expect(data).toMatchObject({
         jsonrpc: '2.0',
         error: expect.objectContaining({
@@ -127,6 +132,9 @@ function expectErrorResponse(data: unknown, expectedCode: number, expectedMessag
             message: expect.stringMatching(expectedMessagePattern)
         })
     });
+    if (options?.expectData) {
+        expect((data as { error: { data?: string } }).error.data).toBeDefined();
+    }
 }
 describe.each(zodTestMatrix)('$zodVersionLabel', (entry: ZodMatrixEntry) => {
     /**
@@ -679,6 +687,28 @@ describe.each(zodTestMatrix)('$zodVersionLabel', (entry: ZodMatrixEntry) => {
             expectErrorResponse(errorData, -32700, /Parse error/);
         });
 
+        it('should include error data in parse error response for unexpected errors', async () => {
+            sessionId = await initializeServer();
+
+            // We can't easily trigger the catch-all error handler, but we can verify
+            // that the JSON parse error includes useful information
+            const response = await fetch(baseUrl, {
+                method: 'POST',
+                headers: {
+                    'Content-Type': 'application/json',
+                    Accept: 'application/json, text/event-stream',
+                    'mcp-session-id': sessionId
+                },
+                body: '{ invalid json }'
+            });
+
+            expect(response.status).toBe(400);
+            const errorData = await response.json();
+            expectErrorResponse(errorData, -32700, /Parse error/);
+            // The error message should contain details about what went wrong
+            expect(errorData.error.message).toContain('Invalid JSON');
+        });
+
         it('should return 400 error for invalid JSON-RPC messages', async () => {
             sessionId = await initializeServer();
 
@@ -1309,9 +1339,6 @@ describe.each(zodTestMatrix)('$zodVersionLabel', (entry: ZodMatrixEntry) => {
             baseUrl = result.baseUrl;
             mcpServer = result.mcpServer;
 
-            // Verify resumability is enabled on the transport
-            expect(transport['_eventStore']).toBeDefined();
-
             // Initialize the server
             const initResponse = await sendPostRequest(baseUrl, TEST_MESSAGES.initialize);
             sessionId = initResponse.headers.get('mcp-session-id') as string;