1use super::model;
2use crate::llm::actions::{
3 protocol_trait::{ActionResult, Protocol, Server},
4 ActionDefinition, Parameter, ParameterDefinition,
5};
6use crate::protocol::log_template::LogTemplate;
7use crate::protocol::{EventType, SpawnContext};
8use crate::state::app_state::AppState;
9use anyhow::{bail, ensure, Context, Result};
10use serde_json::{json, Value};
11use std::sync::LazyLock;
12
13#[derive(Default)]
14pub struct A2aProtocol;
15impl A2aProtocol {
16 pub fn new() -> Self {
17 Self
18 }
19}
20
21pub fn parameter(name: &str, kind: &str, description: &str, required: bool) -> Parameter {
22 Parameter {
23 name: name.into(),
24 type_hint: kind.into(),
25 description: description.into(),
26 required,
27 }
28}
29
30pub fn action(
31 name: &str,
32 description: &str,
33 parameters: Vec<Parameter>,
34 example: Value,
35) -> ActionDefinition {
36 let log_template = match name {
37 "a2a_reply" => LogTemplate::new().with_info("-> A2A reply"),
38 _ => LogTemplate::new().with_info(format!("-> A2A {name}")),
39 };
40 ActionDefinition {
41 name: name.into(),
42 description: description.into(),
43 parameters,
44 example,
45 log_template: Some(log_template),
46 }
47}
48
49pub const ERROR_NAMES: &[(&str, i64)] = &[
50 ("task_not_found", model::TASK_NOT_FOUND),
51 ("task_not_cancelable", model::TASK_NOT_CANCELABLE),
52 ("unsupported_operation", model::UNSUPPORTED_OPERATION),
53 (
54 "content_type_not_supported",
55 model::CONTENT_TYPE_NOT_SUPPORTED,
56 ),
57 ("invalid_params", -32602),
58];
59
60fn reply_action() -> ActionDefinition {
61 action(
62 "a2a_reply",
63 "Answer the pending A2A request. Rust builds the protobuf-JSON envelope, ids, timestamps and (for streaming) the SSE sequence. Supply exactly one of: message {text, parts} (a direct agent reply), task {id, state, text, artifacts} (SendMessage, GetTask, CancelTask), tasks [task, ...] (ListTasks), error {code, message}.",
64 vec![
65 parameter("message", "object", "{text, parts: [{text}|{data}|{url, filename, media_type}]} — a direct agent message", false),
66 parameter("task", "object", "{id (omit for a new task), state: submitted|working|completed|failed|canceled|input_required|rejected|auth_required, text (status message), artifacts: [{name, text, parts}]}", false),
67 parameter("tasks", "array", "ListTasks result: task objects as above (each with its id)", false),
68 parameter("error", "object", "{code: task_not_found|task_not_cancelable|unsupported_operation|content_type_not_supported|invalid_params, message}", false),
69 ],
70 json!({"type":"a2a_reply","task":{"state":"completed","artifacts":[{"name":"answer","text":"42"}]}}),
71 )
72}
73
74pub static MESSAGE_EVENT: LazyLock<EventType> = LazyLock::new(|| {
75 EventType::new(
76 "a2a_message",
77 "A user message via SendMessage or SendStreamingMessage. Answer with a direct message or a task; there is no task store in Rust (keep state in memory if later GetTask calls must find it).",
78 reply_action().example.clone(),
79 )
80 .with_parameters(vec![
81 parameter("method", "string", "SendMessage or SendStreamingMessage", true),
82 parameter("text", "string", "The message's text parts joined by newlines", true),
83 parameter("parts", "array", "Every part: {kind: text|data|file, ...}; inline file bytes are given by size only", true),
84 parameter("message_id", "string", "The caller's message id", true),
85 parameter("context_id", "string", "Conversation context id (Rust generates one when absent)", true),
86 parameter("task_id", "string", "Existing task this message continues, if any", false),
87 ])
88 .with_actions(vec![reply_action()])
89});
90
91pub static TASK_EVENT: LazyLock<EventType> = LazyLock::new(|| {
92 EventType::new("a2a_task_request", "GetTask, CancelTask or ListTasks. Answer from what you remember of earlier tasks, or with task_not_found / task_not_cancelable.", json!({"type":"a2a_reply","error":{"code":"task_not_found","message":"no such task"}}))
93 .with_parameters(vec![
94 parameter("method", "string", "GetTask, CancelTask or ListTasks", true),
95 parameter("task_id", "string", "Task id for GetTask/CancelTask", false),
96 parameter("params", "object", "The request's params (historyLength, contextId, pageSize, ...)", true),
97 ])
98 .with_actions(vec![reply_action()])
99});
100
101fn startup(
102 name: &str,
103 kind: &str,
104 description: &str,
105 example: Value,
106 default: Option<Value>,
107) -> ParameterDefinition {
108 ParameterDefinition {
109 name: name.into(),
110 type_hint: kind.into(),
111 description: description.into(),
112 required: false,
113 example,
114 default,
115 }
116}
117
118impl Protocol for A2aProtocol {
119 fn protocol_name(&self) -> &'static str {
120 "A2A"
121 }
122 fn stack_name(&self) -> &'static str {
123 "ETH>IP>TCP>HTTP>A2A"
124 }
125 fn description(&self) -> &'static str {
126 "Agent2Agent (A2A) protocol 1.0 agent over JSON-RPC with streaming"
127 }
128 fn keywords(&self) -> Vec<&'static str> {
129 vec!["a2a", "agent2agent", "agent card", "ai agent", "json-rpc"]
130 }
131 fn get_async_actions(&self, _: &AppState) -> Vec<ActionDefinition> {
132 vec![]
133 }
134 fn get_sync_actions(&self) -> Vec<ActionDefinition> {
135 vec![reply_action()]
136 }
137 fn get_event_types(&self) -> Vec<EventType> {
138 vec![MESSAGE_EVENT.clone(), TASK_EVENT.clone()]
139 }
140 fn get_startup_parameters(&self) -> Vec<ParameterDefinition> {
141 vec![
142 startup(
143 "agent_name",
144 "string",
145 "Agent card name",
146 json!("Travel Agent"),
147 Some(json!(super::DEFAULT_NAME)),
148 ),
149 startup(
150 "agent_description",
151 "string",
152 "Agent card description",
153 json!("Books trips"),
154 Some(json!(super::DEFAULT_DESCRIPTION)),
155 ),
156 startup(
157 "skills",
158 "array",
159 "Agent card skills: [{id, name, description, tags: [..], examples: [..]}]",
160 json!([{"id":"book","name":"Book","description":"Book a flight","tags":["travel"]}]),
161 None,
162 ),
163 startup(
164 "streaming",
165 "boolean",
166 "Advertise and serve SendStreamingMessage (SSE)",
167 json!(true),
168 Some(json!(super::DEFAULT_STREAMING)),
169 ),
170 ]
171 }
172 fn metadata(&self) -> crate::protocol::metadata::ProtocolMetadataV2 {
173 use crate::protocol::metadata::*;
174 ProtocolMetadataV2::builder()
175 .state(DevelopmentState::Experimental)
176 .privilege_requirement(PrivilegeRequirement::None)
177 .implementation("hyper HTTP/1.1; agent card at /.well-known/agent-card.json; JSON-RPC 2.0 with A2A-Version 1.0; protobuf-JSON messages, tasks and SSE stream responses built and checked in Rust")
178 .llm_control("Every reply: direct messages, task states and artifacts, task lookups and cancellation answers")
179 .e2e_testing("tests/server/a2a: a2a-sdk 1.2.1 client (independent) resolves the card, sends, streams, gets and cancels; JSON-RPC and version refusals")
180 .notes("No task store: GetTask/CancelTask/ListTasks are answered by the handler (persist with memory or SQLite). Push notifications, SubscribeToTask, the extended card, the HTTP+JSON and gRPC bindings and 0.3 compatibility are not implemented. 1 MiB bodies.")
181 .request_only("A2A answers each JSON-RPC request; push notifications are not offered")
182 .answers_on_failure()
183 .max_inbound_bytes(model::MAX_BODY_BYTES)
184 .build()
185 }
186 fn example_prompt(&self) -> &'static str {
187 "A2A agent named Echo that answers every message with its text reversed"
188 }
189 fn get_startup_examples(&self) -> crate::llm::actions::StartupExamples {
190 let llm = json!({"type":"open_server","base_stack":"a2a","port":8080,"instruction":"You are a helpful travel agent"});
191 let mut static_example = llm.clone();
192 static_example["event_handlers"] = json!([
193 {"event_pattern":"a2a_message","handler":{"type":"static","actions":[{"type":"a2a_reply","message":{"text":"Hello from NetGet"}}]}},
194 {"event_pattern":"a2a_task_request","handler":{"type":"static","actions":[{"type":"a2a_reply","error":{"code":"task_not_found","message":"no tasks"}}]}}
195 ]);
196 let mut scripted = static_example.clone();
197 scripted["event_handlers"][0]["handler"] = json!({"type":"script","language":"python","code":"import json,sys\ne=json.load(sys.stdin)['event']\nprint(json.dumps({'actions':[{'type':'a2a_reply','message':{'text':e['text'][::-1]}}]}))"});
198 crate::llm::actions::StartupExamples::new(llm, scripted, static_example)
199 }
200 fn group_name(&self) -> &'static str {
201 "AI & Agents"
202 }
203}
204
205impl Server for A2aProtocol {
206 fn spawn(
207 &self,
208 ctx: SpawnContext,
209 ) -> std::pin::Pin<Box<dyn std::future::Future<Output = Result<std::net::SocketAddr>> + Send>>
210 {
211 Box::pin(super::spawn(ctx))
212 }
213 fn execute_action(&self, v: Value) -> Result<ActionResult> {
214 match v["type"].as_str() {
215 Some("a2a_reply") => {
216 validate_reply(&v)?;
217 Ok(ActionResult::Custom {
218 name: "a2a_reply".into(),
219 data: v,
220 })
221 }
222 _ => bail!("Unknown A2A server action"),
223 }
224 }
225}
226
227pub fn validate_reply(v: &Value) -> Result<()> {
228 ensure!(model::budget_ok(v), "reply exceeds the A2A bounds");
229 let present: Vec<&str> = ["message", "task", "tasks", "error"]
230 .into_iter()
231 .filter(|k| v.get(*k).is_some_and(|x| !x.is_null()))
232 .collect();
233 ensure!(
234 present.len() == 1,
235 "a2a_reply needs exactly one of message, task, tasks, error"
236 );
237 match present[0] {
238 "message" => {
239 model::parts_from(v["message"]["text"].as_str(), v["message"].get("parts"))?;
240 }
241 "task" => {
242 model::task(&v["task"], "x", "x")?;
243 }
244 "tasks" => {
245 for t in v["tasks"].as_array().context("tasks must be an array")? {
246 ensure!(
247 t.get("id").is_some_and(Value::is_string),
248 "each listed task needs its id"
249 );
250 model::task(t, "x", "x")?;
251 }
252 }
253 _ => {
254 let code = v["error"]["code"]
255 .as_str()
256 .context("error.code is required")?;
257 ensure!(
258 ERROR_NAMES.iter().any(|(n, _)| *n == code),
259 "unknown error code '{code}'"
260 );
261 }
262 }
263 Ok(())
264}