a2a_server/
jsonrpc.rs

1// Copyright AGNTCY Contributors (https://github.com/agntcy)
2// SPDX-License-Identifier: Apache-2.0
3use std::sync::Arc;
4
5use a2a::*;
6use a2a_pb::protojson_conv::{self, ProtoJsonPayload};
7use axum::{
8    Json,
9    extract::State,
10    http::{HeaderMap, StatusCode},
11    response::IntoResponse,
12};
13use futures::{StreamExt, stream::BoxStream};
14use serde_json::Value;
15
16use crate::handler::RequestHandler;
17use crate::middleware::{ServiceParams, extract_service_params};
18use crate::sse;
19
20/// Shared state for the JSON-RPC handler.
21pub struct JsonRpcState<H: RequestHandler> {
22    pub handler: Arc<H>,
23}
24
25impl<H: RequestHandler> Clone for JsonRpcState<H> {
26    fn clone(&self) -> Self {
27        JsonRpcState {
28            handler: self.handler.clone(),
29        }
30    }
31}
32
33/// Create an axum router for the JSON-RPC protocol binding.
34///
35/// All requests are dispatched to a single POST endpoint that routes
36/// by the `method` field in the JSON-RPC envelope.
37pub fn jsonrpc_router<H: RequestHandler>(handler: Arc<H>) -> axum::Router {
38    let state = JsonRpcState { handler };
39    axum::Router::new()
40        .route("/", axum::routing::post(handle_jsonrpc::<H>))
41        .with_state(state)
42}
43
44async fn handle_jsonrpc<H: RequestHandler>(
45    State(state): State<JsonRpcState<H>>,
46    headers: HeaderMap,
47    Json(request): Json<JsonRpcRequest>,
48) -> impl IntoResponse {
49    let params = extract_service_params(&headers);
50    let id = request.id.clone();
51    let method = request.method.as_str();
52
53    if request.jsonrpc != "2.0" {
54        return error_response(id, A2AError::invalid_request("invalid jsonrpc version"));
55    }
56
57    if methods::is_streaming(method) {
58        return handle_streaming_request(&state, &params, &request).await;
59    }
60
61    handle_unary_request(&state, &params, &request).await
62}
63
64async fn handle_unary_request<H: RequestHandler>(
65    state: &JsonRpcState<H>,
66    params: &ServiceParams,
67    request: &JsonRpcRequest,
68) -> axum::response::Response {
69    let id = request.id.clone();
70    let raw_params = request.params.clone().unwrap_or(Value::Null);
71
72    let result: Result<Value, A2AError> = match request.method.as_str() {
73        methods::SEND_MESSAGE => match protojson_conv::from_value::<SendMessageRequest>(raw_params)
74        {
75            Ok(req) => state
76                .handler
77                .send_message(params, req)
78                .await
79                .and_then(|r| protojson_value(&r)),
80            Err(e) => Err(parse_error(e)),
81        },
82        methods::GET_TASK => match protojson_conv::from_value::<GetTaskRequest>(raw_params) {
83            Ok(req) => state
84                .handler
85                .get_task(params, req)
86                .await
87                .and_then(|r| protojson_value(&r)),
88            Err(e) => Err(parse_error(e)),
89        },
90        methods::LIST_TASKS => match protojson_conv::from_value::<ListTasksRequest>(raw_params) {
91            Ok(req) => state
92                .handler
93                .list_tasks(params, req)
94                .await
95                .and_then(|r| protojson_value(&r)),
96            Err(e) => Err(parse_error(e)),
97        },
98        methods::CANCEL_TASK => match protojson_conv::from_value::<CancelTaskRequest>(raw_params) {
99            Ok(req) => state
100                .handler
101                .cancel_task(params, req)
102                .await
103                .and_then(|r| protojson_value(&r)),
104            Err(e) => Err(parse_error(e)),
105        },
106        methods::CREATE_PUSH_CONFIG => match parse_create_push_config_request(raw_params) {
107            Ok(req) => state
108                .handler
109                .create_push_config(params, req)
110                .await
111                .and_then(|r| protojson_value(&r)),
112            Err(e) => Err(parse_error(e)),
113        },
114        methods::GET_PUSH_CONFIG => {
115            match protojson_conv::from_value::<GetTaskPushNotificationConfigRequest>(raw_params) {
116                Ok(req) => state
117                    .handler
118                    .get_push_config(params, req)
119                    .await
120                    .and_then(|r| protojson_value(&r)),
121                Err(e) => Err(parse_error(e)),
122            }
123        }
124        methods::LIST_PUSH_CONFIGS => {
125            match protojson_conv::from_value::<ListTaskPushNotificationConfigsRequest>(raw_params) {
126                Ok(req) => state
127                    .handler
128                    .list_push_configs(params, req)
129                    .await
130                    .and_then(|r| protojson_value(&r)),
131                Err(e) => Err(parse_error(e)),
132            }
133        }
134        methods::DELETE_PUSH_CONFIG => {
135            match protojson_conv::from_value::<DeleteTaskPushNotificationConfigRequest>(raw_params)
136            {
137                Ok(req) => state
138                    .handler
139                    .delete_push_config(params, req)
140                    .await
141                    .map(|_| Value::Null),
142                Err(e) => Err(parse_error(e)),
143            }
144        }
145        methods::GET_EXTENDED_AGENT_CARD => {
146            match protojson_conv::from_value::<GetExtendedAgentCardRequest>(raw_params) {
147                Ok(req) => state
148                    .handler
149                    .get_extended_agent_card(params, req)
150                    .await
151                    .and_then(|r| protojson_value(&r)),
152                Err(e) => Err(parse_error(e)),
153            }
154        }
155        "" => Err(A2AError::invalid_request("method is required")),
156        _ => Err(A2AError::method_not_found(&request.method)),
157    };
158
159    match result {
160        Ok(value) => {
161            let resp = JsonRpcResponse::success(id, value);
162            Json(resp).into_response()
163        }
164        Err(e) => error_response(id, e),
165    }
166}
167
168async fn handle_streaming_request<H: RequestHandler>(
169    state: &JsonRpcState<H>,
170    params: &ServiceParams,
171    request: &JsonRpcRequest,
172) -> axum::response::Response {
173    let id = request.id.clone();
174    let raw_params = request.params.clone().unwrap_or(Value::Null);
175
176    match request.method.as_str() {
177        methods::SEND_STREAMING_MESSAGE => {
178            match protojson_conv::from_value::<SendMessageRequest>(raw_params) {
179                Ok(req) => match state.handler.send_streaming_message(params, req).await {
180                    Ok(stream) => {
181                        sse::sse_jsonrpc_stream(id, protojson_stream(stream)).into_response()
182                    }
183                    Err(e) => error_response(id, e),
184                },
185                Err(e) => error_response(id, parse_error(e)),
186            }
187        }
188        methods::SUBSCRIBE_TO_TASK => {
189            match protojson_conv::from_value::<SubscribeToTaskRequest>(raw_params) {
190                Ok(req) => match state.handler.subscribe_to_task(params, req).await {
191                    Ok(stream) => {
192                        sse::sse_jsonrpc_stream(id, protojson_stream(stream)).into_response()
193                    }
194                    Err(e) => error_response(id, e),
195                },
196                Err(e) => error_response(id, parse_error(e)),
197            }
198        }
199        _ => error_response(id, A2AError::method_not_found(&request.method)),
200    }
201}
202
203fn error_response(id: JsonRpcId, err: A2AError) -> axum::response::Response {
204    let resp = JsonRpcResponse::error(id, err.to_jsonrpc_error());
205    (StatusCode::OK, Json(resp)).into_response()
206}
207
208fn protojson_value<T: ProtoJsonPayload>(value: &T) -> Result<Value, A2AError> {
209    protojson_conv::to_value(value)
210        .map_err(|e| A2AError::internal(format!("failed to serialize ProtoJSON payload: {e}")))
211}
212
213fn protojson_stream(
214    stream: BoxStream<'static, Result<StreamResponse, A2AError>>,
215) -> BoxStream<'static, Result<Value, A2AError>> {
216    Box::pin(stream.map(|item| {
217        item.and_then(|value| {
218            protojson_conv::to_value(&value).map_err(|e| {
219                A2AError::internal(format!("failed to serialize ProtoJSON stream payload: {e}"))
220            })
221        })
222    }))
223}
224
225fn parse_create_push_config_request(
226    raw_params: Value,
227) -> Result<TaskPushNotificationConfig, String> {
228    protojson_conv::from_value::<TaskPushNotificationConfig>(raw_params).map_err(|e| e.to_string())
229}
230
231fn parse_error(e: impl std::fmt::Display) -> A2AError {
232    A2AError {
233        code: error_code::PARSE_ERROR,
234        message: format!("invalid params: {e}"),
235        details: None,
236    }
237}
238
239#[cfg(test)]
240mod tests {
241    use super::*;
242    use crate::InMemoryPushConfigStore;
243    use crate::executor::ExecutorContext;
244    use crate::handler::DefaultRequestHandler;
245    use crate::task_store::InMemoryTaskStore;
246    use a2a_pb::protojson_conv;
247    use axum::body::Body;
248    use axum::http::Request;
249
250    use crate::test_util::{CapturingHandler, assert_header_captured, install_crypto_provider};
251    use futures::stream::BoxStream;
252    use http_body_util::BodyExt;
253    use tower::ServiceExt;
254
255    struct EchoExecutor;
256
257    impl crate::AgentExecutor for EchoExecutor {
258        fn execute(
259            &self,
260            ctx: ExecutorContext,
261        ) -> BoxStream<'static, Result<StreamResponse, A2AError>> {
262            let task = Task {
263                id: ctx.task_id.clone(),
264                context_id: ctx.context_id.clone(),
265                status: TaskStatus {
266                    state: TaskState::Completed,
267                    message: ctx.message,
268                    timestamp: None,
269                },
270                artifacts: None,
271                history: None,
272                metadata: None,
273            };
274            Box::pin(futures::stream::once(async move {
275                Ok(StreamResponse::Task(task))
276            }))
277        }
278
279        fn cancel(
280            &self,
281            ctx: ExecutorContext,
282        ) -> BoxStream<'static, Result<StreamResponse, A2AError>> {
283            let task = Task {
284                id: ctx.task_id.clone(),
285                context_id: ctx.context_id.clone(),
286                status: TaskStatus {
287                    state: TaskState::Canceled,
288                    message: None,
289                    timestamp: None,
290                },
291                artifacts: None,
292                history: None,
293                metadata: None,
294            };
295            Box::pin(futures::stream::once(async move {
296                Ok(StreamResponse::Task(task))
297            }))
298        }
299    }
300
301    fn make_app() -> axum::Router {
302        let handler = Arc::new(DefaultRequestHandler::new(
303            EchoExecutor,
304            InMemoryTaskStore::new(),
305        ));
306        jsonrpc_router(handler)
307    }
308
309    fn make_push_app() -> axum::Router {
310        install_crypto_provider();
311        let handler = Arc::new(
312            DefaultRequestHandler::new(EchoExecutor, InMemoryTaskStore::new())
313                .with_push_config_store(InMemoryPushConfigStore::new()),
314        );
315        jsonrpc_router(handler)
316    }
317
318    async fn post_jsonrpc(app: axum::Router, method: &str, params: Value) -> JsonRpcResponse {
319        let rpc = JsonRpcRequest::new(JsonRpcId::Number(1), method, Some(params));
320        let body = serde_json::to_string(&rpc).unwrap();
321        let req = Request::builder()
322            .uri("/")
323            .method("POST")
324            .header("content-type", "application/json")
325            .body(Body::from(body))
326            .unwrap();
327        let resp = app.oneshot(req).await.unwrap();
328        let body = resp.into_body().collect().await.unwrap().to_bytes();
329        serde_json::from_slice(&body).unwrap()
330    }
331
332    #[tokio::test]
333    async fn test_send_message() {
334        let app = make_app();
335        let params = serde_json::json!({
336            "message": {
337                "messageId": "m1",
338                "role": "ROLE_USER",
339                "parts": [{"text": "hello"}]
340            }
341        });
342        let resp = post_jsonrpc(app, methods::SEND_MESSAGE, params).await;
343        assert!(resp.error.is_none(), "unexpected error: {:?}", resp.error);
344        let result =
345            protojson_conv::from_value::<SendMessageResponse>(resp.result.unwrap()).unwrap();
346        assert!(matches!(result, SendMessageResponse::Task(_)));
347    }
348
349    #[tokio::test]
350    async fn test_send_message_legacy_method_rejected() {
351        let app = make_app();
352        let params = serde_json::json!({
353            "message": {
354                "messageId": "m1",
355                "role": "ROLE_USER",
356                "parts": [{"text": "hello"}]
357            }
358        });
359        let resp = post_jsonrpc(app, "message.send", params).await;
360        assert!(resp.error.is_some(), "unexpected result: {:?}", resp.result);
361        assert_eq!(resp.error.unwrap().code, error_code::METHOD_NOT_FOUND);
362    }
363
364    #[tokio::test]
365    async fn test_get_task_not_found() {
366        let app = make_app();
367        let params = serde_json::json!({"id": "nonexistent"});
368        let resp = post_jsonrpc(app, methods::GET_TASK, params).await;
369        assert!(resp.error.is_some());
370        assert_eq!(resp.error.unwrap().code, error_code::TASK_NOT_FOUND);
371    }
372
373    #[tokio::test]
374    async fn test_invalid_method() {
375        let app = make_app();
376        let resp = post_jsonrpc(app, "unknown.method", Value::Null).await;
377        assert!(resp.error.is_some());
378        assert_eq!(resp.error.unwrap().code, error_code::METHOD_NOT_FOUND);
379    }
380
381    #[tokio::test]
382    async fn test_empty_method() {
383        let app = make_app();
384        let resp = post_jsonrpc(app, "", Value::Null).await;
385        assert!(resp.error.is_some());
386    }
387
388    #[tokio::test]
389    async fn test_invalid_jsonrpc_version() {
390        let app = make_app();
391        let rpc = serde_json::json!({
392            "jsonrpc": "1.0",
393            "id": 1,
394            "method": methods::SEND_MESSAGE,
395            "params": {}
396        });
397        let body = serde_json::to_string(&rpc).unwrap();
398        let req = Request::builder()
399            .uri("/")
400            .method("POST")
401            .header("content-type", "application/json")
402            .body(Body::from(body))
403            .unwrap();
404        let resp = app.oneshot(req).await.unwrap();
405        let body = resp.into_body().collect().await.unwrap().to_bytes();
406        let rpc_resp: JsonRpcResponse = serde_json::from_slice(&body).unwrap();
407        assert!(rpc_resp.error.is_some());
408    }
409
410    #[tokio::test]
411    async fn test_invalid_params() {
412        let app = make_app();
413        let params = serde_json::json!({"bogus": true});
414        let resp = post_jsonrpc(app, methods::SEND_MESSAGE, params).await;
415        assert!(resp.error.is_some());
416        assert_eq!(resp.error.unwrap().code, error_code::PARSE_ERROR);
417    }
418
419    #[tokio::test]
420    async fn test_list_tasks() {
421        let app = make_app();
422        let params = serde_json::json!({});
423        let resp = post_jsonrpc(app, methods::LIST_TASKS, params).await;
424        let result = protojson_conv::from_value::<ListTasksResponse>(resp.result.unwrap()).unwrap();
425        assert!(result.tasks.is_empty());
426    }
427
428    #[tokio::test]
429    async fn test_cancel_task_not_found() {
430        let app = make_app();
431        let params = serde_json::json!({"id": "nonexistent"});
432        let resp = post_jsonrpc(app, methods::CANCEL_TASK, params).await;
433        assert!(resp.error.is_some());
434    }
435
436    #[tokio::test]
437    async fn test_create_push_config() {
438        let app = make_push_app();
439        let params = serde_json::json!({
440            "taskId": "t1",
441            "id": "cfg1",
442            "url": "http://example.com/callback"
443        });
444        let resp = post_jsonrpc(app, methods::CREATE_PUSH_CONFIG, params).await;
445        assert!(resp.error.is_none(), "unexpected error: {:?}", resp.error);
446        let result =
447            serde_json::from_value::<TaskPushNotificationConfig>(resp.result.unwrap()).unwrap();
448        assert_eq!(result.task_id, "t1");
449        assert_eq!(result.id.as_deref(), Some("cfg1"));
450        assert_eq!(result.url, "http://example.com/callback");
451    }
452
453    #[tokio::test]
454    async fn test_create_push_config_accepts_flat_config_object() {
455        let app = make_push_app();
456        let params = serde_json::json!({
457            "taskId": "t1",
458            "id": "cfg1",
459            "url": "http://example.com/callback"
460        });
461        let resp = post_jsonrpc(app, methods::CREATE_PUSH_CONFIG, params).await;
462        assert!(resp.error.is_none(), "unexpected error: {:?}", resp.error);
463        let result =
464            serde_json::from_value::<TaskPushNotificationConfig>(resp.result.unwrap()).unwrap();
465        assert_eq!(result.task_id, "t1");
466        assert_eq!(result.id.as_deref(), Some("cfg1"));
467        assert_eq!(result.url, "http://example.com/callback");
468    }
469
470    #[tokio::test]
471    async fn test_get_push_config() {
472        let app = make_app();
473        let params = serde_json::json!({
474            "taskId": "t1",
475            "id": "cfg1"
476        });
477        let resp = post_jsonrpc(app, methods::GET_PUSH_CONFIG, params).await;
478        assert!(resp.error.is_some() || resp.result.is_some());
479    }
480
481    #[tokio::test]
482    async fn test_list_push_configs() {
483        let app = make_push_app();
484        let params = serde_json::json!({
485            "taskId": "t1"
486        });
487        let resp = post_jsonrpc(app, methods::LIST_PUSH_CONFIGS, params).await;
488        assert!(resp.error.is_none(), "unexpected error: {:?}", resp.error);
489        let result =
490            serde_json::from_value::<ListTaskPushNotificationConfigsResponse>(resp.result.unwrap())
491                .unwrap();
492        assert!(result.configs.is_empty());
493    }
494
495    #[tokio::test]
496    async fn test_delete_push_config() {
497        let app = make_app();
498        let params = serde_json::json!({
499            "taskId": "t1",
500            "id": "cfg1"
501        });
502        let resp = post_jsonrpc(app, methods::DELETE_PUSH_CONFIG, params).await;
503        assert!(resp.error.is_some() || resp.result.is_some());
504    }
505
506    #[tokio::test]
507    async fn test_get_extended_agent_card() {
508        let app = make_app();
509        let params = serde_json::json!({});
510        let resp = post_jsonrpc(app, methods::GET_EXTENDED_AGENT_CARD, params).await;
511        // DefaultRequestHandler returns NotSupported
512        assert!(resp.error.is_some() || resp.result.is_some());
513    }
514
515    #[tokio::test]
516    async fn test_streaming_send_message() {
517        let app = make_app();
518        let body = serde_json::json!({
519            "message": {
520                "messageId": "m1",
521                "role": "ROLE_USER",
522                "parts": [{"text": "hello"}]
523            }
524        });
525        let rpc = JsonRpcRequest::new(
526            JsonRpcId::Number(1),
527            methods::SEND_STREAMING_MESSAGE,
528            Some(body),
529        );
530        let req = Request::builder()
531            .uri("/")
532            .method("POST")
533            .header("content-type", "application/json")
534            .header("accept", "text/event-stream")
535            .body(Body::from(serde_json::to_string(&rpc).unwrap()))
536            .unwrap();
537        let resp = app.oneshot(req).await.unwrap();
538        // SSE response => 200
539        assert_eq!(resp.status(), StatusCode::OK);
540    }
541
542    #[tokio::test]
543    async fn test_streaming_subscribe_to_task() {
544        let app = make_app();
545        let rpc = JsonRpcRequest::new(
546            JsonRpcId::Number(1),
547            methods::SUBSCRIBE_TO_TASK,
548            Some(serde_json::json!({"id": "t1"})),
549        );
550        let req = Request::builder()
551            .uri("/")
552            .method("POST")
553            .header("content-type", "application/json")
554            .body(Body::from(serde_json::to_string(&rpc).unwrap()))
555            .unwrap();
556        let resp = app.oneshot(req).await.unwrap();
557        // Subscription may fail (task not found), but routing should work
558        assert_eq!(resp.status(), StatusCode::OK);
559    }
560
561    #[tokio::test]
562    async fn test_streaming_send_message_legacy_method_rejected() {
563        let app = make_app();
564        let body = serde_json::json!({
565            "message": {
566                "messageId": "m1",
567                "role": "ROLE_USER",
568                "parts": [{"text": "hello"}]
569            }
570        });
571        let rpc = JsonRpcRequest::new(JsonRpcId::Number(1), "message.stream", Some(body));
572        let req = Request::builder()
573            .uri("/")
574            .method("POST")
575            .header("content-type", "application/json")
576            .header("accept", "text/event-stream")
577            .body(Body::from(serde_json::to_string(&rpc).unwrap()))
578            .unwrap();
579        let resp = app.oneshot(req).await.unwrap();
580        let body = resp.into_body().collect().await.unwrap().to_bytes();
581        let rpc_resp: JsonRpcResponse = serde_json::from_slice(&body).unwrap();
582        assert!(
583            rpc_resp.error.is_some(),
584            "unexpected result: {:?}",
585            rpc_resp.result
586        );
587        assert_eq!(rpc_resp.error.unwrap().code, error_code::METHOD_NOT_FOUND);
588    }
589
590    #[tokio::test]
591    async fn test_send_message_propagates_headers_as_service_params() {
592        let (handler, captured) = CapturingHandler::new();
593        let app = jsonrpc_router(handler);
594
595        let rpc = JsonRpcRequest::new(
596            JsonRpcId::Number(1),
597            methods::SEND_MESSAGE,
598            Some(serde_json::json!({
599                "message": {
600                    "messageId": "m1",
601                    "role": "ROLE_USER",
602                    "parts": [{"text": "hi"}]
603                }
604            })),
605        );
606        let req = Request::builder()
607            .uri("/")
608            .method("POST")
609            .header("content-type", "application/json")
610            .header("authorization", "Bearer jsonrpc-token")
611            .header("x-tenant-id", "acme")
612            .body(Body::from(serde_json::to_string(&rpc).unwrap()))
613            .unwrap();
614        let resp = app.oneshot(req).await.unwrap();
615        assert_eq!(resp.status(), StatusCode::OK);
616
617        assert_header_captured(
618            &captured,
619            "send_message",
620            "authorization",
621            "Bearer jsonrpc-token",
622        );
623        assert_header_captured(&captured, "send_message", "x-tenant-id", "acme");
624    }
625
626    #[tokio::test]
627    async fn test_streaming_send_message_propagates_headers_as_service_params() {
628        let (handler, captured) = CapturingHandler::new();
629        let app = jsonrpc_router(handler);
630
631        let rpc = JsonRpcRequest::new(
632            JsonRpcId::Number(1),
633            methods::SEND_STREAMING_MESSAGE,
634            Some(serde_json::json!({
635                "message": {
636                    "messageId": "m1",
637                    "role": "ROLE_USER",
638                    "parts": [{"text": "hi"}]
639                }
640            })),
641        );
642        let req = Request::builder()
643            .uri("/")
644            .method("POST")
645            .header("content-type", "application/json")
646            .header("accept", "text/event-stream")
647            .header("authorization", "Bearer jsonrpc-streaming-token")
648            .header("x-tenant-id", "acme")
649            .body(Body::from(serde_json::to_string(&rpc).unwrap()))
650            .unwrap();
651        let resp = app.oneshot(req).await.unwrap();
652        assert_eq!(resp.status(), StatusCode::OK);
653        // Drain the SSE body so the handler is driven to completion.
654        let _ = resp.into_body().collect().await.unwrap().to_bytes();
655
656        assert_header_captured(
657            &captured,
658            "send_streaming_message",
659            "authorization",
660            "Bearer jsonrpc-streaming-token",
661        );
662        assert_header_captured(&captured, "send_streaming_message", "x-tenant-id", "acme");
663    }
664
665    #[tokio::test]
666    async fn test_jsonrpc_unary_methods_propagate_headers_as_service_params() {
667        // Exercise every unary JSON-RPC method to lock in header propagation
668        // across the full match arm in handle_unary_request.
669        let cases: &[(&'static str, &'static str, serde_json::Value)] = &[
670            (
671                methods::GET_TASK,
672                "get_task",
673                serde_json::json!({"id": "t1"}),
674            ),
675            (methods::LIST_TASKS, "list_tasks", serde_json::json!({})),
676            (
677                methods::CANCEL_TASK,
678                "cancel_task",
679                serde_json::json!({"id": "t1"}),
680            ),
681            (
682                methods::CREATE_PUSH_CONFIG,
683                "create_push_config",
684                serde_json::json!({"taskId": "t1", "url": "http://example.com/cb"}),
685            ),
686            (
687                methods::GET_PUSH_CONFIG,
688                "get_push_config",
689                serde_json::json!({"taskId": "t1", "id": "cfg1"}),
690            ),
691            (
692                methods::LIST_PUSH_CONFIGS,
693                "list_push_configs",
694                serde_json::json!({"taskId": "t1"}),
695            ),
696            (
697                methods::DELETE_PUSH_CONFIG,
698                "delete_push_config",
699                serde_json::json!({"taskId": "t1", "id": "cfg1"}),
700            ),
701            (
702                methods::GET_EXTENDED_AGENT_CARD,
703                "get_extended_agent_card",
704                serde_json::json!({}),
705            ),
706        ];
707
708        for (rpc_method, handler_method, params) in cases {
709            let (handler, captured) = CapturingHandler::new();
710            let app = jsonrpc_router(handler);
711            let rpc = JsonRpcRequest::new(JsonRpcId::Number(1), *rpc_method, Some(params.clone()));
712            let token = format!("Bearer token-for-{handler_method}");
713            let req = Request::builder()
714                .uri("/")
715                .method("POST")
716                .header("content-type", "application/json")
717                .header("authorization", &token)
718                .body(Body::from(serde_json::to_string(&rpc).unwrap()))
719                .unwrap();
720            let resp = app.oneshot(req).await.unwrap();
721            assert_eq!(resp.status(), StatusCode::OK);
722            assert_header_captured(&captured, handler_method, "authorization", &token);
723        }
724    }
725
726    #[tokio::test]
727    async fn test_jsonrpc_subscribe_to_task_propagates_headers_as_service_params() {
728        let (handler, captured) = CapturingHandler::new();
729        let app = jsonrpc_router(handler);
730        let rpc = JsonRpcRequest::new(
731            JsonRpcId::Number(1),
732            methods::SUBSCRIBE_TO_TASK,
733            Some(serde_json::json!({"id": "t1"})),
734        );
735        let req = Request::builder()
736            .uri("/")
737            .method("POST")
738            .header("content-type", "application/json")
739            .header("authorization", "Bearer subscribe-token")
740            .body(Body::from(serde_json::to_string(&rpc).unwrap()))
741            .unwrap();
742        let resp = app.oneshot(req).await.unwrap();
743        assert_eq!(resp.status(), StatusCode::OK);
744        let _ = resp.into_body().collect().await.unwrap().to_bytes();
745        assert_header_captured(
746            &captured,
747            "subscribe_to_task",
748            "authorization",
749            "Bearer subscribe-token",
750        );
751    }
752
753    #[tokio::test]
754    async fn test_streaming_invalid_method() {
755        let app = make_app();
756        let rpc = serde_json::json!({
757            "jsonrpc": "2.0",
758            "id": 1,
759            "method": "tasks.subscribe",
760            "params": {"bogus": true}
761        });
762        let req = Request::builder()
763            .uri("/")
764            .method("POST")
765            .header("content-type", "application/json")
766            .body(Body::from(serde_json::to_string(&rpc).unwrap()))
767            .unwrap();
768        let resp = app.oneshot(req).await.unwrap();
769        assert_eq!(resp.status(), StatusCode::OK);
770    }
771}
a2a_server/jsonrpc.rs

a2a_server/
jsonrpc.rs
