diff --git a/server/connector-proxy.test.ts b/server/connector-proxy.test.ts index 60e06f5c8..d6e0fa10b 100644 --- a/server/connector-proxy.test.ts +++ b/server/connector-proxy.test.ts @@ -74,21 +74,130 @@ describe("connector MCP bridge", () => { expect(received.body.resumeKey).toMatch(/^[\w-]{8,100}$/); }); - it("relays ordinary MCP JSON-RPC without exposing upstream headers on stdout", async () => { + it("answers initialize locally so a missing or failing upstream cannot fail the MCP handshake", async () => { + const lines = start({}); + child!.stdin.write(`${JSON.stringify({ + jsonrpc: "2.0", + id: 1, + method: "initialize", + params: { protocolVersion: "2024-11-05" }, + })}\n`); + const reply = await nextJson(lines); + expect(reply).toEqual({ + jsonrpc: "2.0", + id: 1, + result: { + protocolVersion: "2024-11-05", + capabilities: { tools: {} }, + serverInfo: { name: "openmausbot-connectors", version: "1" }, + }, + }); + expect(reply.result).not.toHaveProperty("isError"); + expect(reply.result).not.toHaveProperty("content"); + }); + + it("answers initialize after a bounded wait when the upstream stalls", async () => { + let sawInitialize!: () => void; + const received = new Promise((resolve) => { sawInitialize = resolve; }); + const upstream = await listen((request) => { + request.resume(); + request.on("end", sawInitialize); + // Deliberately never respond. The proxy must abort this request and + // return its local capability result instead of hanging OpenCode. + }); + const lines = start({ OMB_CONNECTOR_UPSTREAM_URL: upstream }); + child!.stdin.write(`${JSON.stringify({ + jsonrpc: "2.0", + id: 11, + method: "initialize", + params: { protocolVersion: "2024-11-05" }, + })}\n`); + + await received; + const reply = await nextJson(lines); + expect(reply).toMatchObject({ + id: 11, + result: { + protocolVersion: "2024-11-05", + capabilities: { tools: {} }, + }, + }); + }); + + it("still opens the upstream MCP session on initialize without echoing secrets", async () => { + let upstreamAuthorization = ""; + let upstreamBody: any = null; + const methods: string[] = []; + const sessionHeaders: string[] = []; + let sawInitialized!: () => void; + const initialized = new Promise((resolve) => { sawInitialized = resolve; }); + const upstream = await listen((request, response) => { + let body = ""; + request.on("data", (chunk) => { body += chunk; }); + request.on("end", () => { + upstreamAuthorization = String(request.headers.authorization ?? ""); + upstreamBody = JSON.parse(body); + methods.push(String(upstreamBody.method ?? "")); + sessionHeaders.push(String(request.headers["mcp-session-id"] ?? "")); + response.writeHead(200, { "content-type": "application/json", "mcp-session-id": "transport-1" }); + response.end(upstreamBody.id === undefined + ? "" + : JSON.stringify({ jsonrpc: "2.0", id: 2, result: { protocolVersion: "2025-06-18" } })); + if (upstreamBody.method === "notifications/initialized") sawInitialized(); + }); + }); + const lines = start({ + OMB_CONNECTOR_UPSTREAM_URL: upstream, + OMB_CONNECTOR_UPSTREAM_HEADERS: JSON.stringify({ authorization: "Bearer upstream-secret" }), + }); + child!.stdin.write(`${JSON.stringify({ jsonrpc: "2.0", id: 2, method: "initialize", params: { protocolVersion: "2024-11-05" } })}\n`); + const reply = await nextJson(lines); + expect(reply.result.protocolVersion).toBe("2024-11-05"); + expect(reply.result.serverInfo).toEqual({ name: "openmausbot-connectors", version: "1" }); + expect(upstreamAuthorization).toBe("Bearer upstream-secret"); + expect(upstreamBody).toMatchObject({ method: "initialize" }); + expect(JSON.stringify(reply)).not.toContain("upstream-secret"); + + child!.stdin.write(`${JSON.stringify({ jsonrpc: "2.0", method: "notifications/initialized", params: {} })}\n`); + await initialized; + expect(methods).toEqual(["initialize", "notifications/initialized"]); + expect(sessionHeaders).toEqual(["", "transport-1"]); + }); + + it("relays tools/list without exposing upstream headers on stdout", async () => { let upstreamAuthorization = ""; const upstream = await listen((request, response) => { upstreamAuthorization = String(request.headers.authorization ?? ""); response.writeHead(200, { "content-type": "application/json", "mcp-session-id": "transport-1" }); - response.end(JSON.stringify({ jsonrpc: "2.0", id: 2, result: { protocolVersion: "2025-06-18" } })); + response.end(JSON.stringify({ + jsonrpc: "2.0", + id: 4, + result: { tools: [{ name: "COMPOSIO_SEARCH_TOOLS" }] }, + })); }); const lines = start({ OMB_CONNECTOR_UPSTREAM_URL: upstream, OMB_CONNECTOR_UPSTREAM_HEADERS: JSON.stringify({ authorization: "Bearer upstream-secret" }), }); - child!.stdin.write(`${JSON.stringify({ jsonrpc: "2.0", id: 2, method: "initialize", params: {} })}\n`); + child!.stdin.write(`${JSON.stringify({ jsonrpc: "2.0", id: 4, method: "tools/list", params: {} })}\n`); const reply = await nextJson(lines); - expect(reply).toEqual({ jsonrpc: "2.0", id: 2, result: { protocolVersion: "2025-06-18" } }); + expect(reply).toEqual({ + jsonrpc: "2.0", + id: 4, + result: { tools: [{ name: "COMPOSIO_SEARCH_TOOLS" }] }, + }); expect(upstreamAuthorization).toBe("Bearer upstream-secret"); expect(JSON.stringify(reply)).not.toContain("upstream-secret"); }); + + it("returns a JSON-RPC error, not a tools result, when a non-call relay fails", async () => { + const lines = start({}); + child!.stdin.write(`${JSON.stringify({ jsonrpc: "2.0", id: 3, method: "tools/list", params: {} })}\n`); + const reply = await nextJson(lines); + expect(reply).toEqual({ + jsonrpc: "2.0", + id: 3, + error: { code: -32000, message: "connected apps are unavailable" }, + }); + }); }); diff --git a/server/connector-proxy.ts b/server/connector-proxy.ts index b3ec0aa49..93c5d54c3 100644 --- a/server/connector-proxy.ts +++ b/server/connector-proxy.ts @@ -17,6 +17,8 @@ const BOT_ID = process.env.OMB_BOT_ID ?? ""; const THREAD_ID = process.env.OMB_THREAD_ID ?? ""; const TOKEN = process.env.OMB_COMMS_TOKEN ?? ""; const MAX_RESPONSE_BYTES = 20 * 1024 * 1024; +const INITIALIZE_RELAY_TIMEOUT_MS = 1_000; +const RELAY_TIMEOUT_MS = 10 * 60_000; function parsedHeaders(): Record { try { @@ -38,6 +40,22 @@ function textResult(id: unknown, text: string, isError = false): Json { return { jsonrpc: "2.0", id, result: { content: [{ type: "text", text }], ...(isError ? { isError: true } : {}) } }; } +function jsonRpcError(id: unknown, message: string): Json { + return { jsonrpc: "2.0", id, error: { code: -32000, message } }; +} + +function initializeResult(id: unknown, protocolVersion: unknown): Json { + return { + jsonrpc: "2.0", + id, + result: { + protocolVersion: typeof protocolVersion === "string" && protocolVersion ? protocolVersion : "2024-11-05", + capabilities: { tools: {} }, + serverInfo: { name: "openmausbot-connectors", version: "1" }, + }, + }; +} + async function readBounded(response: Response): Promise { const declared = Number(response.headers.get("content-length") ?? "0"); if (declared > MAX_RESPONSE_BYTES) throw new Error("connector response exceeded 20 MB"); @@ -78,7 +96,7 @@ function parseUpstream(text: string, id: unknown): Json | null { return frames.findLast((frame) => frame.id === id) ?? frames.at(-1) ?? null; } -async function relay(message: Json): Promise { +async function relay(message: Json, timeoutMs = RELAY_TIMEOUT_MS): Promise { if (!UPSTREAM) throw new Error("connected apps are unavailable"); const response = await fetch(UPSTREAM, { method: "POST", @@ -89,7 +107,7 @@ async function relay(message: Json): Promise { ...(upstreamSessionId ? { "mcp-session-id": upstreamSessionId } : {}), }, body: JSON.stringify(message), - signal: AbortSignal.timeout(10 * 60_000), + signal: AbortSignal.timeout(timeoutMs), }); const nextSession = response.headers.get("mcp-session-id"); if (nextSession) upstreamSessionId = nextSession; @@ -127,6 +145,33 @@ async function showConnectorCards(slugs: string[]): Promise { async function handle(message: Json): Promise { const id = message.id; const method = String(message.method ?? ""); + // OpenCode (and other MCP clients) mark a stdio server failed unless + // initialize returns capabilities/serverInfo. Relaying that handshake to + // Composio can time out, return a newer protocolVersion, or throw when the + // upstream URL never reached the child env — all of which previously + // surfaced as a tools/call-shaped {content,isError} payload. + if (method === "notifications/initialized" || method === "initialized") { + if (UPSTREAM) void relay(message).catch(() => {}); + return; + } + if (method === "initialize") { + if (UPSTREAM) { + try { + // Capture the upstream session id when the service is healthy, but + // never let a stalled provider prevent the local MCP client from + // mounting the connector tools. The client sends initialized only + // after this bounded attempt and the local initialize response. + await relay(message, INITIALIZE_RELAY_TIMEOUT_MS); + } catch { + // Best-effort session setup. The client still needs a valid result. + } + } + if (id !== undefined) { + const params = (message.params ?? {}) as Json; + send(initializeResult(id, params.protocolVersion)); + } + return; + } if (method === "tools/call") { const params = (message.params ?? {}) as Json; const name = String(params.name ?? ""); @@ -144,8 +189,15 @@ async function handle(message: Json): Promise { return; } } - const response = await relay(message); - if (response && id !== undefined) send(response); + try { + const response = await relay(message); + if (response && id !== undefined) send(response); + } catch (error) { + if (id === undefined) return; + const messageText = error instanceof Error ? error.message : String(error); + if (method === "tools/call") send(textResult(id, messageText, true)); + else send(jsonRpcError(id, messageText)); + } } const input = readline.createInterface({ input: process.stdin, terminal: false }); @@ -159,9 +211,11 @@ input.on("line", (line) => { return; } void handle(message).catch((error) => { - if (message.id !== undefined) { - send(textResult(message.id, error instanceof Error ? error.message : String(error), true)); - } + if (message.id === undefined) return; + const method = String(message.method ?? ""); + const messageText = error instanceof Error ? error.message : String(error); + if (method === "tools/call") send(textResult(message.id, messageText, true)); + else send(jsonRpcError(message.id, messageText)); }); }); input.on("close", () => process.exit(0));