Skip to content

fix(core): tolerate optional MCP HTTP event stream failures #24

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
stanley2058 merged 1 commit into from
Aug 1, 2026
Merged

fix(core): tolerate optional MCP HTTP event stream failures #24

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
46 changes: 44 additions & 2 deletions apps/core/src/mcp/registry.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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(),
});
Comment on lines +38 to +43

@coderabbitai coderabbitai Bot Aug 1, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🩺 Stability & Availability | 🟠 Major | ⚡ Quick win

🧩 Analysis chain

🌐 Web query:

Does @ai-sdk/mcp MCPClientError expose a request phase or kind field to distinguish GET SSE stream errors from POST request errors, or only statusCode/url/responseBody?

💡 Result:

No, the @ai-sdk/mcp MCPClientError class does not expose a "request phase kind" field to explicitly distinguish between GET (SSE stream) and POST request errors [1][2]. As of the current implementation, the MCPClientError class exposes the following properties [1][2]: - data: Optional data payload from the error (e.g., from the server's JSON-RPC error response) [1][3]. - code: Optional JSON-RPC error code [1]. - statusCode: Optional HTTP status code, if the error originated from an HTTP transport failure [1][4]. - url: Optional URL of the endpoint that failed [1][4]. - responseBody: Optional body of the failed HTTP response [1][4]. These fields (specifically statusCode, url, and responseBody) were added to provide structured HTTP context for HTTP transport failures, allowing consumers to make decisions based on the HTTP status without needing to parse error message strings [4][5]. However, the class does not include a specific field designed to identify the transport phase (e.g., distinguishing an initial SSE connection/GET request from a tool invocation/POST request) [1][2].

Citations:

Pin the GET-SSE message match to avoid silent behavior change.

@ai-sdk/mcp does not expose a stable request-kind discriminator for MCPClientError; statusCode, url, and responseBody do not distinguish GET-SSE stream failures from POST failures. Add an acceptance test for the exact message.startsWith("MCP HTTP Transport Error: GET SSE failed:") string so a library text change fails CI instead of silently changing whether failures are tolerated.

🤖 Prompt for AI Agents 
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@apps/core/src/mcp/registry.ts` around lines 38 - 43, Add an acceptance test
covering the schema represented by optionalHttpInboundSseErrorSchema, asserting
MCPClientError messages must start with the exact “MCP HTTP Transport Error: GET
SSE failed:” prefix and that a changed prefix is rejected. Keep the existing
statusCode and url validation unchanged.


type RegistryEntry = {
readonly definition: McpServerDefinition;
Expand Down Expand Up @@ -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;
Expand Down Expand Up @@ -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(
Expand All @@ -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);
},
Expand Down
172 changes: 171 additions & 1 deletion apps/core/tests/mcp/registry.test.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,13 +1,15 @@
import { describe, expect, it } from "bun:test";

import {
createMCPClient,
UnauthorizedError,
type MCPClientConfig,
type MCPTransport,
type OAuthClientProvider,
type OAuthTokens,
} from "@ai-sdk/mcp";
import { Experimental_StdioMCPTransport } from "@ai-sdk/mcp/mcp-stdio";
import { z } from "zod";

import {
McpConfigError,
Expand All @@ -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,
Expand Down Expand Up @@ -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<unknown>();
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<void>();
const releaseSessionGet = deferred<void>();
const sessionExpiredError = deferred<unknown>();
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;
Expand Down