diff --git a/crates/aisix-a2a/src/bridge.rs b/crates/aisix-a2a/src/bridge.rs index bb2def0b..3385ab91 100644 --- a/crates/aisix-a2a/src/bridge.rs +++ b/crates/aisix-a2a/src/bridge.rs @@ -11,9 +11,14 @@ //! exposed to the calling client, which presents only its AISIX key. //! //! Wire references (verified against the A2A specification): -//! - Agent card discovery: `https://{domain}/.well-known/agent-card.json`, -//! an RFC 8615 well-known URI resolved at the domain origin. +//! - Agent card discovery: the RFC 8615 well-known URI +//! `https://{domain}/.well-known/agent-card.json`. The spec resolves it at the +//! origin, but real deployments also publish it under the agent's own path +//! prefix, so both bases are tried — see [`HttpBridge::agent_card_urls`]. //! +//! - Every request announces the wire version it speaks in the `A2A-Version` +//! header; an agent that receives none must assume `0.3`. The version is +//! pinned per agent on the `A2aAgent` resource. //! - `message/send` is a JSON-RPC 2.0 method whose envelope differs between the //! A2A 0.3 and 1.0 wire formats. This bridge forwards the caller's request //! verbatim and does not translate between versions, so the method name and @@ -21,9 +26,9 @@ //! use std::sync::OnceLock; -use std::time::Duration; +use std::time::{Duration, Instant}; -use aisix_core::{A2aAgent, A2aAuthType}; +use aisix_core::{A2aAgent, A2aAuthType, A2aProtocolVersion}; use async_trait::async_trait; use futures::StreamExt; use serde::{Deserialize, Serialize}; @@ -39,8 +44,17 @@ pub const DEFAULT_UPSTREAM_TIMEOUT: Duration = Duration::from_secs(30); /// Header carrying the gateway-held key for `api_key` upstream auth. const API_KEY_HEADER: &str = "x-api-key"; -/// Standard RFC 8615 well-known path for an A2A agent card. -const AGENT_CARD_PATH: &str = "/.well-known/agent-card.json"; +/// Header naming the A2A wire version the client speaks. The spec requires a +/// client to send it on every request and requires an agent to read an absent +/// value as `0.3`, so an unlabelled call to a 1.0 agent is not merely untidy — +/// the agent answers `VersionNotSupportedError` and the call never lands. +const VERSION_HEADER: &str = "A2A-Version"; + +/// Well-known agent-card paths, current spec first. RFC 8615 resolves a +/// well-known URI at the origin, but platforms that multiplex tenants under a +/// path prefix publish the card relative to the agent's own path instead, so +/// each of these is tried against both bases — see [`HttpBridge::agent_card_urls`]. +const AGENT_CARD_PATHS: [&str; 2] = ["/.well-known/agent-card.json", "/.well-known/agent.json"]; /// Hard cap on an upstream response body the gateway will buffer. A registered /// agent is semi-trusted, but a compromised or misbehaving one must not be able @@ -129,10 +143,13 @@ impl std::fmt::Debug for A2aAuth { pub struct A2aUpstream { /// The agent's A2A service endpoint, where JSON-RPC requests are sent, e.g. /// `https://agents.example.com/a2a`. The agent card is discovered at the - /// well-known path relative to this URL's origin. + /// well-known paths relative to this URL, then to its origin. pub url: String, /// Upstream authentication, held gateway-side. pub auth: A2aAuth, + /// The wire version this agent speaks, announced to it on every request in + /// the `A2A-Version` header. + pub protocol_version: A2aProtocolVersion, /// Per-operation deadline. Defaults to [`DEFAULT_UPSTREAM_TIMEOUT`]. pub timeout: Duration, } @@ -144,6 +161,7 @@ impl std::fmt::Debug for A2aUpstream { f.debug_struct("A2aUpstream") .field("url", &self.url) .field("auth", &self.auth) + .field("protocol_version", &self.protocol_version) .field("timeout", &self.timeout) .finish() } @@ -199,6 +217,7 @@ pub fn upstream_from_a2a_agent(agent: &A2aAgent) -> A2aUpstream { A2aUpstream { url: agent.url.clone(), auth, + protocol_version: agent.protocol_version, timeout, } } @@ -249,8 +268,12 @@ impl HttpBridge { } } - /// Apply the gateway-held upstream credential to an outgoing request. - fn apply_auth(&self, req: reqwest::RequestBuilder) -> reqwest::RequestBuilder { + /// Apply the gateway-held upstream credential and announce the wire version + /// this agent is pinned to. Both belong on every outgoing request: the + /// credential because the gateway is the one holding it, the version + /// because an agent that receives no `A2A-Version` must assume `0.3`. + fn prepare(&self, req: reqwest::RequestBuilder) -> reqwest::RequestBuilder { + let req = req.header(VERSION_HEADER, self.upstream.protocol_version.as_wire_str()); match &self.upstream.auth { A2aAuth::None => req, A2aAuth::Bearer(token) => req.bearer_auth(token), @@ -258,23 +281,48 @@ impl HttpBridge { } } - /// Resolve the agent-card well-known URI from the service endpoint's origin - /// (RFC 8615): scheme + host + port, with the well-known path. - fn agent_card_url(&self) -> Result { - let mut url = reqwest::Url::parse(&self.upstream.url) + /// Candidate agent-card URIs, most specific first: each well-known path + /// resolved against the registered URL's own path, then against its origin. + /// + /// Both bases are real deployments. An agent that owns its domain publishes + /// at the RFC 8615 origin URI — the only shape this bridge used to try, and + /// what every agent registered before now still serves. A platform that + /// multiplexes tenants under a path prefix (or a self-hosted agent behind an + /// ingress path) publishes relative to its own path instead — the origin + /// there belongs to the platform, not to any one agent. Resolving only one + /// of the two locks out the other, so both are tried, specific first. + /// + /// A registered URL that is already at the origin yields the origin + /// candidates alone — the two bases coincide and probing twice is waste. + fn agent_card_urls(&self) -> Result, A2aError> { + let base = reqwest::Url::parse(&self.upstream.url) .map_err(|e| A2aError::Connect(format!("invalid upstream url: {e}")))?; - url.set_path(AGENT_CARD_PATH); - url.set_query(None); - Ok(url) + let prefix = base.path().trim_end_matches('/').to_string(); + let mut urls = Vec::with_capacity(AGENT_CARD_PATHS.len() * 2); + for path in AGENT_CARD_PATHS { + if !prefix.is_empty() { + let mut relative = base.clone(); + relative.set_path(&format!("{prefix}{path}")); + relative.set_query(None); + urls.push(relative); + } + let mut origin = base.clone(); + origin.set_path(path); + origin.set_query(None); + urls.push(origin); + } + Ok(urls) } -} -#[async_trait] -impl A2aBridge for HttpBridge { - async fn fetch_agent_card(&self) -> Result { - let url = self.agent_card_url()?; + /// Fetch and parse the card at exactly one candidate URI, within whatever + /// is left of the fetch's overall budget. + async fn fetch_card_at( + &self, + url: reqwest::Url, + budget: Duration, + ) -> Result { let resp = self - .apply_auth(self.client.get(url).timeout(self.upstream.timeout)) + .prepare(self.client.get(url).timeout(budget)) .send() .await .map_err(|e| A2aError::Connect(e.to_string()))?; @@ -288,10 +336,48 @@ impl A2aBridge for HttpBridge { serde_json::from_slice::(&bytes) .map_err(|e| A2aError::Request(format!("malformed agent card: {e}"))) } +} + +#[async_trait] +impl A2aBridge for HttpBridge { + async fn fetch_agent_card(&self) -> Result { + // Any failure moves to the next candidate, not just a 404: an upstream + // that does not route the URI answers with whatever its catch-all + // returns (the 405 in #913), and a prefix that happens to resolve can + // still hand back something that is not a card. + // + // `timeout_ms` bounds the card fetch as ONE upstream operation, so the + // whole walk shares a single deadline instead of handing each candidate + // a fresh one. Otherwise a slow or hung upstream stretches the fetch to + // `candidates × timeout_ms` — four times what the operator configured — + // and pins a gateway request for the duration. + let deadline = Instant::now() + self.upstream.timeout; + let mut last_err = None; + for url in self.agent_card_urls()? { + let budget = deadline.saturating_duration_since(Instant::now()); + if budget.is_zero() { + tracing::debug!( + agent_url = %self.upstream.url, + "A2A agent card fetch exhausted its deadline before trying every candidate" + ); + break; + } + match self.fetch_card_at(url.clone(), budget).await { + Ok(card) => return Ok(card), + Err(err) => { + tracing::debug!(%url, error = %err, "A2A agent card candidate did not answer"); + last_err = Some(err); + } + } + } + Err(last_err.unwrap_or_else(|| { + A2aError::Connect("agent card fetch exceeded its timeout".to_string()) + })) + } async fn send(&self, request: &serde_json::Value) -> Result { let resp = self - .apply_auth( + .prepare( self.client .post(&self.upstream.url) .timeout(self.upstream.timeout) @@ -368,22 +454,82 @@ mod tests { let up = A2aUpstream { url: "https://x/a2a".into(), auth: A2aAuth::Bearer("super-secret".into()), + protocol_version: A2aProtocolVersion::V1_0, timeout: DEFAULT_UPSTREAM_TIMEOUT, }; assert!(!format!("{up:?}").contains("super-secret")); } - #[test] - fn agent_card_url_is_origin_well_known() { - let bridge = HttpBridge::new(A2aUpstream { - url: "https://agents.example.com/a2a/v1".into(), + fn bridge_at(url: &str) -> HttpBridge { + HttpBridge::new(A2aUpstream { + url: url.into(), auth: A2aAuth::None, + protocol_version: A2aProtocolVersion::V1_0, timeout: DEFAULT_UPSTREAM_TIMEOUT, - }); + }) + } + + fn card_candidates(url: &str) -> Vec { + bridge_at(url) + .agent_card_urls() + .unwrap() + .iter() + .map(|u| u.as_str().to_string()) + .collect() + } + + #[test] + fn card_candidates_try_the_registered_path_before_the_origin() { + // #913: `set_path` used to replace the path outright, so a path-hosted + // agent was asked for a card URI it never publishes. The prefix now + // survives, and the origin URI stays as the fallback so agents + // registered under the old behaviour keep resolving. + assert_eq!( + card_candidates("https://agents.example.com/v3/a2a/serve/abc"), + vec![ + "https://agents.example.com/v3/a2a/serve/abc/.well-known/agent-card.json", + "https://agents.example.com/.well-known/agent-card.json", + "https://agents.example.com/v3/a2a/serve/abc/.well-known/agent.json", + "https://agents.example.com/.well-known/agent.json", + ] + ); + } + + #[test] + fn card_candidates_collapse_when_the_agent_owns_the_origin() { + // The two bases coincide here, so the relative candidate would be a + // byte-identical second request. A trailing slash is the same case. + let expected = vec![ + "https://agents.example.com/.well-known/agent-card.json", + "https://agents.example.com/.well-known/agent.json", + ]; + assert_eq!(card_candidates("https://agents.example.com"), expected); + assert_eq!(card_candidates("https://agents.example.com/"), expected); + } + + #[test] + fn card_candidates_drop_the_query_and_keep_the_port() { + assert_eq!( + card_candidates("http://127.0.0.1:8080/a2a?tenant=acme"), + vec![ + "http://127.0.0.1:8080/a2a/.well-known/agent-card.json", + "http://127.0.0.1:8080/.well-known/agent-card.json", + "http://127.0.0.1:8080/a2a/.well-known/agent.json", + "http://127.0.0.1:8080/.well-known/agent.json", + ] + ); + } + + #[test] + fn upstream_carries_the_pinned_protocol_version() { + let mut pinned_03 = agent("none"); + pinned_03.protocol_version = A2aProtocolVersion::V0_3; assert_eq!( - bridge.agent_card_url().unwrap().as_str(), - "https://agents.example.com/.well-known/agent-card.json" + upstream_from_a2a_agent(&pinned_03).protocol_version, + A2aProtocolVersion::V0_3 ); + assert_eq!(A2aProtocolVersion::V1_0.as_wire_str(), "1.0"); + assert_eq!(A2aProtocolVersion::V0_3.as_wire_str(), "0.3"); } #[test] diff --git a/crates/aisix-a2a/tests/upstream_roundtrip.rs b/crates/aisix-a2a/tests/upstream_roundtrip.rs index 39c54464..dc2c7e83 100644 --- a/crates/aisix-a2a/tests/upstream_roundtrip.rs +++ b/crates/aisix-a2a/tests/upstream_roundtrip.rs @@ -9,25 +9,47 @@ use std::net::SocketAddr; use std::sync::atomic::{AtomicUsize, Ordering}; use std::sync::Arc; +use std::time::{Duration, Instant}; use aisix_a2a::{A2aAuth, A2aBridge, A2aError, A2aUpstream, HttpBridge, DEFAULT_UPSTREAM_TIMEOUT}; +use aisix_core::A2aProtocolVersion; use axum::http::header::LOCATION; use axum::http::{HeaderMap, StatusCode}; use axum::response::IntoResponse; -use axum::routing::{get, post}; +use axum::routing::{any, get, post}; use axum::{Json, Router}; use serde_json::{json, Value}; +/// An upstream pinned to A2A 1.0 — the default for a registered agent. +fn upstream(url: String, auth: A2aAuth) -> A2aUpstream { + A2aUpstream { + url, + auth, + protocol_version: A2aProtocolVersion::V1_0, + timeout: DEFAULT_UPSTREAM_TIMEOUT, + } +} + +/// Read back the `A2A-Version` an inbound request carried, or `null`. +fn seen_version(headers: &HeaderMap) -> Value { + headers + .get("a2a-version") + .and_then(|v| v.to_str().ok()) + .map(|v| Value::String(v.to_string())) + .unwrap_or(Value::Null) +} + /// A minimal upstream A2A agent: serves its card at the well-known URI and /// answers JSON-RPC by echoing back the request id and the credentials it saw, /// so the test can assert what the gateway forwarded. async fn spawn_agent() -> SocketAddr { - async fn card() -> Json { + async fn card(headers: HeaderMap) -> Json { Json(json!({ "name": "Test Agent", "url": "https://upstream.example.com/a2a", "version": "1.0.0", - "skills": [{"id": "echo", "name": "Echo"}] + "skills": [{"id": "echo", "name": "Echo"}], + "echoed_version": seen_version(&headers), })) } @@ -49,6 +71,7 @@ async fn spawn_agent() -> SocketAddr { "status": {"state": "completed"}, "echoed_auth": auth, "echoed_api_key": api_key, + "echoed_version": seen_version(&headers), } })) } @@ -84,11 +107,10 @@ fn message_send(id: &str) -> Value { #[tokio::test] async fn fetches_card_and_forwards_bearer() { let addr = spawn_agent().await; - let bridge = HttpBridge::new(A2aUpstream { - url: format!("http://{addr}/a2a"), - auth: A2aAuth::Bearer("tok-123".into()), - timeout: DEFAULT_UPSTREAM_TIMEOUT, - }); + let bridge = HttpBridge::new(upstream( + format!("http://{addr}/a2a"), + A2aAuth::Bearer("tok-123".into()), + )); let card = bridge.fetch_agent_card().await.unwrap(); assert_eq!(card.name, "Test Agent"); @@ -105,11 +127,10 @@ async fn fetches_card_and_forwards_bearer() { #[tokio::test] async fn forwards_api_key_header() { let addr = spawn_agent().await; - let bridge = HttpBridge::new(A2aUpstream { - url: format!("http://{addr}/a2a"), - auth: A2aAuth::ApiKey("k-secret".into()), - timeout: DEFAULT_UPSTREAM_TIMEOUT, - }); + let bridge = HttpBridge::new(upstream( + format!("http://{addr}/a2a"), + A2aAuth::ApiKey("k-secret".into()), + )); let resp = bridge.send(&message_send("req-2")).await.unwrap(); assert_eq!(resp["result"]["echoed_api_key"], "k-secret"); @@ -120,11 +141,7 @@ async fn forwards_api_key_header() { #[tokio::test] async fn sends_no_credential_when_none() { let addr = spawn_agent().await; - let bridge = HttpBridge::new(A2aUpstream { - url: format!("http://{addr}/a2a"), - auth: A2aAuth::None, - timeout: DEFAULT_UPSTREAM_TIMEOUT, - }); + let bridge = HttpBridge::new(upstream(format!("http://{addr}/a2a"), A2aAuth::None)); let resp = bridge.send(&message_send("req-3")).await.unwrap(); assert!(resp["result"]["echoed_auth"].is_null()); @@ -165,11 +182,7 @@ async fn spawn_redirect_probe() -> (SocketAddr, Arc) { #[tokio::test] async fn refuses_to_follow_upstream_redirect() { let (addr, secret_hits) = spawn_redirect_probe().await; - let bridge = HttpBridge::new(A2aUpstream { - url: format!("http://{addr}/redirect"), - auth: A2aAuth::None, - timeout: DEFAULT_UPSTREAM_TIMEOUT, - }); + let bridge = HttpBridge::new(upstream(format!("http://{addr}/redirect"), A2aAuth::None)); let err = bridge.send(&message_send("r")).await.unwrap_err(); // The 302 surfaces as a non-success status error — it is NOT followed. @@ -183,3 +196,157 @@ async fn refuses_to_follow_upstream_redirect() { "the redirect target must NOT be fetched — the gateway must not chase upstream redirects" ); } + +/// An upstream that publishes its card ONLY under the agent's own path prefix +/// and answers every unrouted path with the catch-all `405` the real one +/// returns. This is the shape of any platform that multiplexes tenants under a +/// path, and of any self-hosted agent behind an ingress path. +async fn spawn_path_hosted_agent() -> SocketAddr { + async fn card(headers: HeaderMap) -> Json { + Json(json!({ + "name": "Path Hosted Agent", + "url": "https://upstream.example.com/v3/a2a/serve/agent-42", + "protocolVersion": "0.3.0", + "echoed_version": seen_version(&headers), + })) + } + + async fn rpc(Json(body): Json) -> Json { + Json(json!({ + "jsonrpc": "2.0", + "id": body["id"].clone(), + "result": {"kind": "task", "id": "task-path", "status": {"state": "completed"}} + })) + } + + let app = Router::new() + .route( + "/v3/a2a/serve/agent-42/.well-known/agent-card.json", + get(card), + ) + .route("/v3/a2a/serve/agent-42", post(rpc)) + .fallback(any(|| async { StatusCode::METHOD_NOT_ALLOWED })); + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = listener.local_addr().unwrap(); + tokio::spawn(async move { + axum::serve(listener, app.into_make_service()) + .await + .unwrap(); + }); + addr +} + +#[tokio::test] +async fn discovers_a_card_hosted_under_the_agent_path() { + // #913: the registered path used to be discarded, so the bridge asked the + // ORIGIN for a card this agent publishes only under its prefix, and took + // the catch-all 405 as the agent's answer. No configuration could unblock + // it — the one `url` field feeds both the card fetch and the RPC endpoint. + let addr = spawn_path_hosted_agent().await; + let bridge = HttpBridge::new(upstream( + format!("http://{addr}/v3/a2a/serve/agent-42"), + A2aAuth::None, + )); + + let card = bridge.fetch_agent_card().await.unwrap(); + assert_eq!(card.name, "Path Hosted Agent"); + // And the endpoint itself still resolves off the same registered URL. + let resp = bridge.send(&message_send("p-1")).await.unwrap(); + assert_eq!(resp["result"]["id"], "task-path"); +} + +#[tokio::test] +async fn still_discovers_a_card_hosted_at_the_origin() { + // The origin URI remains a candidate, so every agent registered while it + // was the ONLY candidate keeps resolving. `spawn_agent` serves its card + // there and nowhere else. + let addr = spawn_agent().await; + let bridge = HttpBridge::new(upstream(format!("http://{addr}/a2a"), A2aAuth::None)); + + assert_eq!(bridge.fetch_agent_card().await.unwrap().name, "Test Agent"); +} + +#[tokio::test] +async fn reports_a_failure_when_no_candidate_serves_a_card() { + let addr = spawn_path_hosted_agent().await; + let bridge = HttpBridge::new(upstream(format!("http://{addr}/nope"), A2aAuth::None)); + + let err = bridge.fetch_agent_card().await.unwrap_err(); + assert!( + matches!(err, A2aError::Connect(_)), + "exhausting every candidate must surface the upstream failure, got {err:?}" + ); +} + +#[tokio::test] +async fn announces_the_pinned_wire_version_on_every_upstream_call() { + // #911: the gateway sent no `A2A-Version` at all. The spec makes an agent + // read an absent value as 0.3, so a 1.0-pinned agent rejected every call + // with VersionNotSupportedError and `protocol_version` was inert. + let addr = spawn_agent().await; + + let pinned_10 = HttpBridge::new(upstream(format!("http://{addr}/a2a"), A2aAuth::None)); + assert_eq!( + pinned_10.fetch_agent_card().await.unwrap().rest["echoed_version"], + "1.0", + "the card fetch must announce the pinned version too" + ); + assert_eq!( + pinned_10.send(&message_send("v-1")).await.unwrap()["result"]["echoed_version"], + "1.0" + ); + + let pinned_03 = HttpBridge::new(A2aUpstream { + protocol_version: A2aProtocolVersion::V0_3, + ..upstream(format!("http://{addr}/a2a"), A2aAuth::None) + }); + assert_eq!( + pinned_03.fetch_agent_card().await.unwrap().rest["echoed_version"], + "0.3" + ); + assert_eq!( + pinned_03.send(&message_send("v-0")).await.unwrap()["result"]["echoed_version"], + "0.3" + ); +} + +/// An upstream that accepts the connection and never answers, so every +/// candidate URI burns its whole deadline. +async fn spawn_black_hole() -> SocketAddr { + let app = Router::new().fallback(any(|| async { + std::future::pending::<()>().await; + StatusCode::OK + })); + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = listener.local_addr().unwrap(); + tokio::spawn(async move { + axum::serve(listener, app.into_make_service()) + .await + .unwrap(); + }); + addr +} + +#[tokio::test] +async fn the_card_fetch_deadline_covers_the_whole_candidate_walk() { + // `timeout_ms` bounds the card fetch as ONE upstream operation. Walking + // several candidates must not hand each a fresh deadline, or a hung agent + // pins a gateway request for `candidates × timeout_ms`. + let addr = spawn_black_hole().await; + let bridge = HttpBridge::new(A2aUpstream { + timeout: Duration::from_millis(600), + ..upstream(format!("http://{addr}/a2a"), A2aAuth::None) + }); + + let started = Instant::now(); + let err = bridge.fetch_agent_card().await.unwrap_err(); + let elapsed = started.elapsed(); + + assert!(matches!(err, A2aError::Connect(_)), "got {err:?}"); + // Four candidates would be 2.4s if each restarted the clock. The bound is + // loose enough for a slow CI box while staying far below that. + assert!( + elapsed < Duration::from_millis(1500), + "card fetch took {elapsed:?}; the candidate walk must share ONE deadline" + ); +} diff --git a/crates/aisix-core/src/models/a2a_agent.rs b/crates/aisix-core/src/models/a2a_agent.rs index 2ea8d5be..6601d6ca 100644 --- a/crates/aisix-core/src/models/a2a_agent.rs +++ b/crates/aisix-core/src/models/a2a_agent.rs @@ -40,15 +40,19 @@ pub struct A2aAgent { )] pub name: String, - /// The upstream agent's base URL, such as `https://agents.example.com/a2a`. - /// AISIX reaches this URL over HTTP with the A2A JSON-RPC 2.0 protocol and - /// discovers the agent card relative to it. + /// The upstream agent's A2A service endpoint, such as + /// `https://agents.example.com/a2a`, where AISIX sends JSON-RPC 2.0 requests + /// over HTTP. AISIX looks for the agent card at the well-known path under + /// this URL's own path first, then under its origin, so both an agent that + /// owns its domain and one published under a path prefix are reachable + /// without extra configuration. #[schemars(length(min = 1))] pub url: String, - /// The A2A wire-format version AISIX uses for this agent. AISIX pins the - /// version explicitly so the served agent card and accepted requests stay - /// consistent. + /// The A2A wire-format version this agent speaks. AISIX announces it to the + /// agent in the `A2A-Version` header on every request, so it must match what + /// the agent actually serves: an agent reads an absent or mismatched version + /// as a protocol error and rejects the call. #[serde(default)] pub protocol_version: A2aProtocolVersion, @@ -106,6 +110,18 @@ pub enum A2aProtocolVersion { V0_3, } +impl A2aProtocolVersion { + /// The value this version carries on the wire, in the `A2A-Version` header + /// and in the `protocolVersion` field of an agent card. Kept in lockstep + /// with the `serde` renames above, which are the same strings. + pub fn as_wire_str(self) -> &'static str { + match self { + Self::V1_0 => "1.0", + Self::V0_3 => "0.3", + } + } +} + /// How the gateway authenticates to an upstream A2A agent. #[derive( Debug, Clone, Copy, Default, PartialEq, Eq, Serialize, Deserialize, schemars::JsonSchema, diff --git a/crates/aisix-proxy/src/a2a.rs b/crates/aisix-proxy/src/a2a.rs index 67eea0e1..26b5ae36 100644 --- a/crates/aisix-proxy/src/a2a.rs +++ b/crates/aisix-proxy/src/a2a.rs @@ -227,6 +227,7 @@ pub async fn a2a_agent_card( auth: AuthenticatedKey, AisixPath(agent): AisixPath, State(state): State, + uri: axum::http::Uri, headers: HeaderMap, ) -> Response { let snapshot = state.snapshot.load(); @@ -241,6 +242,21 @@ pub async fn a2a_agent_card( ) .into_response(); } + // Resolved BEFORE the upstream is contacted: without a public base there is + // no card this gateway can serve, and finding that out after the fetch only + // wastes an upstream round trip. + let Some(base) = gateway_base(&uri, &headers) else { + tracing::warn!( + agent = %agent, + "cannot derive the gateway's public base for an A2A agent card; refusing to serve one" + ); + return ( + StatusCode::INTERNAL_SERVER_ERROR, + "cannot determine the gateway's public address for this agent card", + ) + .into_response(); + }; + let upstream = upstream_from_a2a_agent(&entry.value); let bridge = HttpBridge::new(upstream); @@ -251,22 +267,56 @@ pub async fn a2a_agent_card( return (StatusCode::BAD_GATEWAY, err.to_string()).into_response(); } }; - // Rewrite the advertised service endpoint to the gateway so downstream - // callers route subsequent requests through `/a2a/`. Derived from - // the request's Host header (and forwarded scheme) since the gateway's - // public URL is not otherwise known here. - if let Some(base) = gateway_base(&headers) { - card.url = format!("{base}/a2a/{agent}"); - } + // Rewrite the advertised service endpoints to the gateway so downstream + // callers route subsequent requests through `/a2a/`. + rewrite_card_urls(&mut card, &format!("{base}/a2a/{agent}")); axum::Json(card).into_response() } -/// Reconstruct the gateway's public base (`scheme://host`) from request -/// headers: the `Host` header, and `X-Forwarded-Proto` when a proxy set it -/// (defaulting to `https`). Returns `None` when no Host header is present, in -/// which case the card's `url` is left as the upstream advertised it. -fn gateway_base(headers: &HeaderMap) -> Option { - let host = headers.get(header::HOST)?.to_str().ok()?; +/// Point every service URL the card advertises at the gateway. +/// +/// The top-level `url` is what a 0.3 caller reads. A 1.0 caller instead picks +/// its endpoint out of `supportedInterfaces` (`additionalInterfaces` on a 0.3 +/// card), so rewriting only the top level leaves it reading the upstream +/// address off the card and calling the agent directly — no auth, no quota, no +/// usage, and the internal address handed to the caller on the way past. +/// +/// Entries are rewritten in place rather than filtered out: this gateway serves +/// JSON-RPC over HTTP only, so an entry naming a transport it does not proxy +/// now points somewhere that will reject the caller. That is the intended +/// trade — failing loudly beats silently bypassing governance. +fn rewrite_card_urls(card: &mut aisix_a2a::AgentCard, gateway_url: &str) { + card.url = gateway_url.to_string(); + for key in ["supportedInterfaces", "additionalInterfaces"] { + let Some(serde_json::Value::Array(interfaces)) = card.rest.get_mut(key) else { + continue; + }; + for interface in interfaces { + if let Some(url) = interface.get_mut("url") { + *url = serde_json::Value::String(gateway_url.to_string()); + } + } + } +} + +/// Reconstruct the gateway's public base (`scheme://host`) for a request: the +/// authority, and `X-Forwarded-Proto` when a proxy set it (defaulting to +/// `https`). +/// +/// The authority comes from the `Host` header, falling back to the request +/// URI's own authority. HTTP/2 carries it there as the `:authority` +/// pseudo-header and sends no `Host` at all — and this listener negotiates h2 +/// whenever `proxy.tls` is set, so header-only lookup finds nothing on exactly +/// the deployments most likely to be in production. +/// +/// `None` means no card can be served: the caller must fail rather than hand +/// back a card still advertising the upstream's own address. +fn gateway_base(uri: &axum::http::Uri, headers: &HeaderMap) -> Option { + let host = headers + .get(header::HOST) + .and_then(|v| v.to_str().ok()) + .filter(|v| !v.is_empty()) + .or_else(|| uri.authority().map(|a| a.as_str()))?; let scheme = headers .get("x-forwarded-proto") .and_then(|v| v.to_str().ok()) @@ -351,22 +401,88 @@ mod tests { #[test] fn gateway_base_uses_forwarded_proto_then_defaults_https() { + let uri = axum::http::Uri::from_static("/a2a/x/.well-known/agent-card.json"); let mut headers = HeaderMap::new(); headers.insert(header::HOST, "gw.example.com".parse().unwrap()); assert_eq!( - gateway_base(&headers).as_deref(), + gateway_base(&uri, &headers).as_deref(), Some("https://gw.example.com") ); headers.insert("x-forwarded-proto", "http".parse().unwrap()); assert_eq!( - gateway_base(&headers).as_deref(), + gateway_base(&uri, &headers).as_deref(), Some("http://gw.example.com") ); } #[test] - fn gateway_base_is_none_without_host() { - assert_eq!(gateway_base(&HeaderMap::new()), None); + fn gateway_base_falls_back_to_the_uri_authority() { + // HTTP/2 sends no `Host` — the authority arrives as the `:authority` + // pseudo-header, which lands on the URI. This listener negotiates h2 + // whenever `proxy.tls` is set, so a header-only lookup finds nothing + // there and the card would otherwise be served still advertising the + // upstream's address. + let h2_uri = axum::http::Uri::from_static("https://gw.example.com/a2a/x"); + assert_eq!( + gateway_base(&h2_uri, &HeaderMap::new()).as_deref(), + Some("https://gw.example.com") + ); + // An empty Host does not shadow a usable authority either. + let mut empty_host = HeaderMap::new(); + empty_host.insert(header::HOST, "".parse().unwrap()); + assert_eq!( + gateway_base(&h2_uri, &empty_host).as_deref(), + Some("https://gw.example.com") + ); + } + + #[test] + fn gateway_base_is_none_without_any_authority() { + let origin_form = axum::http::Uri::from_static("/a2a/x/.well-known/agent-card.json"); + assert_eq!(gateway_base(&origin_form, &HeaderMap::new()), None); + } + + #[test] + fn card_rewrite_leaves_no_upstream_url_for_a_caller_to_follow() { + // Only the top-level `url` used to be rewritten. A 1.0 caller reads its + // endpoint out of `supportedInterfaces` instead, so it went straight to + // the upstream — past auth, quota and usage — and read the internal + // address off the card on the way (#911). + let mut card: aisix_a2a::AgentCard = serde_json::from_str( + r#"{ + "name": "Agent", + "url": "https://internal.upstream/a2a", + "supportedInterfaces": [ + {"url": "https://internal.upstream/a2a", "protocolBinding": "JSONRPC"}, + {"url": "https://internal.upstream/grpc", "protocolBinding": "GRPC"} + ], + "additionalInterfaces": [ + {"url": "https://internal.upstream/rest", "transport": "HTTP+JSON"} + ], + "skills": [{"id": "s1"}] + }"#, + ) + .unwrap(); + + rewrite_card_urls(&mut card, "https://gw.example.com/a2a/billing"); + + let served = serde_json::to_string(&card).unwrap(); + assert!( + !served.contains("internal.upstream"), + "no upstream address may survive anywhere in the served card:\n{served}" + ); + assert_eq!(card.url, "https://gw.example.com/a2a/billing"); + for key in ["supportedInterfaces", "additionalInterfaces"] { + for interface in card.rest[key].as_array().unwrap() { + assert_eq!(interface["url"], "https://gw.example.com/a2a/billing"); + } + } + // Everything the gateway does not own is passed through untouched. + assert_eq!(card.rest["skills"][0]["id"], "s1"); + assert_eq!( + card.rest["supportedInterfaces"][1]["protocolBinding"], + "GRPC" + ); } // ---- endpoint integration tests: drive the real router via oneshot ---- diff --git a/schemas/resources/a2a_agent.schema.json b/schemas/resources/a2a_agent.schema.json index d7c95a8d..050465a9 100644 --- a/schemas/resources/a2a_agent.schema.json +++ b/schemas/resources/a2a_agent.schema.json @@ -149,7 +149,7 @@ } ], "default": "1.0", - "description": "The A2A wire-format version AISIX uses for this agent. AISIX pins the version explicitly so the served agent card and accepted requests stay consistent." + "description": "The A2A wire-format version this agent speaks. AISIX announces it to the agent in the `A2A-Version` header on every request, so it must match what the agent actually serves: an agent reads an absent or mismatched version as a protocol error and rejects the call." }, "secret": { "description": "Credential AISIX uses to authenticate to the upstream agent. For `bearer`, AISIX sends it as `Authorization: Bearer `; for `api_key`, AISIX sends it as `x-api-key: `. Leave unset for `none`.", @@ -168,7 +168,7 @@ ] }, "url": { - "description": "The upstream agent's base URL, such as `https://agents.example.com/a2a`. AISIX reaches this URL over HTTP with the A2A JSON-RPC 2.0 protocol and discovers the agent card relative to it.", + "description": "The upstream agent's A2A service endpoint, such as `https://agents.example.com/a2a`, where AISIX sends JSON-RPC 2.0 requests over HTTP. AISIX looks for the agent card at the well-known path under this URL's own path first, then under its origin, so both an agent that owns its domain and one published under a path prefix are reachable without extra configuration.", "minLength": 1, "type": "string" } diff --git a/tests/e2e/src/cases/a2a-gateway-e2e.test.ts b/tests/e2e/src/cases/a2a-gateway-e2e.test.ts new file mode 100644 index 00000000..3f6961d5 --- /dev/null +++ b/tests/e2e/src/cases/a2a-gateway-e2e.test.ts @@ -0,0 +1,267 @@ +import { createHash, randomUUID } from "node:crypto"; +import { afterAll, beforeAll, describe, expect, test } from "vitest"; +import { + EtcdClient, + SeedClient, + spawnApp, + startA2aUpstream, + waitConfigPropagation, + type A2aUpstream, + type SpawnedApp, +} from "../harness/index.js"; + +// E2E: the `/a2a/{agent}` gateway endpoint against a real gateway + etcd + real +// upstream A2A agents. The A2A MVP (#717) deferred endpoint-level e2e; this is +// it, written around the two faults that made the endpoint unusable in +// practice. +// +// Pinned contract: +// - every upstream call announces the agent's pinned wire version in +// `A2A-Version`, on the card fetch as well as the JSON-RPC call (#911) — +// without it an agent must read the call as 0.3 and reject a 1.0 body; +// - a `0.3`-pinned agent is announced as 0.3, so the pin is what is sent +// rather than a constant; +// - an agent whose card is published under its own path prefix resolves, and +// the catch-all 405 its platform returns at the origin is not mistaken for +// the agent's answer (#913); +// - the served card carries NO upstream address: the top-level `url` and +// every `supportedInterfaces[].url` point back at the gateway, so a 1.0 +// caller (which reads its endpoint out of `supportedInterfaces`) cannot +// route around the gateway; +// - the upstream credential is presented by the gateway and never reaches +// the caller; the caller's own key never reaches the upstream; +// - per-agent ACL and unknown/disabled agents still gate the endpoint. + +const KEY_ALLOWED = "sk-a2a-e2e-allowed"; +const KEY_NO_AGENTS = "sk-a2a-e2e-no-agents"; + +const sha256 = (s: string) => createHash("sha256").update(s).digest("hex"); + +const messageSend = (id: string) => ({ + jsonrpc: "2.0", + id, + method: "message/send", + params: { + message: { + role: "user", + parts: [{ kind: "text", text: "invoice 42" }], + messageId: "m-1", + }, + }, +}); + +describe("a2a gateway e2e: /a2a/{agent}", () => { + let app: SpawnedApp | undefined; + let rootHosted: A2aUpstream | undefined; + let pathHosted: A2aUpstream | undefined; + let etcdReachable = false; + let seed: SeedClient; + + // The gate responses (401 / 403 / 404) are plain text, not JSON-RPC + // envelopes, so the body is parsed opportunistically. + const readBody = (text: string): Record | undefined => { + if (!text) return undefined; + try { + return JSON.parse(text) as Record; + } catch { + return undefined; + } + }; + + const call = async (path: string, token: string, body: unknown) => { + const res = await fetch(`${app!.proxyUrl}${path}`, { + method: "POST", + headers: { + authorization: `Bearer ${token}`, + "content-type": "application/json", + }, + body: JSON.stringify(body), + }); + const text = await res.text(); + return { status: res.status, raw: text, json: readBody(text) }; + }; + + const fetchCard = async (agent: string, token: string) => { + const res = await fetch( + `${app!.proxyUrl}/a2a/${agent}/.well-known/agent-card.json`, + { headers: { authorization: `Bearer ${token}` } }, + ); + const text = await res.text(); + return { status: res.status, raw: text, json: readBody(text) }; + }; + + beforeAll(async () => { + const etcd = new EtcdClient(); + etcdReachable = await etcd.ping(); + if (!etcdReachable) return; + + rootHosted = await startA2aUpstream({ + cardMount: "origin", + token: "upstream-secret-tok", + }); + pathHosted = await startA2aUpstream({ cardMount: "path" }); + app = await spawnApp(); + seed = new SeedClient(etcd, app.etcdPrefix); + + await seed.update("a2a_agents", randomUUID(), { + name: "invoices", + url: rootHosted.url, + protocol_version: "1.0", + auth_type: "bearer", + secret: "upstream-secret-tok", + enabled: true, + }); + await seed.update("a2a_agents", randomUUID(), { + name: "legacy", + url: rootHosted.url, + protocol_version: "0.3", + auth_type: "bearer", + secret: "upstream-secret-tok", + enabled: true, + }); + await seed.update("a2a_agents", randomUUID(), { + name: "tenant", + url: pathHosted.url, + protocol_version: "1.0", + auth_type: "none", + enabled: true, + }); + await seed.update("a2a_agents", randomUUID(), { + name: "retired", + url: rootHosted.url, + protocol_version: "1.0", + auth_type: "none", + enabled: false, + }); + + await seed.createApiKey({ + key_hash: sha256(KEY_ALLOWED), + allowed_models: [], + allowed_agents: ["*"], + }); + await seed.createApiKey({ + key_hash: sha256(KEY_NO_AGENTS), + allowed_models: [], + }); + + // The gate proves only that the seeded rows reached the DP snapshot, and + // deliberately asserts none of the behaviour under test: a 404 would mean + // the agent row has not landed, a 401 that the key row has not. Anything + // else — including a 403 or a 502 — means both are present, and any defect + // in discovery, the version header, the card rewrite or the ACL then fails + // its own test by name instead of surfacing as a 60s propagation timeout. + await waitConfigPropagation(async () => { + const probe = await call("/a2a/invoices", KEY_ALLOWED, messageSend("gate")); + return probe.status !== 404 && probe.status !== 401; + }); + }, 60_000); + + afterAll(async () => { + await app?.exit(); + await rootHosted?.close(); + await pathHosted?.close(); + }); + + test("announces the pinned wire version on every upstream call", async (ctx) => { + if (!etcdReachable || !app || !rootHosted) return ctx.skip(); + rootHosted.requests.length = 0; + + await fetchCard("invoices", KEY_ALLOWED); + const reply = await call("/a2a/invoices", KEY_ALLOWED, messageSend("e2e-1")); + + expect(reply.status).toBe(200); + expect(reply.json?.result?.sawVersion).toBe("1.0"); + expect(reply.json?.result?.sawMethod).toBe("message/send"); + // A card fetch is an A2A request like any other, so EVERY hop carries the + // version — including the candidate probes that miss. + expect(rootHosted.requests.map((r) => r.version)).not.toContain(null); + expect(rootHosted.requests.every((r) => r.version === "1.0")).toBe(true); + expect(rootHosted.requests.some((r) => r.httpMethod === "GET")).toBe(true); + expect(rootHosted.requests.some((r) => r.httpMethod === "POST")).toBe(true); + }); + + test("announces 0.3 for an agent pinned to 0.3", async (ctx) => { + if (!etcdReachable || !app || !rootHosted) return ctx.skip(); + rootHosted.requests.length = 0; + + const reply = await call("/a2a/legacy", KEY_ALLOWED, messageSend("e2e-2")); + + expect(reply.status).toBe(200); + expect(reply.json?.result?.sawVersion).toBe("0.3"); + expect(rootHosted.requests.at(-1)?.version).toBe("0.3"); + }); + + test("reaches an agent whose card is published under a path prefix", async (ctx) => { + if (!etcdReachable || !app || !pathHosted) return ctx.skip(); + pathHosted.requests.length = 0; + + const card = await fetchCard("tenant", KEY_ALLOWED); + expect(card.status).toBe(200); + expect(card.json?.name).toBe("Invoice Processor"); + + // Resolved on the first candidate: the agent's own path prefix is tried + // before the origin, so the platform's catch-all 405 at the origin is never + // even reached — which is exactly what used to be mistaken for the agent's + // answer when the origin was the ONLY candidate. + const cardPaths = pathHosted.requests + .filter((r) => r.httpMethod === "GET") + .map((r) => r.path); + expect(cardPaths).toEqual([ + "/v3/agents/serve/tenant-42/.well-known/agent-card.json", + ]); + + const reply = await call("/a2a/tenant", KEY_ALLOWED, messageSend("e2e-3")); + expect(reply.status).toBe(200); + expect(reply.json?.result?.id).toBe("task-e2e-1"); + }); + + test("the served card points every endpoint back at the gateway", async (ctx) => { + if (!etcdReachable || !app) return ctx.skip(); + + const card = await fetchCard("invoices", KEY_ALLOWED); + + expect(card.status).toBe(200); + expect(card.raw).not.toContain("internal.upstream.invalid"); + expect(card.json?.url).toMatch(/\/a2a\/invoices$/); + for (const iface of card.json?.supportedInterfaces ?? []) { + expect(iface.url).toMatch(/\/a2a\/invoices$/); + } + // Everything the gateway does not own survives the rewrite. + expect(card.json?.skills?.[0]?.id).toBe("invoice"); + expect(card.json?.version).toBe("2.1.0"); + }); + + test("the gateway holds the upstream credential and the caller never sees it", async (ctx) => { + if (!etcdReachable || !app || !rootHosted) return ctx.skip(); + rootHosted.requests.length = 0; + + const reply = await call("/a2a/invoices", KEY_ALLOWED, messageSend("e2e-4")); + + expect(reply.status).toBe(200); + const seen = rootHosted.requests.at(-1); + expect(seen?.authorization).toBe("Bearer upstream-secret-tok"); + // The caller's own AISIX key must never be forwarded as the upstream token. + expect(seen?.authorization).not.toContain(KEY_ALLOWED); + expect(JSON.stringify(reply.json)).not.toContain("upstream-secret-tok"); + }); + + test("gates on the per-agent ACL and on agent existence", async (ctx) => { + if (!etcdReachable || !app) return ctx.skip(); + + const denied = await call("/a2a/invoices", KEY_NO_AGENTS, messageSend("x")); + expect(denied.status).toBe(403); + + const unknown = await call("/a2a/ghost", KEY_ALLOWED, messageSend("x")); + expect(unknown.status).toBe(404); + + const disabled = await call("/a2a/retired", KEY_ALLOWED, messageSend("x")); + expect(disabled.status).toBe(404); + + const unauthenticated = await fetch(`${app.proxyUrl}/a2a/invoices`, { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify(messageSend("x")), + }); + expect(unauthenticated.status).toBe(401); + }); +}); diff --git a/tests/e2e/src/harness/index.ts b/tests/e2e/src/harness/index.ts index 7830d1fb..4b28316e 100644 --- a/tests/e2e/src/harness/index.ts +++ b/tests/e2e/src/harness/index.ts @@ -5,6 +5,12 @@ export { EtcdClient } from "./etcd.js"; export { SeedClient } from "./seed.js"; export { startOpenAiUpstream, type OpenAiUpstream, type ReceivedRequest } from "./upstream-openai.js"; export { startMcpUpstream, type McpUpstream } from "./upstream-mcp.js"; +export { + startA2aUpstream, + type A2aUpstream, + type A2aCardMount, + type A2aReceivedRequest, +} from "./upstream-a2a.js"; export { startRestUpstream, type RestUpstream } from "./upstream-rest.js"; export { pickFreePort, pickFreePorts } from "./ports.js"; export { diff --git a/tests/e2e/src/harness/upstream-a2a.ts b/tests/e2e/src/harness/upstream-a2a.ts new file mode 100644 index 00000000..4cf6704b --- /dev/null +++ b/tests/e2e/src/harness/upstream-a2a.ts @@ -0,0 +1,181 @@ +import { + createServer, + type IncomingMessage, + type Server as HttpServer, + type ServerResponse, +} from "node:http"; + +/** One request the stub agent received, as the gateway sent it. */ +export interface A2aReceivedRequest { + httpMethod: string; + path: string; + /** The `A2A-Version` header, or `null` when the gateway sent none. */ + version: string | null; + authorization: string | null; + apiKey: string | null; + body?: Record; +} + +/** + * Where the agent publishes its card. + * + * - `origin` — at the RFC 8615 origin URI, the shape of an agent that owns its + * whole domain. This is the only shape the gateway used to be able to reach. + * - `path` — under the service endpoint's own path prefix, with every other + * path answering `405` from a catch-all. This is the shape of any platform + * that multiplexes tenants under a prefix, and of a self-hosted agent behind + * an ingress path (api7/aisix#913). + */ +export type A2aCardMount = "origin" | "path"; + +export interface A2aUpstream { + /** Register this as the agent's `url`: the A2A service endpoint. */ + url: string; + /** Every request the agent received, in arrival order. */ + requests: A2aReceivedRequest[]; + close(): Promise; +} + +export interface A2aUpstreamOptions { + cardMount?: A2aCardMount; + /** When set, the agent is registered with `auth_type: bearer` and this token. */ + token?: string; +} + +const PATH_PREFIX = "/v3/agents/serve/tenant-42"; + +/** + * A stub upstream A2A agent: serves an agent card and answers JSON-RPC, while + * recording what the gateway actually sent it — the wire version it announced + * and the credential it presented. + * + * The card it serves deliberately advertises an unreachable `https://` address + * both at the top level and inside `supportedInterfaces`, so a test can prove + * the gateway rewrote every one of them before handing the card to a caller. + */ +export async function startA2aUpstream( + options: A2aUpstreamOptions = {}, +): Promise { + const mount: A2aCardMount = options.cardMount ?? "origin"; + const requests: A2aReceivedRequest[] = []; + const servicePath = mount === "path" ? PATH_PREFIX : "/a2a"; + const cardPath = + mount === "path" + ? `${PATH_PREFIX}/.well-known/agent-card.json` + : "/.well-known/agent-card.json"; + + const httpServer: HttpServer = createServer((req, res) => { + handle(req, res, { + requests, + servicePath, + cardPath, + token: options.token, + }).catch((err: unknown) => { + // `handle` rejects on a malformed body, and on a write to an already + // closed socket. Unhandled, that terminates the test process; worse, the + // request never gets a response, so a stub fault reads as a gateway + // timeout instead of what it is. Answer on the wire either way. + if (!res.headersSent) { + res.writeHead(500, { "content-type": "application/json" }); + } + res.end(JSON.stringify({ error: `a2a stub failed: ${String(err)}` })); + }); + }); + await new Promise((resolve) => + httpServer.listen(0, "127.0.0.1", resolve), + ); + const address = httpServer.address(); + if (address === null || typeof address === "string") { + throw new Error("a2a upstream: no listen address"); + } + + return { + url: `http://127.0.0.1:${address.port}${servicePath}`, + requests, + close: () => new Promise((resolve) => httpServer.close(() => resolve())), + }; +} + +async function handle( + req: IncomingMessage, + res: ServerResponse, + ctx: { + requests: A2aReceivedRequest[]; + servicePath: string; + cardPath: string; + token?: string; + }, +): Promise { + const path = new URL(req.url ?? "/", "http://127.0.0.1").pathname; + const header = (name: string): string | null => { + const value = req.headers[name]; + return typeof value === "string" ? value : null; + }; + const send = (status: number, body: unknown): void => { + res.writeHead(status, { "content-type": "application/json" }); + res.end(JSON.stringify(body)); + }; + + let body: Record | undefined; + if (req.method === "POST") { + const chunks: Buffer[] = []; + for await (const chunk of req) chunks.push(chunk as Buffer); + const raw = Buffer.concat(chunks).toString("utf8"); + body = raw ? (JSON.parse(raw) as Record) : undefined; + } + + ctx.requests.push({ + httpMethod: req.method ?? "GET", + path, + version: header("a2a-version"), + authorization: header("authorization"), + apiKey: header("x-api-key"), + body, + }); + + if (ctx.token !== undefined && header("authorization") !== `Bearer ${ctx.token}`) { + send(401, { error: "unauthorized" }); + return; + } + + if (req.method === "GET" && path === ctx.cardPath) { + send(200, { + name: "Invoice Processor", + description: "Stub agent for the A2A gateway e2e.", + protocolVersion: "1.0", + version: "2.1.0", + url: "https://internal.upstream.invalid/a2a", + supportedInterfaces: [ + { + url: "https://internal.upstream.invalid/a2a", + protocolBinding: "JSONRPC", + protocolVersion: "1.0", + }, + ], + skills: [{ id: "invoice", name: "Process invoice", tags: ["billing"] }], + }); + return; + } + + if (req.method === "POST" && path === ctx.servicePath) { + send(200, { + jsonrpc: "2.0", + id: body?.id ?? null, + result: { + kind: "task", + id: "task-e2e-1", + status: { state: "completed" }, + // Echoed so the gateway's forwarding can be asserted from the caller + // side as well as from `requests`. + sawVersion: header("a2a-version"), + sawMethod: body?.method ?? null, + }, + }); + return; + } + + // The catch-all a real path-hosting agent platform answers with: not a 404, + // which is what made the original report look like a missing card rather + // than a mis-built URL. + send(405, { error: "method not allowed" }); +}