Transport for web standard fetch APIs: FetchSSEServerTransport

[kentcdodds]

Is your feature request related to a problem? Please describe.

I don't want to work with Node's http request/response APIs.

Describe the solution you'd like

I would like to work with web standard fetch request/response APIs. In my case, I'm integrating with React Router (framework mode), but I imagine this would be helpful to anyone using Hono.js, other modern frameworks, Cloudflare, Bun, Deno, etc.

Describe alternatives you've considered

I've built a transport myself that seems to work:

import { type Transport } from '@modelcontextprotocol/sdk/shared/transport.js'
import {
	type JSONRPCMessage,
	JSONRPCMessageSchema,
} from '@modelcontextprotocol/sdk/types.js'

/**
 * Server transport for SSE using Standard Fetch: this will send messages over
 * an SSE connection and receive messages from HTTP POST requests.
 */
export class FetchSSEServerTransport implements Transport {
	#stream?: ReadableStreamDefaultController<string>
	#sessionId: string
	#endpoint: string

	onclose?: () => void
	onerror?: (error: Error) => void
	onmessage?: (message: JSONRPCMessage) => void

	/**
	 * Creates a new SSE server transport, which will direct the client to POST
	 * messages to the relative or absolute URL identified by `endpoint`.
	 */
	constructor(endpoint: string, sessionId?: string | null) {
		this.#endpoint = endpoint
		this.#sessionId = sessionId ?? crypto.randomUUID()
	}

	/**
	 * Starts processing messages on the transport.
	 * This is called by the Server class and should not be called directly.
	 */
	async start(): Promise<void> {
		if (this.#stream) {
			throw new Error(
				'FetchSSEServerTransport already started! If using Server class, note that connect() calls start() automatically.',
			)
		}
	}

	/**
	 * Handles the initial SSE connection request.
	 * This should be called from your Remix loader to establish the SSE stream.
	 */
	async handleSSERequest(request: Request): Promise<Response> {
		const stream = new ReadableStream<string>({
			start: (controller) => {
				this.#stream = controller

				// Send headers
				controller.enqueue(': ping\n\n') // Keep connection alive

				// Send the endpoint event
				controller.enqueue(
					`event: endpoint\ndata: ${encodeURI(
						this.#endpoint,
					)}?sessionId=${this.#sessionId}\n\n`,
				)

				// Handle cleanup when the connection closes
				request.signal.addEventListener('abort', () => {
					controller.close()
					this.#stream = undefined
					this.onclose?.()
				})
			},
			cancel: () => {
				this.#stream = undefined
				this.onclose?.()
			},
		})

		return new Response(stream, {
			headers: {
				'Content-Type': 'text/event-stream',
				'Cache-Control': 'no-cache',
				Connection: 'keep-alive',
				'Mcp-Session-Id': this.#sessionId,
			},
		})
	}

	/**
	 * Handles incoming POST messages.
	 * This should be called from your Remix action to handle incoming messages.
	 */
	async handlePostMessage(request: Request): Promise<Response> {
		if (!this.#stream) {
			const message = 'SSE connection not established'
			return new Response(message, { status: 500 })
		}

		let body: unknown
		try {
			const contentType = request.headers.get('content-type')
			if (contentType !== 'application/json') {
				throw new Error(`Unsupported content-type: ${contentType}`)
			}

			body = await request.json()
		} catch (error) {
			this.onerror?.(error as Error)
			return new Response(String(error), { status: 400 })
		}

		try {
			await this.handleMessage(body)
		} catch (error) {
			console.error(error)
			return new Response(`Invalid message: ${body}`, { status: 400 })
		}

		return new Response('Accepted', { status: 202 })
	}

	/**
	 * Handle a client message, regardless of how it arrived.
	 */
	async handleMessage(message: unknown): Promise<void> {
		let parsedMessage: JSONRPCMessage
		try {
			parsedMessage = JSONRPCMessageSchema.parse(message)
		} catch (error) {
			this.onerror?.(error as Error)
			throw error
		}

		this.onmessage?.(parsedMessage)
	}

	async close(): Promise<void> {
		this.#stream?.close()
		this.#stream = undefined
		this.onclose?.()
	}

	async send(message: JSONRPCMessage): Promise<void> {
		if (!this.#stream) {
			throw new Error('Not connected')
		}

		// Send the message through the event stream
		this.#stream.enqueue(`event: message\ndata: ${JSON.stringify(message)}\n\n`)
	}

	/**
	 * Returns the session ID for this transport.
	 * This can be used to route incoming POST requests.
	 */
	get sessionId(): string {
		return this.#sessionId
	}
}

Feel free to use this.

[kentcdodds]

Note: I'm using SSE here because all clients I can test this with don't support the new streaming protocol, but I would love to have the SDK have a transport that supports the streaming protocol with Request/Response as well :)

[rogervila]

Thank you @kentcdodds!

Could you provide an example using Hono?

[LuisMalhadas]

Thank you @kentcdodds!

Could you provide an example using Hono?

I had to make some changes to the @kentcdodds code:

import { type Transport } from '@modelcontextprotocol/sdk/shared/transport.js'
import {
	type JSONRPCMessage,
	JSONRPCMessageSchema,
} from '@modelcontextprotocol/sdk/types.js'

/**
 * Server transport for SSE using Standard Fetch: this will send messages over
 * an SSE connection and receive messages from HTTP POST requests.
 */
export class FetchSSEServerTransport implements Transport {
	stream?: ReadableStreamDefaultController<string>
	heartbeat?: ReturnType<typeof setInterval>
	_sessionId: string
	endpoint: string

	onclose?: () => void
	onerror?: (error: Error) => void
	onmessage?: (message: JSONRPCMessage) => void

	/**
	 * Creates a new SSE server transport, which will direct the client to POST
	 * messages to the relative or absolute URL identified by `endpoint`.
	 */
	constructor(endpoint: string, sessionId?: string | null) {
		this.endpoint = endpoint
		this._sessionId = sessionId ?? crypto.randomUUID()
	}

	/**
	 * Starts processing messages on the transport.
	 * This is called by the Server class and should not be called directly.
	 */
	async start(): Promise<void> {
		console.warn('[FetchSSEServerTransport] start() is a no-op; handled by handleSSERequest()')
		/*if (this.stream) {
			throw new Error(
				'FetchSSEServerTransport already started! If using Server class, note that connect() calls start() automatically.',
			)
		}*/
	}
	

	/**
	 * Handles the initial SSE connection request.
	 * This should be called from your Remix loader to establish the SSE stream.
	 */
	async handleSSERequest(request: Request): Promise<Response> {
		const stream = new ReadableStream<string>({
			start: (controller) => {
				this.stream = controller

				// Send headers
				controller.enqueue(`event: heartbeat\ndata: init\n\n`)
				controller.enqueue(`retry: 5000\n\n`)

				// Send the endpoint event
				controller.enqueue(
					`event: endpoint\ndata: ${encodeURI(
						this.endpoint,
					)}?sessionId=${this.sessionId}\n\n`,
				)

				// Start heartbeat
				this.heartbeat = setInterval(() => {
					try {
						this.stream?.enqueue(`event: heartbeat\ndata: ping\n\n`)
					} catch (err) {
						console.warn('Heartbeat enqueue failed:', err)
					}
				}, 10000)

				// Handle cleanup when the connection closes
				request.signal.addEventListener('abort', () => {
					console.log('SSE connection aborted by client')
					console.log('>> request.signal.aborted:', request.signal.aborted)
					clearInterval(this.heartbeat)
					controller.close()
					this.stream = undefined
					this.onclose?.()
				})
				
			},
			cancel: () => {
				console.log('SSE stream canceled')
				clearInterval(this.heartbeat)
				this.stream = undefined
				this.onclose?.()
			},
		})

		const encoder = new TextEncoderStream()
		stream.pipeTo(encoder.writable)
		return new Response(encoder.readable, {
			headers: {
				'Content-Type': 'text/event-stream',
				'Cache-Control': 'no-cache',
				'Connection': 'keep-alive',
				'Transfer-Encoding': 'chunked',
				'Mcp-Session-Id': this.sessionId,
			},
		})
	}

	/**
	 * Handles incoming POST messages.
	 * This should be called from your Remix action to handle incoming messages.
	 */
	async handlePostMessage(request: Request): Promise<Response> {
		if (!this.stream) {
			const message = 'SSE connection not established'
			return new Response(message, { status: 500 })
		}

		let body: unknown
		try {
			const contentType = request.headers.get('content-type')
			if (contentType !== 'application/json') {
				throw new Error(`Unsupported content-type: ${contentType}`)
			}

			body = await request.json()
		} catch (error) {
			this.onerror?.(error as Error)
			return new Response(String(error), { status: 400 })
		}

		try {
			await this.handleMessage(body)
		} catch (error) {
			console.error(error)
			return new Response(`Invalid message: ${body}`, { status: 400 })
		}

		return new Response('Accepted', { status: 202 })
	}

	/**
	 * Handle a client message, regardless of how it arrived.
	 */
	async handleMessage(message: unknown): Promise<void> {
		let parsedMessage: JSONRPCMessage
		try {
			parsedMessage = JSONRPCMessageSchema.parse(message)
		} catch (error) {
			this.onerror?.(error as Error)
			throw error
		}

		this.onmessage?.(parsedMessage)
	}

	async close(): Promise<void> {
		clearInterval(this.heartbeat)
		this.stream?.close()
		this.stream = undefined
		this.onclose?.()
	}

	async send(message: JSONRPCMessage): Promise<void> {
		if (!this.stream) {
			throw new Error('Not connected')
		}

		// Send the message through the event stream
		try {
			this.stream.enqueue(`event: message\ndata: ${JSON.stringify(message)}\n\n`)
		} catch (err) {
			this.onerror?.(new Error(`Failed to send message: ${err}`))
		}
		
	}

	/**
	 * Returns the session ID for this transport.
	 * This can be used to route incoming POST requests.
	 */
	get sessionId(): string {
		return this._sessionId
	}
}

The example using Hono:

import { McpServer } from '@modelcontextprotocol/sdk/server/mcp.js'
import { FetchSSEServerTransport } from './FetchSSEServerTransport'
import { Hono } from 'hono'
import { z } from 'zod'

const server = new McpServer({
    name: "accessibles-mcp-server",
    version: "1.0.0"
})
  
const app = new Hono()

server.tool(
    "easy2read",
    { text: z.string() },
    async ({ text }) => {
        const response = await fetch(process.env.ACCESSIBLES_API+'/easy2read', {
            method: 'POST',
            headers: {
                'Content-Type': 'application/json',
                'Authorization': process.env.ACCESSIBLES_API_KEY || '',
            },
            body: JSON.stringify({ 
                'text': text,
                'language': '',
                'streaming': false 
            })
        })

        if (!response.ok) {
            throw new Error(`API request failed with status ${response.status}`)
        }

        const data = await response.text()
        return {
            content: [
                {
                    type: "text",
                    text: data || "No result available"
                }
            ]
        }
    }
)
  
const transport: FetchSSEServerTransport = new FetchSSEServerTransport('/mcp')
  
// Setup routes for the server
const setupServer = async () => {
    await server.connect(transport)
}

// Handle the Server-Sent Events stream (client connects via GET)
app.get('/mcp/sse', async (c) => {
    return await transport.handleSSERequest(c.req.raw)
})

// Handle client messages sent to the server (via POST)
app.post('/mcp', async (c) => {
    try {
        return await transport.handlePostMessage(c.req.raw)
    } catch (error) {
        console.error('Error handling POST message:', error)
        return c.json({
            jsonrpc: '2.0',
            error: {
                code: -32603,
                message: 'Internal server error',
            },
            id: null,
        }, 500)
    }
})

app.delete('/mcp', async (c) => {
    console.log('Received DELETE MCP request')
    return c.json({
        jsonrpc: "2.0",
        error: {
            code: -32000,
            message: "Method not allowed."
        },
        id: null
    }, 405)
})

// Start the server
setupServer().then(() => {
    Bun.serve({
        port: process.env.PORT,
        fetch: app.fetch,
        idleTimeout: 0,
        error(err) {
            console.error('Server error:', err)
        },
    })
    console.log(`MCP Streamable HTTP Server listening on port ${process.env.PORT}`)
}).catch(error => {
    console.error('Failed to set up the server:', error)
    process.exit(1)
})

I still have an issue where every time i refresh the MCP Inspector v0.10.2 webapp it kills the mcp server...
Any insights are very welcomed! Everything is very experimental yet!

[neviaumi]

For who are interested to make the official transport work.

You may checkout https://github.com/mhart/fetch-to-node

This project has provide an adapter to convert Web Fetch Request to NodeJS Incoming HTTP

With this you can just call the official transport without implement your own.

Check my experimental implementation on Deno + Hono here.

[Hopsken]

For people who looking for an example of a stateless streamable http transport using web standard Fetch and Stream API.

Here is what I managed to create in my personal project.

src/server/webStreamableTransport.ts

The implementation handles security / header / protocol checks as recommended.

[TomasHubelbauer]

Using web APIs would be great as it would make the SDK compatible with Bun's built in server, Bun.serve.

[EzzatOmar]

For who are interested to make the official transport work.

You may checkout https://github.com/mhart/fetch-to-node

This project has provide an adapter to convert Web Fetch Request to NodeJS Incoming HTTP

With this you can just call the official transport without implement your own.

Check my experimental implementation on Deno + Hono here.

Do you face any limitations with this approach?

[mattzcarey]

We made this for the agents sdk also. Will try to push it upstream for v2

[KKonstantinov]

As mentioned, this is planned for #809 and a PoC implemented in #1209 .v2 will likely do web standards only and then map to it via plug and play middlewares.

Closing this for now as going through clean-up and identifying v2 issues.