Parse SSE responses in the stdio bridge

scripts/bridge-serialization-test.mjs

@@ -12,6 +12,8 @@ const root = resolve(dirname(fileURLToPath(import.meta.url)), "..");
 await testSerializedForwarding();
 await testBridgeGeneratedErrorPreservesId();
 await testToolCallIsNeverReplayedAfterServerError();
+await testSseResponseEmitsEachEventAsJsonLine();
+await testSseStreamWithoutResponseFailsClosed();
 
 async function testSerializedForwarding() {
   let requests = 0;
@@ -128,6 +130,58 @@ async function testToolCallIsNeverReplayedAfterServerError() {
   await server.close();
 }
 
+async function testSseResponseEmitsEachEventAsJsonLine() {
+  const server = await startServer((req, res) => {
+    collectBody(req, () => {
+      res.writeHead(200, { "Content-Type": "text/event-stream" });
+      res.write("event: message\r\n");
+      res.write('data: {"jsonrpc":"2.0","method":"notifications/progress","params":{"progress":1}}\r\n\r\n');
+      res.write(": keep-alive\n\n");
+      res.write("event: message\n");
+      res.write('data: {"jsonrpc":"2.0","result":{"ok":true},"id":7}\n\n');
+      res.end();
+    });
+  });
+
+  const bridge = spawnBridge(server.url);
+  bridge.write({ jsonrpc: "2.0", method: "tools/call", id: 7 });
+
+  const lines = await bridge.waitForLines(2);
+  const [notification, response] = lines.map((line) => JSON.parse(line));
+
+  assert.equal(notification.method, "notifications/progress");
+  assert.equal(notification.id, undefined);
+  assert.deepEqual(response, { jsonrpc: "2.0", result: { ok: true }, id: 7 });
+
+  bridge.end();
+  await bridge.waitForExit();
+  await server.close();
+}
+
+async function testSseStreamWithoutResponseFailsClosed() {
+  const server = await startServer((req, res) => {
+    collectBody(req, () => {
+      res.writeHead(200, { "Content-Type": "text/event-stream" });
+      res.write('data: {"jsonrpc":"2.0","method":"notifications/progress","params":{}}\n\n');
+      res.end();
+    });
+  });
+
+  const bridge = spawnBridge(server.url);
+  bridge.write({ jsonrpc: "2.0", method: "tools/call", id: "sse-orphan" });
+
+  const lines = await bridge.waitForLines(2);
+  const response = JSON.parse(lines[1]);
+
+  assert.equal(response.id, "sse-orphan");
+  assert.equal(response.error.code, -32000);
+  assert.match(response.error.message, /stream ended/i);
+
+  bridge.end();
+  await bridge.waitForExit();
+  await server.close();
+}
+
 async function startServer(handler) {
   const server = http.createServer(handler);
 

src/index.ts

@@ -15,7 +15,9 @@ let mcpSessionId: string | undefined;
 
 type JsonRpcId = string | number | null;
 
-async function forward(line: string): Promise<string> {
+type EmitMessage = (message: string) => Promise<void>;
+
+async function forward(line: string, emit: EmitMessage): Promise<void> {
   const requestId = extractRequestId(line);
   try {
     const headers: Record<string, string> = {
@@ -49,26 +51,120 @@ async function forward(line: string): Promise<string> {
 
     if (res.status === 202) {
       // JSON-RPC notification — no response body
-      return "";
+      return;
     }
 
     if (res.ok) {
-      return await res.text();
+      const contentType = res.headers.get("Content-Type") || "";
+      if (contentType.includes("text/event-stream")) {
+        await forwardEventStream(res, requestId, emit);
+        return;
+      }
+
+      const text = await res.text();
+      if (text) {
+        await emit(text);
+      }
+      return;
     }
 
     if (res.status === 401 || res.status === 403) {
       const authHint = mode === "bearer"
         ? "Check your NITROSEND_BEARER_TOKEN (may be expired — re-authenticate via OAuth)"
         : "Check your NITROSEND_API_KEY";
       console.error(`Auth error (${res.status}): ${authHint}`);
-      return jsonRpcError(-32000, `Authentication failed (${res.status})`, requestId);
+      await emit(jsonRpcError(-32000, `Authentication failed (${res.status})`, requestId));
+      return;
     }
 
-    return jsonRpcError(-32000, `API returned ${res.status}`, requestId);
+    await emit(jsonRpcError(-32000, `API returned ${res.status}`, requestId));
   } catch (err) {
     const message = (err as Error).message || "Unknown network error";
     console.error(`Network error: ${message}`);
-    return jsonRpcError(-32000, `Network error: ${message}`, requestId);
+    await emit(jsonRpcError(-32000, `Network error: ${message}`, requestId));
+  }
+}
+
+// A Streamable HTTP server may answer a POST with an SSE stream that carries
+// related notifications before the final response. Each event's data is one
+// JSON-RPC message; re-serialize it so stdout stays one message per line no
+// matter how the server framed the event.
+async function forwardEventStream(
+  res: Response,
+  requestId: JsonRpcId,
+  emit: EmitMessage
+): Promise<void> {
+  let respondedToRequest = requestId === null;
+  let dataLines: string[] = [];
+
+  const dispatch = async (): Promise<void> => {
+    if (dataLines.length === 0) return;
+    const data = dataLines.join("\n");
+    dataLines = [];
+
+    let message: unknown;
+    try {
+      message = JSON.parse(data);
+    } catch {
+      console.error("Ignoring non-JSON event stream data");
+      return;
+    }
+
+    if (
+      message !== null &&
+      typeof message === "object" &&
+      (message as { id?: unknown }).id === requestId
+    ) {
+      respondedToRequest = true;
+    }
+
+    await emit(JSON.stringify(message));
+  };
+
+  const handleLine = async (rawLine: string): Promise<void> => {
+    const line = rawLine.endsWith("\r") ? rawLine.slice(0, -1) : rawLine;
+    if (line === "") {
+      await dispatch();
+      return;
+    }
+    if (line.startsWith("data:")) {
+      dataLines.push(line.startsWith("data: ") ? line.slice(6) : line.slice(5));
+    }
+    // event:, id:, retry: and comment lines carry no JSON-RPC payload
+  };
+
+  try {
+    if (res.body) {
+      const decoder = new TextDecoder();
+      let buffered = "";
+
+      for await (const chunk of res.body) {
+        buffered += decoder.decode(chunk as Uint8Array, { stream: true });
+
+        let newline: number;
+        while ((newline = buffered.indexOf("\n")) !== -1) {
+          const rawLine = buffered.slice(0, newline);
+          buffered = buffered.slice(newline + 1);
+          await handleLine(rawLine);
+        }
+      }
+
+      buffered += decoder.decode();
+      if (buffered) {
+        await handleLine(buffered);
+      }
+      await dispatch();
+    }
+  } catch (err) {
+    console.error(`Event stream error: ${(err as Error).message}`);
+  }
+
+  if (!respondedToRequest) {
+    // The stream closed without answering the request; fail closed so the
+    // client is not left waiting on an id that will never resolve.
+    await emit(
+      jsonRpcError(-32000, "Event stream ended before a response arrived", requestId)
+    );
   }
 }
 
@@ -116,10 +212,9 @@ rl.on("close", () => {
 });
 
 async function processLine(line: string): Promise<void> {
-  const response = await forward(line);
-  if (response) {
-    await writeStdout(response + "\n");
-  }
+  await forward(line, async (message) => {
+    await writeStdout(message + "\n");
+  });
 }
 
 async function writeStdout(text: string): Promise<void> {