diff --git a/apps/core/src/mcp/registry.ts b/apps/core/src/mcp/registry.ts index d495ad16..db663a83 100644 --- a/apps/core/src/mcp/registry.ts +++ b/apps/core/src/mcp/registry.ts @@ -35,6 +35,12 @@ const mcpApplicationErrorSchema = z.object({ name: z.literal("MCPClientError"), code: z.number(), }); +const optionalHttpInboundSseErrorSchema = z.object({ + name: z.literal("MCPClientError"), + message: z.string().startsWith("MCP HTTP Transport Error: GET SSE failed:"), + statusCode: z.number().int(), + url: z.string(), +}); type RegistryEntry = { readonly definition: McpServerDefinition; @@ -200,6 +206,31 @@ function isCustomTransport(transport: MCPClientConfig["transport"]): transport i ); } +function observeHttpSessionExpiration( + transport: MCPClientConfig["transport"], + onSessionExpired: () => void, +): MCPClientConfig["transport"] { + if (isCustomTransport(transport) || transport.type !== "http") return transport; + const sdkOnSessionExpired = transport.onSessionExpired; + return { + ...transport, + onSessionExpired: (sessionId) => { + sdkOnSessionExpired?.(sessionId); + onSessionExpired(); + }, + }; +} + +function isOptionalHttpInboundSseError( + definition: McpServerDefinition, + sessionExpired: boolean, + error: unknown, +): boolean { + if (definition.transportConfig.transport !== "http" || sessionExpired) return false; + const parsed = optionalHttpInboundSseErrorSchema.safeParse(error); + return parsed.success && parsed.data.url === new URL(definition.transportConfig.url).href; +} + export class McpRegistry implements McpRegistryApi { private readonly configPath: string; private readonly initDeadlineMs: number; @@ -509,12 +540,18 @@ export class McpRegistry implements McpRegistryApi { let phase: McpRegistryPhase = "configuration"; let sensitiveValues: readonly string[] = []; let client: McpRegistryClient | undefined; - const holder: { client?: McpRegistryClient; terminalError?: unknown } = {}; + const holder: { + client?: McpRegistryClient; + terminalError?: unknown; + sessionExpired?: boolean; + } = {}; try { const resolved = prefetchedTransport ?? (await this.resolveTransport(definition)); sensitiveValues = resolved.sensitiveValues; - const transport = this.createTransport(resolved.input); + const transport = observeHttpSessionExpiration(this.createTransport(resolved.input), () => { + holder.sessionExpired = true; + }); phase = "connection"; const candidate = await this.withDeadline( @@ -525,6 +562,11 @@ export class McpRegistry implements McpRegistryApi { clientName: `lilac-mcp-${definition.id}`, maxRetries: 0, onUncaughtError: (error) => { + if ( + isOptionalHttpInboundSseError(definition, holder.sessionExpired === true, error) + ) { + return; + } holder.terminalError = error; if (holder.client) this.handleTerminalFailure(definition.id, holder.client, error); }, diff --git a/apps/core/tests/mcp/registry.test.ts b/apps/core/tests/mcp/registry.test.ts index aae6ad4c..a1a56eb5 100644 --- a/apps/core/tests/mcp/registry.test.ts +++ b/apps/core/tests/mcp/registry.test.ts @@ -1,6 +1,7 @@ import { describe, expect, it } from "bun:test"; import { + createMCPClient, UnauthorizedError, type MCPClientConfig, type MCPTransport, @@ -8,6 +9,7 @@ import { type OAuthTokens, } from "@ai-sdk/mcp"; import { Experimental_StdioMCPTransport } from "@ai-sdk/mcp/mcp-stdio"; +import { z } from "zod"; import { McpConfigError, @@ -27,6 +29,12 @@ import { stdioDefinition, } from "./fixtures/registry-fixture"; +const mcpHttpRequestSchema = z.object({ + jsonrpc: z.literal("2.0"), + id: z.union([z.string(), z.number()]).optional(), + method: z.string(), +}); + function fakeAuthProvider(tokens: OAuthTokens | undefined): OAuthClientProvider { return { tokens: async () => tokens, @@ -155,14 +163,176 @@ describe("McpRegistry startup and discovery", () => { const localConfig = factory.configs.find((value) => value.clientName === "lilac-mcp-local"); const remoteConfig = factory.configs.find((value) => value.clientName === "lilac-mcp-remote"); expect(localConfig?.transport).toBeInstanceOf(Experimental_StdioMCPTransport); - expect(remoteConfig?.transport).toEqual({ + expect(remoteConfig?.transport).toMatchObject({ type: "http", url: "https://example.invalid/mcp", headers: { Authorization: "Bearer http-secret" }, + onSessionExpired: expect.any(Function), }); await registry.shutdown(); }); + it("keeps native HTTP available when the optional inbound SSE stream is unavailable", async () => { + const inboundError = deferred(); + const server = Bun.serve({ + hostname: "127.0.0.1", + port: 0, + async fetch(request) { + if (request.method === "GET") { + return new Response("Session not found", { status: 404 }); + } + + const message = mcpHttpRequestSchema.parse(await request.json()); + if (message.method === "initialize" && message.id !== undefined) { + return Response.json({ + jsonrpc: "2.0", + id: message.id, + result: { + protocolVersion: "2025-11-25", + capabilities: { tools: {} }, + serverInfo: { name: "native-http-test", version: "1.0.0" }, + }, + }); + } + if (message.method === "tools/list" && message.id !== undefined) { + return Response.json({ + jsonrpc: "2.0", + id: message.id, + result: { + tools: [{ name: "native-http-tool", inputSchema: { type: "object" } }], + }, + }); + } + return new Response(null, { status: 202 }); + }, + }); + const registry = new McpRegistry({ + configPath: "/data/mcp-config.yaml", + dependencies: { + readConfig: async () => + configSnapshot(mcpConfig([httpDefinition("native-http", server.url.toString())])), + createClient: async (config) => + createMCPClient({ + ...config, + onUncaughtError: (error) => { + inboundError.resolve(error); + config.onUncaughtError?.(error); + }, + }), + }, + }); + + try { + const initializing = registry.init(); + await expect(inboundError.promise).resolves.toMatchObject({ + name: "MCPClientError", + message: "MCP HTTP Transport Error: GET SSE failed: 404 Not Found", + statusCode: 404, + url: server.url.toString(), + }); + await initializing; + + expect(registry.list()).toEqual([ + { + serverId: "native-http", + transport: "http", + status: "available", + toolCount: 1, + }, + ]); + expect(registry.getTools().map((tool) => tool.rawName)).toEqual(["native-http-tool"]); + } finally { + await registry.shutdown(); + server.stop(true); + } + }); + + it("retires native HTTP when an established session expires on the inbound stream", async () => { + const sessionGetStarted = deferred(); + const releaseSessionGet = deferred(); + const sessionExpiredError = deferred(); + const sessionId = "native-http-session"; + const server = Bun.serve({ + hostname: "127.0.0.1", + port: 0, + async fetch(request) { + if (request.method === "GET") { + if (request.headers.get("mcp-session-id") !== sessionId) { + return new Response(null, { status: 405 }); + } + sessionGetStarted.resolve(); + await releaseSessionGet.promise; + return new Response("Session not found", { status: 404 }); + } + + const message = mcpHttpRequestSchema.parse(await request.json()); + if (message.method === "initialize" && message.id !== undefined) { + return Response.json( + { + jsonrpc: "2.0", + id: message.id, + result: { + protocolVersion: "2025-11-25", + capabilities: { tools: {} }, + serverInfo: { name: "native-http-session-test", version: "1.0.0" }, + }, + }, + { headers: { "mcp-session-id": sessionId } }, + ); + } + if (message.method === "tools/list" && message.id !== undefined) { + return Response.json({ + jsonrpc: "2.0", + id: message.id, + result: { + tools: [{ name: "stateful-http-tool", inputSchema: { type: "object" } }], + }, + }); + } + return new Response(null, { status: 202 }); + }, + }); + const registry = new McpRegistry({ + configPath: "/data/mcp-config.yaml", + dependencies: { + readConfig: async () => + configSnapshot(mcpConfig([httpDefinition("stateful-http", server.url.toString())])), + createClient: async (config) => + createMCPClient({ + ...config, + onUncaughtError: (error) => { + sessionExpiredError.resolve(error); + config.onUncaughtError?.(error); + }, + }), + }, + }); + + try { + await registry.init(); + await sessionGetStarted.promise; + expect(registry.list()[0]).toMatchObject({ status: "available", toolCount: 1 }); + + releaseSessionGet.resolve(); + await expect(sessionExpiredError.promise).resolves.toMatchObject({ + name: "MCPClientError", + message: "MCP HTTP Transport Error: GET SSE failed: 404 Not Found", + statusCode: 404, + url: server.url.toString(), + }); + expect(registry.list()[0]).toMatchObject({ + serverId: "stateful-http", + status: "unavailable", + phase: "runtime", + }); + expect(registry.getTools()).toEqual([]); + } finally { + releaseSessionGet.resolve(); + await registry.shutdown(); + server.stop(true); + } + }); + it("injects resolved transport and auth-provider dependencies", async () => { const transportInputs: McpRegistryTransportInput[] = []; let authProviderCalls = 0;