Fix a2amcp login error reporting

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>

.gitignore

@@ -1,5 +1,6 @@
 node_modules/
 dist/
+.codegraph/
 *.log
 *.tsbuildinfo
 .DS_Store

README.md

@@ -4,6 +4,21 @@ MCP gateway for a2acloud agents. It runs locally as a stdio MCP server and
 exposes deployed A2A agents as tools to Claude Code, Cursor, and other MCP
 clients.
 
+This package is the **local stdio gateway**. It is different from the hosted
+OAuth MCP connector endpoints used by ChatGPT/Claude web connectors:
+
+```text
+Local stdio gateway:        npx -y a2amcp
+Remote standard MCP:        https://<agent>.a2acloud.io/mcp
+Remote connector MCP:       https://<agent>.a2acloud.io/connector-mcp
+Remote orchestrator MCP:    https://api.a2acloud.io/connector-mcp
+```
+
+Use `a2amcp` for editor clients that launch a local MCP server process. Use the
+remote `/connector-mcp` URL for hosted connector UIs that need OAuth login,
+Dynamic Client Registration, async job polling, and structured approval/input
+interrupts.
+
 ## Quickstart
 
 ```bash
@@ -30,7 +45,7 @@ Restart the client after adding or removing agents.
 ## Commands
 
 ```bash
-a2amcp login              # save control-plane credentials
+a2amcp login              # sign in through Keycloak and save OAuth tokens
 a2amcp whoami             # show current account
 a2amcp agents             # list agents visible to your account
 a2amcp add <name>         # expose one agent through this gateway
@@ -41,3 +56,28 @@ a2amcp logout             # clear local credentials
 ```
 
 Running `a2amcp` with no command starts the MCP gateway over stdio.
+
+`a2amcp login` uses Keycloak Authorization Code + PKCE. It starts a local
+loopback callback server, opens the browser, exchanges the code for a Keycloak
+access token, and stores the token in `~/.a2a/credentials.json`. For headless
+machines, pass an existing Keycloak access token:
+
+```bash
+a2amcp login --token "$KEYCLOAK_ACCESS_TOKEN"
+```
+
+## Hosted OAuth Connectors
+
+ChatGPT and Claude-style hosted connectors should not run this stdio gateway.
+Configure them directly with the remote connector URL:
+
+```text
+https://api.a2acloud.io/connector-mcp
+https://<agent>.a2acloud.io/connector-mcp
+```
+
+Choose OAuth authentication. The server advertises protected-resource metadata,
+the client dynamically registers with Keycloak, and the browser consent flow
+issues tokens scoped to MCP access. Long-running connector calls return a
+`job_id`; the client should poll `chat_result` on the orchestrator connector or
+`job_result` on a leaf-agent connector.

package-lock.json

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

package.json

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

@@ -1,7 +1,4 @@
-/**
+/** Thin control-plane client used with Keycloak bearer tokens. */
- * Thin control-plane client. Mirrors the relevant subset of
- * a2a_pack.cli.api_client.ControlPlaneClient.
- */
 export class ApiError extends Error {
   constructor(public status: number, message: string) {
     super(`API ${status}: ${message}`);
@@ -15,6 +12,7 @@ export interface AgentRow {
   status: string;
   url?: string;
   description?: string;
+  card?: Record<string, any>;
 }
 
 export class ControlPlaneClient {
@@ -26,8 +24,11 @@ export class ControlPlaneClient {
   }
 
   private headers(): Record<string, string> {
-    const h: Record<string, string> = { "content-type": "application/json" };
+    const h: Record<string, string> = {
-    if (this.token) h["authorization"] = `bearer ${this.token}`;
+      accept: "application/json",
+      "content-type": "application/json",
+    };
+    if (this.token) h["authorization"] = `Bearer ${this.token}`;
     return h;
   }
 
@@ -37,32 +38,12 @@ export class ControlPlaneClient {
       headers: this.headers(),
       body: body !== undefined ? JSON.stringify(body) : undefined,
     });
+    const text = await resp.text();
     if (!resp.ok) {
-      let detail: string;
+      throw new ApiError(resp.status, errorDetail(text, resp.statusText));
-      try {
-        const j: any = await resp.json();
-        detail = j?.detail ?? JSON.stringify(j);
-      } catch {
-        detail = await resp.text();
-      }
-      throw new ApiError(resp.status, detail);
     }
     if (resp.status === 204) return undefined as T;
-    return (await resp.json()) as T;
+    return (text ? JSON.parse(text) : undefined) as T;
-  }
-
-  login(email: string, password: string) {
-    return this.request<{
-      access_token: string;
-      user: { id: number; email: string };
-    }>("POST", "/v1/auth/login", { email, password });
-  }
-
-  signup(email: string, password: string) {
-    return this.request<{
-      access_token: string;
-      user: { id: number; email: string };
-    }>("POST", "/v1/auth/signup", { email, password });
   }
 
   me() {
@@ -77,3 +58,16 @@ export class ControlPlaneClient {
     return this.request<AgentRow>("GET", `/v1/agents/${encodeURIComponent(name)}`);
   }
 }
+
+function errorDetail(text: string, fallback: string): string {
+  if (!text.trim()) return fallback;
+  try {
+    const parsed = JSON.parse(text);
+    const detail = parsed?.detail;
+    if (typeof detail === "string") return detail;
+    if (detail !== undefined) return JSON.stringify(detail);
+    return JSON.stringify(parsed);
+  } catch {
+    return text;
+  }
+}

src/cli.ts

@@ -21,9 +21,17 @@ import {
   clearCredentials,
   loadCredentials,
   resolveApiUrl,
-  saveCredentials,
 } from "./credentials.js";
 import { runStdio } from "./gateway.js";
+import {
+  DEFAULT_OAUTH_CLIENT_ID,
+  DEFAULT_OAUTH_ISSUER,
+  DEFAULT_OAUTH_SCOPE,
+  DEFAULT_REDIRECT_PORT,
+  loginWithAccessToken,
+  loginWithBrowser,
+  refreshCredentialsIfNeeded,
+} from "./oauth.js";
 
 const program = new Command();
 
@@ -34,8 +42,9 @@ program
 
 async function client(apiOverride?: string): Promise<ControlPlaneClient> {
   const creds = await loadCredentials();
+  const refreshed = creds ? await refreshCredentialsIfNeeded(creds) : null;
   const apiUrl = resolveApiUrl(apiOverride ?? creds?.apiUrl);
-  return new ControlPlaneClient(apiUrl, creds?.token ?? null);
+  return new ControlPlaneClient(apiUrl, refreshed?.token ?? null);
 }
 
 function fail(msg: string, code = 1): never {
@@ -43,6 +52,30 @@ function fail(msg: string, code = 1): never {
   process.exit(code);
 }
 
+function normalizeMcpPath(path: unknown): string {
+  if (typeof path !== "string" || !path.trim()) return "/mcp";
+  const trimmed = path.trim();
+  return trimmed.startsWith("/") ? trimmed : `/${trimmed}`;
+}
+
+function mcpTargetForAgent(
+  row: { name: string; url?: string; card?: Record<string, any> },
+  apiUrl: string,
+): { url?: string; mcpPath: string } {
+  const imported = row.card?.capabilities?.a2a_cloud_import;
+  const mcp = imported?.mcp;
+  if (mcp?.mode === "generated") {
+    return {
+      url: `${apiUrl.replace(/\/+$/, "")}/v1/agents/${encodeURIComponent(row.name)}`,
+      mcpPath: normalizeMcpPath(mcp.path),
+    };
+  }
+  return {
+    url: mcp?.base_url || row.url,
+    mcpPath: normalizeMcpPath(mcp?.path || row.card?.mcp_endpoint),
+  };
+}
+
 async function prompt(label: string, opts: { hidden?: boolean } = {}): Promise<string> {
   const rl = readline.createInterface({ input, output, terminal: true });
   try {
@@ -82,25 +115,42 @@ program
 
 program
   .command("login")
-  .description("Authenticate against the control plane.")
+  .description("Authenticate with Keycloak and cache a bearer token.")
-  .option("--email <email>")
-  .option("--password <password>")
   .option("--api <url>", "Override control plane URL", DEFAULT_API_URL)
-  .action(async (opts: { email?: string; password?: string; api: string }) => {
+  .option("--issuer <url>", "Keycloak realm issuer", DEFAULT_OAUTH_ISSUER)
-    const email = opts.email ?? (await prompt("email: "));
+  .option("--client-id <id>", "Keycloak public client id", DEFAULT_OAUTH_CLIENT_ID)
-    const password = opts.password ?? (await prompt("password: ", { hidden: true }));
+  .option("--scope <scope>", "OAuth scopes to request", DEFAULT_OAUTH_SCOPE)
+  .option("--port <port>", "Loopback callback port", String(DEFAULT_REDIRECT_PORT))
+  .option("--token <token>", "Save an already-issued Keycloak access token")
+  .option("--no-open", "Print the login URL without opening a browser")
+  .action(async (opts: {
+    api: string;
+    issuer: string;
+    clientId: string;
+    scope: string;
+    port: string;
+    token?: string;
+    open?: boolean;
+  }) => {
     const api = resolveApiUrl(opts.api);
     try {
-      const out = await new ControlPlaneClient(api, null).login(email, password);
+      const common = {
-      const bucket = `user-${out.user.id}-files`;
-      await saveCredentials({
         apiUrl: api,
-        token: out.access_token,
+        issuer: opts.issuer,
-        email: out.user.email,
+        clientId: opts.clientId,
-        userId: out.user.id,
+        scope: opts.scope,
-        bucket,
+      };
-      });
+      const creds = opts.token
-      output.write(`logged in as ${out.user.email} @ ${api}\n`);
+        ? await loginWithAccessToken(opts.token, common)
+        : await loginWithBrowser({
+            ...common,
+            port: Number.parseInt(opts.port, 10),
+            openBrowser: opts.open !== false,
+            onAuthorizationUrl: (url) => {
+              output.write(`Open this URL to sign in with Keycloak:\n${url}\n`);
+            },
+          });
+      output.write(`logged in as ${creds.email} @ ${api}\n`);
     } catch (err) {
       fail(err instanceof ApiError ? err.message : (err as Error).message);
     }
@@ -160,7 +210,9 @@ program
       output.write("(no agents added; try `a2amcp add <name>`)\n");
       return;
     }
-    for (const a of cfg.agents) output.write(`  ${a.name}  ${a.url}\n`);
+    for (const a of cfg.agents) {
+      output.write(`  ${a.name}  ${a.url}${a.mcpPath ?? "/mcp"}\n`);
+    }
   });
 
 program
@@ -171,11 +223,17 @@ program
   .action(
     async (name: string, opts: { url?: string; api?: string }) => {
       let url = opts.url;
+      let mcpPath = "/mcp";
       let description: string | undefined;
       if (!url) {
         try {
-          const row = await (await client(opts.api)).getAgent(name);
+          const creds = await loadCredentials();
-          url = row.url;
+          const refreshed = creds ? await refreshCredentialsIfNeeded(creds) : null;
+          const apiUrl = resolveApiUrl(opts.api ?? creds?.apiUrl);
+          const row = await new ControlPlaneClient(apiUrl, refreshed?.token ?? null).getAgent(name);
+          const target = mcpTargetForAgent(row, apiUrl);
+          url = target.url;
+          mcpPath = target.mcpPath;
           description = row.description;
         } catch (err) {
           fail(err instanceof ApiError ? err.message : (err as Error).message);
@@ -185,10 +243,11 @@ program
       await addAgent({
         name,
         url,
+        mcpPath,
         description,
         addedAt: new Date().toISOString(),
       });
-      output.write(`added ${name} -> ${url}\n`);
+      output.write(`added ${name} -> ${url.replace(/\/+$/, "")}${mcpPath}\n`);
       output.write("restart your MCP client to pick up the new tools.\n");
     },
   );
@@ -210,6 +269,7 @@ program
   .action(async () => {
     const cfg = await loadConfig();
     const creds = await loadCredentials();
+    const refreshed = creds ? await refreshCredentialsIfNeeded(creds) : null;
     if (cfg.agents.length === 0) {
       output.write("(no agents enabled)\n");
       return;
@@ -219,7 +279,8 @@ program
       const u = new UpstreamAgent({
         name: a.name,
         url: a.url,
-        token: creds?.token ?? null,
+        mcpPath: a.mcpPath,
+        token: refreshed?.token ?? null,
       });
       try {
         const tools = await u.listTools();

src/config.ts

@@ -12,6 +12,7 @@ import path from "node:path";
 export interface EnabledAgent {
   name: string;
   url: string;
+  mcpPath?: string;
   description?: string;
   addedAt: string;
 }

src/credentials.ts

@@ -1,8 +1,7 @@
 /**
- * Reader for ~/.a2a/credentials.json — same file the Python `a2a` CLI writes.
+ * Reader/writer for ~/.a2a/credentials.json — same file the Python `a2a` CLI
- *
+ * writes. New logins store Keycloak access/refresh tokens; older token-only
- * We deliberately *only read* this file (and overwrite it on `a2amcp login`)
+ * files are still readable.
- * so a single login flow is shared between the Python and Node tooling.
  */
 import { promises as fs } from "node:fs";
 import os from "node:os";
@@ -16,6 +15,11 @@ export interface Credentials {
   email: string;
   userId?: number;
   bucket?: string;
+  refreshToken?: string;
+  expiresAt?: number;
+  issuer?: string;
+  clientId?: string;
+  scope?: string;
 }
 
 const credsDir = () => path.join(os.homedir(), ".a2a");
@@ -32,6 +36,12 @@ export async function loadCredentials(): Promise<Credentials | null> {
       email: data.email ?? "",
       userId: typeof data.user_id === "number" ? data.user_id : undefined,
       bucket: typeof data.bucket === "string" ? data.bucket : undefined,
+      refreshToken:
+        typeof data.refresh_token === "string" ? data.refresh_token : undefined,
+      expiresAt: typeof data.expires_at === "number" ? data.expires_at : undefined,
+      issuer: typeof data.oauth_issuer === "string" ? data.oauth_issuer : undefined,
+      clientId: typeof data.client_id === "string" ? data.client_id : undefined,
+      scope: typeof data.scope === "string" ? data.scope : undefined,
     };
   } catch (err: any) {
     if (err?.code === "ENOENT") return null;
@@ -48,6 +58,11 @@ export async function saveCredentials(c: Credentials): Promise<void> {
   };
   if (c.userId !== undefined) out.user_id = c.userId;
   if (c.bucket !== undefined) out.bucket = c.bucket;
+  if (c.refreshToken !== undefined) out.refresh_token = c.refreshToken;
+  if (c.expiresAt !== undefined) out.expires_at = c.expiresAt;
+  if (c.issuer !== undefined) out.oauth_issuer = c.issuer;
+  if (c.clientId !== undefined) out.client_id = c.clientId;
+  if (c.scope !== undefined) out.scope = c.scope;
   await fs.writeFile(credsFile(), JSON.stringify(out), { mode: 0o600 });
 }
 

src/gateway.ts

@@ -11,11 +11,14 @@ import { Server } from "@modelcontextprotocol/sdk/server/index.js";
 import { StdioServerTransport } from "@modelcontextprotocol/sdk/server/stdio.js";
 import {
   CallToolRequestSchema,
+  ElicitResultSchema,
   ListToolsRequestSchema,
+  ResultSchema,
 } from "@modelcontextprotocol/sdk/types.js";
 
 import { loadConfig, type EnabledAgent } from "./config.js";
-import { loadCredentials } from "./credentials.js";
+import { loadCredentials, type Credentials } from "./credentials.js";
+import { refreshCredentialsIfNeeded } from "./oauth.js";
 import { UpstreamAgent, UpstreamError, type UpstreamTool } from "./upstream.js";
 
 const require = createRequire(import.meta.url);
@@ -28,6 +31,10 @@ export interface GatewayOptions {
   agents?: EnabledAgent[];
   /** Override the bearer token (otherwise read from ~/.a2a/credentials.json). */
   token?: string | null;
+  /** Test/advanced override for one-shot upstream JSON-RPC calls. */
+  upstreamRequestTimeoutMs?: number;
+  /** Test/advanced override for idle SSE upstream tools/call streams. */
+  upstreamStreamIdleTimeoutMs?: number;
   /** Optional override for the underlying MCP Server (tests). */
   server?: Server;
 }
@@ -39,13 +46,18 @@ export interface GatewayHandle {
 
 export async function buildGateway(opts: GatewayOptions = {}): Promise<GatewayHandle> {
   const cfg = opts.agents ?? (await loadConfig()).agents;
-  const creds = opts.token !== undefined ? null : await loadCredentials();
+  let creds: Credentials | null = opts.token !== undefined ? null : await loadCredentials();
+  async function refreshedToken(): Promise<string | null> {
+    if (opts.token !== undefined) return opts.token;
+    if (!creds) return null;
+    creds = await refreshCredentialsIfNeeded(creds);
+    return creds.token;
+  }
   const token = opts.token !== undefined ? opts.token : creds?.token ?? null;
   // Forward CP credentials + bucket as params._meta so upstream agents
   // 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;
 
@@ -54,8 +66,12 @@ export async function buildGateway(opts: GatewayOptions = {}): Promise<GatewayHa
     upstreams.set(
       a.name,
       new UpstreamAgent({
-        name: a.name, url: a.url, token,
+        name: a.name, url: a.url, mcpPath: a.mcpPath, token,
-        cpJwt, cpUrl, bucket,
+        tokenProvider: refreshedToken,
+        cpJwtProvider: refreshedToken,
+        cpUrl, bucket,
+        requestTimeoutMs: opts.upstreamRequestTimeoutMs,
+        streamIdleTimeoutMs: opts.upstreamStreamIdleTimeoutMs,
       }),
     );
   }
@@ -95,7 +111,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) {
@@ -107,10 +123,42 @@ 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
+            }
+          },
+          onRequest: async (upstreamReq) => {
+            const schema = upstreamReq.method === "elicitation/create"
+              ? ElicitResultSchema
+              : ResultSchema;
+            return extra.sendRequest(
+              {
+                method: upstreamReq.method,
+                params: upstreamReq.params,
+              } as any,
+              schema as any,
+            );
+          },
+        },
       );
       // Pass through the upstream's CallToolResult as-is. Upstreams already
       // populate content / structuredContent / isError per the MCP spec.

src/index.ts

@@ -14,3 +14,13 @@ export {
   saveCredentials,
 } from "./credentials.js";
 export type { Credentials } from "./credentials.js";
+export {
+  DEFAULT_OAUTH_CLIENT_ID,
+  DEFAULT_OAUTH_ISSUER,
+  DEFAULT_OAUTH_SCOPE,
+  DEFAULT_REDIRECT_PORT,
+  createPkcePair,
+  loginWithAccessToken,
+  loginWithBrowser,
+  refreshCredentialsIfNeeded,
+} from "./oauth.js";

src/oauth.ts

@@ -0,0 +1,349 @@
+import crypto from "node:crypto";
+import http from "node:http";
+import { spawn } from "node:child_process";
+import type { AddressInfo } from "node:net";
+
+import { ControlPlaneClient } from "./api.js";
+import {
+  DEFAULT_API_URL,
+  type Credentials,
+  saveCredentials,
+} from "./credentials.js";
+
+export const DEFAULT_OAUTH_ISSUER = "https://auth.a2acloud.io/realms/a2acloud";
+export const DEFAULT_OAUTH_CLIENT_ID = "a2acloud-cli";
+export const DEFAULT_OAUTH_SCOPE = "openid email offline_access mcp:invoke agent:read";
+export const DEFAULT_REDIRECT_PORT = 41873;
+const CALLBACK_PATH = "/callback";
+
+type OpenIdConfiguration = {
+  authorization_endpoint: string;
+  token_endpoint: string;
+};
+
+type TokenResponse = {
+  access_token?: string;
+  refresh_token?: string;
+  expires_in?: number;
+  scope?: string;
+  error?: string;
+  error_description?: string;
+};
+
+export type BrowserLoginOptions = {
+  apiUrl?: string;
+  issuer?: string;
+  clientId?: string;
+  scope?: string;
+  port?: number;
+  openBrowser?: boolean;
+  timeoutMs?: number;
+  onAuthorizationUrl?: (url: string) => void;
+};
+
+export async function loginWithBrowser(
+  opts: BrowserLoginOptions = {},
+): Promise<Credentials> {
+  const apiUrl = (opts.apiUrl || DEFAULT_API_URL).replace(/\/+$/, "");
+  const issuer = (opts.issuer || DEFAULT_OAUTH_ISSUER).replace(/\/+$/, "");
+  const clientId = opts.clientId || DEFAULT_OAUTH_CLIENT_ID;
+  const scope = opts.scope || DEFAULT_OAUTH_SCOPE;
+  const cfg = await discoverOpenIdConfiguration(issuer);
+  const pkce = createPkcePair();
+  const state = randomUrlSafe(32);
+  const callback = await listenForCallback({
+    port: opts.port ?? DEFAULT_REDIRECT_PORT,
+    state,
+    timeoutMs: opts.timeoutMs ?? 180_000,
+  });
+  try {
+    const redirectUri = callback.redirectUri;
+    const authorizationUrl = buildAuthorizationUrl({
+      authorizationEndpoint: cfg.authorization_endpoint,
+      clientId,
+      redirectUri,
+      scope,
+      state,
+      codeChallenge: pkce.challenge,
+    });
+    opts.onAuthorizationUrl?.(authorizationUrl);
+    if (opts.openBrowser !== false) openBrowser(authorizationUrl);
+    const code = await callback.waitForCode;
+    const token = await exchangeAuthorizationCode({
+      tokenEndpoint: cfg.token_endpoint,
+      clientId,
+      redirectUri,
+      code,
+      codeVerifier: pkce.verifier,
+    });
+    const creds = await credentialsFromToken({
+      apiUrl,
+      token,
+      issuer,
+      clientId,
+      scope,
+    });
+    await saveCredentials(creds);
+    return creds;
+  } finally {
+    await callback.close();
+  }
+}
+
+export async function loginWithAccessToken(
+  token: string,
+  opts: {
+    apiUrl?: string;
+    issuer?: string;
+    clientId?: string;
+    scope?: string;
+  } = {},
+): Promise<Credentials> {
+  const apiUrl = (opts.apiUrl || DEFAULT_API_URL).replace(/\/+$/, "");
+  const creds = await credentialsFromToken({
+    apiUrl,
+    token: { access_token: token },
+    issuer: opts.issuer,
+    clientId: opts.clientId,
+    scope: opts.scope,
+  });
+  await saveCredentials(creds);
+  return creds;
+}
+
+export async function refreshCredentialsIfNeeded(
+  creds: Credentials,
+  minTtlSeconds = 60,
+): Promise<Credentials> {
+  if (!creds.refreshToken || !creds.issuer || !creds.clientId) return creds;
+  if (creds.expiresAt && creds.expiresAt - nowSeconds() > minTtlSeconds) return creds;
+
+  const cfg = await discoverOpenIdConfiguration(creds.issuer);
+  const body = new URLSearchParams({
+    grant_type: "refresh_token",
+    client_id: creds.clientId,
+    refresh_token: creds.refreshToken,
+  });
+  const resp = await fetch(cfg.token_endpoint, {
+    method: "POST",
+    headers: { "content-type": "application/x-www-form-urlencoded" },
+    body,
+  });
+  const token = await parseTokenResponse(resp);
+  const next: Credentials = {
+    ...creds,
+    token: token.access_token!,
+    refreshToken: token.refresh_token || creds.refreshToken,
+    expiresAt: expiresAt(token.expires_in),
+    scope: token.scope || creds.scope,
+  };
+  await saveCredentials(next);
+  return next;
+}
+
+export async function discoverOpenIdConfiguration(
+  issuer: string,
+): Promise<OpenIdConfiguration> {
+  const url = `${issuer.replace(/\/+$/, "")}/.well-known/openid-configuration`;
+  const resp = await fetch(url, { headers: { accept: "application/json" } });
+  if (!resp.ok) throw new Error(`OpenID discovery failed: ${resp.status}`);
+  const data = (await resp.json()) as Partial<OpenIdConfiguration>;
+  if (!data.authorization_endpoint || !data.token_endpoint) {
+    throw new Error("OpenID discovery document is missing authorization/token endpoints");
+  }
+  return {
+    authorization_endpoint: data.authorization_endpoint,
+    token_endpoint: data.token_endpoint,
+  };
+}
+
+export function createPkcePair(): { verifier: string; challenge: string } {
+  const verifier = randomUrlSafe(64);
+  const challenge = crypto
+    .createHash("sha256")
+    .update(verifier)
+    .digest("base64url");
+  return { verifier, challenge };
+}
+
+export function buildAuthorizationUrl(opts: {
+  authorizationEndpoint: string;
+  clientId: string;
+  redirectUri: string;
+  scope: string;
+  state: string;
+  codeChallenge: string;
+}): string {
+  const url = new URL(opts.authorizationEndpoint);
+  url.searchParams.set("client_id", opts.clientId);
+  url.searchParams.set("redirect_uri", opts.redirectUri);
+  url.searchParams.set("response_type", "code");
+  url.searchParams.set("scope", opts.scope);
+  url.searchParams.set("state", opts.state);
+  url.searchParams.set("code_challenge", opts.codeChallenge);
+  url.searchParams.set("code_challenge_method", "S256");
+  return url.toString();
+}
+
+async function exchangeAuthorizationCode(opts: {
+  tokenEndpoint: string;
+  clientId: string;
+  redirectUri: string;
+  code: string;
+  codeVerifier: string;
+}): Promise<TokenResponse> {
+  const body = new URLSearchParams({
+    grant_type: "authorization_code",
+    client_id: opts.clientId,
+    redirect_uri: opts.redirectUri,
+    code: opts.code,
+    code_verifier: opts.codeVerifier,
+  });
+  const resp = await fetch(opts.tokenEndpoint, {
+    method: "POST",
+    headers: { "content-type": "application/x-www-form-urlencoded" },
+    body,
+  });
+  return parseTokenResponse(resp);
+}
+
+async function parseTokenResponse(resp: Response): Promise<TokenResponse> {
+  const text = await resp.text();
+  const data = parseJsonObject(text) as TokenResponse;
+  if (!resp.ok || !data.access_token) {
+    const message = data.error_description || data.error || text.trim() || resp.statusText;
+    throw new Error(`Keycloak token exchange failed: ${message}`);
+  }
+  return data;
+}
+
+async function credentialsFromToken(opts: {
+  apiUrl: string;
+  token: TokenResponse;
+  issuer?: string;
+  clientId?: string;
+  scope?: string;
+}): Promise<Credentials> {
+  const me = await new ControlPlaneClient(opts.apiUrl, opts.token.access_token!).me();
+  return {
+    apiUrl: opts.apiUrl,
+    token: opts.token.access_token!,
+    email: me.email,
+    userId: me.id,
+    bucket: `user-${me.id}-files`,
+    refreshToken: opts.token.refresh_token,
+    expiresAt: expiresAt(opts.token.expires_in),
+    issuer: opts.issuer,
+    clientId: opts.clientId,
+    scope: opts.token.scope || opts.scope,
+  };
+}
+
+function listenForCallback(opts: {
+  port: number;
+  state: string;
+  timeoutMs: number;
+}): Promise<{
+  redirectUri: string;
+  waitForCode: Promise<string>;
+  close: () => Promise<void>;
+}> {
+  let resolveCode!: (code: string) => void;
+  let rejectCode!: (err: Error) => void;
+  const waitForCode = new Promise<string>((resolve, reject) => {
+    resolveCode = resolve;
+    rejectCode = reject;
+  });
+
+  const server = http.createServer((req, res) => {
+    const url = new URL(req.url || "/", `http://${req.headers.host}`);
+    if (url.pathname !== CALLBACK_PATH) {
+      res.statusCode = 404;
+      res.end("not found");
+      return;
+    }
+    const error = url.searchParams.get("error");
+    const errorDescription = url.searchParams.get("error_description");
+    const code = url.searchParams.get("code");
+    const state = url.searchParams.get("state");
+    if (error) {
+      rejectCode(new Error(errorDescription || error));
+      finishBrowser(res, false);
+      return;
+    }
+    if (!code || state !== opts.state) {
+      rejectCode(new Error("invalid OAuth callback"));
+      finishBrowser(res, false);
+      return;
+    }
+    resolveCode(code);
+    finishBrowser(res, true);
+  });
+
+  const timer = setTimeout(() => {
+    rejectCode(new Error("timed out waiting for Keycloak login"));
+  }, opts.timeoutMs);
+
+  return new Promise((resolve, reject) => {
+    server.on("error", reject);
+    server.listen(opts.port, "127.0.0.1", () => {
+      const { port } = server.address() as AddressInfo;
+      resolve({
+        redirectUri: `http://127.0.0.1:${port}${CALLBACK_PATH}`,
+        waitForCode,
+        close: () =>
+          new Promise((done) => {
+            clearTimeout(timer);
+            server.close(() => done());
+          }),
+      });
+    });
+  });
+}
+
+function finishBrowser(res: http.ServerResponse, ok: boolean): void {
+  res.statusCode = ok ? 200 : 400;
+  res.setHeader("content-type", "text/html; charset=utf-8");
+  res.end(
+    `<html><body><h1>${ok ? "a2amcp is connected" : "a2amcp login failed"}</h1>` +
+      "<p>You can close this tab and return to your terminal.</p></body></html>",
+  );
+}
+
+function openBrowser(url: string): void {
+  const cmd =
+    process.platform === "darwin"
+      ? "open"
+      : process.platform === "win32"
+        ? "cmd"
+        : "xdg-open";
+  const args =
+    process.platform === "win32"
+      ? ["/c", "start", "", url]
+      : [url];
+  const child = spawn(cmd, args, { detached: true, stdio: "ignore" });
+  child.on("error", () => {});
+  child.unref();
+}
+
+function randomUrlSafe(bytes: number): string {
+  return crypto.randomBytes(bytes).toString("base64url");
+}
+
+function expiresAt(expiresIn: number | undefined): number | undefined {
+  return typeof expiresIn === "number" ? nowSeconds() + expiresIn : undefined;
+}
+
+function nowSeconds(): number {
+  return Math.floor(Date.now() / 1000);
+}
+
+function parseJsonObject(text: string): Record<string, unknown> {
+  if (!text.trim()) return {};
+  try {
+    const parsed = JSON.parse(text);
+    return parsed && typeof parsed === "object" ? parsed : {};
+  } catch {
+    return {};
+  }
+}

src/upstream.ts

@@ -7,15 +7,22 @@
  * header without coupling to a transport class.
  */
 const JSON_RPC = "2.0";
+const DEFAULT_RPC_TIMEOUT_MS = 30_000;
+const DEFAULT_STREAM_IDLE_TIMEOUT_MS = 120_000;
 
 export interface UpstreamConfig {
   name: string;
   url: string;
+  mcpPath?: string;
   token: string | null;
+  tokenProvider?: () => Promise<string | null>;
+  requestTimeoutMs?: number;
+  streamIdleTimeoutMs?: number;
   /** 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;
+  cpJwtProvider?: () => Promise<string | null>;
   /** Paired with ``cpJwt``. The upstream agent uses this as the base URL
    *  when calling back into /v1/me/*. */
   cpUrl?: string | null;
@@ -39,6 +46,12 @@ export interface UpstreamCallResult {
   isError?: boolean;
 }
 
+export interface UpstreamServerRequest {
+  id: string | number;
+  method: string;
+  params?: Record<string, unknown>;
+}
+
 export class UpstreamError extends Error {
   constructor(
     public agent: string,
@@ -52,43 +65,87 @@ export class UpstreamError extends Error {
 
 let _nextId = 1;
 
+function normalizeMcpPath(path: string | undefined): string {
+  if (!path) return "/mcp";
+  const trimmed = path.trim();
+  if (!trimmed) return "/mcp";
+  return trimmed.startsWith("/") ? trimmed : `/${trimmed}`;
+}
+
 export class UpstreamAgent {
   constructor(private cfg: UpstreamConfig) {
     this.cfg.url = cfg.url.replace(/\/+$/, "");
+    this.cfg.mcpPath = normalizeMcpPath(cfg.mcpPath);
   }
 
   get name(): string {
     return this.cfg.name;
   }
 
+  private mcpUrl(): string {
+    return `${this.cfg.url}${this.cfg.mcpPath ?? "/mcp"}`;
+  }
+
+  private async bearerToken(): Promise<string | null> {
+    if (this.cfg.tokenProvider) return this.cfg.tokenProvider();
+    return this.cfg.token ?? null;
+  }
+
+  private async cpJwt(): Promise<string | null> {
+    if (this.cfg.cpJwtProvider) return this.cfg.cpJwtProvider();
+    return this.cfg.cpJwt ?? null;
+  }
+
   private async rpc<T>(method: string, params: unknown): Promise<T> {
     const id = _nextId++;
+    const timeoutMs = this.cfg.requestTimeoutMs ?? DEFAULT_RPC_TIMEOUT_MS;
+    const controller = new AbortController();
+    let timedOut = false;
+    const timeout = setTimeout(() => {
+      timedOut = true;
+      controller.abort();
+    }, timeoutMs);
     const headers: Record<string, string> = {
       "content-type": "application/json",
       accept: "application/json",
     };
-    if (this.cfg.token) headers["authorization"] = `Bearer ${this.cfg.token}`;
+    const token = await this.bearerToken();
+    if (token) headers["authorization"] = `Bearer ${token}`;
 
-    const resp = await fetch(`${this.cfg.url}/mcp`, {
+    try {
-      method: "POST",
+      const resp = await fetch(this.mcpUrl(), {
-      headers,
+        method: "POST",
-      body: JSON.stringify({ jsonrpc: JSON_RPC, id, method, params }),
+        headers,
-    });
+        body: JSON.stringify({ jsonrpc: JSON_RPC, id, method, params }),
+        signal: controller.signal,
+      });
 
-    if (!resp.ok) {
+      if (!resp.ok) {
-      const text = await resp.text().catch(() => "");
+        const text = await resp.text().catch(() => "");
-      throw new UpstreamError(this.cfg.name, resp.status, text || resp.statusText);
+        throw new UpstreamError(this.cfg.name, resp.status, text || resp.statusText);
+      }
+
+      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 T;
+    } catch (err) {
+      if (timedOut) {
+        throw new UpstreamError(
+          this.cfg.name,
+          504,
+          `request timed out after ${timeoutMs}ms`,
+        );
+      }
+      throw err;
+    } finally {
+      clearTimeout(timeout);
     }
-
-    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 T;
   }
 
   async listTools(): Promise<UpstreamTool[]> {
@@ -99,14 +156,27 @@ export class UpstreamAgent {
   async callTool(
     name: string,
     arguments_: Record<string, unknown>,
+    opts: {
+      onNotification?: (notif: {
+        method: string;
+        params?: Record<string, unknown>;
+      }) => void | Promise<void>;
+      onRequest?: (req: UpstreamServerRequest) => unknown | Promise<unknown>;
+      progressToken?: string | number;
+    } = {},
   ): Promise<UpstreamCallResult> {
     const meta: Record<string, unknown> = {};
-    if (this.cfg.cpJwt) meta.cp_jwt = this.cfg.cpJwt;
+    const cpJwt = await this.cpJwt();
+    if (cpJwt) meta.cp_jwt = cpJwt;
     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);
+    return this.streamingCallTool(params, {
+      onNotification: opts.onNotification,
+      onRequest: opts.onRequest,
+    });
   }
 
   /**
@@ -115,31 +185,66 @@ export class UpstreamAgent {
    * 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
+   * and a final tools/call result; here we relay server-to-client
-   * the final result. (Elicit relaying to the MCP client is not wired
+   * requests through the MCP client and return the final result.
-   * 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>,
+    handlers: {
+      onNotification?: (notif: {
+        method: string;
+        params?: Record<string, unknown>;
+      }) => void | Promise<void>;
+      onRequest?: (req: UpstreamServerRequest) => unknown | Promise<unknown>;
+    } = {},
   ): Promise<UpstreamCallResult> {
     const id = _nextId++;
+    const idleTimeoutMs =
+      this.cfg.streamIdleTimeoutMs ?? DEFAULT_STREAM_IDLE_TIMEOUT_MS;
+    const controller = new AbortController();
+    let timedOut = false;
+    let idleTimer: ReturnType<typeof setTimeout> | undefined;
+    const resetIdleTimer = () => {
+      if (idleTimer) clearTimeout(idleTimer);
+      idleTimer = setTimeout(() => {
+        timedOut = true;
+        controller.abort();
+      }, idleTimeoutMs);
+    };
+    resetIdleTimer();
     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 token = await this.bearerToken();
+    if (token) headers["authorization"] = `Bearer ${token}`;
 
-    const resp = await fetch(`${this.cfg.url}/mcp`, {
+    let resp: Response;
-      method: "POST",
+    try {
-      headers,
+      resp = await fetch(this.mcpUrl(), {
-      body: JSON.stringify({
+        method: "POST",
-        jsonrpc: JSON_RPC, id, method: "tools/call", params,
+        headers,
-      }),
+        body: JSON.stringify({
-    });
+          jsonrpc: JSON_RPC, id, method: "tools/call", params,
+        }),
+        signal: controller.signal,
+      });
+    } catch (err) {
+      clearTimeout(idleTimer);
+      if (timedOut) {
+        throw new UpstreamError(
+          this.cfg.name,
+          504,
+          `stream idle timeout after ${idleTimeoutMs}ms`,
+        );
+      }
+      throw err;
+    }
+    resetIdleTimer();
 
     if (!resp.ok) {
       const text = await resp.text().catch(() => "");
+      clearTimeout(idleTimer);
       throw new UpstreamError(this.cfg.name, resp.status, text || resp.statusText);
     }
 
@@ -148,15 +253,20 @@ export class UpstreamAgent {
       // Server fell back to JSON (older a2a-pack or non-SSE upstream).
       const env: any = await resp.json();
       if (env?.error) {
+        clearTimeout(idleTimer);
         throw new UpstreamError(
           this.cfg.name, 500,
           `${env.error.code}: ${env.error.message}`,
         );
       }
+      clearTimeout(idleTimer);
       return env?.result as UpstreamCallResult;
     }
 
+    const upstreamSessionId = resp.headers.get("mcp-session-id");
+
     if (!resp.body) {
+      clearTimeout(idleTimer);
       throw new UpstreamError(this.cfg.name, 502, "empty stream body");
     }
 
@@ -165,8 +275,24 @@ export class UpstreamAgent {
     let buf = "";
     try {
       while (true) {
-        const { done, value } = await reader.read();
+        let done: boolean;
-        if (value) buf += decoder.decode(value, { stream: true });
+        let value: Uint8Array | undefined;
+        try {
+          ({ done, value } = await reader.read());
+        } catch (err) {
+          if (timedOut) {
+            throw new UpstreamError(
+              this.cfg.name,
+              504,
+              `stream idle timeout after ${idleTimeoutMs}ms`,
+            );
+          }
+          throw err;
+        }
+        if (value) {
+          resetIdleTimer();
+          buf += decoder.decode(value, { stream: true });
+        }
         let sep = buf.indexOf("\n\n");
         while (sep !== -1) {
           const frame = buf.slice(0, sep);
@@ -177,6 +303,10 @@ export class UpstreamAgent {
           }
           if (dataLines.length > 0) {
             const raw = dataLines.join("\n");
+            if (raw === "[DONE]") {
+              sep = buf.indexOf("\n\n");
+              continue;
+            }
             let msg: any;
             try { msg = JSON.parse(raw); } catch { msg = null; }
             if (msg && msg.id === id) {
@@ -188,16 +318,85 @@ export class UpstreamAgent {
               }
               return msg.result as UpstreamCallResult;
             }
-            // else: server→client request (elicitation/create) or another
+            // Otherwise it's a server→client message. Notifications
-            // frame; ignore — elicit relaying not wired yet.
+            // (no id) are best-effort. Requests (has id, has method), such
+            // as elicitation/create, must be relayed and answered upstream.
+            if (msg && typeof msg.method === "string" && !("id" in msg) && handlers.onNotification) {
+              try {
+                await handlers.onNotification({
+                  method: msg.method,
+                  params: msg.params as Record<string, unknown> | undefined,
+                });
+              } catch {
+                // Best-effort; never break the stream on a notify error.
+              }
+            } else if (msg && typeof msg.method === "string" && "id" in msg) {
+              await this.relayServerRequest(upstreamSessionId, {
+                id: msg.id as string | number,
+                method: msg.method,
+                params: msg.params as Record<string, unknown> | undefined,
+              }, handlers.onRequest);
+              resetIdleTimer();
+            }
           }
           sep = buf.indexOf("\n\n");
         }
         if (done) break;
       }
     } finally {
+      clearTimeout(idleTimer);
       try { reader.releaseLock(); } catch { /* ignore */ }
     }
     throw new UpstreamError(this.cfg.name, 502, "stream ended without result");
   }
+
+  private async relayServerRequest(
+    upstreamSessionId: string | null,
+    req: UpstreamServerRequest,
+    onRequest?: (req: UpstreamServerRequest) => unknown | Promise<unknown>,
+  ): Promise<void> {
+    if (!upstreamSessionId) {
+      throw new UpstreamError(
+        this.cfg.name,
+        502,
+        `upstream request ${req.method} missing Mcp-Session-Id`,
+      );
+    }
+    let envelope: Record<string, unknown>;
+    try {
+      if (!onRequest) throw new Error(`no client request relay for ${req.method}`);
+      const result = await onRequest(req);
+      envelope = { jsonrpc: JSON_RPC, id: req.id, result };
+    } catch (err) {
+      envelope = {
+        jsonrpc: JSON_RPC,
+        id: req.id,
+        error: {
+          code: -32603,
+          message: err instanceof Error ? err.message : String(err),
+        },
+      };
+    }
+
+    const headers: Record<string, string> = {
+      "content-type": "application/json",
+      accept: "application/json",
+      "mcp-session-id": upstreamSessionId,
+    };
+    const token = await this.bearerToken();
+    if (token) headers.authorization = `Bearer ${token}`;
+    const resp = await fetch(this.mcpUrl(), {
+      method: "POST",
+      headers,
+      body: JSON.stringify(envelope),
+    });
+    if (!resp.ok) {
+      const text = await resp.text().catch(() => "");
+      throw new UpstreamError(
+        this.cfg.name,
+        resp.status,
+        text || `failed to deliver response for ${req.method}`,
+      );
+    }
+  }
 }

tests/api.test.ts

@@ -0,0 +1,50 @@
+import { test, beforeEach, afterEach } from "node:test";
+import assert from "node:assert/strict";
+
+import { ApiError, ControlPlaneClient } from "../src/api.js";
+
+const originalFetch = globalThis.fetch;
+
+beforeEach(() => {
+  globalThis.fetch = originalFetch;
+});
+
+afterEach(() => {
+  globalThis.fetch = originalFetch;
+});
+
+test("ControlPlaneClient reports non-JSON error bodies without rereading them", async () => {
+  globalThis.fetch = (async () =>
+    new Response("<h1>upstream unavailable</h1>", {
+      status: 502,
+      statusText: "Bad Gateway",
+      headers: { "content-type": "text/html" },
+    })) as typeof fetch;
+
+  await assert.rejects(
+    () => new ControlPlaneClient("https://api.example.test", "token").me(),
+    (err) => {
+      assert.ok(err instanceof ApiError);
+      assert.equal(err.status, 502);
+      assert.match(err.message, /upstream unavailable/);
+      assert.doesNotMatch(err.message, /Body is unusable/);
+      return true;
+    },
+  );
+});
+
+test("ControlPlaneClient forwards bearer tokens and extracts JSON error detail", async () => {
+  globalThis.fetch = (async (_input: RequestInfo | URL, init?: RequestInit) => {
+    const headers = init?.headers as Record<string, string>;
+    assert.equal(headers.authorization, "Bearer access-token");
+    return new Response(JSON.stringify({ detail: "keycloak email is not verified" }), {
+      status: 403,
+      headers: { "content-type": "application/json" },
+    });
+  }) as typeof fetch;
+
+  await assert.rejects(
+    () => new ControlPlaneClient("https://api.example.test", "access-token").me(),
+    /API 403: keycloak email is not verified/,
+  );
+});

tests/config.test.ts

@@ -5,6 +5,7 @@ import path from "node:path";
 import { promises as fs } from "node:fs";
 
 import { addAgent, loadConfig, removeAgent } from "../src/config.js";
+import { loadCredentials, saveCredentials } from "../src/credentials.js";
 
 let originalHome: string | undefined;
 let tmpHome: string;
@@ -44,3 +45,31 @@ test("removeAgent reports hit/miss", async () => {
   const cfg = await loadConfig();
   assert.equal(cfg.agents.length, 0);
 });
+
+test("credentials preserve Keycloak token metadata", async () => {
+  await saveCredentials({
+    apiUrl: "https://api.example.test",
+    token: "access-token",
+    refreshToken: "refresh-token",
+    expiresAt: 12345,
+    email: "dev@example.test",
+    userId: 7,
+    bucket: "user-7-files",
+    issuer: "https://auth.example.test/realms/a2a",
+    clientId: "a2acloud-cli",
+    scope: "openid email mcp:invoke",
+  });
+
+  assert.deepEqual(await loadCredentials(), {
+    apiUrl: "https://api.example.test",
+    token: "access-token",
+    refreshToken: "refresh-token",
+    expiresAt: 12345,
+    email: "dev@example.test",
+    userId: 7,
+    bucket: "user-7-files",
+    issuer: "https://auth.example.test/realms/a2a",
+    clientId: "a2acloud-cli",
+    scope: "openid email mcp:invoke",
+  });
+});

tests/gateway.test.ts

@@ -9,7 +9,7 @@ import {
   ListToolsRequestSchema,
 } from "@modelcontextprotocol/sdk/types.js";
 
-import { buildGateway, SEP } from "../src/gateway.js";
+import { buildGateway, SEP, type GatewayOptions } from "../src/gateway.js";
 import type { EnabledAgent } from "../src/config.js";
 
 /**
@@ -70,25 +70,65 @@ function startFakeAgent(opts: {
   });
 }
 
+function startSseAgent(opts: {
+  onCall: (body: any, res: http.ServerResponse) => void;
+}): Promise<{ url: string; close: () => Promise<void>; requests: any[] }> {
+  const requests: any[] = [];
+  return new Promise((resolve) => {
+    const server = http.createServer((req, res) => {
+      if (req.method !== "POST" || req.url !== "/mcp") {
+        res.statusCode = 404;
+        res.end();
+        return;
+      }
+      const chunks: Buffer[] = [];
+      req.on("data", (c) => chunks.push(c));
+      req.on("end", () => {
+        const body = JSON.parse(Buffer.concat(chunks).toString("utf8"));
+        requests.push({ body });
+        if (body.method !== "tools/call") {
+          res.setHeader("content-type", "application/json");
+          res.end(JSON.stringify({ jsonrpc: "2.0", id: body.id, result: {} }));
+          return;
+        }
+        res.writeHead(200, { "content-type": "text/event-stream" });
+        opts.onCall(body, res);
+      });
+    });
+    server.listen(0, "127.0.0.1", () => {
+      const { port } = server.address() as AddressInfo;
+      resolve({
+        url: `http://127.0.0.1:${port}`,
+        requests,
+        close: () => new Promise((r) => server.close(() => r())),
+      });
+    });
+  });
+}
+
 /** Drive an in-process gateway Server via the SDK's request handlers. */
 async function callGateway(
   agents: EnabledAgent[],
   token: string | null,
   method: "tools/list" | "tools/call",
   params: any,
+  opts: {
+    gateway?: Partial<Omit<GatewayOptions, "agents" | "token" | "server">>;
+    extra?: any;
+  } = {},
 ) {
   const mcpServer = new Server(
     { name: "test", version: "0.0.0" },
     { capabilities: { tools: { listChanged: false } } },
   );
-  await buildGateway({ agents, token, server: mcpServer });
+  await buildGateway({ agents, token, server: mcpServer, ...opts.gateway });
   // Reach into the server's registered handlers — the SDK exposes them via
   // a private map, so we route through the public schema-keyed lookup we
   // know is wired up.
   const schema = method === "tools/list" ? ListToolsRequestSchema : CallToolRequestSchema;
   const handler = (mcpServer as any)._requestHandlers.get(schema.shape.method.value);
   if (!handler) throw new Error("handler not registered");
-  return handler({ method, params }, {} as any);
+  return handler({ method, params }, opts.extra ?? ({} as any));
 }
 
 test("tools/list aggregates upstream tools with agent prefix", async () => {
@@ -209,3 +249,168 @@ test("tools/call surfaces upstream JSON-RPC error as soft error", async () => {
     await upstream.close();
   }
 });
+
+test("tools/call forwards upstream SSE progress notifications", async () => {
+  const upstream = await startSseAgent({
+    onCall: (body, res) => {
+      res.write(
+        `data: ${JSON.stringify({
+          jsonrpc: "2.0",
+          method: "notifications/progress",
+          params: { progressToken: "p1", message: "halfway" },
+        })}\n\n`,
+      );
+      res.end(
+        `data: ${JSON.stringify({
+          jsonrpc: "2.0",
+          id: body.id,
+          result: {
+            content: [{ type: "text", text: "done" }],
+            isError: false,
+          },
+        })}\n\n`,
+      );
+    },
+  });
+  const notifications: any[] = [];
+  try {
+    const result = await callGateway(
+      [{ name: "streamer", url: upstream.url, addedAt: "now" }],
+      null,
+      "tools/call",
+      { name: `streamer${SEP}build`, arguments: {} },
+      {
+        extra: {
+          requestId: "req-1",
+          sendNotification: async (notification: any) => {
+            notifications.push(notification);
+          },
+        },
+      },
+    );
+    assert.equal(result.isError, false);
+    assert.deepEqual(notifications, [
+      {
+        method: "notifications/progress",
+        params: { progressToken: "p1", message: "halfway" },
+      },
+    ]);
+  } finally {
+    await upstream.close();
+  }
+});
+
+test("tools/call reports upstream SSE idle timeout", async () => {
+  const upstream = await startSseAgent({
+    onCall: (_body, res) => {
+      res.write(": connected\n\n");
+    },
+  });
+  try {
+    const result = await callGateway(
+      [{ name: "slow", url: upstream.url, addedAt: "now" }],
+      null,
+      "tools/call",
+      { name: `slow${SEP}build`, arguments: {} },
+      { gateway: { upstreamStreamIdleTimeoutMs: 25 } },
+    );
+    assert.equal(result.isError, true);
+    assert.match(result.content[0].text, /stream idle timeout/);
+  } finally {
+    await upstream.close();
+  }
+});
+
+test("tools/call relays upstream elicitation requests", async () => {
+  let resolveResponse: (() => void) | undefined;
+  const responseSeen = new Promise<void>((resolve) => { resolveResponse = resolve; });
+  let upstreamResponse: any = null;
+  const requests: any[] = [];
+  const server = http.createServer((req, res) => {
+    if (req.method !== "POST" || req.url !== "/mcp") {
+      res.statusCode = 404;
+      res.end();
+      return;
+    }
+    const chunks: Buffer[] = [];
+    req.on("data", (c) => chunks.push(c));
+    req.on("end", () => {
+      const body = JSON.parse(Buffer.concat(chunks).toString("utf8"));
+      requests.push({ headers: req.headers, body });
+      if (body.method === "tools/call") {
+        res.writeHead(200, {
+          "content-type": "text/event-stream",
+          "Mcp-Session-Id": "sess-1",
+        });
+        res.write(
+          `data: ${JSON.stringify({
+            jsonrpc: "2.0",
+            id: "ask-1",
+            method: "elicitation/create",
+            params: { message: "Name?", requestedSchema: { type: "object" } },
+          })}\n\n`,
+        );
+        void responseSeen.then(() => {
+          res.end(
+            `data: ${JSON.stringify({
+              jsonrpc: "2.0",
+              id: body.id,
+              result: {
+                content: [{ type: "text", text: upstreamResponse.result.content.answer }],
+                isError: false,
+              },
+            })}\n\n`,
+          );
+        });
+        return;
+      }
+      if (body.id === "ask-1" && req.headers["mcp-session-id"] === "sess-1") {
+        upstreamResponse = body;
+        res.statusCode = 202;
+        res.end();
+        resolveResponse?.();
+        return;
+      }
+      res.setHeader("content-type", "application/json");
+      res.end(JSON.stringify({ jsonrpc: "2.0", id: body.id, result: {} }));
+    });
+  });
+  const upstream = await new Promise<{ url: string; close: () => Promise<void> }>((resolve) => {
+    server.listen(0, "127.0.0.1", () => {
+      const { port } = server.address() as AddressInfo;
+      resolve({
+        url: `http://127.0.0.1:${port}`,
+        close: () => new Promise((r) => server.close(() => r())),
+      });
+    });
+  });
+  const clientRequests: any[] = [];
+  try {
+    const result = await callGateway(
+      [{ name: "human", url: upstream.url, addedAt: "now" }],
+      null,
+      "tools/call",
+      { name: `human${SEP}run_demo`, arguments: {} },
+      {
+        extra: {
+          requestId: "req-elicitation",
+          sendNotification: async () => {},
+          sendRequest: async (request: any) => {
+            clientRequests.push(request);
+            return { action: "accept", content: { answer: "Alice" } };
+          },
+        },
+      },
+    );
+    assert.equal(result.isError, false);
+    assert.equal(result.content[0].text, "Alice");
+    assert.equal(clientRequests[0].method, "elicitation/create");
+    assert.deepEqual(upstreamResponse.result, {
+      action: "accept",
+      content: { answer: "Alice" },
+    });
+    assert.ok(requests.some((r) => r.body.id === "ask-1"));
+  } finally {
+    await upstream.close();
+  }
+});

tests/oauth.test.ts

@@ -0,0 +1,95 @@
+import { test, beforeEach, afterEach } from "node:test";
+import assert from "node:assert/strict";
+import os from "node:os";
+import path from "node:path";
+import { promises as fs } from "node:fs";
+
+import { loadCredentials, saveCredentials } from "../src/credentials.js";
+import {
+  buildAuthorizationUrl,
+  createPkcePair,
+  refreshCredentialsIfNeeded,
+} from "../src/oauth.js";
+
+let originalHome: string | undefined;
+let tmpHome: string;
+const originalFetch = globalThis.fetch;
+
+beforeEach(async () => {
+  originalHome = process.env.HOME;
+  tmpHome = await fs.mkdtemp(path.join(os.tmpdir(), "a2a-mcp-oauth-"));
+  process.env.HOME = tmpHome;
+  globalThis.fetch = originalFetch;
+});
+
+afterEach(async () => {
+  if (originalHome !== undefined) process.env.HOME = originalHome;
+  else delete process.env.HOME;
+  globalThis.fetch = originalFetch;
+  await fs.rm(tmpHome, { recursive: true, force: true });
+});
+
+test("PKCE authorization URL requests Keycloak code flow", () => {
+  const pkce = createPkcePair();
+  assert.ok(pkce.verifier.length > 40);
+  assert.ok(pkce.challenge.length > 40);
+
+  const url = new URL(
+    buildAuthorizationUrl({
+      authorizationEndpoint: "https://auth.example.test/auth",
+      clientId: "a2acloud-cli",
+      redirectUri: "http://127.0.0.1:41873/callback",
+      scope: "openid email mcp:invoke",
+      state: "state-123",
+      codeChallenge: pkce.challenge,
+    }),
+  );
+
+  assert.equal(url.searchParams.get("client_id"), "a2acloud-cli");
+  assert.equal(url.searchParams.get("response_type"), "code");
+  assert.equal(url.searchParams.get("code_challenge_method"), "S256");
+  assert.equal(url.searchParams.get("scope"), "openid email mcp:invoke");
+});
+
+test("refreshCredentialsIfNeeded refreshes expired Keycloak access token", async () => {
+  const calls: Array<{ url: string; body?: string }> = [];
+  globalThis.fetch = (async (input: RequestInfo | URL, init?: RequestInit) => {
+    const url = String(input);
+    calls.push({ url, body: init?.body?.toString() });
+    if (url.endsWith("/.well-known/openid-configuration")) {
+      return new Response(
+        JSON.stringify({
+          authorization_endpoint: "https://auth.example.test/auth",
+          token_endpoint: "https://auth.example.test/token",
+        }),
+        { status: 200, headers: { "content-type": "application/json" } },
+      );
+    }
+    return new Response(
+      JSON.stringify({
+        access_token: "fresh-token",
+        refresh_token: "fresh-refresh",
+        expires_in: 3600,
+        scope: "openid email mcp:invoke",
+      }),
+      { status: 200, headers: { "content-type": "application/json" } },
+    );
+  }) as typeof fetch;
+
+  await saveCredentials({
+    apiUrl: "https://api.example.test",
+    token: "old-token",
+    refreshToken: "refresh-token",
+    expiresAt: 1,
+    email: "dev@example.test",
+    issuer: "https://auth.example.test",
+    clientId: "a2acloud-cli",
+  });
+
+  const refreshed = await refreshCredentialsIfNeeded((await loadCredentials())!);
+
+  assert.equal(refreshed.token, "fresh-token");
+  assert.equal(refreshed.refreshToken, "fresh-refresh");
+  assert.equal((await loadCredentials())!.token, "fresh-token");
+  assert.ok(calls.some((call) => call.body?.includes("grant_type=refresh_token")));
+});