release: 0.1.6 — relay notifications/progress to MCP client

Long-running upstream skills emit MCP notifications/progress on the SSE
stream. Without forwarding, the gateway swallows them and the MCP
client (Claude Code, Cursor) aborts the tools/call on its idle timer
after ~60s — even though the HTTP stream stays warm.

Gateway now passes the SDK's extra.sendNotification through to
UpstreamAgent.callTool. Any non-response frame with a method but no id
becomes a notification on the client side. Each tools/call carries a
progressToken (client-supplied or request-id fallback) so the client
can correlate.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

package-lock.json

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

package.json

@@ -1,6 +1,6 @@
 {
   "name": "a2amcp",
-  "version": "0.1.3",
+  "version": "0.1.6",
   "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": {

src/api.ts

@@ -52,23 +52,21 @@ export class ControlPlaneClient {
   }
 
   login(email: string, password: string) {
-    return this.request<{ access_token: string; user: { email: string } }>(
+    return this.request<{
-      "POST",
+      access_token: string;
-      "/v1/auth/login",
+      user: { id: number; email: string };
-      { email, password },
+    }>("POST", "/v1/auth/login", { email, password });
-    );
   }
 
   signup(email: string, password: string) {
-    return this.request<{ access_token: string; user: { email: string } }>(
+    return this.request<{
-      "POST",
+      access_token: string;
-      "/v1/auth/signup",
+      user: { id: number; email: string };
-      { email, password },
+    }>("POST", "/v1/auth/signup", { email, password });
-    );
   }
 
   me() {
-    return this.request<{ email: string }>("GET", "/v1/me");
+    return this.request<{ id: number; email: string }>("GET", "/v1/me");
   }
 
   listAgents() {

src/cli.ts

@@ -92,7 +92,14 @@ program
     const api = resolveApiUrl(opts.api);
     try {
       const out = await new ControlPlaneClient(api, null).login(email, password);
-      await saveCredentials({ apiUrl: api, token: out.access_token, email: out.user.email });
+      const bucket = `user-${out.user.id}-files`;
+      await saveCredentials({
+        apiUrl: api,
+        token: out.access_token,
+        email: out.user.email,
+        userId: out.user.id,
+        bucket,
+      });
       output.write(`logged in as ${out.user.email} @ ${api}\n`);
     } catch (err) {
       fail(err instanceof ApiError ? err.message : (err as Error).message);

src/credentials.ts

@@ -14,6 +14,8 @@ export interface Credentials {
   apiUrl: string;
   token: string;
   email: string;
+  userId?: number;
+  bucket?: string;
 }
 
 const credsDir = () => path.join(os.homedir(), ".a2a");
@@ -28,6 +30,8 @@ export async function loadCredentials(): Promise<Credentials | null> {
       apiUrl: data.api_url ?? DEFAULT_API_URL,
       token: data.token,
       email: data.email ?? "",
+      userId: typeof data.user_id === "number" ? data.user_id : undefined,
+      bucket: typeof data.bucket === "string" ? data.bucket : undefined,
     };
   } catch (err: any) {
     if (err?.code === "ENOENT") return null;
@@ -37,11 +41,14 @@ export async function loadCredentials(): Promise<Credentials | null> {
 
 export async function saveCredentials(c: Credentials): Promise<void> {
   await fs.mkdir(credsDir(), { recursive: true });
-  await fs.writeFile(
+  const out: Record<string, unknown> = {
-    credsFile(),
+    api_url: c.apiUrl,
-    JSON.stringify({ api_url: c.apiUrl, token: c.token, email: c.email }),
+    token: c.token,
-    { mode: 0o600 },
+    email: c.email,
-  );
+  };
+  if (c.userId !== undefined) out.user_id = c.userId;
+  if (c.bucket !== undefined) out.bucket = c.bucket;
+  await fs.writeFile(credsFile(), JSON.stringify(out), { mode: 0o600 });
 }
 
 export async function clearCredentials(): Promise<boolean> {

src/gateway.ts

@@ -39,14 +39,25 @@ export interface GatewayHandle {
 
 export async function buildGateway(opts: GatewayOptions = {}): Promise<GatewayHandle> {
   const cfg = opts.agents ?? (await loadConfig()).agents;
-  const token =
+  const creds = opts.token !== undefined ? null : await loadCredentials();
-    opts.token !== undefined
+  const token = opts.token !== undefined ? opts.token : creds?.token ?? null;
-      ? opts.token
+  // Forward CP credentials + bucket as params._meta so upstream agents
-      : (await loadCredentials())?.token ?? null;
+  // can act on behalf of the caller (mirrors the platform-orchestrator
+  // call shape). When the user isn't logged in, _meta is omitted and
+  // upstream sees the same unauthenticated context as before.
+  const cpJwt = creds?.token ?? null;
+  const cpUrl = creds?.apiUrl ?? null;
+  const bucket = creds?.bucket ?? null;
 
   const upstreams = new Map<string, UpstreamAgent>();
   for (const a of cfg) {
-    upstreams.set(a.name, new UpstreamAgent({ name: a.name, url: a.url, token }));
+    upstreams.set(
+      a.name,
+      new UpstreamAgent({
+        name: a.name, url: a.url, token,
+        cpJwt, cpUrl, bucket,
+      }),
+    );
   }
 
   const server =
@@ -84,7 +95,7 @@ export async function buildGateway(opts: GatewayOptions = {}): Promise<GatewayHa
     return { tools };
   });
 
-  server.setRequestHandler(CallToolRequestSchema, async (req) => {
+  server.setRequestHandler(CallToolRequestSchema, async (req, extra) => {
     const fullName = req.params.name;
     const idx = fullName.indexOf(SEP);
     if (idx < 0) {
@@ -96,10 +107,30 @@ export async function buildGateway(opts: GatewayOptions = {}): Promise<GatewayHa
     if (!upstream) {
       return _softError(`unknown agent: ${agentName}`) as any;
     }
+    // Honor the client's progressToken if it supplied one; else fall
+    // back to the request id. Either way, when upstream emits
+    // notifications/progress we forward them via the SDK's
+    // sendNotification so clients like Claude Code reset their idle
+    // tool-call timer and don't abort long builds.
+    const clientToken = (req.params._meta as any)?.progressToken as
+      | string
+      | number
+      | undefined;
+    const progressToken = clientToken ?? (extra.requestId as string | number);
     try {
       const result = await upstream.callTool(
         toolName,
         (req.params.arguments as Record<string, unknown>) ?? {},
+        {
+          progressToken,
+          onNotification: async (notif) => {
+            try {
+              await extra.sendNotification(notif as any);
+            } catch {
+              // best effort
+            }
+          },
+        },
       );
       // Pass through the upstream's CallToolResult as-is. Upstreams already
       // populate content / structuredContent / isError per the MCP spec.

src/upstream.ts

@@ -12,6 +12,18 @@ export interface UpstreamConfig {
   name: string;
   url: string;
   token: string | null;
+  /** Forwarded to the upstream as ``params._meta.cp_jwt`` on tools/call.
+   *  Lets the upstream agent act on the caller's behalf (file/agent CRUD
+   *  on /v1/me/*) — same path the platform orchestrator uses. */
+  cpJwt?: string | null;
+  /** Paired with ``cpJwt``. The upstream agent uses this as the base URL
+   *  when calling back into /v1/me/*. */
+  cpUrl?: string | null;
+  /** User's MinIO bucket (``user-<id>-files``). Forwarded as
+   *  ``params._meta.bucket`` so agents that touch the workspace
+   *  (agent-builder, file tools) get a context whose
+   *  ``ctx.workspace.bucket`` resolves to the right bucket. */
+  bucket?: string | null;
 }
 
 export interface UpstreamTool {
@@ -87,10 +99,130 @@ export class UpstreamAgent {
   async callTool(
     name: string,
     arguments_: Record<string, unknown>,
+    opts: {
+      onNotification?: (notif: {
+        method: string;
+        params?: Record<string, unknown>;
+      }) => void | Promise<void>;
+      progressToken?: string | number;
+    } = {},
   ): Promise<UpstreamCallResult> {
-    return this.rpc<UpstreamCallResult>("tools/call", {
+    const meta: Record<string, unknown> = {};
-      name,
+    if (this.cfg.cpJwt) meta.cp_jwt = this.cfg.cpJwt;
-      arguments: arguments_,
+    if (this.cfg.cpUrl) meta.cp_url = this.cfg.cpUrl;
+    if (this.cfg.bucket) meta.bucket = this.cfg.bucket;
+    if (opts.progressToken !== undefined) meta.progressToken = opts.progressToken;
+    const params: Record<string, unknown> = { name, arguments: arguments_ };
+    if (Object.keys(meta).length > 0) params._meta = meta;
+    return this.streamingCallTool(params, opts.onNotification);
+  }
+
+  /**
+   * 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>,
+    onNotification?: (notif: {
+      method: string;
+      params?: Record<string, unknown>;
+    }) => void | Promise<void>,
+  ): 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;
+            }
+            // Otherwise it's a server→client message. Notifications
+            // (no id) → forward to the MCP client so long-running
+            // skills keep the client's tool-call timer alive. Requests
+            // (has id, has method) — e.g. elicitation/create — are
+            // not relayed yet, so they will time out upstream.
+            if (msg && typeof msg.method === "string" && !("id" in msg) && onNotification) {
+              try {
+                await onNotification({
+                  method: msg.method,
+                  params: msg.params as Record<string, unknown> | undefined,
+                });
+              } catch {
+                // Best-effort; never break the stream on a notify error.
+              }
+            }
+          }
+          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");
   }
 }