1 Commits

Author SHA1 Message Date
09b588326c release: 0.1.5 — SSE-stream tools/call to dodge ingress timeout
All checks were successful
build / test (push) Successful in 28s
publish / npm (push) Successful in 32s
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>
2026-05-15 21:12:53 -03:00
3 changed files with 96 additions and 4 deletions

4
package-lock.json generated
View File

@@ -1,12 +1,12 @@
{ {
"name": "a2amcp", "name": "a2amcp",
"version": "0.1.4", "version": "0.1.5",
"lockfileVersion": 3, "lockfileVersion": 3,
"requires": true, "requires": true,
"packages": { "packages": {
"": { "": {
"name": "a2amcp", "name": "a2amcp",
"version": "0.1.4", "version": "0.1.5",
"license": "Apache-2.0", "license": "Apache-2.0",
"dependencies": { "dependencies": {
"@modelcontextprotocol/sdk": "^1.0.4", "@modelcontextprotocol/sdk": "^1.0.4",

View File

@@ -1,6 +1,6 @@
{ {
"name": "a2amcp", "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.", "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", "type": "module",
"bin": { "bin": {

View File

@@ -106,6 +106,98 @@ export class UpstreamAgent {
if (this.cfg.bucket) meta.bucket = this.cfg.bucket; if (this.cfg.bucket) meta.bucket = this.cfg.bucket;
const params: Record<string, unknown> = { name, arguments: arguments_ }; const params: Record<string, unknown> = { name, arguments: arguments_ };
if (Object.keys(meta).length > 0) params._meta = meta; 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");
} }
} }