From 09b588326c7266acbd668cb5a83774d440769b7d Mon Sep 17 00:00:00 2001
From: Robert <bobseares@gmail.com>
Date: Fri, 15 May 2026 21:12:53 -0300
Subject: [PATCH] =?UTF-8?q?release:=200.1.5=20=E2=80=94=20SSE-stream=20too?=
 =?UTF-8?q?ls/call=20to=20dodge=20ingress=20timeout?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

agent-builder.build (and any long-running skill) takes minutes — LLM
loop + sandbox tests + deploy + wait-for-live-card. A plain JSON POST
to /mcp gets killed by ingress idle timeouts (~60s on traefik) and the
gateway sees "fetch failed" while the upstream is still running.

UpstreamAgent.callTool now negotiates ``Accept: text/event-stream`` on
tools/call. We drain SSE frames, return the final tools/call result,
and fall back to plain JSON if the upstream doesn't speak SSE (older
a2a-pack). Elicit relaying isn't wired yet; for now any
elicitation/create frame is ignored.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
---
 package-lock.json |  4 +-
 package.json      |  2 +-
 src/upstream.ts   | 94 ++++++++++++++++++++++++++++++++++++++++++++++-
 3 files changed, 96 insertions(+), 4 deletions(-)

diff --git a/package-lock.json b/package-lock.json
index ef945db..40370fe 100644
--- a/package-lock.json
+++ b/package-lock.json
@@ -1,12 +1,12 @@
 {
   "name": "a2amcp",
-  "version": "0.1.4",
+  "version": "0.1.5",
   "lockfileVersion": 3,
   "requires": true,
   "packages": {
     "": {
       "name": "a2amcp",
-      "version": "0.1.4",
+      "version": "0.1.5",
       "license": "Apache-2.0",
       "dependencies": {
         "@modelcontextprotocol/sdk": "^1.0.4",
diff --git a/package.json b/package.json
index 6aaa58c..957482c 100644
--- a/package.json
+++ b/package.json
@@ -1,6 +1,6 @@
 {
   "name": "a2amcp",
-  "version": "0.1.4",
+  "version": "0.1.5",
   "description": "MCP gateway for a2acloud agents. Run as a local MCP server; expose any number of deployed A2A agents to Claude Code, Cursor, and other MCP clients.",
   "type": "module",
   "bin": {
diff --git a/src/upstream.ts b/src/upstream.ts
index e89c6f6..248dc49 100644
--- a/src/upstream.ts
+++ b/src/upstream.ts
@@ -106,6 +106,98 @@ export class UpstreamAgent {
     if (this.cfg.bucket) meta.bucket = this.cfg.bucket;
     const params: Record<string, unknown> = { name, arguments: arguments_ };
     if (Object.keys(meta).length > 0) params._meta = meta;
-    return this.rpc<UpstreamCallResult>("tools/call", params);
+    return this.streamingCallTool(params);
+  }
+
+  /**
+   * tools/call always negotiates SSE on the wire — upstream agents can
+   * take minutes (agent-builder.build runs an LLM loop + sandbox tests +
+   * a deploy + a wait-for-live-card poll), and a plain JSON POST gets
+   * killed by ingress idle timeouts (~60s default on traefik). The
+   * upstream's SSE response interleaves any elicitation/create requests
+   * and a final tools/call result; here we drain the stream and return
+   * the final result. (Elicit relaying to the MCP client is not wired
+   * yet — any elicit that fires will time out on the upstream side and
+   * the skill will surface a soft error.)
+   */
+  private async streamingCallTool(
+    params: Record<string, unknown>,
+  ): Promise<UpstreamCallResult> {
+    const id = _nextId++;
+    const headers: Record<string, string> = {
+      "content-type": "application/json",
+      accept: "text/event-stream, application/json",
+    };
+    if (this.cfg.token) headers["authorization"] = `Bearer ${this.cfg.token}`;
+
+    const resp = await fetch(`${this.cfg.url}/mcp`, {
+      method: "POST",
+      headers,
+      body: JSON.stringify({
+        jsonrpc: JSON_RPC, id, method: "tools/call", params,
+      }),
+    });
+
+    if (!resp.ok) {
+      const text = await resp.text().catch(() => "");
+      throw new UpstreamError(this.cfg.name, resp.status, text || resp.statusText);
+    }
+
+    const ctype = resp.headers.get("content-type") || "";
+    if (!ctype.includes("text/event-stream")) {
+      // Server fell back to JSON (older a2a-pack or non-SSE upstream).
+      const env: any = await resp.json();
+      if (env?.error) {
+        throw new UpstreamError(
+          this.cfg.name, 500,
+          `${env.error.code}: ${env.error.message}`,
+        );
+      }
+      return env?.result as UpstreamCallResult;
+    }
+
+    if (!resp.body) {
+      throw new UpstreamError(this.cfg.name, 502, "empty stream body");
+    }
+
+    const reader = resp.body.getReader();
+    const decoder = new TextDecoder();
+    let buf = "";
+    try {
+      while (true) {
+        const { done, value } = await reader.read();
+        if (value) buf += decoder.decode(value, { stream: true });
+        let sep = buf.indexOf("\n\n");
+        while (sep !== -1) {
+          const frame = buf.slice(0, sep);
+          buf = buf.slice(sep + 2);
+          const dataLines: string[] = [];
+          for (const ln of frame.split("\n")) {
+            if (ln.startsWith("data:")) dataLines.push(ln.slice(5).trimStart());
+          }
+          if (dataLines.length > 0) {
+            const raw = dataLines.join("\n");
+            let msg: any;
+            try { msg = JSON.parse(raw); } catch { msg = null; }
+            if (msg && msg.id === id) {
+              if (msg.error) {
+                throw new UpstreamError(
+                  this.cfg.name, 500,
+                  `${msg.error.code}: ${msg.error.message}`,
+                );
+              }
+              return msg.result as UpstreamCallResult;
+            }
+            // else: server→client request (elicitation/create) or another
+            // frame; ignore — elicit relaying not wired yet.
+          }
+          sep = buf.indexOf("\n\n");
+        }
+        if (done) break;
+      }
+    } finally {
+      try { reader.releaseLock(); } catch { /* ignore */ }
+    }
+    throw new UpstreamError(this.cfg.name, 502, "stream ended without result");
   }
 }