a2a_agents/control_plane/
client.rs

1//! HTTP client adapter for the control-plane API — the other side of
2//! [`control_plane_router`](super::control_plane_router).
3//!
4//! Until this existed the control plane could only be driven with `curl`, which
5//! made the daemon story unusable without either Terraform or a hand-rolled
6//! request. `a2a deploy/ps/logs/stop` are thin printers over this type, and the
7//! Terraform provider will target the same routes.
8//!
9//! It is a plain adapter, not a port implementation: nothing in the inner layers
10//! needs to *call* a control plane, so introducing a trait here would be a port
11//! with one implementation and no consumer.
12
13use reqwest::StatusCode;
14use thiserror::Error;
15
16use super::wire::{AgentLogs, AgentStatus, ApiErrorBody, DeployRequest, ListQuery};
17use super::{DeployedAgent, ListFilter};
18use crate::registry::AgentId;
19
20/// Why a control-plane call failed.
21///
22/// The distinctions are the ones a caller acts on differently: a bad token is
23/// fixed by the operator, a missing agent by naming another one, an unreachable
24/// URL by starting the control plane — and everything else is the control
25/// plane's own diagnosis, which is worth passing through verbatim (it is where
26/// "line 4: unknown key `http_prot`" and "references env vars that are not
27/// allowed" come from).
28#[derive(Debug, Error)]
29pub enum ControlPlaneClientError {
30    /// The control plane could not be reached at all.
31    #[error("could not reach the control plane at {url}: {source}")]
32    Unreachable {
33        /// The base URL that was dialled.
34        url: String,
35        /// The underlying transport error.
36        #[source]
37        source: reqwest::Error,
38    },
39
40    /// The request was rejected for want of a valid bearer token.
41    #[error(
42        "the control plane rejected the token \
43         (pass --token, or set A2A_CONTROL_PLANE_TOKEN)"
44    )]
45    Unauthorized,
46
47    /// No agent with that id is deployed.
48    #[error("no agent '{0}' is deployed (see `a2a ps`)")]
49    NotFound(AgentId),
50
51    /// This control plane cannot serve the request — e.g. logs from a runtime
52    /// that does not capture them.
53    #[error("{0}")]
54    Unsupported(String),
55
56    /// The control plane refused or failed the request, and said why.
57    #[error("control plane returned {status}: {message}")]
58    Api {
59        /// The HTTP status it answered with.
60        status: StatusCode,
61        /// Its own description of the failure.
62        message: String,
63    },
64
65    /// A response arrived but was not the shape this client expects.
66    #[error("could not read the control plane's response: {0}")]
67    Malformed(#[source] reqwest::Error),
68}
69
70/// Calls the control-plane API on behalf of the CLI.
71///
72/// Cheap to clone (`reqwest::Client` is an `Arc` inside). Holds the bearer token
73/// the API requires; `None` is only useful against a `--no-auth` control plane.
74#[derive(Debug, Clone)]
75pub struct ControlPlaneClient {
76    base_url: String,
77    token: Option<String>,
78    http: reqwest::Client,
79}
80
81impl ControlPlaneClient {
82    /// Target the control plane at `base_url` (e.g. `http://127.0.0.1:9090`).
83    ///
84    /// A trailing slash is trimmed, so a pasted URL and a typed one behave the
85    /// same.
86    pub fn new(base_url: impl Into<String>) -> Self {
87        let base_url = base_url.into().trim_end_matches('/').to_string();
88        Self {
89            base_url,
90            token: None,
91            http: reqwest::Client::new(),
92        }
93    }
94
95    /// Present this bearer token on every request.
96    pub fn with_token(mut self, token: impl Into<String>) -> Self {
97        self.token = Some(token.into());
98        self
99    }
100
101    /// Present `token` if there is one — the `Option`-shaped form of
102    /// [`with_token`](Self::with_token), so a caller resolving a flag or an env
103    /// var does not have to branch.
104    pub fn with_optional_token(self, token: Option<impl Into<String>>) -> Self {
105        match token {
106            Some(token) => self.with_token(token),
107            None => self,
108        }
109    }
110
111    /// The control plane this client talks to.
112    pub fn base_url(&self) -> &str {
113        &self.base_url
114    }
115
116    /// Deploy an agent from **raw, unexpanded** config TOML.
117    ///
118    /// Deliberately takes the text as it sits on disk: `${VAR}` refs are the
119    /// control plane's to resolve, against its own environment and its
120    /// allowlist. Expanding them here would put the operator's secrets on the
121    /// wire and require the deploying machine to hold them.
122    pub async fn deploy(
123        &self,
124        config_toml: impl Into<String>,
125    ) -> Result<DeployedAgent, ControlPlaneClientError> {
126        let request = self.http.post(self.url("/agents")).json(&DeployRequest {
127            config_toml: config_toml.into(),
128        });
129        self.send(request, None).await
130    }
131
132    /// Deployed agents with their endpoint and current health.
133    ///
134    /// [`ListFilter::Live`] hides agents that have been stopped; the control
135    /// plane keeps their entries (their logs are still readable) but they are
136    /// not part of what is running.
137    pub async fn list(
138        &self,
139        filter: ListFilter,
140    ) -> Result<Vec<DeployedAgent>, ControlPlaneClientError> {
141        let request = self.http.get(self.url("/agents")).query(&ListQuery {
142            all: filter == ListFilter::All,
143        });
144        self.send(request, None).await
145    }
146
147    /// One agent's current health.
148    pub async fn status(&self, id: &AgentId) -> Result<AgentStatus, ControlPlaneClientError> {
149        let request = self.http.get(self.url(&format!("/agents/{id}")));
150        self.send(request, Some(id)).await
151    }
152
153    /// An agent's captured output, oldest line first, limited to the last `tail`
154    /// lines when given.
155    pub async fn logs(
156        &self,
157        id: &AgentId,
158        tail: Option<usize>,
159    ) -> Result<AgentLogs, ControlPlaneClientError> {
160        let mut request = self.http.get(self.url(&format!("/agents/{id}/logs")));
161        if let Some(tail) = tail {
162            request = request.query(&[("tail", tail)]);
163        }
164        self.send(request, Some(id)).await
165    }
166
167    /// Stop an agent and remove it from discovery.
168    pub async fn undeploy(&self, id: &AgentId) -> Result<(), ControlPlaneClientError> {
169        let request = self.http.delete(self.url(&format!("/agents/{id}")));
170        // 204 No Content: there is no body to decode, so this one cannot go
171        // through `send`.
172        let response = self.execute(request).await?;
173        self.check(response, Some(id)).await.map(drop)
174    }
175
176    /// Absolute URL for an API path.
177    fn url(&self, path: &str) -> String {
178        format!("{}{path}", self.base_url)
179    }
180
181    /// Send a request expecting a JSON body of type `T`.
182    ///
183    /// `id` is the agent the call was about, so a 404 can name it.
184    async fn send<T: serde::de::DeserializeOwned>(
185        &self,
186        request: reqwest::RequestBuilder,
187        id: Option<&AgentId>,
188    ) -> Result<T, ControlPlaneClientError> {
189        let response = self.execute(request).await?;
190        let response = self.check(response, id).await?;
191        response
192            .json()
193            .await
194            .map_err(ControlPlaneClientError::Malformed)
195    }
196
197    /// Attach credentials and dial.
198    async fn execute(
199        &self,
200        request: reqwest::RequestBuilder,
201    ) -> Result<reqwest::Response, ControlPlaneClientError> {
202        let request = match &self.token {
203            Some(token) => request.bearer_auth(token),
204            None => request,
205        };
206        request
207            .send()
208            .await
209            .map_err(|source| ControlPlaneClientError::Unreachable {
210                url: self.base_url.clone(),
211                source,
212            })
213    }
214
215    /// Turn a non-2xx response into the most actionable error available.
216    ///
217    /// The body is read for the API's own `error` message rather than reporting
218    /// the status alone: the control plane is the only party that knows *which*
219    /// key was unknown or *which* variable was disallowed, and dropping that
220    /// would leave the operator with a bare 400.
221    async fn check(
222        &self,
223        response: reqwest::Response,
224        id: Option<&AgentId>,
225    ) -> Result<reqwest::Response, ControlPlaneClientError> {
226        let status = response.status();
227        if status.is_success() {
228            return Ok(response);
229        }
230        let message = response
231            .json::<ApiErrorBody>()
232            .await
233            .map(|body| body.error)
234            .unwrap_or_else(|_| status.to_string());
235
236        Err(match (status, id) {
237            (StatusCode::UNAUTHORIZED, _) => ControlPlaneClientError::Unauthorized,
238            (StatusCode::NOT_FOUND, Some(id)) => ControlPlaneClientError::NotFound(id.clone()),
239            (StatusCode::NOT_IMPLEMENTED, _) => ControlPlaneClientError::Unsupported(message),
240            _ => ControlPlaneClientError::Api { status, message },
241        })
242    }
243}
244
245#[cfg(test)]
246mod tests {
247    use super::*;
248
249    #[test]
250    fn urls_are_built_without_doubling_the_separator() {
251        let client = ControlPlaneClient::new("http://127.0.0.1:9090/");
252        assert_eq!(client.base_url(), "http://127.0.0.1:9090");
253        assert_eq!(client.url("/agents"), "http://127.0.0.1:9090/agents");
254        assert_eq!(
255            client.url("/agents/weather-agent/logs"),
256            "http://127.0.0.1:9090/agents/weather-agent/logs"
257        );
258    }
259
260    #[test]
261    fn an_absent_token_leaves_the_client_unauthenticated() {
262        let client = ControlPlaneClient::new("http://x").with_optional_token(None::<String>);
263        assert!(client.token.is_none());
264        let client = ControlPlaneClient::new("http://x").with_optional_token(Some("t"));
265        assert_eq!(client.token.as_deref(), Some("t"));
266    }
267}
a2a_agents/control_plane/client.rs

a2a_agents/control_plane/
client.rs
