a2a_server/push/
sender.rs

1// Copyright AGNTCY Contributors (https://github.com/agntcy)
2// SPDX-License-Identifier: Apache-2.0
3use a2a::*;
4use std::time::Duration;
5
6/// HTTP push notification sender.
7///
8/// Sends A2A streaming events to a push notification endpoint over HTTP POST.
9pub struct HttpPushSender {
10    client: reqwest::Client,
11    fail_on_error: bool,
12}
13
14/// Configuration for [`HttpPushSender`].
15pub struct HttpPushSenderConfig {
16    /// HTTP request timeout (default: 30s).
17    pub timeout: Duration,
18    /// If true, push sending errors abort execution.
19    pub fail_on_error: bool,
20}
21
22impl Default for HttpPushSenderConfig {
23    fn default() -> Self {
24        HttpPushSenderConfig {
25            timeout: Duration::from_secs(30),
26            fail_on_error: false,
27        }
28    }
29}
30
31impl HttpPushSender {
32    pub fn new(config: Option<HttpPushSenderConfig>) -> Self {
33        let config = config.unwrap_or_default();
34        let client = reqwest::Client::builder()
35            .timeout(config.timeout)
36            .build()
37            .expect("failed to create HTTP client");
38        HttpPushSender {
39            client,
40            fail_on_error: config.fail_on_error,
41        }
42    }
43
44    /// Send an event to the push notification endpoint.
45    pub async fn send_push(
46        &self,
47        config: &TaskPushNotificationConfig,
48        event: StreamResponse,
49    ) -> Result<(), A2AError> {
50        let body = match serde_json::to_vec(&event) {
51            Ok(b) => b,
52            Err(e) => return self.handle_error(format!("failed to serialize event: {e}")),
53        };
54
55        let mut request = self
56            .client
57            .post(&config.url)
58            .header("Content-Type", "application/json")
59            .body(body);
60
61        if let Some(ref token) = config.token {
62            request = request.header("A2A-Notification-Token", token);
63        }
64
65        if let Some(ref auth) = config.authentication {
66            if let Some(ref creds) = auth.credentials {
67                match auth.scheme.to_lowercase().as_str() {
68                    "bearer" => {
69                        request = request.header("Authorization", format!("Bearer {creds}"));
70                    }
71                    "basic" => {
72                        request = request.header("Authorization", format!("Basic {creds}"));
73                    }
74                    _ => {}
75                }
76            }
77        }
78
79        match request.send().await {
80            Ok(resp) => {
81                if !resp.status().is_success() {
82                    return self
83                        .handle_error(format!("push endpoint returned status: {}", resp.status()));
84                }
85                Ok(())
86            }
87            Err(e) => self.handle_error(format!("failed to send push notification: {e}")),
88        }
89    }
90
91    fn handle_error(&self, msg: String) -> Result<(), A2AError> {
92        if self.fail_on_error {
93            Err(A2AError::internal(&msg))
94        } else {
95            tracing::error!("{}", msg);
96            Ok(())
97        }
98    }
99}
100
101#[cfg(test)]
102mod tests {
103    use super::*;
104
105    use crate::test_util::install_crypto_provider;
106
107    #[test]
108    fn test_default_config() {
109        let config = HttpPushSenderConfig::default();
110        assert_eq!(config.timeout, Duration::from_secs(30));
111        assert!(!config.fail_on_error);
112    }
113
114    #[test]
115    fn test_sender_new_default() {
116        install_crypto_provider();
117        let sender = HttpPushSender::new(None);
118        assert!(!sender.fail_on_error);
119    }
120
121    #[test]
122    fn test_sender_new_custom() {
123        install_crypto_provider();
124        let config = HttpPushSenderConfig {
125            timeout: Duration::from_secs(10),
126            fail_on_error: true,
127        };
128        let sender = HttpPushSender::new(Some(config));
129        assert!(sender.fail_on_error);
130    }
131
132    #[test]
133    fn test_handle_error_fail_on_error() {
134        install_crypto_provider();
135        let sender = HttpPushSender::new(Some(HttpPushSenderConfig {
136            timeout: Duration::from_secs(5),
137            fail_on_error: true,
138        }));
139        let result = sender.handle_error("test error".to_string());
140        assert!(result.is_err());
141    }
142
143    #[test]
144    fn test_handle_error_ignore() {
145        install_crypto_provider();
146        let sender = HttpPushSender::new(None);
147        let result = sender.handle_error("test error".to_string());
148        assert!(result.is_ok());
149    }
150
151    #[tokio::test]
152    async fn test_send_push_connection_refused_no_fail() {
153        use a2a::event::*;
154        install_crypto_provider();
155        let sender = HttpPushSender::new(None);
156        let config = TaskPushNotificationConfig {
157            task_id: String::new(),
158            url: "http://127.0.0.1:1/callback".to_string(),
159            id: None,
160            token: Some("tok".to_string()),
161            authentication: Some(AuthenticationInfo {
162                scheme: "bearer".to_string(),
163                credentials: Some("secret".to_string()),
164            }),
165            tenant: None,
166        };
167        let event = StreamResponse::StatusUpdate(TaskStatusUpdateEvent {
168            task_id: "t1".into(),
169            context_id: "c1".into(),
170            status: TaskStatus {
171                state: TaskState::Working,
172                message: None,
173                timestamp: None,
174            },
175            metadata: None,
176        });
177        let result = sender.send_push(&config, event).await;
178        // Should be Ok because fail_on_error is false
179        assert!(result.is_ok());
180    }
181
182    #[tokio::test]
183    async fn test_send_push_connection_refused_fail() {
184        use a2a::event::*;
185        install_crypto_provider();
186        let sender = HttpPushSender::new(Some(HttpPushSenderConfig {
187            timeout: std::time::Duration::from_millis(100),
188            fail_on_error: true,
189        }));
190        let config = TaskPushNotificationConfig {
191            task_id: String::new(),
192            url: "http://127.0.0.1:1/callback".to_string(),
193            id: None,
194            token: None,
195            authentication: None,
196            tenant: None,
197        };
198        let event = StreamResponse::StatusUpdate(TaskStatusUpdateEvent {
199            task_id: "t1".into(),
200            context_id: "c1".into(),
201            status: TaskStatus {
202                state: TaskState::Working,
203                message: None,
204                timestamp: None,
205            },
206            metadata: None,
207        });
208        let result = sender.send_push(&config, event).await;
209        assert!(result.is_err());
210    }
211
212    #[tokio::test]
213    async fn test_send_push_with_basic_auth() {
214        use a2a::event::*;
215        install_crypto_provider();
216        let sender = HttpPushSender::new(None);
217        let config = TaskPushNotificationConfig {
218            task_id: String::new(),
219            url: "http://127.0.0.1:1/callback".to_string(),
220            id: None,
221            token: None,
222            authentication: Some(AuthenticationInfo {
223                scheme: "basic".to_string(),
224                credentials: Some("dXNlcjpwYXNz".to_string()),
225            }),
226            tenant: None,
227        };
228        let event = StreamResponse::StatusUpdate(TaskStatusUpdateEvent {
229            task_id: "t1".into(),
230            context_id: "c1".into(),
231            status: TaskStatus {
232                state: TaskState::Working,
233                message: None,
234                timestamp: None,
235            },
236            metadata: None,
237        });
238        // Will fail to connect but exercises basic auth header path
239        let result = sender.send_push(&config, event).await;
240        assert!(result.is_ok()); // fail_on_error=false
241    }
242
243    #[tokio::test]
244    async fn test_send_push_with_unknown_auth_scheme() {
245        use a2a::event::*;
246        install_crypto_provider();
247        let sender = HttpPushSender::new(None);
248        let config = TaskPushNotificationConfig {
249            task_id: String::new(),
250            url: "http://127.0.0.1:1/callback".to_string(),
251            id: None,
252            token: None,
253            authentication: Some(AuthenticationInfo {
254                scheme: "custom".to_string(),
255                credentials: Some("cred".to_string()),
256            }),
257            tenant: None,
258        };
259        let event = StreamResponse::StatusUpdate(TaskStatusUpdateEvent {
260            task_id: "t1".into(),
261            context_id: "c1".into(),
262            status: TaskStatus {
263                state: TaskState::Working,
264                message: None,
265                timestamp: None,
266            },
267            metadata: None,
268        });
269        let result = sender.send_push(&config, event).await;
270        assert!(result.is_ok());
271    }
272}
a2a_server/push/sender.rs

a2a_server/push/
sender.rs
