a2a_server/
lib.rs

1// Copyright AGNTCY Contributors (https://github.com/agntcy)
2// Copyright A2A Contributors (https://github.com/a2aproject)
3// SPDX-License-Identifier: Apache-2.0
4pub mod agent_card;
5pub mod executor;
6pub mod handler;
7pub mod jsonrpc;
8pub mod middleware;
9pub mod pagination;
10pub mod push;
11pub mod rest;
12pub mod sse;
13pub mod task_store;
14
15// Push-URL validation internals reached by the fuzz target in fuzz/ — see
16// #238. Thin wrappers, not re-exports: the originals stay crate-private.
17#[cfg(feature = "fuzzing")]
18#[doc(hidden)]
19pub mod fuzz_support {
20    use a2a::A2AError;
21
22    pub fn validate_push_url(url: &str) -> Result<(), A2AError> {
23        crate::push::sender::validate_push_url(url)
24    }
25
26    pub fn is_blocked_ip(ip: std::net::IpAddr) -> bool {
27        crate::push::sender::is_blocked_ip(ip)
28    }
29}
30
31#[cfg(any(feature = "rustls-tls", feature = "rustls-no-provider"))]
32pub mod tls;
33
34pub use agent_card::{AgentCardProducer, StaticAgentCard, WELL_KNOWN_AGENT_CARD_PATH};
35pub use executor::{AgentExecutor, ExecutorContext};
36pub use handler::{
37    DefaultRequestHandler, ExtendedAgentCardResolver, RequestAuthorizer, RequestHandler,
38};
39pub use middleware::{CallContext, CallInterceptor, InterceptedHandler, ServiceParams, User};
40pub use push::{HttpPushSender, HttpPushSenderConfig, InMemoryPushConfigStore, PushConfigStore};
41pub use task_store::{InMemoryTaskStore, TaskStore};
42
43/// Build an `INTERNAL_ERROR` whose client-visible message is generic, while
44/// the real cause is logged server-side (BUG-12 / CWE-209).
45///
46/// Call this where the **server itself** failed — a serialization fault, a
47/// broken invariant — and never at a response boundary. A2A gives an
48/// `AgentExecutor` no code other than `INTERNAL_ERROR` to report its own
49/// failure with (`error_code` has nothing between `INVALID_AGENT_RESPONSE`
50/// and the JSON-RPC generics), and `A2AError` carries no marker for where an
51/// error came from. Sanitizing by code at the boundary therefore cannot tell
52/// a serialization fault from an agent saying why it failed, and would
53/// replace both with "Internal error".
54pub(crate) fn sanitized_internal_error(detail: impl std::fmt::Display) -> a2a::A2AError {
55    tracing::error!(error = %detail, "internal error; returning a generic message to the client");
56    a2a::A2AError::internal("Internal error")
57}
58
59#[cfg(test)]
60pub(crate) mod test_util {
61    use std::collections::HashMap;
62    use std::sync::{Arc, Mutex};
63
64    use a2a::*;
65    use async_trait::async_trait;
66    use futures::stream::BoxStream;
67
68    use crate::handler::RequestHandler;
69    use crate::middleware::ServiceParams;
70
71    pub fn install_crypto_provider() {
72        #[cfg(feature = "rustls-tls")]
73        {
74            let _ = rustls::crypto::aws_lc_rs::default_provider().install_default();
75        }
76    }
77
78    pub(crate) type CapturedParams = Arc<Mutex<HashMap<&'static str, ServiceParams>>>;
79
80    /// `RequestHandler` stub that records the `ServiceParams` it sees per method.
81    ///
82    /// Used by transport-layer tests to assert that incoming request headers
83    /// are surfaced through to the `RequestHandler` boundary — which is exactly
84    /// what the HTTP / JSON-RPC transports are responsible for. Returns minimal
85    /// stub responses so tests don't depend on `DefaultRequestHandler` behavior.
86    pub(crate) struct CapturingHandler {
87        captured: CapturedParams,
88    }
89
90    impl CapturingHandler {
91        pub fn new() -> (Arc<Self>, CapturedParams) {
92            let captured = Arc::new(Mutex::new(HashMap::new()));
93            let handler = Arc::new(CapturingHandler {
94                captured: captured.clone(),
95            });
96            (handler, captured)
97        }
98
99        fn record(&self, method: &'static str, params: &ServiceParams) {
100            self.captured.lock().unwrap().insert(method, params.clone());
101        }
102    }
103
104    #[async_trait]
105    impl RequestHandler for CapturingHandler {
106        async fn send_message(
107            &self,
108            params: &ServiceParams,
109            _req: SendMessageRequest,
110        ) -> Result<SendMessageResponse, A2AError> {
111            self.record("send_message", params);
112            Err(A2AError::internal("stub"))
113        }
114
115        async fn send_streaming_message(
116            &self,
117            params: &ServiceParams,
118            _req: SendMessageRequest,
119        ) -> Result<BoxStream<'static, Result<StreamResponse, A2AError>>, A2AError> {
120            self.record("send_streaming_message", params);
121            Ok(Box::pin(futures::stream::empty()))
122        }
123
124        async fn get_task(
125            &self,
126            params: &ServiceParams,
127            _req: GetTaskRequest,
128        ) -> Result<Task, A2AError> {
129            self.record("get_task", params);
130            Err(A2AError::task_not_found("stub"))
131        }
132
133        async fn list_tasks(
134            &self,
135            params: &ServiceParams,
136            _req: ListTasksRequest,
137        ) -> Result<ListTasksResponse, A2AError> {
138            self.record("list_tasks", params);
139            Ok(ListTasksResponse {
140                tasks: Vec::new(),
141                next_page_token: String::new(),
142                page_size: 0,
143                total_size: 0,
144            })
145        }
146
147        async fn cancel_task(
148            &self,
149            params: &ServiceParams,
150            _req: CancelTaskRequest,
151        ) -> Result<Task, A2AError> {
152            self.record("cancel_task", params);
153            Err(A2AError::task_not_found("stub"))
154        }
155
156        async fn subscribe_to_task(
157            &self,
158            params: &ServiceParams,
159            _req: SubscribeToTaskRequest,
160        ) -> Result<BoxStream<'static, Result<StreamResponse, A2AError>>, A2AError> {
161            self.record("subscribe_to_task", params);
162            Ok(Box::pin(futures::stream::empty()))
163        }
164
165        async fn create_push_config(
166            &self,
167            params: &ServiceParams,
168            req: TaskPushNotificationConfig,
169        ) -> Result<TaskPushNotificationConfig, A2AError> {
170            self.record("create_push_config", params);
171            Ok(req)
172        }
173
174        async fn get_push_config(
175            &self,
176            params: &ServiceParams,
177            _req: GetTaskPushNotificationConfigRequest,
178        ) -> Result<TaskPushNotificationConfig, A2AError> {
179            self.record("get_push_config", params);
180            Err(A2AError::internal("stub"))
181        }
182
183        async fn list_push_configs(
184            &self,
185            params: &ServiceParams,
186            _req: ListTaskPushNotificationConfigsRequest,
187        ) -> Result<ListTaskPushNotificationConfigsResponse, A2AError> {
188            self.record("list_push_configs", params);
189            Ok(ListTaskPushNotificationConfigsResponse {
190                configs: Vec::new(),
191                next_page_token: None,
192            })
193        }
194
195        async fn delete_push_config(
196            &self,
197            params: &ServiceParams,
198            _req: DeleteTaskPushNotificationConfigRequest,
199        ) -> Result<(), A2AError> {
200            self.record("delete_push_config", params);
201            Ok(())
202        }
203
204        async fn get_extended_agent_card(
205            &self,
206            params: &ServiceParams,
207            _req: GetExtendedAgentCardRequest,
208        ) -> Result<AgentCard, A2AError> {
209            self.record("get_extended_agent_card", params);
210            Err(A2AError::internal("stub"))
211        }
212    }
213
214    /// Assert that `method` was called with a service-params entry matching `header = expected`.
215    #[track_caller]
216    pub(crate) fn assert_header_captured(
217        captured: &CapturedParams,
218        method: &'static str,
219        header: &str,
220        expected: &str,
221    ) {
222        let captured = captured.lock().unwrap();
223        let params = captured
224            .get(method)
225            .unwrap_or_else(|| panic!("{method} should have been called; captured={captured:?}"));
226        assert_eq!(
227            params.get(header),
228            Some(&vec![expected.to_string()]),
229            "{header} should be propagated to {method}; got {params:?}",
230        );
231    }
232}
a2a_server/lib.rs

a2a_server/
lib.rs
