1use crate::llm::actions::{
2 client_trait::{Client, ClientActionResult},
3 protocol_trait::Protocol,
4 ActionDefinition, ParameterDefinition,
5};
6use crate::protocol::{ConnectContext, EventType};
7use crate::server::a2a::actions::{action, parameter};
8use anyhow::{bail, ensure, Context, Result};
9use serde_json::{json, Value};
10use std::sync::LazyLock;
11
12#[derive(Default)]
13pub struct A2aClientProtocol;
14impl A2aClientProtocol {
15 pub fn new() -> Self {
16 Self
17 }
18}
19
20fn send_action() -> ActionDefinition {
21 action(
22 "a2a_send_message",
23 "Send a user message to the agent (SendMessage, or SendStreamingMessage when stream is true and the card advertises streaming). Rust builds the message, ids and A2A-Version header.",
24 vec![
25 parameter("text", "string", "What to tell the agent, sent as the message's text part (1 byte to 256 KiB)", true),
26 parameter("data", "object", "Optional structured data part", false),
27 parameter("context_id", "string", "Continue this conversation context", false),
28 parameter("task_id", "string", "Continue this task (e.g. after input_required)", false),
29 parameter("stream", "boolean", "Use SendStreamingMessage and collect the SSE events", false),
30 ],
31 json!({"type":"a2a_send_message","text":"Book a flight to Lisbon"}),
32 )
33}
34fn get_action() -> ActionDefinition {
35 action(
36 "a2a_get_task",
37 "Fetch a task by id (GetTask)",
38 vec![
39 parameter("task_id", "string", "Task id returned earlier", true),
40 parameter(
41 "history_length",
42 "number",
43 "Messages of history to include",
44 false,
45 ),
46 ],
47 json!({"type":"a2a_get_task","task_id":"<task id>"}),
48 )
49}
50fn cancel_action() -> ActionDefinition {
51 action(
52 "a2a_cancel_task",
53 "Ask the agent to cancel a task (CancelTask)",
54 vec![parameter("task_id", "string", "Task id to cancel", true)],
55 json!({"type":"a2a_cancel_task","task_id":"<task id>"}),
56 )
57}
58fn list_action() -> ActionDefinition {
59 action(
60 "a2a_list_tasks",
61 "List the agent's tasks (ListTasks)",
62 vec![
63 parameter("context_id", "string", "Only tasks of this context", false),
64 parameter(
65 "page_size",
66 "number",
67 "Maximum tasks to return in one page, 1 to 100 (the agent's default when omitted)",
68 false,
69 ),
70 ],
71 json!({"type":"a2a_list_tasks"}),
72 )
73}
74fn disconnect_action() -> ActionDefinition {
75 action(
76 "disconnect",
77 "Stop this A2A client",
78 vec![],
79 json!({"type":"disconnect"}),
80 )
81}
82fn actions() -> Vec<ActionDefinition> {
83 vec![
84 send_action(),
85 get_action(),
86 cancel_action(),
87 list_action(),
88 disconnect_action(),
89 ]
90}
91
92pub static CONNECTED_EVENT: LazyLock<EventType> = LazyLock::new(|| {
93 EventType::new(
94 "a2a_connected",
95 "The agent card was fetched and offers a JSON-RPC interface for protocol 1.0",
96 send_action().example.clone(),
97 )
98 .with_parameters(vec![
99 parameter("name", "string", "Agent name from the card", true),
100 parameter("description", "string", "Agent description", true),
101 parameter("skills", "array", "Skills the agent advertises", true),
102 parameter("streaming", "boolean", "Whether the agent streams", true),
103 parameter(
104 "rpc_url",
105 "string",
106 "JSON-RPC endpoint the client will use",
107 true,
108 ),
109 ])
110 .with_actions(actions())
111});
112pub static RESPONSE_EVENT: LazyLock<EventType> = LazyLock::new(|| {
113 EventType::new(
114 "a2a_response",
115 "The agent's answer to the last request",
116 disconnect_action().example.clone(),
117 )
118 .with_parameters(vec![
119 parameter("method", "string", "The JSON-RPC method sent", true),
120 parameter(
121 "result",
122 "object",
123 "The result: {task}|{message} for sends, a task for Get/Cancel, {tasks} for List",
124 false,
125 ),
126 parameter(
127 "error",
128 "object",
129 "JSON-RPC error {code, message}, e.g. -32001 task not found",
130 false,
131 ),
132 parameter(
133 "stream_events",
134 "array",
135 "For streaming sends: every StreamResponse received in order (up to 256)",
136 false,
137 ),
138 parameter(
139 "task_id",
140 "string",
141 "Task id the answer is about, when there is one",
142 false,
143 ),
144 parameter(
145 "state",
146 "string",
147 "Last known task state (completed, working, canceled, ...)",
148 false,
149 ),
150 ])
151 .with_actions(actions())
152});
153
154impl Protocol for A2aClientProtocol {
155 fn protocol_name(&self) -> &'static str {
156 "A2A"
157 }
158 fn stack_name(&self) -> &'static str {
159 "ETH>IP>TCP>HTTP>A2A"
160 }
161 fn keywords(&self) -> Vec<&'static str> {
162 vec!["a2a", "agent2agent", "agent client", "json-rpc"]
163 }
164 fn description(&self) -> &'static str {
165 "A2A 1.0 client: resolves an agent card, sends and streams messages, manages tasks"
166 }
167 fn get_async_actions(&self, _: &crate::state::app_state::AppState) -> Vec<ActionDefinition> {
168 actions()
169 }
170 fn get_sync_actions(&self) -> Vec<ActionDefinition> {
171 vec![]
172 }
173 fn get_event_types(&self) -> Vec<EventType> {
174 vec![CONNECTED_EVENT.clone(), RESPONSE_EVENT.clone()]
175 }
176 fn get_startup_parameters(&self) -> Vec<ParameterDefinition> {
177 vec![ParameterDefinition {
178 name: "allow_card_redirect".into(),
179 type_hint: "boolean".into(),
180 description: "Follow an agent card whose JSON-RPC url names a different host than remote_addr (off: such a card is refused, so a client pointed at one agent cannot be sent elsewhere)".into(),
181 required: false,
182 example: json!(true),
183 default: Some(json!(super::DEFAULT_ALLOW_REDIRECT)),
184 }]
185 }
186 fn metadata(&self) -> crate::protocol::metadata::ProtocolMetadataV2 {
187 use crate::protocol::metadata::*;
188 ProtocolMetadataV2::builder()
189 .state(DevelopmentState::Experimental)
190 .privilege_requirement(PrivilegeRequirement::None)
191 .implementation("Shared http_fetch client (reqwest natively, no redirects); agent card resolution; JSON-RPC 2.0 with A2A-Version 1.0; SSE stream collection; results checked against the 1.0 shapes")
192 .llm_control("What to ask the agent, whether to stream, and which tasks to fetch, list or cancel")
193 .e2e_testing("tests/client/a2a: an a2a-sdk 1.2.1 agent (independent) answers a direct message, a streamed task, GetTask and CancelTask; NetGet pair and refusals")
194 .notes("JSON-RPC binding only (no HTTP+JSON or gRPC, no push notifications, no 0.3). Streams are collected whole (1 MiB, 256 events, 30 s). A card pointing at another host is refused unless allow_card_redirect.")
195 .max_inbound_bytes(crate::server::a2a::model::MAX_BODY_BYTES)
196 .build()
197 }
198 fn example_prompt(&self) -> &'static str {
199 "Ask the A2A agent at 127.0.0.1:8080 to summarise today's tasks"
200 }
201 fn get_startup_examples(&self) -> crate::llm::actions::StartupExamples {
202 let llm = json!({"type":"open_client","protocol":"a2a","remote_addr":"127.0.0.1:8080","instruction":"Ask the agent for the weather in Lisbon"});
203 let mut static_example = llm.clone();
204 static_example["event_handlers"] = json!([
205 {"event_pattern":"a2a_connected","handler":{"type":"static","actions":[send_action().example]}},
206 {"event_pattern":"a2a_response","handler":{"type":"static","actions":[]}}
207 ]);
208 let mut scripted = static_example.clone();
209 scripted["event_handlers"][1]["handler"] = json!({"type":"script","language":"python","code":"import json\nprint(json.dumps({'actions':[]}))"});
210 crate::llm::actions::StartupExamples::new(llm, scripted, static_example)
211 }
212 fn group_name(&self) -> &'static str {
213 "AI & Agents"
214 }
215}
216
217impl Client for A2aClientProtocol {
218 fn connect(
219 &self,
220 ctx: ConnectContext,
221 ) -> std::pin::Pin<Box<dyn std::future::Future<Output = Result<std::net::SocketAddr>> + Send>>
222 {
223 Box::pin(super::connect(ctx))
224 }
225 fn execute_action(&self, v: Value) -> Result<ClientActionResult> {
226 let id_ok = |k: &str| -> Result<()> {
227 if let Some(x) = v.get(k).filter(|x| !x.is_null()) {
228 let s = x
229 .as_str()
230 .with_context(|| format!("{k} must be a string"))?;
231 ensure!(
232 !s.is_empty() && s.len() <= 256,
233 "{k} must be 1..=256 characters"
234 );
235 }
236 Ok(())
237 };
238 match v["type"].as_str() {
239 Some("a2a_send_message") => {
240 let t = v["text"].as_str().context("text is required")?;
241 ensure!(
242 !t.is_empty() && t.len() <= crate::server::a2a::model::MAX_TEXT_BYTES,
243 "text must be 1 byte..256 KiB"
244 );
245 id_ok("context_id")?;
246 id_ok("task_id")?;
247 }
248 Some("a2a_get_task" | "a2a_cancel_task") => {
249 ensure!(
250 v.get("task_id").is_some_and(|t| t.is_string()),
251 "task_id is required"
252 );
253 id_ok("task_id")?;
254 }
255 Some("a2a_list_tasks") => {
256 if let Some(n) = v.get("page_size").and_then(Value::as_u64) {
257 ensure!((1..=100).contains(&n), "page_size must be 1..=100");
258 }
259 }
260 Some("disconnect") => return Ok(ClientActionResult::Disconnect),
261 _ => bail!("Unknown A2A client action"),
262 }
263 Ok(ClientActionResult::Custom {
264 name: v["type"].as_str().unwrap_or_default().into(),
265 data: v,
266 })
267 }
268}