netget/server/a2a/
model.rs

1//! A2A protocol 1.0, JSON-RPC binding: the wire shapes (protobuf-JSON, camelCase) both roles
2//! build and check. Pinned to the released 1.0 specification as implemented by a2a-sdk 1.2.1.
3use anyhow::{bail, ensure, Context, Result};
4use serde_json::{json, Map, Value};
5
6pub const PROTOCOL_VERSION: &str = "1.0";
7pub const VERSION_HEADER: &str = "A2A-Version";
8pub const CARD_PATH: &str = "/.well-known/agent-card.json";
9pub const MAX_BODY_BYTES: usize = 1024 * 1024;
10pub const MAX_PARTS: usize = 64;
11pub const MAX_TEXT_BYTES: usize = 256 * 1024;
12
13pub const TASK_NOT_FOUND: i64 = -32001;
14pub const TASK_NOT_CANCELABLE: i64 = -32002;
15pub const PUSH_NOT_SUPPORTED: i64 = -32003;
16pub const UNSUPPORTED_OPERATION: i64 = -32004;
17pub const CONTENT_TYPE_NOT_SUPPORTED: i64 = -32005;
18pub const INVALID_AGENT_RESPONSE: i64 = -32006;
19pub const EXTENDED_CARD_NOT_CONFIGURED: i64 = -32007;
20pub const VERSION_NOT_SUPPORTED: i64 = -32009;
21
22pub const STATES: &[(&str, &str)] = &[
23    ("submitted", "TASK_STATE_SUBMITTED"),
24    ("working", "TASK_STATE_WORKING"),
25    ("completed", "TASK_STATE_COMPLETED"),
26    ("failed", "TASK_STATE_FAILED"),
27    ("canceled", "TASK_STATE_CANCELED"),
28    ("input_required", "TASK_STATE_INPUT_REQUIRED"),
29    ("rejected", "TASK_STATE_REJECTED"),
30    ("auth_required", "TASK_STATE_AUTH_REQUIRED"),
31];
32
33pub fn state_wire(short: &str) -> Result<&'static str> {
34    STATES.iter().find(|(s, _)| *s == short).map(|(_, w)| *w).with_context(|| format!("unknown task state '{short}'; use one of submitted, working, completed, failed, canceled, input_required, rejected, auth_required"))
35}
36pub fn state_short(wire: &str) -> Option<&'static str> {
37    STATES.iter().find(|(_, w)| *w == wire).map(|(s, _)| *s)
38}
39pub fn terminal(short: &str) -> bool {
40    matches!(short, "completed" | "failed" | "canceled" | "rejected")
41}
42
43pub fn budget_ok(v: &Value) -> bool {
44    crate::utils::json_budget::within_budget(v, MAX_BODY_BYTES, 50_000, 32)
45}
46
47pub fn new_id() -> String {
48    uuid::Uuid::new_v4().to_string()
49}
50
51/// `A2A-Version` must name 1.0 (patch ignored); missing means 0.3, which is refused.
52pub fn version_ok(header: Option<&str>) -> bool {
53    header.is_some_and(|v| {
54        let mut parts = v.trim().split('.');
55        parts.next() == Some("1")
56            && parts.next() == Some("0")
57            && parts.all(|p| p.bytes().all(|b| b.is_ascii_digit()))
58    })
59}
60
61pub fn rpc_result(id: &Value, result: Value) -> Value {
62    json!({"jsonrpc": "2.0", "id": id, "result": result})
63}
64pub fn rpc_error(id: &Value, code: i64, message: &str) -> Value {
65    json!({"jsonrpc": "2.0", "id": id, "error": {"code": code, "message": message}})
66}
67
68/// One SSE event carrying a JSON-RPC response.
69pub fn sse(value: &Value) -> String {
70    format!("data: {}\n\n", value)
71}
72
73/// Parts as handed to a handler: text and structured data as-is, files by reference/size.
74fn parts_view(parts: &[Value]) -> Result<(String, Vec<Value>)> {
75    ensure!(
76        !parts.is_empty() && parts.len() <= MAX_PARTS,
77        "a message carries 1..=64 parts"
78    );
79    let mut text = String::new();
80    let mut view = Vec::new();
81    for p in parts {
82        let o = p.as_object().context("each part must be an object")?;
83        if let Some(t) = o.get("text").and_then(Value::as_str) {
84            ensure!(t.len() <= MAX_TEXT_BYTES, "text part exceeds 256 KiB");
85            if !text.is_empty() {
86                text.push('\n');
87            }
88            text.push_str(t);
89            view.push(json!({"kind": "text", "text": t}));
90        } else if let Some(d) = o.get("data") {
91            view.push(json!({"kind": "data", "data": d}));
92        } else if let Some(u) = o.get("url").and_then(Value::as_str) {
93            view.push(json!({"kind": "file", "url": u, "filename": o.get("filename"), "media_type": o.get("mediaType")}));
94        } else if let Some(r) = o.get("raw").and_then(Value::as_str) {
95            // Inline bytes are not handed to a model; their size and type are.
96            view.push(json!({"kind": "file", "inline_bytes": r.len() / 4 * 3, "filename": o.get("filename"), "media_type": o.get("mediaType")}));
97        } else {
98            bail!("a part needs text, data, url or raw");
99        }
100    }
101    ensure!(text.len() <= MAX_TEXT_BYTES, "message text exceeds 256 KiB");
102    Ok((text, view))
103}
104
105/// A message sent to the agent, checked.
106pub struct Incoming {
107    pub message_id: String,
108    pub context_id: Option<String>,
109    pub task_id: Option<String>,
110    pub text: String,
111    pub parts: Vec<Value>,
112}
113
114pub fn incoming_message(params: &Value) -> Result<Incoming> {
115    let m = params
116        .get("message")
117        .and_then(Value::as_object)
118        .context("params.message is required")?;
119    let message_id = m
120        .get("messageId")
121        .and_then(Value::as_str)
122        .filter(|s| !s.is_empty() && s.len() <= 256)
123        .context("message.messageId is required")?;
124    ensure!(
125        m.get("role").and_then(Value::as_str) == Some("ROLE_USER"),
126        "message.role must be ROLE_USER"
127    );
128    let parts = m
129        .get("parts")
130        .and_then(Value::as_array)
131        .context("message.parts is required")?;
132    let (text, view) = parts_view(parts)?;
133    let id = |k: &str| -> Result<Option<String>> {
134        match m.get(k) {
135            None | Some(Value::Null) => Ok(None),
136            Some(Value::String(s)) if !s.is_empty() && s.len() <= 256 => Ok(Some(s.clone())),
137            _ => bail!("message.{k} must be a non-empty string"),
138        }
139    };
140    Ok(Incoming {
141        message_id: message_id.into(),
142        context_id: id("contextId")?,
143        task_id: id("taskId")?,
144        text,
145        parts: view,
146    })
147}
148
149/// Parts a handler supplies: text, data or file references (never inline bytes).
150pub fn parts_from(text: Option<&str>, parts: Option<&Value>) -> Result<Vec<Value>> {
151    let mut out = Vec::new();
152    if let Some(t) = text {
153        ensure!(t.len() <= MAX_TEXT_BYTES, "text exceeds 256 KiB");
154        out.push(json!({"text": t}));
155    }
156    if let Some(list) = parts.filter(|p| !p.is_null()) {
157        for p in list.as_array().context("parts must be an array")? {
158            let o = p.as_object().context("each part must be an object")?;
159            let part = if let Some(t) = o.get("text").and_then(Value::as_str) {
160                json!({"text": t})
161            } else if let Some(d) = o.get("data") {
162                json!({"data": d})
163            } else if let Some(u) = o.get("url").and_then(Value::as_str) {
164                ensure!(
165                    u.starts_with("http://") || u.starts_with("https://"),
166                    "file parts reference an http(s) url"
167                );
168                let mut f = json!({"url": u});
169                if let Some(n) = o.get("filename").and_then(Value::as_str) {
170                    f["filename"] = json!(n);
171                }
172                if let Some(m) = o.get("media_type").and_then(Value::as_str) {
173                    f["mediaType"] = json!(m);
174                }
175                f
176            } else {
177                bail!("a part needs text, data or url");
178            };
179            out.push(part);
180        }
181    }
182    ensure!(
183        !out.is_empty() && out.len() <= MAX_PARTS,
184        "an answer carries 1..=64 parts"
185    );
186    Ok(out)
187}
188
189pub fn agent_message(
190    text: Option<&str>,
191    parts: Option<&Value>,
192    context_id: &str,
193    task_id: Option<&str>,
194) -> Result<Value> {
195    let mut m = json!({"messageId": new_id(), "contextId": context_id, "role": "ROLE_AGENT", "parts": parts_from(text, parts)?});
196    if let Some(t) = task_id {
197        m["taskId"] = json!(t);
198    }
199    Ok(m)
200}
201
202pub fn timestamp() -> String {
203    let now = crate::utils::clock::SystemTime::now()
204        .duration_since(crate::utils::clock::UNIX_EPOCH)
205        .unwrap_or_default();
206    let secs = now.as_secs();
207    let days = (secs / 86_400) as i64;
208    let rem = secs % 86_400;
209    let z = days + 719_468;
210    let era = z.div_euclid(146_097);
211    let doe = z - era * 146_097;
212    let yoe = (doe - doe / 1460 + doe / 36_524 - doe / 146_096) / 365;
213    let doy = doe - (365 * yoe + yoe / 4 - yoe / 100);
214    let mp = (5 * doy + 2) / 153;
215    let d = doy - (153 * mp + 2) / 5 + 1;
216    let m = if mp < 10 { mp + 3 } else { mp - 9 };
217    let y = yoe + era * 400 + i64::from(m <= 2);
218    format!(
219        "{y:04}-{m:02}-{d:02}T{:02}:{:02}:{:02}Z",
220        rem / 3600,
221        rem % 3600 / 60,
222        rem % 60
223    )
224}
225
226/// A task from a handler's `{id, state, text, artifacts}` description.
227pub fn task(spec: &Value, default_id: &str, context_id: &str) -> Result<Value> {
228    let o = spec.as_object().context("task must be an object")?;
229    let id = o.get("id").and_then(Value::as_str).unwrap_or(default_id);
230    ensure!(
231        !id.is_empty() && id.len() <= 256,
232        "task id must be 1..=256 characters"
233    );
234    let state = state_wire(
235        o.get("state")
236            .and_then(Value::as_str)
237            .context("task.state is required")?,
238    )?;
239    let mut status = json!({"state": state, "timestamp": timestamp()});
240    if let Some(t) = o.get("text").and_then(Value::as_str) {
241        status["message"] = agent_message(Some(t), None, context_id, Some(id))?;
242    }
243    let mut t = json!({"id": id, "contextId": o.get("context_id").and_then(Value::as_str).unwrap_or(context_id), "status": status});
244    if let Some(list) = o.get("artifacts").filter(|a| !a.is_null()) {
245        let mut arts = Vec::new();
246        for a in list.as_array().context("artifacts must be an array")? {
247            ensure!(arts.len() < MAX_PARTS, "too many artifacts");
248            let mut art = json!({"artifactId": a.get("id").and_then(Value::as_str).map(str::to_owned).unwrap_or_else(new_id), "parts": parts_from(a.get("text").and_then(Value::as_str), a.get("parts"))?});
249            if let Some(n) = a.get("name").and_then(Value::as_str) {
250                art["name"] = json!(n);
251            }
252            arts.push(art);
253        }
254        t["artifacts"] = Value::Array(arts);
255    }
256    ensure!(budget_ok(&t), "task exceeds the A2A bounds");
257    Ok(t)
258}
259
260/// The SSE sequence for a streamed task: snapshot, working, artifacts, final status.
261pub fn stream_events(id: &Value, task: &Value) -> Vec<Value> {
262    let task_id = task["id"].clone();
263    let context_id = task["contextId"].clone();
264    let mut out = vec![rpc_result(
265        id,
266        json!({"task": {"id": task_id, "contextId": context_id, "status": {"state": "TASK_STATE_SUBMITTED"}}}),
267    )];
268    let final_state = task["status"]["state"]
269        .as_str()
270        .unwrap_or("TASK_STATE_COMPLETED");
271    if final_state != "TASK_STATE_SUBMITTED" {
272        out.push(rpc_result(id, json!({"statusUpdate": {"taskId": task_id, "contextId": context_id, "status": {"state": "TASK_STATE_WORKING", "timestamp": timestamp()}}})));
273    }
274    if let Some(arts) = task["artifacts"].as_array() {
275        for a in arts {
276            out.push(rpc_result(id, json!({"artifactUpdate": {"taskId": task_id, "contextId": context_id, "artifact": a, "lastChunk": true}})));
277        }
278    }
279    if !matches!(final_state, "TASK_STATE_SUBMITTED" | "TASK_STATE_WORKING") {
280        out.push(rpc_result(id, json!({"statusUpdate": {"taskId": task_id, "contextId": context_id, "status": task["status"]}})));
281    }
282    out
283}
284
285/// Agent card built from configuration.
286pub fn card(
287    name: &str,
288    description: &str,
289    version: &str,
290    url: &str,
291    streaming: bool,
292    skills: &[Value],
293) -> Value {
294    json!({
295        "name": name,
296        "description": description,
297        "version": version,
298        "supportedInterfaces": [{"url": url, "protocolBinding": "JSONRPC", "protocolVersion": PROTOCOL_VERSION}],
299        "capabilities": {"streaming": streaming, "pushNotifications": false},
300        "defaultInputModes": ["text/plain", "application/json"],
301        "defaultOutputModes": ["text/plain", "application/json"],
302        "skills": skills,
303    })
304}
305
306/// Client-side: the JSON-RPC interface a card offers for protocol 1.0.
307pub fn card_rpc_url(card: &Value) -> Result<String> {
308    let interfaces = card
309        .get("supportedInterfaces")
310        .and_then(Value::as_array)
311        .context("agent card lacks supportedInterfaces")?;
312    let iface = interfaces
313        .iter()
314        .find(|i| {
315            i["protocolBinding"] == "JSONRPC"
316                && i["protocolVersion"]
317                    .as_str()
318                    .is_some_and(|v| version_ok(Some(v)))
319        })
320        .context("agent card offers no JSON-RPC interface for protocol 1.0")?;
321    let url = iface["url"].as_str().context("interface url missing")?;
322    ensure!(
323        url.starts_with("http://") || url.starts_with("https://"),
324        "interface url must be http(s)"
325    );
326    Ok(url.to_owned())
327}
328
329/// Client-side check of a result for `method`.
330pub fn check_result(method: &str, result: &Value) -> Result<()> {
331    ensure!(budget_ok(result), "result exceeds the A2A bounds");
332    let o: &Map<String, Value> = result.as_object().context("result must be an object")?;
333    match method {
334        "SendMessage" | "SendStreamingMessage" => ensure!(
335            ["task", "message", "statusUpdate", "artifactUpdate"]
336                .iter()
337                .any(|k| o.contains_key(*k)),
338            "a send result carries task, message, statusUpdate or artifactUpdate"
339        ),
340        "GetTask" | "CancelTask" => ensure!(
341            o.get("id").is_some_and(Value::is_string)
342                && o.get("status").is_some_and(Value::is_object),
343            "a task result needs id and status"
344        ),
345        "ListTasks" => ensure!(
346            o.get("tasks").is_none_or(Value::is_array),
347            "tasks must be an array"
348        ),
349        _ => {}
350    }
351    Ok(())
352}
netget/server/a2a/model.rs

netget/server/a2a/
model.rs
