rust_web_server/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.
51pub(crate) const KEYS: &[&str] = &["upstream", "timeout_ms", "audit_log"];
52
53/// `section` is the action's, e.g. `route[0].action`.
54pub(crate) fn parse(map: &SectionMap, section: &str) -> A2aGatewayConfig {
55    let base = format!("{section}.a2a");
56    A2aGatewayConfig {
57        upstream: get_str(map, &base, "upstream").trim_end_matches('/').to_string(),
58        timeout_ms: get_u64(map, &base, "timeout_ms", 300_000),
59        audit_log: get_str(map, &base, "audit_log").eq_ignore_ascii_case("true"),
60        headers_env: map.get(&format!("{base}.headers_env")).cloned().unwrap_or_default(),
61    }
62}
63
64/// Semantic checks for `rws --check` and startup.
65pub(crate) fn validate(config: &A2aGatewayConfig, route_path: Option<&str>, label: &str, report: &mut ConfigReport) {
66    if route_path.is_none() {
67        report.error(format!("{label}: an a2a route needs [route.match] path, e.g. \"/agents/billing/*\""));
68    }
69    if config.upstream.is_empty() {
70        report.error(format!("{label}: [route.action.a2a] needs upstream, the agent's base URL"));
71    } else if super::health::parse_backend_url(&config.upstream).is_none() || !(config.upstream.starts_with("http://") || config.upstream.starts_with("https://")) {
72        report.error(format!("{label}: a2a upstream {:?} isn't an http(s) URL", config.upstream));
73    }
74    for (header, env) in &config.headers_env {
75        if std::env::var(env).map(|v| v.is_empty()).unwrap_or(true) && !crate::config_check::offline() {
76            report.warn(format!("{label}: a2a header {header} comes from {env}, which is unset or empty"));
77        }
78    }
79}
80
81struct Card {
82    /// The agent's card as it sent it.
83    json: JsonValue,
84    /// Its JSON-RPC endpoint.
85    endpoint: String,
86    fetched: Instant,
87}
88
89/// The route's application.
90pub(crate) struct A2aGateway {
91    config: A2aGatewayConfig,
92    /// The route's path without a trailing `/*` or `/`.
93    prefix: String,
94    card: Mutex<Option<Card>>,
95}
96
97impl A2aGateway {
98    pub(crate) fn new(config: A2aGatewayConfig, route_path: &str) -> A2aGateway {
99        let prefix = route_path.trim_end_matches('*').trim_end_matches('/').to_string();
100        A2aGateway { config, prefix, card: Mutex::new(None) }
101    }
102
103    fn headers(&self) -> Vec<(String, String)> {
104        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()
105    }
106
107    /// The agent's card and JSON-RPC endpoint, cached for [`CARD_TTL`].
108    fn card(&self) -> Result<(JsonValue, String), String> {
109        let mut cached = self.card.lock().unwrap_or_else(|e| e.into_inner());
110        if let Some(card) = cached.as_ref().filter(|c| c.fetched.elapsed() < CARD_TTL) {
111            return Ok((card.json.clone(), card.endpoint.clone()));
112        }
113        let url = format!("{}{}", self.config.upstream, crate::a2a::AGENT_CARD_PATH);
114        let client = crate::http_client::Client::new().max_redirects(0);
115        let mut request = client.get(&url).timeout_ms(10_000);
116        for (name, value) in self.headers() {
117            request = request.header(&name, &value);
118        }
119        let response = request.send().map_err(|e| format!("fetching the agent's card: {e}"))?;
120        if !response.is_success() {
121            return Err(format!("the agent's card answered HTTP {}", response.status()));
122        }
123        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}"))?;
124        let endpoint = json
125            .get("supportedInterfaces")
126            .and_then(JsonValue::as_array)
127            .and_then(|list| list.iter().find(|i| i.get("protocolBinding").and_then(JsonValue::as_str) == Some("JSONRPC")))
128            .and_then(|i| i.get("url"))
129            .and_then(JsonValue::as_str)
130            .map(str::to_string)
131            .ok_or("the agent's card lists no JSONRPC interface")?;
132        *cached = Some(Card { json: json.clone(), endpoint: endpoint.clone(), fetched: Instant::now() });
133        Ok((json, endpoint))
134    }
135
136    /// The card, with only a JSON-RPC interface: this route.
137    fn public_card(&self, request: &Request, connection: &ConnectionInfo) -> Response {
138        let (card, _) = match self.card() {
139            Ok(card) => card,
140            Err(e) => return unavailable(&e),
141        };
142        let host = request.header(crate::header::Header::_HOST).map(|h| h.value.to_string()).unwrap_or_else(|| format!("{}:{}", connection.server.ip, connection.server.port));
143        let public = format!("{}://{host}{}", if connection.tls { "https" } else { "http" }, if self.prefix.is_empty() { "/" } else { &self.prefix });
144        let JsonValue::Object(mut fields) = card else { return unavailable("the agent's card isn't an object") };
145        for (key, value) in fields.iter_mut() {
146            if key == "supportedInterfaces" {
147                let version = value
148                    .as_array()
149                    .and_then(|l| l.iter().find(|i| i.get("protocolBinding").and_then(JsonValue::as_str) == Some("JSONRPC")))
150                    .and_then(|i| i.get("protocolVersion"))
151                    .and_then(JsonValue::as_str)
152                    .unwrap_or(crate::a2a::PROTOCOL_VERSION)
153                    .to_string();
154                *value = JsonValue::Array(vec![JsonValue::Object(vec![
155                    ("url".to_string(), JsonValue::String(public.clone())),
156                    ("protocolBinding".to_string(), JsonValue::String("JSONRPC".to_string())),
157                    ("protocolVersion".to_string(), JsonValue::String(version)),
158                ])]);
159            }
160        }
161        let mut response = Response::json(STATUS_CODE_REASON_PHRASE.n200_ok, crate::a2a::json_text(&JsonValue::Object(fields)).into_bytes());
162        response.headers.push(crate::header::Header::new("Cache-Control", "public, max-age=60".to_string()));
163        response
164    }
165
166    /// The JSON-RPC request, forwarded to the agent (streams pass through).
167    fn forward(&self, request: &Request, connection: &ConnectionInfo) -> Response {
168        let started = Instant::now();
169        let (_, endpoint) = match self.card() {
170            Ok(card) => card,
171            Err(e) => return unavailable(&e),
172        };
173        let Some((host, port, tls)) = super::health::parse_backend_url(&endpoint) else {
174            return unavailable(&format!("the agent's JSON-RPC URL {endpoint:?} can't be used"));
175        };
176        let path = endpoint.splitn(4, '/').nth(3).map(|p| format!("/{p}")).unwrap_or_else(|| "/".to_string());
177        let mut upstream = request.clone();
178        upstream.request_uri = path;
179        upstream.headers.retain(|h| !h.name.eq_ignore_ascii_case("authorization") && !h.name.eq_ignore_ascii_case("host"));
180        upstream.headers.push(crate::header::Header::new("Host", if (tls && port == 443) || (!tls && port == 80) { host.clone() } else { format!("{host}:{port}") }));
181        for (name, value) in self.headers() {
182            upstream.headers.retain(|h| !h.name.eq_ignore_ascii_case(&name));
183            upstream.headers.push(crate::header::Header::new(name, value));
184        }
185        let timeout = Duration::from_millis(self.config.timeout_ms.max(1));
186        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);
187        let response = match result {
188            Ok(mut response) => {
189                crate::proxy::status::received(&mut response, Some(&endpoint));
190                response
191            }
192            Err(e) => {
193                let mut response = crate::error::problem(STATUS_CODE_REASON_PHRASE.n502_bad_gateway, "");
194                crate::proxy::status::failed_with(&mut response, &e, Some(&endpoint));
195                response
196            }
197        };
198        if self.config.audit_log {
199            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();
200            let line = audit_line(&connection.client.ip, &method, &self.config.upstream, response.status_code, started.elapsed());
201            crate::log::access_sink::write_line(|out| out.extend_from_slice(line.as_bytes()));
202        }
203        response
204    }
205}
206
207fn unavailable(reason: &str) -> Response {
208    let mut response = crate::error::problem(STATUS_CODE_REASON_PHRASE.n503_service_unavailable, reason);
209    response.headers.push(crate::header::Header::new("Retry-After", "10".to_string()));
210    response
211}
212
213/// One audit log line (without the newline).
214fn audit_line(client: &str, method: &str, upstream: &str, status: i16, duration: Duration) -> String {
215    let mut time = String::new();
216    crate::log::iso8601_now(|iso| time.push_str(iso));
217    let q = crate::a2a::quote_json;
218    format!(
219        r#"{{"time":"{time}","event":"a2a_call","client":{},"method":{},"upstream":{},"status":{status},"duration_ms":{:.3}}}"#,
220        q(client),
221        q(method),
222        q(upstream),
223        duration.as_secs_f64() * 1000.0
224    )
225}
226
227impl Application for A2aGateway {
228    fn execute(&self, request: &Request, connection: &ConnectionInfo) -> Result<Response, String> {
229        let path = crate::url::URL::request_target_path(&request.request_uri).unwrap_or("");
230        let rest = path.strip_prefix(self.prefix.as_str()).unwrap_or(path);
231        if rest == crate::a2a::AGENT_CARD_PATH && (request.method == "GET" || request.method == "HEAD") {
232            return Ok(self.public_card(request, connection));
233        }
234        if rest.is_empty() || rest == "/" {
235            if request.method == "POST" {
236                return Ok(self.forward(request, connection));
237            }
238            let mut response = crate::error::problem(STATUS_CODE_REASON_PHRASE.n405_method_not_allowed, "A2A requests are JSON-RPC POSTs");
239            response.headers.push(crate::header::Header::new("Allow", "POST".to_string()));
240            return Ok(response);
241        }
242        Ok(crate::error::problem(STATUS_CODE_REASON_PHRASE.n404_not_found, ""))
243    }
244}
245
246#[cfg(test)]
247mod tests;
rust_web_server/proxy_config/a2a_gateway.rs

rust_web_server/proxy_config/
a2a_gateway.rs
