release: 0.1.5 — SSE-stream tools/call to dodge ingress timeout
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>
This commit is contained in:
4
package-lock.json
generated
4
package-lock.json
generated
@@ -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",
|
||||
|
||||
@@ -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": {
|
||||
|
||||
@@ -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");
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user