rws_proxy/proxy_config/
a2a_gateway.rs

1//! `[route.action] type = "a2a"`: one public address for an A2A agent
2//! running elsewhere (18.18.0, #65).
3//!
4//! ```toml
5//! [[route]]
6//!   [route.match]
7//!   path = "/agents/billing/*"
8//!
9//!   [route.action]
10//!   type = "a2a"
11//!
12//!   [route.action.a2a]
13//!   upstream    = "http://10.0.0.7:9000"     # the agent; its card at <upstream>/.well-known/agent-card.json
14//!   timeout_ms  = 300000                     # per request (SendMessage may wait for its task)
15//!   audit_log   = true
16//!   headers_env = { Authorization = "BILLING_AGENT_AUTHORIZATION" }  # sent upstream, from the environment
17//! ```
18//!
19//! `GET <path>/.well-known/agent-card.json` serves the agent's card with its
20//! JSON-RPC interface pointing here (the agent's own address stays
21//! private); a `POST` to `<path>` goes to the agent's JSON-RPC endpoint,
22//! streams included. The route's middleware (auth, rate limits, IP filter)
23//! applies as to any route; the caller's `Authorization` isn't passed on,
24//! the configured headers are.
25
26use std::sync::Mutex;
27use std::time::{Duration, Instant};
28
29use crate::application::Application;
30use crate::config_check::ConfigReport;
31use crate::request::Request;
32use crate::response::{Response, STATUS_CODE_REASON_PHRASE};
33use crate::server::ConnectionInfo;
34use crate::service_discovery::json_lite::{self, JsonValue};
35
36use super::parser::{get_str, get_u64, SectionMap};
37
38/// How long the agent's card is kept before it's fetched again.
39const CARD_TTL: Duration = Duration::from_secs(60);
40
41#[derive(Debug, Clone, Default, PartialEq)]
42pub struct A2aGatewayConfig {
43    pub upstream: String,
44    pub timeout_ms: u64,
45    pub audit_log: bool,
46    /// `(header, environment variable holding its value)`.
47    pub headers_env: Vec<(String, String)>,
48}
49
50/// The keys `check_unknown_keys` accepts.
51#[doc(hidden)]
52pub const KEYS: &[&str] = &["upstream", "timeout_ms", "audit_log"];
53
54/// `section` is the action's, e.g. `route[0].action`.
55#[doc(hidden)]
56pub fn parse(map: &SectionMap, section: &str) -> A2aGatewayConfig {
57    let base = format!("{section}.a2a");
58    A2aGatewayConfig {
59        upstream: get_str(map, &base, "upstream").trim_end_matches('/').to_string(),
60        timeout_ms: get_u64(map, &base, "timeout_ms", 300_000),
61        audit_log: get_str(map, &base, "audit_log").eq_ignore_ascii_case("true"),
62        headers_env: map.get(&format!("{base}.headers_env")).cloned().unwrap_or_default(),
63    }
64}
65
66/// Semantic checks for `rws --check` and startup.
67#[doc(hidden)]
68pub fn validate(config: &A2aGatewayConfig, route_path: Option<&str>, label: &str, report: &mut ConfigReport) {
69    if route_path.is_none() {
70        report.error(format!("{label}: an a2a route needs [route.match] path, e.g. \"/agents/billing/*\""));
71    }
72    if config.upstream.is_empty() {
73        report.error(format!("{label}: [route.action.a2a] needs upstream, the agent's base URL"));
74    } else if super::health::parse_backend_url(&config.upstream).is_none() || !(config.upstream.starts_with("http://") || config.upstream.starts_with("https://")) {
75        report.error(format!("{label}: a2a upstream {:?} isn't an http(s) URL", config.upstream));
76    }
77    for (header, env) in &config.headers_env {
78        if std::env::var(env).map(|v| v.is_empty()).unwrap_or(true) && !crate::config_check::offline() {
79            report.warn(format!("{label}: a2a header {header} comes from {env}, which is unset or empty"));
80        }
81    }
82}
83
84struct Card {
85    /// The agent's card as it sent it.
86    json: JsonValue,
87    /// Its JSON-RPC endpoint.
88    endpoint: String,
89    fetched: Instant,
90}
91
92/// The route's application.
93#[doc(hidden)]
94pub struct A2aGateway {
95    config: A2aGatewayConfig,
96    /// The route's path without a trailing `/*` or `/`.
97    prefix: String,
98    card: Mutex<Option<Card>>,
99}
100
101impl A2aGateway {
102    #[doc(hidden)]
103    pub fn new(config: A2aGatewayConfig, route_path: &str) -> A2aGateway {
104        let prefix = route_path.trim_end_matches('*').trim_end_matches('/').to_string();
105        A2aGateway { config, prefix, card: Mutex::new(None) }
106    }
107
108    fn headers(&self) -> Vec<(String, String)> {
109        self.config.headers_env.iter().filter_map(|(h, env)| std::env::var(env).ok().filter(|v| !v.is_empty()).map(|v| (h.clone(), v))).collect()
110    }
111
112    /// The agent's card and JSON-RPC endpoint, cached for [`CARD_TTL`].
113    fn card(&self) -> Result<(JsonValue, String), String> {
114        let mut cached = self.card.lock().unwrap_or_else(|e| e.into_inner());
115        if let Some(card) = cached.as_ref().filter(|c| c.fetched.elapsed() < CARD_TTL) {
116            return Ok((card.json.clone(), card.endpoint.clone()));
117        }
118        let url = format!("{}{}", self.config.upstream, crate::a2a::AGENT_CARD_PATH);
119        let client = crate::http_client::Client::new().max_redirects(0);
120        let mut request = client.get(&url).timeout_ms(10_000);
121        for (name, value) in self.headers() {
122            request = request.header(&name, &value);
123        }
124        let response = request.send().map_err(|e| format!("fetching the agent's card: {e}"))?;
125        if !response.is_success() {
126            return Err(format!("the agent's card answered HTTP {}", response.status()));
127        }
128        let json = json_lite::parse(&response.text().map_err(|e| e.to_string())?).map_err(|e| format!("the agent's card isn't JSON: {e}"))?;
129        let endpoint = json
130            .get("supportedInterfaces")
131            .and_then(JsonValue::as_array)
132            .and_then(|list| list.iter().find(|i| i.get("protocolBinding").and_then(JsonValue::as_str) == Some("JSONRPC")))
133            .and_then(|i| i.get("url"))
134            .and_then(JsonValue::as_str)
135            .map(str::to_string)
136            .ok_or("the agent's card lists no JSONRPC interface")?;
137        *cached = Some(Card { json: json.clone(), endpoint: endpoint.clone(), fetched: Instant::now() });
138        Ok((json, endpoint))
139    }
140
141    /// The card, with only a JSON-RPC interface: this route.
142    fn public_card(&self, request: &Request, connection: &ConnectionInfo) -> Response {
143        let (card, _) = match self.card() {
144            Ok(card) => card,
145            Err(e) => return unavailable(&e),
146        };
147        let host = request.header(crate::header::Header::_HOST).map(|h| h.value.to_string()).unwrap_or_else(|| format!("{}:{}", connection.server.ip, connection.server.port));
148        let public = format!("{}://{host}{}", if connection.tls { "https" } else { "http" }, if self.prefix.is_empty() { "/" } else { &self.prefix });
149        let JsonValue::Object(mut fields) = card else { return unavailable("the agent's card isn't an object") };
150        for (key, value) in fields.iter_mut() {
151            if key == "supportedInterfaces" {
152                let version = value
153                    .as_array()
154                    .and_then(|l| l.iter().find(|i| i.get("protocolBinding").and_then(JsonValue::as_str) == Some("JSONRPC")))
155                    .and_then(|i| i.get("protocolVersion"))
156                    .and_then(JsonValue::as_str)
157                    .unwrap_or(crate::a2a::PROTOCOL_VERSION)
158                    .to_string();
159                *value = JsonValue::Array(vec![JsonValue::Object(vec![
160                    ("url".to_string(), JsonValue::String(public.clone())),
161                    ("protocolBinding".to_string(), JsonValue::String("JSONRPC".to_string())),
162                    ("protocolVersion".to_string(), JsonValue::String(version)),
163                ])]);
164            }
165        }
166        let mut response = Response::json(STATUS_CODE_REASON_PHRASE.n200_ok, crate::a2a::json_text(&JsonValue::Object(fields)).into_bytes());
167        response.headers.push(crate::header::Header::new("Cache-Control", "public, max-age=60".to_string()));
168        response
169    }
170
171    /// The JSON-RPC request, forwarded to the agent (streams pass through).
172    fn forward(&self, request: &Request, connection: &ConnectionInfo) -> Response {
173        let started = Instant::now();
174        let (_, endpoint) = match self.card() {
175            Ok(card) => card,
176            Err(e) => return unavailable(&e),
177        };
178        let Some((host, port, tls)) = super::health::parse_backend_url(&endpoint) else {
179            return unavailable(&format!("the agent's JSON-RPC URL {endpoint:?} can't be used"));
180        };
181        let path = endpoint.splitn(4, '/').nth(3).map(|p| format!("/{p}")).unwrap_or_else(|| "/".to_string());
182        let mut upstream = request.clone();
183        upstream.request_uri = path;
184        upstream.headers.retain(|h| !h.name.eq_ignore_ascii_case("authorization") && !h.name.eq_ignore_ascii_case("host"));
185        upstream.headers.push(crate::header::Header::new("Host", if (tls && port == 443) || (!tls && port == 80) { host.clone() } else { format!("{host}:{port}") }));
186        for (name, value) in self.headers() {
187            upstream.headers.retain(|h| !h.name.eq_ignore_ascii_case(&name));
188            upstream.headers.push(crate::header::Header::new(name, value));
189        }
190        let timeout = Duration::from_millis(self.config.timeout_ms.max(1));
191        let result = crate::proxy::forward_upstream_trusting(crate::proxy::shared_pool(), &upstream, crate::proxy::Peer::of(connection), &host, port, tls, None, timeout.min(Duration::from_secs(10)), timeout);
192        let response = match result {
193            Ok(mut response) => {
194                crate::proxy::status::received(&mut response, Some(&endpoint));
195                response
196            }
197            Err(e) => {
198                let mut response = crate::error::problem(STATUS_CODE_REASON_PHRASE.n502_bad_gateway, "");
199                crate::proxy::status::failed_with(&mut response, &e, Some(&endpoint));
200                response
201            }
202        };
203        if self.config.audit_log {
204            let method = std::str::from_utf8(&request.body).ok().and_then(|b| json_lite::parse(b).ok()).and_then(|v| v.get("method").and_then(JsonValue::as_str).map(str::to_string)).unwrap_or_default();
205            let line = audit_line(&crate::ip_filter::client_ip(request, &connection.client.ip), &method, &self.config.upstream, response.status_code, started.elapsed());
206            crate::log::access_sink::write_line(|out| out.extend_from_slice(line.as_bytes()));
207        }
208        response
209    }
210}
211
212fn unavailable(reason: &str) -> Response {
213    let mut response = crate::error::problem(STATUS_CODE_REASON_PHRASE.n503_service_unavailable, reason);
214    response.headers.push(crate::header::Header::new("Retry-After", "10".to_string()));
215    response
216}
217
218/// One audit log line (without the newline).
219fn audit_line(client: &str, method: &str, upstream: &str, status: i16, duration: Duration) -> String {
220    let mut time = String::new();
221    crate::log::iso8601_now(|iso| time.push_str(iso));
222    let q = crate::a2a::quote_json;
223    format!(
224        r#"{{"time":"{time}","event":"a2a_call","client":{},"method":{},"upstream":{},"status":{status},"duration_ms":{:.3}}}"#,
225        q(client),
226        q(method),
227        q(upstream),
228        duration.as_secs_f64() * 1000.0
229    )
230}
231
232impl Application for A2aGateway {
233    fn execute(&self, request: &Request, connection: &ConnectionInfo) -> Result<Response, String> {
234        let path = crate::url::URL::request_target_path(&request.request_uri).unwrap_or("");
235        let rest = path.strip_prefix(self.prefix.as_str()).unwrap_or(path);
236        if rest == crate::a2a::AGENT_CARD_PATH && (request.method == "GET" || request.method == "HEAD") {
237            return Ok(self.public_card(request, connection));
238        }
239        if rest.is_empty() || rest == "/" {
240            if request.method == "POST" {
241                return Ok(self.forward(request, connection));
242            }
243            let mut response = crate::error::problem(STATUS_CODE_REASON_PHRASE.n405_method_not_allowed, "A2A requests are JSON-RPC POSTs");
244            response.headers.push(crate::header::Header::new("Allow", "POST".to_string()));
245            return Ok(response);
246        }
247        Ok(crate::error::problem(STATUS_CODE_REASON_PHRASE.n404_not_found, ""))
248    }
249}
250
251#[cfg(test)]
252mod tests;
rws_proxy/proxy_config/a2a_gateway.rs

rws_proxy/proxy_config/
a2a_gateway.rs
