fix(composio): answer MCP initialize locally so OpenCode can mount tools

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<void>((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<void>((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" },
+    });
+  });
 });

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<string, string> {
   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<string> {
   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<Json | null> {
+async function relay(message: Json, timeoutMs = RELAY_TIMEOUT_MS): Promise<Json | null> {
   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<Json | null> {
       ...(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<void> {
 async function handle(message: Json): Promise<void> {
   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<void> {
       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));