microclaw_engine/tools/
a2a.rs

1use async_trait::async_trait;
2use serde_json::json;
3
4use super::{schema_object, Tool, ToolResult};
5use crate::a2a::{
6    normalize_base_url, normalize_peer_name, A2AMessageRequest, A2AMessageResponse,
7    A2A_AGENT_CARD_PATH, A2A_MESSAGE_PATH, A2A_PROTOCOL_VERSION,
8};
9use crate::config::Config;
10use crate::http_client::default_llm_user_agent;
11use crate::internal::tool_runtime::runtime::HIGH_RISK_APPROVED_KEY as A2A_HIGH_RISK_APPROVED_KEY;
12use microclaw_core::llm_types::ToolDefinition;
13
14pub struct A2AListPeersTool {
15    config: Config,
16}
17
18impl A2AListPeersTool {
19    pub fn new(config: &Config) -> Self {
20        Self {
21            config: config.clone(),
22        }
23    }
24}
25
26pub struct A2ASendTool {
27    client: reqwest::Client,
28    config: Config,
29}
30
31impl A2ASendTool {
32    pub fn new(config: &Config) -> Self {
33        let user_agent = format!("{}/a2a", default_llm_user_agent());
34        let client = reqwest::Client::builder()
35            .user_agent(user_agent)
36            .build()
37            .unwrap_or_else(|_| reqwest::Client::new());
38        Self {
39            client,
40            config: config.clone(),
41        }
42    }
43}
44
45#[async_trait]
46impl Tool for A2AListPeersTool {
47    fn name(&self) -> &str {
48        "a2a_list_peers"
49    }
50
51    fn definition(&self) -> ToolDefinition {
52        ToolDefinition {
53            name: self.name().into(),
54            description:
55                "List configured agent-to-agent peers that can receive remote tasks over HTTP."
56                    .into(),
57            input_schema: schema_object(json!({}), &[]),
58        }
59    }
60
61    async fn execute(&self, _input: serde_json::Value) -> ToolResult {
62        if !self.config.a2a.enabled {
63            return ToolResult::error("A2A is disabled in config (`a2a.enabled: true`).".into());
64        }
65        let peers = self
66            .config
67            .a2a
68            .peers
69            .iter()
70            .filter(|(_, peer)| peer.enabled)
71            .map(|(name, peer)| {
72                json!({
73                    "peer": name,
74                    "base_url": peer.base_url,
75                    "message_endpoint": format!("{}{}", peer.base_url, A2A_MESSAGE_PATH),
76                    "agent_card_endpoint": format!("{}{}", peer.base_url, A2A_AGENT_CARD_PATH),
77                    "default_session_key": peer.default_session_key,
78                    "description": peer.description,
79                    "has_bearer_token": peer.bearer_token.is_some(),
80                })
81            })
82            .collect::<Vec<_>>();
83        ToolResult::success(
84            serde_json::to_string_pretty(&json!({
85                "protocol_version": A2A_PROTOCOL_VERSION,
86                "peers": peers
87            }))
88            .unwrap_or_else(|_| "{\"peers\":[]}".to_string()),
89        )
90    }
91}
92
93#[async_trait]
94impl Tool for A2ASendTool {
95    fn name(&self) -> &str {
96        "a2a_send"
97    }
98
99    fn definition(&self) -> ToolDefinition {
100        ToolDefinition {
101            name: self.name().into(),
102            description:
103                "Send a task or question to a configured remote MicroClaw peer via the A2A HTTP protocol."
104                    .into(),
105            input_schema: schema_object(
106                json!({
107                    "peer": {
108                        "type": "string",
109                        "description": "Configured peer name from `a2a.peers`."
110                    },
111                    "message": {
112                        "type": "string",
113                        "description": "The task or question to send to the remote agent."
114                    },
115                    "session_key": {
116                        "type": "string",
117                        "description": "Optional remote session key. Defaults to the peer's configured default or `a2a:<peer>`."
118                    },
119                    "timeout_secs": {
120                        "type": "integer",
121                        "description": "HTTP timeout in seconds."
122                    }
123                }),
124                &["peer", "message"],
125            ),
126        }
127    }
128
129    async fn execute(&self, input: serde_json::Value) -> ToolResult {
130        if !self.config.a2a.enabled {
131            return ToolResult::error("A2A is disabled in config (`a2a.enabled: true`).".into());
132        }
133
134        let Some(peer_name) = input.get("peer").and_then(|v| v.as_str()) else {
135            return ToolResult::error("Missing required parameter: peer".into());
136        };
137        let Some(message) = input.get("message").and_then(|v| v.as_str()) else {
138            return ToolResult::error("Missing required parameter: message".into());
139        };
140        let message = message.trim();
141        if message.is_empty() {
142            return ToolResult::error("Parameter `message` cannot be empty".into());
143        }
144        let Some(peer_key) = normalize_peer_name(peer_name) else {
145            return ToolResult::error("Parameter `peer` cannot be empty".into());
146        };
147        let Some(peer) = self.config.a2a.peers.get(&peer_key) else {
148            return ToolResult::error(format!("Unknown A2A peer: {peer_name}"));
149        };
150        if !peer.enabled {
151            return ToolResult::error(format!("A2A peer `{peer_name}` is disabled"));
152        }
153        if peer.trust == crate::config::A2ATrust::Sandboxed {
154            // A sandboxed peer makes this specific send high-risk regardless
155            // of which chat invoked it (same shape as the bash
156            // dangerous-pattern gate): the executor sees `approval_required`
157            // and pauses for explicit operator approval before anything
158            // leaves for the peer.
159            let already_approved = input
160                .get(A2A_HIGH_RISK_APPROVED_KEY)
161                .and_then(|v| v.as_bool())
162                .unwrap_or(false);
163            if !already_approved {
164                return ToolResult::error(format!(
165                    "Approval required: A2A peer `{peer_name}` has trust tier `sandboxed`. \
166                     Operator must explicitly approve before this message is sent."
167                ))
168                .with_error_type("approval_required");
169            }
170        }
171        let Some(base_url) = normalize_base_url(&peer.base_url) else {
172            return ToolResult::error(format!("A2A peer `{peer_name}` has invalid base_url"));
173        };
174        let session_key = input
175            .get("session_key")
176            .and_then(|v| v.as_str())
177            .map(str::trim)
178            .filter(|v| !v.is_empty())
179            .map(ToOwned::to_owned)
180            .or_else(|| peer.default_session_key.clone())
181            .unwrap_or_else(|| format!("a2a:{peer_key}"));
182        let timeout_secs = input
183            .get("timeout_secs")
184            .and_then(|v| v.as_u64())
185            .unwrap_or_else(|| self.config.tool_timeout_secs(self.name(), 60));
186        let body = A2AMessageRequest {
187            session_key: Some(session_key.clone()),
188            sender_name: None,
189            source_agent: Some(crate::a2a::local_agent_name(&self.config)),
190            source_url: self.config.a2a.public_base_url.clone(),
191            message: message.to_string(),
192        };
193
194        let mut request = self
195            .client
196            .post(format!("{base_url}{A2A_MESSAGE_PATH}"))
197            .timeout(std::time::Duration::from_secs(timeout_secs))
198            .header("x-microclaw-a2a-version", A2A_PROTOCOL_VERSION)
199            .json(&body);
200        if let Some(token) = peer.bearer_token.as_deref() {
201            request = request.bearer_auth(token);
202        }
203        let response = match request.send().await {
204            Ok(resp) => resp,
205            Err(err) => {
206                return ToolResult::error(format!("A2A request to `{peer_name}` failed: {err}"))
207            }
208        };
209        let status = response.status();
210        let body_text = match response.text().await {
211            Ok(text) => text,
212            Err(err) => {
213                return ToolResult::error(format!(
214                    "A2A peer `{peer_name}` returned unreadable body: {err}"
215                ))
216            }
217        };
218        if !status.is_success() {
219            return ToolResult::error(format!(
220                "A2A peer `{peer_name}` returned HTTP {}: {}",
221                status.as_u16(),
222                body_text.trim()
223            ))
224            .with_status_code(status.as_u16().into());
225        }
226        let parsed: A2AMessageResponse = match serde_json::from_str(&body_text) {
227            Ok(body) => body,
228            Err(err) => {
229                return ToolResult::error(format!(
230                    "A2A peer `{peer_name}` returned invalid JSON: {err}"
231                ))
232            }
233        };
234        ToolResult::success(
235            serde_json::to_string_pretty(&json!({
236                "peer": peer_key,
237                "protocol_version": parsed.protocol_version,
238                "agent_name": parsed.agent_name,
239                "session_key": parsed.session_key,
240                "response": parsed.response
241            }))
242            .unwrap_or(parsed.response),
243        )
244    }
245}
246
247#[cfg(test)]
248mod tests {
249    use super::*;
250    use tokio::io::{AsyncReadExt, AsyncWriteExt};
251    use tokio::net::TcpListener;
252
253    #[tokio::test]
254    async fn test_a2a_list_peers_returns_enabled_peers() {
255        let mut cfg = Config::test_defaults();
256        cfg.a2a.enabled = true;
257        cfg.a2a.peers.insert(
258            "planner".into(),
259            crate::config::A2APeerConfig {
260                enabled: true,
261                base_url: "https://planner.example.com".into(),
262                bearer_token: Some("secret".into()),
263                description: Some("plans".into()),
264                default_session_key: Some("a2a:planner".into()),
265                trust: Default::default(),
266            },
267        );
268        let tool = A2AListPeersTool::new(&cfg);
269        let result = tool.execute(json!({})).await;
270        assert!(!result.is_error);
271        assert!(result.content.contains("\"peer\": \"planner\""));
272    }
273
274    #[tokio::test]
275    async fn test_a2a_send_to_sandboxed_peer_requires_approval() {
276        let mut cfg = Config::test_defaults();
277        cfg.a2a.enabled = true;
278        cfg.a2a.peers.insert(
279            "risky".into(),
280            crate::config::A2APeerConfig {
281                enabled: true,
282                base_url: "https://risky.example.com".into(),
283                bearer_token: None,
284                description: None,
285                default_session_key: None,
286                trust: crate::config::A2ATrust::Sandboxed,
287            },
288        );
289        let tool = A2ASendTool::new(&cfg);
290        let result = tool
291            .execute(json!({"peer": "risky", "message": "do work"}))
292            .await;
293        assert!(result.is_error);
294        assert_eq!(result.error_type.as_deref(), Some("approval_required"));
295        // No network request was attempted: the gate fires before the URL
296        // is even normalized.
297        assert!(result.content.contains("sandboxed"));
298    }
299
300    #[tokio::test]
301    async fn test_a2a_send_posts_to_peer() {
302        let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
303        let addr = listener.local_addr().unwrap();
304        tokio::spawn(async move {
305            let (mut stream, _) = listener.accept().await.unwrap();
306            let mut request = Vec::new();
307            let mut buffer = [0_u8; 4096];
308            loop {
309                let bytes_read = stream.read(&mut buffer).await.unwrap();
310                if bytes_read == 0 {
311                    break;
312                }
313                request.extend_from_slice(&buffer[..bytes_read]);
314                if request.windows(4).any(|window| window == b"\r\n\r\n") && request.ends_with(b"}")
315                {
316                    break;
317                }
318            }
319            let request = String::from_utf8(request).unwrap();
320            assert!(request.starts_with(&format!("POST {A2A_MESSAGE_PATH} HTTP/1.1")));
321            assert!(request
322                .to_ascii_lowercase()
323                .contains("authorization: bearer secret"));
324            assert!(request.contains("\"message\":\"do work\""));
325
326            let response = serde_json::to_vec(&A2AMessageResponse {
327                ok: true,
328                protocol_version: A2A_PROTOCOL_VERSION.to_string(),
329                agent_name: "Worker".into(),
330                session_key: "a2a:worker".into(),
331                response: "done".into(),
332            })
333            .unwrap();
334            let headers = format!(
335                "HTTP/1.1 200 OK\r\ncontent-type: application/json\r\ncontent-length: {}\r\nconnection: close\r\n\r\n",
336                response.len()
337            );
338            stream.write_all(headers.as_bytes()).await.unwrap();
339            stream.write_all(&response).await.unwrap();
340        });
341
342        let mut cfg = Config::test_defaults();
343        cfg.a2a.enabled = true;
344        cfg.a2a.agent_name = Some("Planner".into());
345        cfg.a2a.peers.insert(
346            "worker".into(),
347            crate::config::A2APeerConfig {
348                enabled: true,
349                base_url: format!("http://{}", addr),
350                bearer_token: Some("secret".into()),
351                description: None,
352                default_session_key: None,
353                trust: Default::default(),
354            },
355        );
356        let tool = A2ASendTool::new(&cfg);
357        let result = tool
358            .execute(json!({"peer":"worker","message":"do work","timeout_secs":5}))
359            .await;
360        assert!(!result.is_error, "{}", result.content);
361        assert!(result.content.contains("\"response\": \"done\""));
362    }
363}
microclaw_engine/tools/a2a.rs

microclaw_engine/tools/
a2a.rs
