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

a2a_server/
lib.rs
