a2a_server/
jsonrpc.rs

1// Copyright AGNTCY Contributors (https://github.com/agntcy)
2// Copyright A2A Contributors (https://github.com/a2aproject)
3// SPDX-License-Identifier: Apache-2.0
4use std::sync::Arc;
5
6use a2a::*;
7use a2a_pb::protojson_conv::{self, ProtoJsonPayload};
8use axum::{
9    Json,
10    body::Bytes,
11    extract::State,
12    http::{HeaderMap, StatusCode},
13    response::IntoResponse,
14};
15use futures::{StreamExt, stream::BoxStream};
16use serde_json::Value;
17
18use crate::handler::RequestHandler;
19use crate::middleware::{ServiceParams, extract_service_params};
20use crate::sse;
21
22/// Shared state for the JSON-RPC handler.
23pub struct JsonRpcState<H: RequestHandler> {
24    pub handler: Arc<H>,
25}
26
27impl<H: RequestHandler> Clone for JsonRpcState<H> {
28    fn clone(&self) -> Self {
29        JsonRpcState {
30            handler: self.handler.clone(),
31        }
32    }
33}
34
35/// Largest request body either binding accepts.
36///
37/// Without an explicit bound a single large POST is read into memory. The
38/// limit matches a2a-go's `MaxSSETokenSize` (`internal/sse/sse.go`), so a
39/// payload one SDK accepts is not rejected by the other.
40pub const MAX_REQUEST_BODY_BYTES: usize = 10 * 1024 * 1024;
41
42/// Create an axum router for the JSON-RPC protocol binding.
43///
44/// All requests are dispatched to a single POST endpoint that routes
45/// by the `method` field in the JSON-RPC envelope.
46pub fn jsonrpc_router<H: RequestHandler>(handler: Arc<H>) -> axum::Router {
47    let state = JsonRpcState { handler };
48    axum::Router::new()
49        .route("/", axum::routing::post(handle_jsonrpc::<H>))
50        .layer(axum::extract::DefaultBodyLimit::max(MAX_REQUEST_BODY_BYTES))
51        .with_state(state)
52}
53
54async fn handle_jsonrpc<H: RequestHandler>(
55    State(state): State<JsonRpcState<H>>,
56    headers: HeaderMap,
57    body: Bytes,
58) -> impl IntoResponse {
59    let params = extract_service_params(&headers);
60
61    // A `Json<JsonRpcRequest>` extractor would reject a malformed body with
62    // axum's own error response before this handler runs at all, which is
63    // the wrong shape for JSON-RPC over HTTP: the status stays 200 and the
64    // error belongs inside the envelope regardless of what failed.
65    let request: JsonRpcRequest = match serde_json::from_slice(&body) {
66        Ok(request) => request,
67        Err(error) => return malformed_request_response(&body, error),
68    };
69    let id = request.id.clone();
70    let method = request.method.as_str();
71
72    if request.jsonrpc != "2.0" {
73        return error_response(id, A2AError::invalid_request("invalid jsonrpc version"));
74    }
75
76    if let Err(error) = check_a2a_version(&params) {
77        return error_response(id, error);
78    }
79
80    if methods::is_streaming(method) {
81        return handle_streaming_request(&state, &params, &request).await;
82    }
83
84    handle_unary_request(&state, &params, &request).await
85}
86
87/// A body that failed to deserialize into [`JsonRpcRequest`]: invalid JSON
88/// syntax is `ParseError` (-32700); well-formed JSON with the wrong shape
89/// (e.g. a missing `method`) is `InvalidRequestError` (-32600). The id is
90/// recovered on a best-effort basis for the latter, since JSON-RPC 2.0 asks
91/// for it to be echoed back when it can be determined; a syntax error, by
92/// definition, cannot.
93fn malformed_request_response(body: &[u8], error: serde_json::Error) -> axum::response::Response {
94    if !error.is_data() {
95        return error_response(
96            JsonRpcId::Null,
97            A2AError {
98                code: error_code::PARSE_ERROR,
99                message: format!("parse error: {error}"),
100                details: None,
101            },
102        );
103    }
104    let id = serde_json::from_slice::<Value>(body)
105        .ok()
106        .and_then(|v| v.get("id").cloned())
107        .and_then(|v| serde_json::from_value(v).ok())
108        .unwrap_or(JsonRpcId::Null);
109    error_response(
110        id,
111        A2AError::invalid_request(format!("invalid request: {error}")),
112    )
113}
114
115async fn handle_unary_request<H: RequestHandler>(
116    state: &JsonRpcState<H>,
117    params: &ServiceParams,
118    request: &JsonRpcRequest,
119) -> axum::response::Response {
120    let id = request.id.clone();
121    let raw_params = request.params.clone().unwrap_or(Value::Null);
122
123    let result: Result<Value, A2AError> = match request.method.as_str() {
124        methods::SEND_MESSAGE => match protojson_conv::from_value::<SendMessageRequest>(raw_params)
125        {
126            Ok(req) => state
127                .handler
128                .send_message(params, req)
129                .await
130                .and_then(|r| protojson_value(&r)),
131            Err(e) => Err(invalid_params(e)),
132        },
133        methods::GET_TASK => match protojson_conv::from_value::<GetTaskRequest>(raw_params) {
134            Ok(req) => state
135                .handler
136                .get_task(params, req)
137                .await
138                .and_then(|r| protojson_value(&r)),
139            Err(e) => Err(invalid_params(e)),
140        },
141        methods::LIST_TASKS => match protojson_conv::from_value::<ListTasksRequest>(raw_params) {
142            Ok(req) => state
143                .handler
144                .list_tasks(params, req)
145                .await
146                .and_then(|r| protojson_value(&r)),
147            Err(e) => Err(invalid_params(e)),
148        },
149        methods::CANCEL_TASK => match protojson_conv::from_value::<CancelTaskRequest>(raw_params) {
150            Ok(req) => state
151                .handler
152                .cancel_task(params, req)
153                .await
154                .and_then(|r| protojson_value(&r)),
155            Err(e) => Err(invalid_params(e)),
156        },
157        methods::CREATE_PUSH_CONFIG => match parse_create_push_config_request(raw_params) {
158            Ok(req) => state
159                .handler
160                .create_push_config(params, req)
161                .await
162                .and_then(|r| protojson_value(&r)),
163            Err(e) => Err(invalid_params(e)),
164        },
165        methods::GET_PUSH_CONFIG => {
166            match protojson_conv::from_value::<GetTaskPushNotificationConfigRequest>(raw_params) {
167                Ok(req) => state
168                    .handler
169                    .get_push_config(params, req)
170                    .await
171                    .and_then(|r| protojson_value(&r)),
172                Err(e) => Err(invalid_params(e)),
173            }
174        }
175        methods::LIST_PUSH_CONFIGS => {
176            match protojson_conv::from_value::<ListTaskPushNotificationConfigsRequest>(raw_params) {
177                Ok(req) => state
178                    .handler
179                    .list_push_configs(params, req)
180                    .await
181                    .and_then(|r| protojson_value(&r)),
182                Err(e) => Err(invalid_params(e)),
183            }
184        }
185        methods::DELETE_PUSH_CONFIG => {
186            match protojson_conv::from_value::<DeleteTaskPushNotificationConfigRequest>(raw_params)
187            {
188                Ok(req) => state
189                    .handler
190                    .delete_push_config(params, req)
191                    .await
192                    .map(|_| Value::Null),
193                Err(e) => Err(invalid_params(e)),
194            }
195        }
196        methods::GET_EXTENDED_AGENT_CARD => {
197            match protojson_conv::from_value::<GetExtendedAgentCardRequest>(raw_params) {
198                Ok(req) => state
199                    .handler
200                    .get_extended_agent_card(params, req)
201                    .await
202                    .and_then(|r| protojson_value(&r)),
203                Err(e) => Err(invalid_params(e)),
204            }
205        }
206        "" => Err(A2AError::invalid_request("method is required")),
207        _ => Err(A2AError::method_not_found(&request.method)),
208    };
209
210    match result {
211        Ok(value) => {
212            let resp = JsonRpcResponse::success(id, value);
213            Json(resp).into_response()
214        }
215        Err(e) => error_response(id, e),
216    }
217}
218
219async fn handle_streaming_request<H: RequestHandler>(
220    state: &JsonRpcState<H>,
221    params: &ServiceParams,
222    request: &JsonRpcRequest,
223) -> axum::response::Response {
224    let id = request.id.clone();
225    let raw_params = request.params.clone().unwrap_or(Value::Null);
226
227    match request.method.as_str() {
228        methods::SEND_STREAMING_MESSAGE => {
229            match protojson_conv::from_value::<SendMessageRequest>(raw_params) {
230                Ok(req) => match state.handler.send_streaming_message(params, req).await {
231                    Ok(stream) => {
232                        sse::sse_jsonrpc_stream(id, protojson_stream(stream)).into_response()
233                    }
234                    Err(e) => error_response(id, e),
235                },
236                Err(e) => error_response(id, invalid_params(e)),
237            }
238        }
239        methods::SUBSCRIBE_TO_TASK => {
240            match protojson_conv::from_value::<SubscribeToTaskRequest>(raw_params) {
241                Ok(req) => match state.handler.subscribe_to_task(params, req).await {
242                    Ok(stream) => {
243                        sse::sse_jsonrpc_stream(id, protojson_stream(stream)).into_response()
244                    }
245                    Err(e) => error_response(id, e),
246                },
247                Err(e) => error_response(id, invalid_params(e)),
248            }
249        }
250        _ => error_response(id, A2AError::method_not_found(&request.method)),
251    }
252}
253
254fn error_response(id: JsonRpcId, err: A2AError) -> axum::response::Response {
255    let resp = JsonRpcResponse::error(id, err.to_jsonrpc_error());
256    (StatusCode::OK, Json(resp)).into_response()
257}
258
259fn protojson_value<T: ProtoJsonPayload>(value: &T) -> Result<Value, A2AError> {
260    protojson_conv::to_value(value)
261        .map_err(|e| A2AError::internal(format!("failed to serialize ProtoJSON payload: {e}")))
262}
263
264fn protojson_stream(
265    stream: BoxStream<'static, Result<StreamResponse, A2AError>>,
266) -> BoxStream<'static, Result<Value, A2AError>> {
267    Box::pin(stream.map(|item| {
268        item.and_then(|value| {
269            protojson_conv::to_value(&value).map_err(|e| {
270                A2AError::internal(format!("failed to serialize ProtoJSON stream payload: {e}"))
271            })
272        })
273    }))
274}
275
276fn parse_create_push_config_request(
277    raw_params: Value,
278) -> Result<TaskPushNotificationConfig, String> {
279    protojson_conv::from_value::<TaskPushNotificationConfig>(raw_params).map_err(|e| e.to_string())
280}
281
282/// A2A §9.5: `params` that fail schema validation are
283/// `InvalidParamsError` (-32602). `PARSE_ERROR` (-32700) is reserved for a
284/// body that is not valid JSON, which `malformed_request_response` handles
285/// before this point.
286fn invalid_params(e: impl std::fmt::Display) -> A2AError {
287    A2AError::invalid_params(format!("invalid params: {e}"))
288}
289
290/// The major version this build implements (§3.6.2). Minor is accepted
291/// unconditionally since nothing here branches on it.
292const SUPPORTED_MAJOR_VERSION: u32 = 1;
293
294/// Mirrors a2acli's own `parse_protocol_version`: tolerant of a bare major,
295/// extra trailing components, and surrounding whitespace.
296fn parse_major_version(value: &str) -> Option<u32> {
297    value.trim().split('.').next()?.parse().ok()
298}
299
300/// §3.6.2: reject a request whose `A2A-Version` header names an
301/// unsupported major version. §3.6.1/§3.6.2 both require an absent header
302/// to be interpreted as "0.3", not as "whatever this interface happens to
303/// serve" -- so it's rejected here exactly like any other unsupported
304/// version, rather than silently processed as this build's version.
305fn check_a2a_version(params: &ServiceParams) -> Result<(), A2AError> {
306    let requested = params
307        .get("a2a-version")
308        .and_then(|v| v.first())
309        .map(String::as_str)
310        .unwrap_or("0.3");
311    match parse_major_version(requested) {
312        Some(SUPPORTED_MAJOR_VERSION) => Ok(()),
313        _ => Err(A2AError::version_not_supported(requested)),
314    }
315}
316
317#[cfg(test)]
318mod tests {
319    use super::*;
320    use crate::InMemoryPushConfigStore;
321    use crate::executor::ExecutorContext;
322    use crate::handler::DefaultRequestHandler;
323    use crate::task_store::InMemoryTaskStore;
324    use a2a_pb::protojson_conv;
325    use axum::body::Body;
326    use axum::http::Request;
327
328    use crate::test_util::{CapturingHandler, assert_header_captured, install_crypto_provider};
329    use futures::stream::BoxStream;
330    use http_body_util::BodyExt;
331    use tower::ServiceExt;
332
333    struct EchoExecutor;
334
335    impl crate::AgentExecutor for EchoExecutor {
336        fn execute(
337            &self,
338            ctx: ExecutorContext,
339        ) -> BoxStream<'static, Result<StreamResponse, A2AError>> {
340            let task = Task {
341                id: ctx.task_id.clone(),
342                context_id: ctx.context_id.clone(),
343                status: TaskStatus {
344                    state: TaskState::Completed,
345                    message: ctx.message,
346                    timestamp: None,
347                },
348                artifacts: None,
349                history: None,
350                metadata: None,
351            };
352            Box::pin(futures::stream::once(async move {
353                Ok(StreamResponse::Task(task))
354            }))
355        }
356
357        fn cancel(
358            &self,
359            ctx: ExecutorContext,
360        ) -> BoxStream<'static, Result<StreamResponse, A2AError>> {
361            let task = Task {
362                id: ctx.task_id.clone(),
363                context_id: ctx.context_id.clone(),
364                status: TaskStatus {
365                    state: TaskState::Canceled,
366                    message: None,
367                    timestamp: None,
368                },
369                artifacts: None,
370                history: None,
371                metadata: None,
372            };
373            Box::pin(futures::stream::once(async move {
374                Ok(StreamResponse::Task(task))
375            }))
376        }
377    }
378
379    fn make_app() -> axum::Router {
380        let handler = Arc::new(DefaultRequestHandler::new(
381            EchoExecutor,
382            InMemoryTaskStore::new(),
383        ));
384        jsonrpc_router(handler)
385    }
386
387    fn make_push_app() -> axum::Router {
388        install_crypto_provider();
389        let handler = Arc::new(
390            DefaultRequestHandler::new(EchoExecutor, InMemoryTaskStore::new())
391                .with_push_config_store(InMemoryPushConfigStore::new()),
392        );
393        jsonrpc_router(handler)
394    }
395
396    async fn post_jsonrpc(app: axum::Router, method: &str, params: Value) -> JsonRpcResponse {
397        let rpc = JsonRpcRequest::new(JsonRpcId::Number(1), method, Some(params));
398        let body = serde_json::to_string(&rpc).unwrap();
399        let req = Request::builder()
400            .uri("/")
401            .method("POST")
402            .header("content-type", "application/json")
403            // A supported version, so tests using this helper for something
404            // other than version negotiation itself aren't confounded by
405            // §3.6.2's now-enforced absent-header rejection.
406            .header("a2a-version", "1.0")
407            .body(Body::from(body))
408            .unwrap();
409        let resp = app.oneshot(req).await.unwrap();
410        let body = resp.into_body().collect().await.unwrap().to_bytes();
411        serde_json::from_slice(&body).unwrap()
412    }
413
414    /// An unbounded body is read into memory; the limit rejects it at the
415    /// framework boundary instead, before any handler runs.
416    #[tokio::test]
417    async fn test_request_body_over_the_limit_is_rejected() {
418        let app = make_app();
419        let oversized = Body::from("x".repeat(MAX_REQUEST_BODY_BYTES + 1));
420        let req = Request::builder()
421            .uri("/")
422            .method("POST")
423            .header("content-type", "application/json")
424            .body(oversized)
425            .unwrap();
426
427        let resp = app.oneshot(req).await.unwrap();
428        assert_eq!(resp.status(), axum::http::StatusCode::PAYLOAD_TOO_LARGE);
429    }
430
431    #[tokio::test]
432    async fn test_send_message() {
433        let app = make_app();
434        let params = serde_json::json!({
435            "message": {
436                "messageId": "m1",
437                "role": "ROLE_USER",
438                "parts": [{"text": "hello"}]
439            }
440        });
441        let resp = post_jsonrpc(app, methods::SEND_MESSAGE, params).await;
442        assert!(resp.error.is_none(), "unexpected error: {:?}", resp.error);
443        let result =
444            protojson_conv::from_value::<SendMessageResponse>(resp.result.unwrap()).unwrap();
445        assert!(matches!(result, SendMessageResponse::Task(_)));
446    }
447
448    #[tokio::test]
449    async fn test_send_message_legacy_method_rejected() {
450        let app = make_app();
451        let params = serde_json::json!({
452            "message": {
453                "messageId": "m1",
454                "role": "ROLE_USER",
455                "parts": [{"text": "hello"}]
456            }
457        });
458        let resp = post_jsonrpc(app, "message.send", params).await;
459        assert!(resp.error.is_some(), "unexpected result: {:?}", resp.result);
460        assert_eq!(resp.error.unwrap().code, error_code::METHOD_NOT_FOUND);
461    }
462
463    #[tokio::test]
464    async fn test_get_task_not_found() {
465        let app = make_app();
466        let params = serde_json::json!({"id": "nonexistent"});
467        let resp = post_jsonrpc(app, methods::GET_TASK, params).await;
468        assert!(resp.error.is_some());
469        assert_eq!(resp.error.unwrap().code, error_code::TASK_NOT_FOUND);
470    }
471
472    #[tokio::test]
473    async fn test_invalid_method() {
474        let app = make_app();
475        let resp = post_jsonrpc(app, "unknown.method", Value::Null).await;
476        assert!(resp.error.is_some());
477        assert_eq!(resp.error.unwrap().code, error_code::METHOD_NOT_FOUND);
478    }
479
480    #[tokio::test]
481    async fn test_error_response_boundary_does_not_rewrite_internal_errors() {
482        // The boundary no longer sanitizes: an INTERNAL_ERROR keeps its
483        // message, since the boundary cannot tell an executor's own failure
484        // report from a server fault. Sanitization happens at the raise site
485        // via `sanitized_internal_error`.
486        let resp = error_response(JsonRpcId::Number(1), A2AError::internal("boom details"));
487        let body = resp.into_body().collect().await.unwrap().to_bytes();
488        let rpc_resp: JsonRpcResponse = serde_json::from_slice(&body).unwrap();
489        let error = rpc_resp.error.unwrap();
490        assert_eq!(error.code, error_code::INTERNAL_ERROR);
491        assert_eq!(error.message, "boom details");
492
493        // Non-internal errors keep their message (client-validation feedback).
494        let resp = error_response(JsonRpcId::Number(1), A2AError::task_not_found("t1"));
495        let body = resp.into_body().collect().await.unwrap().to_bytes();
496        let rpc_resp: JsonRpcResponse = serde_json::from_slice(&body).unwrap();
497        assert_eq!(rpc_resp.error.unwrap().message, "task not found: t1");
498    }
499
500    #[tokio::test]
501    async fn test_empty_method() {
502        let app = make_app();
503        let resp = post_jsonrpc(app, "", Value::Null).await;
504        assert!(resp.error.is_some());
505    }
506
507    #[tokio::test]
508    async fn test_invalid_jsonrpc_version() {
509        let app = make_app();
510        let rpc = serde_json::json!({
511            "jsonrpc": "1.0",
512            "id": 1,
513            "method": methods::SEND_MESSAGE,
514            "params": {}
515        });
516        let body = serde_json::to_string(&rpc).unwrap();
517        let req = Request::builder()
518            .uri("/")
519            .method("POST")
520            .header("content-type", "application/json")
521            .body(Body::from(body))
522            .unwrap();
523        let resp = app.oneshot(req).await.unwrap();
524        let body = resp.into_body().collect().await.unwrap().to_bytes();
525        let rpc_resp: JsonRpcResponse = serde_json::from_slice(&body).unwrap();
526        assert!(rpc_resp.error.is_some());
527    }
528
529    /// A `Json<T>` extractor would have axum reject this with a 422 before
530    /// the handler runs; JSON-RPC over HTTP always answers 200 with the
531    /// error inside the envelope, regardless of what about the body failed.
532    #[tokio::test]
533    async fn test_missing_method_field_is_invalid_request_not_a_422() {
534        let app = make_app();
535        let req = Request::builder()
536            .uri("/")
537            .method("POST")
538            .header("content-type", "application/json")
539            .body(Body::from(r#"{"jsonrpc":"2.0","id":1,"params":{}}"#))
540            .unwrap();
541        let resp = app.oneshot(req).await.unwrap();
542        assert_eq!(resp.status(), StatusCode::OK);
543        let body = resp.into_body().collect().await.unwrap().to_bytes();
544        let rpc_resp: JsonRpcResponse = serde_json::from_slice(&body).unwrap();
545        let code = rpc_resp.error.expect("expected an error").code;
546        assert!(
547            code == error_code::INVALID_REQUEST || code == error_code::INVALID_PARAMS,
548            "got {code}"
549        );
550        assert_eq!(rpc_resp.id, JsonRpcId::Number(1), "id should be recovered");
551    }
552
553    #[tokio::test]
554    async fn test_invalid_json_syntax_is_parse_error_not_a_400() {
555        let app = make_app();
556        let req = Request::builder()
557            .uri("/")
558            .method("POST")
559            .header("content-type", "application/json")
560            .body(Body::from("{this is not valid json"))
561            .unwrap();
562        let resp = app.oneshot(req).await.unwrap();
563        assert_eq!(resp.status(), StatusCode::OK);
564        let body = resp.into_body().collect().await.unwrap().to_bytes();
565        let rpc_resp: JsonRpcResponse = serde_json::from_slice(&body).unwrap();
566        assert_eq!(
567            rpc_resp.error.expect("expected an error").code,
568            error_code::PARSE_ERROR
569        );
570        assert_eq!(rpc_resp.id, JsonRpcId::Null, "id cannot be recovered");
571    }
572
573    /// A2A §3.3.2 / §9.5: parameters that fail to decode are
574    /// `InvalidParamsError` (-32602). `JSONParseError` (-32700) is reserved
575    /// for a body that is not valid JSON at all, which
576    /// `test_invalid_json_syntax_is_parse_error_not_a_400` covers.
577    #[tokio::test]
578    async fn test_invalid_params() {
579        let app = make_app();
580        // Type mismatch: `message` must be an object. A merely-unknown field
581        // is not invalid params; it is ignored per spec §5.7.
582        let params = serde_json::json!({"message": "not-an-object"});
583        let resp = post_jsonrpc(app, methods::SEND_MESSAGE, params).await;
584        let error = resp.error.expect("expected an error");
585        assert_eq!(error.code, error_code::INVALID_PARAMS);
586    }
587
588    /// A2A §9.5: a `params` payload that fails to decode is
589    /// `InvalidParamsError` (-32602) for *every* method, not just
590    /// `message/send`. Each arm decodes its own request type, so guard the
591    /// whole table rather than a single representative method.
592    #[tokio::test]
593    async fn test_invalid_params_is_reported_for_every_method() {
594        // Each payload is a well-formed JSON object whose one recognized
595        // field has the wrong type, so decoding fails before the handler.
596        let cases = [
597            (methods::GET_TASK, serde_json::json!({ "id": 123 })),
598            (
599                methods::LIST_TASKS,
600                serde_json::json!({ "pageSize": "not-a-number" }),
601            ),
602            (methods::CANCEL_TASK, serde_json::json!({ "id": 123 })),
603            (
604                methods::CREATE_PUSH_CONFIG,
605                serde_json::json!({ "taskId": 123 }),
606            ),
607            (
608                methods::GET_PUSH_CONFIG,
609                serde_json::json!({ "taskId": 123 }),
610            ),
611            (
612                methods::LIST_PUSH_CONFIGS,
613                serde_json::json!({ "taskId": 123 }),
614            ),
615            (
616                methods::DELETE_PUSH_CONFIG,
617                serde_json::json!({ "taskId": 123 }),
618            ),
619            (
620                methods::GET_EXTENDED_AGENT_CARD,
621                serde_json::json!({ "tenant": 123 }),
622            ),
623        ];
624        for (method, params) in cases {
625            let resp = post_jsonrpc(make_app(), method, params).await;
626            let error = resp
627                .error
628                .unwrap_or_else(|| panic!("{method}: expected an error, got {:?}", resp.result));
629            assert_eq!(error.code, error_code::INVALID_PARAMS, "method {method}");
630        }
631    }
632
633    /// Spec §5.7: unrecognized fields in request params are ignored for
634    /// forward compatibility instead of being rejected.
635    #[tokio::test]
636    async fn test_unknown_params_fields_are_ignored() {
637        let app = make_app();
638        let params = serde_json::json!({
639            "message": {
640                "messageId": "m1",
641                "role": "ROLE_USER",
642                "parts": [{"text": "hi"}],
643                "futureField": true
644            },
645            "anotherFutureField": {"nested": 1}
646        });
647        let resp = post_jsonrpc(app, methods::SEND_MESSAGE, params).await;
648        assert!(
649            resp.error.is_none(),
650            "unknown fields must be ignored: {:?}",
651            resp.error
652        );
653        assert!(resp.result.is_some());
654    }
655
656    #[tokio::test]
657    async fn test_list_tasks() {
658        let app = make_app();
659        let params = serde_json::json!({});
660        let resp = post_jsonrpc(app, methods::LIST_TASKS, params).await;
661        let result = protojson_conv::from_value::<ListTasksResponse>(resp.result.unwrap()).unwrap();
662        assert!(result.tasks.is_empty());
663    }
664
665    #[tokio::test]
666    async fn test_cancel_task_not_found() {
667        let app = make_app();
668        let params = serde_json::json!({"id": "nonexistent"});
669        let resp = post_jsonrpc(app, methods::CANCEL_TASK, params).await;
670        assert!(resp.error.is_some());
671    }
672
673    #[tokio::test]
674    async fn test_create_push_config() {
675        let app = make_push_app();
676        let params = serde_json::json!({
677            "taskId": "t1",
678            "id": "cfg1",
679            "url": "http://example.com/callback"
680        });
681        let resp = post_jsonrpc(app, methods::CREATE_PUSH_CONFIG, params).await;
682        assert!(resp.error.is_none(), "unexpected error: {:?}", resp.error);
683        let result =
684            serde_json::from_value::<TaskPushNotificationConfig>(resp.result.unwrap()).unwrap();
685        assert_eq!(result.task_id, "t1");
686        assert_eq!(result.id.as_deref(), Some("cfg1"));
687        assert_eq!(result.url, "http://example.com/callback");
688    }
689
690    #[tokio::test]
691    async fn test_create_push_config_accepts_flat_config_object() {
692        let app = make_push_app();
693        let params = serde_json::json!({
694            "taskId": "t1",
695            "id": "cfg1",
696            "url": "http://example.com/callback"
697        });
698        let resp = post_jsonrpc(app, methods::CREATE_PUSH_CONFIG, params).await;
699        assert!(resp.error.is_none(), "unexpected error: {:?}", resp.error);
700        let result =
701            serde_json::from_value::<TaskPushNotificationConfig>(resp.result.unwrap()).unwrap();
702        assert_eq!(result.task_id, "t1");
703        assert_eq!(result.id.as_deref(), Some("cfg1"));
704        assert_eq!(result.url, "http://example.com/callback");
705    }
706
707    #[tokio::test]
708    async fn test_get_push_config() {
709        let app = make_app();
710        let params = serde_json::json!({
711            "taskId": "t1",
712            "id": "cfg1"
713        });
714        let resp = post_jsonrpc(app, methods::GET_PUSH_CONFIG, params).await;
715        assert!(resp.error.is_some() || resp.result.is_some());
716    }
717
718    #[tokio::test]
719    async fn test_list_push_configs() {
720        let app = make_push_app();
721        let params = serde_json::json!({
722            "taskId": "t1"
723        });
724        let resp = post_jsonrpc(app, methods::LIST_PUSH_CONFIGS, params).await;
725        assert!(resp.error.is_none(), "unexpected error: {:?}", resp.error);
726        let result =
727            serde_json::from_value::<ListTaskPushNotificationConfigsResponse>(resp.result.unwrap())
728                .unwrap();
729        assert!(result.configs.is_empty());
730    }
731
732    #[tokio::test]
733    async fn test_delete_push_config() {
734        let app = make_app();
735        let params = serde_json::json!({
736            "taskId": "t1",
737            "id": "cfg1"
738        });
739        let resp = post_jsonrpc(app, methods::DELETE_PUSH_CONFIG, params).await;
740        assert!(resp.error.is_some() || resp.result.is_some());
741    }
742
743    #[tokio::test]
744    async fn test_get_extended_agent_card() {
745        let app = make_app();
746        let params = serde_json::json!({});
747        let resp = post_jsonrpc(app, methods::GET_EXTENDED_AGENT_CARD, params).await;
748        // DefaultRequestHandler returns NotSupported
749        assert!(resp.error.is_some() || resp.result.is_some());
750    }
751
752    #[tokio::test]
753    async fn test_streaming_send_message() {
754        let app = make_app();
755        let body = serde_json::json!({
756            "message": {
757                "messageId": "m1",
758                "role": "ROLE_USER",
759                "parts": [{"text": "hello"}]
760            }
761        });
762        let rpc = JsonRpcRequest::new(
763            JsonRpcId::Number(1),
764            methods::SEND_STREAMING_MESSAGE,
765            Some(body),
766        );
767        let req = Request::builder()
768            .uri("/")
769            .method("POST")
770            .header("content-type", "application/json")
771            .header("accept", "text/event-stream")
772            .body(Body::from(serde_json::to_string(&rpc).unwrap()))
773            .unwrap();
774        let resp = app.oneshot(req).await.unwrap();
775        // SSE response => 200
776        assert_eq!(resp.status(), StatusCode::OK);
777    }
778
779    #[tokio::test]
780    async fn test_streaming_subscribe_to_task() {
781        let app = make_app();
782        let rpc = JsonRpcRequest::new(
783            JsonRpcId::Number(1),
784            methods::SUBSCRIBE_TO_TASK,
785            Some(serde_json::json!({"id": "t1"})),
786        );
787        let req = Request::builder()
788            .uri("/")
789            .method("POST")
790            .header("content-type", "application/json")
791            .body(Body::from(serde_json::to_string(&rpc).unwrap()))
792            .unwrap();
793        let resp = app.oneshot(req).await.unwrap();
794        // Subscription may fail (task not found), but routing should work
795        assert_eq!(resp.status(), StatusCode::OK);
796    }
797
798    /// A params-decode failure on the streaming binding must carry the same
799    /// code as the unary binding: `InvalidParamsError` (-32602), for both
800    /// streaming methods.
801    #[tokio::test]
802    async fn test_streaming_invalid_params_use_invalid_params_error_code() {
803        let cases = [
804            (
805                methods::SEND_STREAMING_MESSAGE,
806                serde_json::json!({"message": "not-an-object"}),
807            ),
808            (methods::SUBSCRIBE_TO_TASK, serde_json::json!({"id": 123})),
809        ];
810        for (method, params) in cases {
811            let app = make_app();
812            let rpc = JsonRpcRequest::new(JsonRpcId::Number(1), method, Some(params));
813            let req = Request::builder()
814                .uri("/")
815                .method("POST")
816                .header("content-type", "application/json")
817                // A supported version, so this test exercises params decoding
818                // rather than §3.6.2's absent-header rejection.
819                .header("a2a-version", "1.0")
820                .body(Body::from(serde_json::to_string(&rpc).unwrap()))
821                .unwrap();
822            let resp = app.oneshot(req).await.unwrap();
823            assert_eq!(resp.status(), StatusCode::OK);
824            let body = resp.into_body().collect().await.unwrap().to_bytes();
825            let parsed: JsonRpcResponse = serde_json::from_slice(&body).unwrap();
826            let error = parsed.error.expect("expected an error");
827            assert_eq!(error.code, error_code::INVALID_PARAMS, "method {method}");
828        }
829    }
830
831    #[tokio::test]
832    async fn test_streaming_send_message_legacy_method_rejected() {
833        let app = make_app();
834        let body = serde_json::json!({
835            "message": {
836                "messageId": "m1",
837                "role": "ROLE_USER",
838                "parts": [{"text": "hello"}]
839            }
840        });
841        let rpc = JsonRpcRequest::new(JsonRpcId::Number(1), "message.stream", Some(body));
842        let req = Request::builder()
843            .uri("/")
844            .method("POST")
845            .header("content-type", "application/json")
846            .header("accept", "text/event-stream")
847            .header("a2a-version", "1.0")
848            .body(Body::from(serde_json::to_string(&rpc).unwrap()))
849            .unwrap();
850        let resp = app.oneshot(req).await.unwrap();
851        let body = resp.into_body().collect().await.unwrap().to_bytes();
852        let rpc_resp: JsonRpcResponse = serde_json::from_slice(&body).unwrap();
853        assert!(
854            rpc_resp.error.is_some(),
855            "unexpected result: {:?}",
856            rpc_resp.result
857        );
858        assert_eq!(rpc_resp.error.unwrap().code, error_code::METHOD_NOT_FOUND);
859    }
860
861    #[tokio::test]
862    async fn test_send_message_propagates_headers_as_service_params() {
863        let (handler, captured) = CapturingHandler::new();
864        let app = jsonrpc_router(handler);
865
866        let rpc = JsonRpcRequest::new(
867            JsonRpcId::Number(1),
868            methods::SEND_MESSAGE,
869            Some(serde_json::json!({
870                "message": {
871                    "messageId": "m1",
872                    "role": "ROLE_USER",
873                    "parts": [{"text": "hi"}]
874                }
875            })),
876        );
877        let req = Request::builder()
878            .uri("/")
879            .method("POST")
880            .header("content-type", "application/json")
881            .header("authorization", "Bearer jsonrpc-token")
882            .header("x-tenant-id", "acme")
883            .header("a2a-version", "1.0")
884            .body(Body::from(serde_json::to_string(&rpc).unwrap()))
885            .unwrap();
886        let resp = app.oneshot(req).await.unwrap();
887        assert_eq!(resp.status(), StatusCode::OK);
888
889        assert_header_captured(
890            &captured,
891            "send_message",
892            "authorization",
893            "Bearer jsonrpc-token",
894        );
895        assert_header_captured(&captured, "send_message", "x-tenant-id", "acme");
896    }
897
898    #[tokio::test]
899    async fn test_unsupported_a2a_version_is_rejected() {
900        let app = make_app();
901        let rpc = JsonRpcRequest::new(
902            JsonRpcId::Number(1),
903            methods::SEND_MESSAGE,
904            Some(serde_json::json!({
905                "message": {"messageId": "m1", "role": "ROLE_USER", "parts": [{"text": "hi"}]}
906            })),
907        );
908        let req = Request::builder()
909            .uri("/")
910            .method("POST")
911            .header("content-type", "application/json")
912            .header("a2a-version", "99.0")
913            .body(Body::from(serde_json::to_string(&rpc).unwrap()))
914            .unwrap();
915        let resp = app.oneshot(req).await.unwrap();
916        assert_eq!(resp.status(), StatusCode::OK);
917        let body = resp.into_body().collect().await.unwrap().to_bytes();
918        let parsed: JsonRpcResponse = serde_json::from_slice(&body).unwrap();
919        assert_eq!(
920            parsed.error.expect("expected an error").code,
921            error_code::VERSION_NOT_SUPPORTED
922        );
923    }
924
925    #[tokio::test]
926    async fn test_missing_a2a_version_is_rejected() {
927        // §3.6.1/§3.6.2: an absent A2A-Version header MUST be interpreted
928        // as "0.3", not as this interface's own version -- so this server,
929        // which only speaks 1.0, rejects it the same as any other
930        // unsupported version. No a2a-version header at all, unlike
931        // `post_jsonrpc`, which sets one so other tests aren't confounded
932        // by this same check.
933        let app = make_app();
934        let rpc = JsonRpcRequest::new(
935            JsonRpcId::Number(1),
936            methods::SEND_MESSAGE,
937            Some(serde_json::json!({
938                "message": {"messageId": "m1", "role": "ROLE_USER", "parts": [{"text": "hi"}]}
939            })),
940        );
941        let req = Request::builder()
942            .uri("/")
943            .method("POST")
944            .header("content-type", "application/json")
945            .body(Body::from(serde_json::to_string(&rpc).unwrap()))
946            .unwrap();
947        let resp = app.oneshot(req).await.unwrap();
948        assert_eq!(resp.status(), StatusCode::OK);
949        let body = resp.into_body().collect().await.unwrap().to_bytes();
950        let parsed: JsonRpcResponse = serde_json::from_slice(&body).unwrap();
951        assert_eq!(
952            parsed.error.expect("expected an error").code,
953            error_code::VERSION_NOT_SUPPORTED
954        );
955    }
956
957    #[tokio::test]
958    async fn test_supported_a2a_version_is_accepted() {
959        let app = make_app();
960        let rpc = JsonRpcRequest::new(
961            JsonRpcId::Number(1),
962            methods::SEND_MESSAGE,
963            Some(serde_json::json!({
964                "message": {"messageId": "m1", "role": "ROLE_USER", "parts": [{"text": "hi"}]}
965            })),
966        );
967        let req = Request::builder()
968            .uri("/")
969            .method("POST")
970            .header("content-type", "application/json")
971            .header("a2a-version", "1.0")
972            .body(Body::from(serde_json::to_string(&rpc).unwrap()))
973            .unwrap();
974        let resp = app.oneshot(req).await.unwrap();
975        assert_eq!(resp.status(), StatusCode::OK);
976        let body = resp.into_body().collect().await.unwrap().to_bytes();
977        let parsed: JsonRpcResponse = serde_json::from_slice(&body).unwrap();
978        assert!(
979            parsed.error.is_none(),
980            "expected success, got {:?}",
981            parsed.error
982        );
983    }
984
985    #[tokio::test]
986    async fn test_streaming_send_message_propagates_headers_as_service_params() {
987        let (handler, captured) = CapturingHandler::new();
988        let app = jsonrpc_router(handler);
989
990        let rpc = JsonRpcRequest::new(
991            JsonRpcId::Number(1),
992            methods::SEND_STREAMING_MESSAGE,
993            Some(serde_json::json!({
994                "message": {
995                    "messageId": "m1",
996                    "role": "ROLE_USER",
997                    "parts": [{"text": "hi"}]
998                }
999            })),
1000        );
1001        let req = Request::builder()
1002            .uri("/")
1003            .method("POST")
1004            .header("content-type", "application/json")
1005            .header("accept", "text/event-stream")
1006            .header("authorization", "Bearer jsonrpc-streaming-token")
1007            .header("x-tenant-id", "acme")
1008            .header("a2a-version", "1.0")
1009            .body(Body::from(serde_json::to_string(&rpc).unwrap()))
1010            .unwrap();
1011        let resp = app.oneshot(req).await.unwrap();
1012        assert_eq!(resp.status(), StatusCode::OK);
1013        // Drain the SSE body so the handler is driven to completion.
1014        let _ = resp.into_body().collect().await.unwrap().to_bytes();
1015
1016        assert_header_captured(
1017            &captured,
1018            "send_streaming_message",
1019            "authorization",
1020            "Bearer jsonrpc-streaming-token",
1021        );
1022        assert_header_captured(&captured, "send_streaming_message", "x-tenant-id", "acme");
1023    }
1024
1025    #[tokio::test]
1026    async fn test_jsonrpc_unary_methods_propagate_headers_as_service_params() {
1027        // Exercise every unary JSON-RPC method to lock in header propagation
1028        // across the full match arm in handle_unary_request.
1029        let cases: &[(&'static str, &'static str, serde_json::Value)] = &[
1030            (
1031                methods::GET_TASK,
1032                "get_task",
1033                serde_json::json!({"id": "t1"}),
1034            ),
1035            (methods::LIST_TASKS, "list_tasks", serde_json::json!({})),
1036            (
1037                methods::CANCEL_TASK,
1038                "cancel_task",
1039                serde_json::json!({"id": "t1"}),
1040            ),
1041            (
1042                methods::CREATE_PUSH_CONFIG,
1043                "create_push_config",
1044                serde_json::json!({"taskId": "t1", "url": "http://example.com/cb"}),
1045            ),
1046            (
1047                methods::GET_PUSH_CONFIG,
1048                "get_push_config",
1049                serde_json::json!({"taskId": "t1", "id": "cfg1"}),
1050            ),
1051            (
1052                methods::LIST_PUSH_CONFIGS,
1053                "list_push_configs",
1054                serde_json::json!({"taskId": "t1"}),
1055            ),
1056            (
1057                methods::DELETE_PUSH_CONFIG,
1058                "delete_push_config",
1059                serde_json::json!({"taskId": "t1", "id": "cfg1"}),
1060            ),
1061            (
1062                methods::GET_EXTENDED_AGENT_CARD,
1063                "get_extended_agent_card",
1064                serde_json::json!({}),
1065            ),
1066        ];
1067
1068        for (rpc_method, handler_method, params) in cases {
1069            let (handler, captured) = CapturingHandler::new();
1070            let app = jsonrpc_router(handler);
1071            let rpc = JsonRpcRequest::new(JsonRpcId::Number(1), *rpc_method, Some(params.clone()));
1072            let token = format!("Bearer token-for-{handler_method}");
1073            let req = Request::builder()
1074                .uri("/")
1075                .method("POST")
1076                .header("content-type", "application/json")
1077                .header("authorization", &token)
1078                .header("a2a-version", "1.0")
1079                .body(Body::from(serde_json::to_string(&rpc).unwrap()))
1080                .unwrap();
1081            let resp = app.oneshot(req).await.unwrap();
1082            assert_eq!(resp.status(), StatusCode::OK);
1083            assert_header_captured(&captured, handler_method, "authorization", &token);
1084        }
1085    }
1086
1087    #[tokio::test]
1088    async fn test_jsonrpc_subscribe_to_task_propagates_headers_as_service_params() {
1089        let (handler, captured) = CapturingHandler::new();
1090        let app = jsonrpc_router(handler);
1091        let rpc = JsonRpcRequest::new(
1092            JsonRpcId::Number(1),
1093            methods::SUBSCRIBE_TO_TASK,
1094            Some(serde_json::json!({"id": "t1"})),
1095        );
1096        let req = Request::builder()
1097            .uri("/")
1098            .method("POST")
1099            .header("content-type", "application/json")
1100            .header("authorization", "Bearer subscribe-token")
1101            .header("a2a-version", "1.0")
1102            .body(Body::from(serde_json::to_string(&rpc).unwrap()))
1103            .unwrap();
1104        let resp = app.oneshot(req).await.unwrap();
1105        assert_eq!(resp.status(), StatusCode::OK);
1106        let _ = resp.into_body().collect().await.unwrap().to_bytes();
1107        assert_header_captured(
1108            &captured,
1109            "subscribe_to_task",
1110            "authorization",
1111            "Bearer subscribe-token",
1112        );
1113    }
1114
1115    #[tokio::test]
1116    async fn test_streaming_invalid_method() {
1117        let app = make_app();
1118        let rpc = serde_json::json!({
1119            "jsonrpc": "2.0",
1120            "id": 1,
1121            "method": "tasks.subscribe",
1122            "params": {"bogus": true}
1123        });
1124        let req = Request::builder()
1125            .uri("/")
1126            .method("POST")
1127            .header("content-type", "application/json")
1128            .body(Body::from(serde_json::to_string(&rpc).unwrap()))
1129            .unwrap();
1130        let resp = app.oneshot(req).await.unwrap();
1131        assert_eq!(resp.status(), StatusCode::OK);
1132    }
1133}
a2a_server/jsonrpc.rs

a2a_server/
jsonrpc.rs
