a2a_server/
lib.rs

1// Copyright AGNTCY Contributors (https://github.com/agntcy)
2// SPDX-License-Identifier: Apache-2.0
3pub mod agent_card;
4pub mod executor;
5pub mod handler;
6pub mod jsonrpc;
7pub mod middleware;
8pub mod pagination;
9pub mod push;
10pub mod rest;
11pub mod sse;
12pub mod task_store;
13
14#[cfg(any(feature = "rustls-tls", feature = "rustls-no-provider"))]
15pub mod tls;
16
17pub use agent_card::{AgentCardProducer, StaticAgentCard, WELL_KNOWN_AGENT_CARD_PATH};
18pub use executor::{AgentExecutor, ExecutorContext};
19pub use handler::{
20    DefaultRequestHandler, ExtendedAgentCardResolver, RequestAuthorizer, RequestHandler,
21};
22pub use middleware::{CallContext, CallInterceptor, InterceptedHandler, ServiceParams, User};
23pub use push::{HttpPushSender, HttpPushSenderConfig, InMemoryPushConfigStore, PushConfigStore};
24pub use task_store::{InMemoryTaskStore, TaskStore};
25
26#[cfg(test)]
27pub(crate) mod test_util {
28    use std::collections::HashMap;
29    use std::sync::{Arc, Mutex};
30
31    use a2a::*;
32    use async_trait::async_trait;
33    use futures::stream::BoxStream;
34
35    use crate::handler::RequestHandler;
36    use crate::middleware::ServiceParams;
37
38    pub fn install_crypto_provider() {
39        #[cfg(feature = "rustls-tls")]
40        {
41            let _ = rustls::crypto::aws_lc_rs::default_provider().install_default();
42        }
43    }
44
45    pub(crate) type CapturedParams = Arc<Mutex<HashMap<&'static str, ServiceParams>>>;
46
47    /// `RequestHandler` stub that records the `ServiceParams` it sees per method.
48    ///
49    /// Used by transport-layer tests to assert that incoming request headers
50    /// are surfaced through to the `RequestHandler` boundary — which is exactly
51    /// what the HTTP / JSON-RPC transports are responsible for. Returns minimal
52    /// stub responses so tests don't depend on `DefaultRequestHandler` behavior.
53    pub(crate) struct CapturingHandler {
54        captured: CapturedParams,
55    }
56
57    impl CapturingHandler {
58        pub fn new() -> (Arc<Self>, CapturedParams) {
59            let captured = Arc::new(Mutex::new(HashMap::new()));
60            let handler = Arc::new(CapturingHandler {
61                captured: captured.clone(),
62            });
63            (handler, captured)
64        }
65
66        fn record(&self, method: &'static str, params: &ServiceParams) {
67            self.captured.lock().unwrap().insert(method, params.clone());
68        }
69    }
70
71    #[async_trait]
72    impl RequestHandler for CapturingHandler {
73        async fn send_message(
74            &self,
75            params: &ServiceParams,
76            _req: SendMessageRequest,
77        ) -> Result<SendMessageResponse, A2AError> {
78            self.record("send_message", params);
79            Err(A2AError::internal("stub"))
80        }
81
82        async fn send_streaming_message(
83            &self,
84            params: &ServiceParams,
85            _req: SendMessageRequest,
86        ) -> Result<BoxStream<'static, Result<StreamResponse, A2AError>>, A2AError> {
87            self.record("send_streaming_message", params);
88            Ok(Box::pin(futures::stream::empty()))
89        }
90
91        async fn get_task(
92            &self,
93            params: &ServiceParams,
94            _req: GetTaskRequest,
95        ) -> Result<Task, A2AError> {
96            self.record("get_task", params);
97            Err(A2AError::task_not_found("stub"))
98        }
99
100        async fn list_tasks(
101            &self,
102            params: &ServiceParams,
103            _req: ListTasksRequest,
104        ) -> Result<ListTasksResponse, A2AError> {
105            self.record("list_tasks", params);
106            Ok(ListTasksResponse {
107                tasks: Vec::new(),
108                next_page_token: String::new(),
109                page_size: 0,
110                total_size: 0,
111            })
112        }
113
114        async fn cancel_task(
115            &self,
116            params: &ServiceParams,
117            _req: CancelTaskRequest,
118        ) -> Result<Task, A2AError> {
119            self.record("cancel_task", params);
120            Err(A2AError::task_not_found("stub"))
121        }
122
123        async fn subscribe_to_task(
124            &self,
125            params: &ServiceParams,
126            _req: SubscribeToTaskRequest,
127        ) -> Result<BoxStream<'static, Result<StreamResponse, A2AError>>, A2AError> {
128            self.record("subscribe_to_task", params);
129            Ok(Box::pin(futures::stream::empty()))
130        }
131
132        async fn create_push_config(
133            &self,
134            params: &ServiceParams,
135            req: TaskPushNotificationConfig,
136        ) -> Result<TaskPushNotificationConfig, A2AError> {
137            self.record("create_push_config", params);
138            Ok(req)
139        }
140
141        async fn get_push_config(
142            &self,
143            params: &ServiceParams,
144            _req: GetTaskPushNotificationConfigRequest,
145        ) -> Result<TaskPushNotificationConfig, A2AError> {
146            self.record("get_push_config", params);
147            Err(A2AError::internal("stub"))
148        }
149
150        async fn list_push_configs(
151            &self,
152            params: &ServiceParams,
153            _req: ListTaskPushNotificationConfigsRequest,
154        ) -> Result<ListTaskPushNotificationConfigsResponse, A2AError> {
155            self.record("list_push_configs", params);
156            Ok(ListTaskPushNotificationConfigsResponse {
157                configs: Vec::new(),
158                next_page_token: None,
159            })
160        }
161
162        async fn delete_push_config(
163            &self,
164            params: &ServiceParams,
165            _req: DeleteTaskPushNotificationConfigRequest,
166        ) -> Result<(), A2AError> {
167            self.record("delete_push_config", params);
168            Ok(())
169        }
170
171        async fn get_extended_agent_card(
172            &self,
173            params: &ServiceParams,
174            _req: GetExtendedAgentCardRequest,
175        ) -> Result<AgentCard, A2AError> {
176            self.record("get_extended_agent_card", params);
177            Err(A2AError::internal("stub"))
178        }
179    }
180
181    /// Assert that `method` was called with a service-params entry matching `header = expected`.
182    #[track_caller]
183    pub(crate) fn assert_header_captured(
184        captured: &CapturedParams,
185        method: &'static str,
186        header: &str,
187        expected: &str,
188    ) {
189        let captured = captured.lock().unwrap();
190        let params = captured
191            .get(method)
192            .unwrap_or_else(|| panic!("{method} should have been called; captured={captured:?}"));
193        assert_eq!(
194            params.get(header),
195            Some(&vec![expected.to_string()]),
196            "{header} should be propagated to {method}; got {params:?}",
197        );
198    }
199}
a2a_server/lib.rs

a2a_server/
lib.rs
