Add stateless version of mcp

Signed-off-by: Marcos Candeia <[email protected]>

Author: mcandeia

deno.json

@@ -1,6 +1,6 @@
 {
   "name": "@deco/mcp",
-  "version": "0.2.0",
+  "version": "0.2.1",
   "exports": "./mod.ts",
   "tasks": {
     "check": "deno fmt && deno lint && deno check mod.ts"

mcp/server.ts

@@ -12,6 +12,7 @@ import {
   ListToolsRequestSchema,
 } from "@modelcontextprotocol/sdk/types.js";
 import { SSEServerTransport } from "./sse.ts";
+import { StatelessServerTransport } from "./stateless.ts";
 import { dereferenceSchema } from "./utils.ts";
 import { WebSocketServerTransport } from "./websocket.ts";
 
@@ -244,51 +245,52 @@ export function mcpServer<TManifest extends AppManifest>(
 
   return async (c: Context, next: Next) => {
     const path = new URL(c.req.url).pathname;
+    const basePath = options?.basePath ?? "";
 
+    // Handle WebSocket upgrade if requested
     if (
-      path === `${options?.basePath ?? ""}/mcp/ws` &&
+      path === `${basePath}/mcp/ws` &&
       c.req.raw.headers.get("upgrade") === "websocket"
     ) {
       const { response, socket } = Deno.upgradeWebSocket(c.req.raw);
-
       const transport = new WebSocketServerTransport();
-
       transport.acceptWebSocket(socket);
       mcp.server.connect(transport);
-
       return response;
     }
 
-    if (path === `${options?.basePath ?? ""}/mcp/sse`) {
+    // Legacy SSE endpoint for backwards compatibility
+    if (path === `${basePath}/mcp/sse`) {
       const transport = new SSEServerTransport(
-        `${options?.basePath ?? ""}${MESSAGES_ENDPOINT}`,
+        `${basePath}${MESSAGES_ENDPOINT}`,
       );
       transports.set(transport.sessionId, transport);
-
       transport.onclose = () => {
         transports.delete(transport.sessionId);
       };
-
       const response = transport.createSSEResponse();
       mcp.server.connect(transport);
-
       return response;
     }
 
+    // Main message endpoint - handles both stateless requests and SSE upgrades
     if (path === `${options?.basePath ?? ""}${MESSAGES_ENDPOINT}`) {
       const sessionId = c.req.query("sessionId");
-      if (!sessionId) {
-        return c.json({ error: "Missing sessionId" }, 400);
-      }
+      if (sessionId) {
+        const transport = transports.get(sessionId);
+        if (!transport) {
+          return c.json({ error: "Invalid session" }, 404);
+        }
 
-      const transport = transports.get(sessionId);
-      if (!transport) {
-        return c.json({ error: "Invalid session" }, 404);
+        return await transport.handlePostMessage(c.req.raw);
       }
-
-      return await transport.handlePostMessage(c.req.raw);
+      // For stateless transport
+      const transport = new StatelessServerTransport();
+      mcp.server.connect(transport);
+      const response = await transport.handleMessage(c.req.raw);
+      transport.close(); // Close the transport after handling the message
+      return response;
     }
-
     await next();
   };
 }

mcp/stateless-client.ts

@@ -0,0 +1,191 @@
+import {
+  type JSONRPCMessage,
+  JSONRPCMessageSchema,
+} from "@modelcontextprotocol/sdk/types.js";
+import type { Transport } from "@modelcontextprotocol/sdk/shared/transport.js";
+
+import {
+  auth,
+  type OAuthClientProvider,
+  UnauthorizedError,
+} from "@modelcontextprotocol/sdk/client/auth.js";
+/**
+ * Configuration options for the `StatelessClientTransport`.
+ */
+export type StatelessClientTransportOptions = {
+  /**
+   * An OAuth client provider to use for authentication.
+   */
+  authProvider?: OAuthClientProvider;
+
+  /**
+   * Customizes requests to the server.
+   */
+  requestInit?: RequestInit;
+};
+
+/**
+ * Client transport for Stateless HTTP: this will communicate with the server using HTTP requests
+ * and handle both immediate responses and streaming responses when needed.
+ */
+export class StatelessClientTransport implements Transport {
+  private _abortController?: AbortController;
+  private _eventSource?: EventSource;
+  private _authProvider?: OAuthClientProvider;
+  private _requestInit?: RequestInit;
+
+  onclose?: () => void;
+  onerror?: (error: Error) => void;
+  onmessage?: (message: JSONRPCMessage) => void;
+
+  constructor(
+    private _url: URL,
+    opts?: StatelessClientTransportOptions,
+  ) {
+    this._authProvider = opts?.authProvider;
+    this._requestInit = opts?.requestInit;
+  }
+
+  private async _commonHeaders(): Promise<HeadersInit> {
+    const headers: HeadersInit = {};
+    if (this._authProvider) {
+      const tokens = await this._authProvider.tokens();
+      if (tokens) {
+        headers["Authorization"] = `Bearer ${tokens.access_token}`;
+      }
+    }
+    return headers;
+  }
+
+  private async _handleStreamingResponse(response: Response) {
+    const reader = response.body?.getReader();
+    if (!reader) {
+      throw new Error("No response body");
+    }
+
+    const decoder = new TextDecoder();
+    let buffer = "";
+
+    try {
+      while (true) {
+        const { done, value } = await reader.read();
+        if (done) break;
+
+        buffer += decoder.decode(value, { stream: true });
+
+        // Process complete SSE messages
+        const messages = buffer.split("\n\n");
+        buffer = messages.pop() || ""; // Keep incomplete message in buffer
+
+        for (const message of messages) {
+          if (!message.trim()) continue;
+
+          const lines = message.split("\n");
+          const data = lines.find((line) => line.startsWith("data: "))?.slice(
+            6,
+          );
+
+          if (data) {
+            try {
+              const parsed = JSONRPCMessageSchema.parse(JSON.parse(data));
+              this.onmessage?.(parsed);
+            } catch (error) {
+              this.onerror?.(error as Error);
+            }
+          }
+        }
+      }
+    } finally {
+      reader.releaseLock();
+    }
+  }
+
+  private async _authThenSend(message: JSONRPCMessage): Promise<void> {
+    if (!this._authProvider) {
+      throw new UnauthorizedError("No auth provider");
+    }
+
+    const result = await auth(this._authProvider, { serverUrl: this._url });
+    if (result !== "AUTHORIZED") {
+      throw new UnauthorizedError();
+    }
+
+    await this.send(message);
+  }
+
+  start(): Promise<void> {
+    // No persistent connection needed
+    return Promise.resolve();
+  }
+
+  close(): Promise<void> {
+    this._abortController?.abort();
+    this._eventSource?.close();
+    this.onclose?.();
+    return Promise.resolve();
+  }
+
+  async send(message: JSONRPCMessage): Promise<void> {
+    try {
+      const commonHeaders = await this._commonHeaders();
+      const headers = new Headers({
+        ...commonHeaders,
+        ...this._requestInit?.headers,
+      });
+      headers.set("content-type", "application/json");
+
+      const init = {
+        ...this._requestInit,
+        method: "POST",
+        headers,
+        body: JSON.stringify(message),
+        signal: this._abortController?.signal,
+      };
+
+      const response = await fetch(this._url, init);
+
+      if (!response.ok) {
+        if (response.status === 401 && this._authProvider) {
+          await this._authThenSend(message);
+          return;
+        }
+
+        const text = await response.text().catch(() => null);
+        throw new Error(
+          `Error POSTing to endpoint (HTTP ${response.status}): ${text}`,
+        );
+      }
+
+      // Handle streaming responses
+      if (response.headers.get("content-type")?.includes("text/event-stream")) {
+        await this._handleStreamingResponse(response);
+        return;
+      }
+
+      // Handle immediate JSON responses
+      const responseData = await response.json();
+      const responseMessage = JSONRPCMessageSchema.parse(responseData);
+      this.onmessage?.(responseMessage);
+    } catch (error) {
+      this.onerror?.(error as Error);
+      throw error;
+    }
+  }
+
+  /**
+   * Call this method after the user has finished authorizing via their user agent and is redirected back to the MCP client application.
+   */
+  async finishAuth(authorizationCode: string): Promise<void> {
+    if (!this._authProvider) {
+      throw new UnauthorizedError("No auth provider");
+    }
+
+    const result = await auth(this._authProvider, {
+      serverUrl: this._url,
+      authorizationCode,
+    });
+    if (result !== "AUTHORIZED") {
+      throw new UnauthorizedError("Failed to authorize");
+    }
+  }
+}

mcp/stateless.ts

@@ -0,0 +1,119 @@
+import type { Transport } from "@modelcontextprotocol/sdk/shared/transport.js";
+import {
+  type JSONRPCMessage,
+  JSONRPCMessageSchema,
+} from "@modelcontextprotocol/sdk/types.js";
+import { ServerSentEventStream } from "@std/http";
+
+/**
+ * Server transport for Stateless HTTP: this will handle messages over plain HTTP requests
+ * with optional SSE upgrade for streaming responses.
+ */
+export class StatelessServerTransport implements Transport {
+  private _controller?: ReadableStreamDefaultController;
+  private _responseResolver?: (response: Response) => void;
+
+  onclose?: () => void;
+  onerror?: (error: Error) => void;
+  onmessage?: (message: JSONRPCMessage) => void;
+
+  /**
+   * Creates an SSE response for streaming
+   */
+  private createStreamingResponse(): Response {
+    const stream = new ReadableStream({
+      start: (controller) => {
+        this._controller = controller;
+      },
+      cancel: () => {
+        this._controller = undefined;
+        this.onclose?.();
+      },
+    }).pipeThrough(new ServerSentEventStream());
+
+    return new Response(stream, {
+      headers: {
+        "Content-Type": "text/event-stream",
+        "Cache-Control": "no-cache",
+        "Connection": "keep-alive",
+        "Access-Control-Allow-Origin": "*",
+      },
+    });
+  }
+
+  /**
+   * Handles incoming HTTP messages
+   */
+  async handleMessage(request: Request): Promise<Response> {
+    try {
+      const contentTypeHeader = request.headers.get("content-type");
+      if (!contentTypeHeader?.includes("application/json")) {
+        throw new Error("Unsupported content-type: Expected application/json");
+      }
+
+      const body = await request.json();
+      const message = JSONRPCMessageSchema.parse(body);
+
+      // Handle the message
+      this.onmessage?.(message);
+
+      // Create a promise that will be resolved with the response
+      const { promise, resolve } = Promise.withResolvers<Response>();
+      this._responseResolver = resolve;
+
+      // For requests that need streaming responses, immediately return SSE stream
+      if (this.shouldUpgradeToStreaming(message)) {
+        resolve(this.createStreamingResponse());
+      }
+
+      return promise;
+    } catch (error) {
+      this.onerror?.(error as Error);
+      return new Response(String(error), { status: 400 });
+    }
+  }
+
+  /**
+   * Determines if a request should be upgraded to streaming based on message type
+   */
+  private shouldUpgradeToStreaming(message: JSONRPCMessage): boolean {
+    console.log({ message });
+    // Implement logic to determine if streaming is needed
+    // For example, based on method name or parameters
+    return false; // Default to non-streaming
+  }
+
+  start(): Promise<void> {
+    return Promise.resolve();
+  }
+
+  close(): Promise<void> {
+    this._controller?.close();
+    this._controller = undefined;
+    this.onclose?.();
+    return Promise.resolve();
+  }
+
+  send(message: JSONRPCMessage): Promise<void> {
+    // If we have a controller, this is a streaming response
+    if (this._controller) {
+      this._controller.enqueue({
+        event: "message",
+        data: JSON.stringify(message),
+        id: Date.now().toString(),
+      });
+    } else if (this._responseResolver) {
+      // For non-streaming responses, resolve with a single JSON response
+      this._responseResolver(
+        new Response(JSON.stringify(message), {
+          headers: {
+            "Content-Type": "application/json",
+          },
+        }),
+      );
+      this._responseResolver = undefined;
+    }
+
+    return Promise.resolve();
+  }
+}