fix: bootstrapping bug with streamable http transport

Author: felixnorden

examples/email-mcp/src/index.ts

@@ -11,16 +11,21 @@ export default {
     _ctx: ExecutionContext,
   ): Promise<Response> {
     const url = new URL(request.url);
-    let sessionIdStr = url.searchParams.get("sessionId");
 
-    const id = sessionIdStr
-      ? env.EMAIL_MCP_SERVER.idFromString(sessionIdStr)
-      : env.EMAIL_MCP_SERVER.idFromName("default-email-session");
+    // Dynamically generate sessionId if it isn't provided to allocate a session
+    const sessionId =
+      request.headers.get("mcp-session-id") ?? crypto.randomUUID();
 
-    url.searchParams.set("sessionId", id.toString());
+    const reqClone = request.clone();
+
+    const json = await request.json();
+
+    console.log("Request:", { headers: request.headers, json });
+
+    const id = env.EMAIL_MCP_SERVER.idFromName(sessionId);
 
     return env.EMAIL_MCP_SERVER.get(id).fetch(
-      new Request(url.toString(), request.clone()),
+      new Request(url.toString(), reqClone),
     );
   },
 };

examples/email-mcp/test/email-mcp-client.test.ts

@@ -69,6 +69,7 @@ describe("Email MCP Client Integration Tests", () => {
 
     const transport = createTransport(ctx);
     await client.connect(transport);
+    console.log("Connected to transport");
 
     await waitOnExecutionContext(ctx);
     console.log(`Client connection test passed!`);

examples/email-mcp/wrangler.jsonc

@@ -1,33 +1,29 @@
 {
-  "$schema": "https://raw.githubusercontent.com/cloudflare/workers-sdk/main/packages/wrangler/config-schema.json",
+  "$schema": "./node_modules/wrangler/config-schema.json",
   "name": "email-mcp",
   "main": "src/index.ts",
   "compatibility_date": "2025-03-10",
   "account_id": "7541b433fbcea05edcdd0b86dbdd41fc",
+  "compatibility_flags": ["nodejs_compat"],
 
   "durable_objects": {
-    "bindings": [
-      { "name": "EMAIL_MCP_SERVER", "class_name": "EmailMcpServer" }
-    ]
+    "bindings": [{ "name": "EMAIL_MCP_SERVER", "class_name": "EmailMcpServer" }]
   },
 
   // Register the Durable Object class with SQLite support
-  "migrations": [
-    { "tag": "v1", "new_sqlite_classes": ["EmailMcpServer"] }
-  ],
+  "migrations": [{ "tag": "v1", "new_sqlite_classes": ["EmailMcpServer"] }],
 
   // Cloudflare Email Workers "Send Email" binding
-  "send_email": [
-    { "name": "SEND_EMAIL" }
-  ],
+  "send_email": [{ "name": "SEND_EMAIL" }],
 
   "vars": {
     "MAIL_FROM": "no-reply@nullshot.ai",
     "ALLOWED_RECIPIENTS": "raymond@xavalabs.com,@nullshot.ai"
   },
 
-  // logs/traces 
+  // logs/traces
   "observability": {
     "enabled": true
   }
-}
+}
+

packages/mcp/AGENTS.md

@@ -0,0 +1,31 @@
+# AGENTS.md - MCP Package
+
+## Build/Run Commands
+- Build: `pnpm run build` (uses tsc)
+- Dev mode: `pnpm run dev` (wrangler dev)
+- Run tests: `pnpm run test` (vitest)
+- Run single test: `pnpm run test src/mcp/sse-transport.test.ts`
+- Typecheck: `pnpm run type-check` (tsc --noEmit)
+- Deploy: `pnpm run deploy` (wrangler deploy)
+
+## Code Style Guidelines
+- Use double quotes for strings, no semicolons
+- Use tabs for indentation, strict TypeScript
+- No unused locals/parameters, exact optional property types
+- Use ES2021 target and ES2022 modules for Cloudflare Workers
+- Import paths use relative paths for local imports
+
+## Naming Conventions
+- Files: kebab-case, Types: PascalCase
+- Functions/variables: camelCase, Constants: UPPER_SNAKE_CASE
+- Test files: *.test.ts, Tools/Resources: snake_case
+
+## Error Handling & Testing
+- Use try/catch for async operations with meaningful messages
+- Use vitest with Cloudflare Workers pool, tests alongside source files
+- Tests run with isolated storage disabled for Durable Objects
+
+## Development Practices
+- Follow MCP patterns from ../../.cursor/rules/mcp-development.mdc
+- Extend McpHonoServerDO, use Zod schemas with .describe()
+- Always call super.setupRoutes() when overriding setupRoutes()

packages/mcp/src/mcp/hono-server.ts

@@ -1,5 +1,6 @@
 import { Hono } from 'hono';
 import { McpServerDO, SSE_MESSAGE_ENDPOINT, WEBSOCKET_ENDPOINT, MCP_SUBPROTOCOL } from './server';
+import { cors } from 'hono/cors';
 
 // Support both Cloudflare and Hono environments
 export abstract class McpHonoServerDO<Env extends Record<string, any> = Record<string, any>> extends McpServerDO<Env> {
@@ -8,6 +9,13 @@ export abstract class McpHonoServerDO<Env extends Record<string, any> = Record<s
 	public constructor(ctx: DurableObjectState, env: Env) {
 		super(ctx, env);
 		this.app = new Hono<{ Bindings: Env }>();
+		this.app.use(
+			cors({
+				origin: '*',
+				exposeHeaders: ['mcp-session-id', 'mcp-protocol-version'],
+				allowHeaders: ['Content-Type', 'mcp-session-id', 'mcp-protocol-version'],
+			}),
+		);
 		this.setupRoutes(this.app);
 	}
 

packages/mcp/src/mcp/server.ts

@@ -21,7 +21,7 @@ interface WebSocketAttachment {
 
 /**
  * McpDurableServer is a Durable Object implementation of an MCP server.
- * It supports SSE connections for event streaming and WebSocket connections with hibernation.
+ * It supports all transport connections for event streaming and WebSocket connections with hibernation.
  */
 export abstract class McpServerDO<Env = unknown> extends DurableObject<Env> {
 	private server: IMcpServer;

packages/test-utils/src/mcp/WorkerStreamableHTTPClientTransport.ts

@@ -16,12 +16,15 @@ export class WorkerStreamableHTTPClientTransport extends StreamableHTTPClientTra
       fetchUrl: RequestInfo | URL,
       fetchInit: RequestInit = {},
     ) => {
-      console.log(`[Debug] Fetching from: ${fetchUrl}`);
+      console.log(
+        `[Debug] Fetching from: ${fetchUrl}`,
+        JSON.stringify(fetchInit, null, 2),
+      );
       // add auth headers
       const workerOptions = {
         ...fetchInit,
         headers: {
-          ...fetchInit?.headers,
+          ...fetchInit.headers,
           "Content-Type": "application/json",
           Accept: "application/json, text/event-stream",
         },
@@ -32,58 +35,28 @@ export class WorkerStreamableHTTPClientTransport extends StreamableHTTPClientTra
       const request = new Request(fetchUrl.toString(), workerOptions);
 
       // Pass the Request object to the worker.fetch method
-      return await SELF.fetch(request);
+      const response = await SELF.fetch(request);
+      const resClone = response.clone();
+      console.log("Response:", {
+        headers: response.headers,
+        json: await response.json(),
+      });
+      return resClone;
     };
 
     // Initialize the parent StreamableHTTPClientTransport with our custom fetch
     super(url, { fetch: fetchOverride });
     this.ctx = ctx;
   }
 
-  /**
-   * Override the send method to direct requests to our worker
-   */
-  async send(message: JSONRPCMessage): Promise<void> {
-    console.log(
-      `[Debug] Sending message to worker: ${JSON.stringify(message)}`,
-    );
-    // Call the internal method to get the endpoint
-    // @ts-ignore
-    const endpoint = this._url;
-
-    if (!endpoint) {
-      throw new Error("Not connected");
-    }
-
-    try {
-      // Set up headers - we would normally get these from _commonHeaders
-      // but we can't access it due to it being private
-      const headers = new Headers();
-      headers.set("content-type", "application/json");
-      headers.set("accept", "application/json, text/event-stream");
-
-      const init = {
-        method: "POST",
-        headers,
-        body: JSON.stringify(message),
-      };
-
-      console.log(`Sending message to worker: ${JSON.stringify(message)}`);
-
-      // Use our worker fetch instead of regular fetch
-      const request = new Request(endpoint.toString(), init);
-
-      const response = await SELF.fetch(request);
-
-      if (!response.ok) {
-        const text = await response.text().catch(() => null);
-        throw new Error(
-          `Error POSTing to endpoint (HTTP ${response.status}): ${text}`,
-        );
-      }
-    } catch (error) {
-      this.onerror?.(error as Error);
-      throw error;
-    }
+  async send(
+    message: JSONRPCMessage | JSONRPCMessage[],
+    options?: {
+      resumptionToken?: string;
+      onresumptiontoken?: (token: string) => void;
+    },
+  ): Promise<void> {
+    console.log("Session:", { sessionId: this.sessionId });
+    await super.send(message, options);
   }
 }