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

a2a_server/
lib.rs
