a2a_client/
rest.rs

1// Copyright AGNTCY Contributors (https://github.com/agntcy)
2// Copyright A2A Contributors (https://github.com/a2aproject)
3// SPDX-License-Identifier: Apache-2.0
4use a2a::*;
5use a2a_pb::protojson_conv::{self, ProtoJsonPayload};
6use async_trait::async_trait;
7use futures::stream::BoxStream;
8use reqwest::Client;
9use serde::Deserialize;
10use serde_json::Value;
11
12use crate::push_config_compat::{
13    deserialize_list_task_push_notification_configs_response,
14    deserialize_task_push_notification_config,
15};
16use crate::transport::{ServiceParams, Transport, TransportFactory};
17use crate::wire;
18
19const REST_SEND_MESSAGE_PATH: &str = "/message:send";
20const REST_STREAM_MESSAGE_PATH: &str = "/message:stream";
21const REST_EXTENDED_AGENT_CARD_PATH: &str = "/extendedAgentCard";
22
23#[derive(Debug, Deserialize)]
24struct RestErrorEnvelope {
25    error: RestErrorStatus,
26}
27
28#[derive(Debug, Deserialize)]
29struct RestErrorStatus {
30    message: String,
31
32    #[serde(default)]
33    details: Vec<TypedDetail>,
34}
35
36/// REST (HTTP+JSON) transport implementation.
37///
38/// Maps A2A operations to RESTful HTTP endpoints.
39pub struct RestTransport {
40    client: Client,
41    base_url: String,
42}
43
44impl RestTransport {
45    /// Build a `RestTransport` from a pre-constructed `reqwest::Client`.
46    pub fn new(client: Client, base_url: String) -> Self {
47        let base_url = base_url.trim_end_matches('/').to_string();
48        RestTransport { client, base_url }
49    }
50
51    fn build_request(
52        &self,
53        method: reqwest::Method,
54        path: &str,
55        params: &ServiceParams,
56    ) -> reqwest::RequestBuilder {
57        let url = format!("{}{}", self.base_url, path);
58        let mut builder = self.client.request(method, &url);
59        for (key, values) in params {
60            for v in values {
61                builder = builder.header(key, v);
62            }
63        }
64        builder
65    }
66
67    fn build_request_with_query(
68        &self,
69        method: reqwest::Method,
70        path: &str,
71        params: &ServiceParams,
72        query: &[(String, String)],
73    ) -> reqwest::RequestBuilder {
74        let builder = self.build_request(method, path, params);
75        if query.is_empty() {
76            builder
77        } else {
78            builder.query(query)
79        }
80    }
81
82    async fn send(&self, builder: reqwest::RequestBuilder) -> Result<reqwest::Response, A2AError> {
83        // Every REST request goes through here, so wire logging for this
84        // binding needs exactly one call site.
85        wire::send(&self.client, builder).await
86    }
87
88    async fn into_rest_error(resp: reqwest::Response) -> A2AError {
89        let status = resp.status();
90        let body = resp.text().await.unwrap_or_default();
91        // A failing response is the one a person reaches for --debug to see,
92        // so log it on the way to being turned into an error.
93        wire::log_response(status, &body);
94        parse_rest_error(status, &body)
95    }
96
97    async fn post_value<Req>(
98        &self,
99        path: &str,
100        params: &ServiceParams,
101        body: &Req,
102    ) -> Result<Value, A2AError>
103    where
104        Req: ProtoJsonPayload,
105    {
106        let payload = protojson_conv::to_value(body).map_err(|e| {
107            A2AError::internal(format!("failed to serialize request as ProtoJSON: {e}"))
108        })?;
109        let resp = self
110            .send(
111                self.build_request(reqwest::Method::POST, path, params)
112                    .json(&payload),
113            )
114            .await?;
115
116        if !resp.status().is_success() {
117            return Err(Self::into_rest_error(resp).await);
118        }
119        let body = wire::response_text(resp)
120            .await
121            .map_err(|e| A2AError::internal(format!("failed to parse response: {e}")))?;
122        let payload = serde_json::from_str::<Value>(&body)
123            .map_err(|e| A2AError::internal(format!("failed to parse response: {e}")))?;
124
125        Ok(payload)
126    }
127
128    async fn post_json<Req, Resp>(
129        &self,
130        path: &str,
131        params: &ServiceParams,
132        body: &Req,
133    ) -> Result<Resp, A2AError>
134    where
135        Req: ProtoJsonPayload,
136        Resp: ProtoJsonPayload,
137    {
138        let payload = self.post_value(path, params, body).await?;
139
140        protojson_conv::from_value(payload).map_err(|e| {
141            A2AError::internal(format!("failed to deserialize response as ProtoJSON: {e}"))
142        })
143    }
144
145    async fn get_value(
146        &self,
147        path: &str,
148        params: &ServiceParams,
149        query: &[(String, String)],
150    ) -> Result<Value, A2AError> {
151        let resp = self
152            .send(self.build_request_with_query(reqwest::Method::GET, path, params, query))
153            .await?;
154
155        if !resp.status().is_success() {
156            return Err(Self::into_rest_error(resp).await);
157        }
158        let body = wire::response_text(resp)
159            .await
160            .map_err(|e| A2AError::internal(format!("failed to parse response: {e}")))?;
161        let payload = serde_json::from_str::<Value>(&body)
162            .map_err(|e| A2AError::internal(format!("failed to parse response: {e}")))?;
163
164        Ok(payload)
165    }
166
167    async fn get_json<Resp>(
168        &self,
169        path: &str,
170        params: &ServiceParams,
171        query: &[(String, String)],
172    ) -> Result<Resp, A2AError>
173    where
174        Resp: ProtoJsonPayload,
175    {
176        let payload = self.get_value(path, params, query).await?;
177
178        protojson_conv::from_value(payload).map_err(|e| {
179            A2AError::internal(format!("failed to deserialize response as ProtoJSON: {e}"))
180        })
181    }
182
183    async fn post_streaming<Req>(
184        &self,
185        path: &str,
186        params: &ServiceParams,
187        body: &Req,
188    ) -> Result<BoxStream<'static, Result<StreamResponse, A2AError>>, A2AError>
189    where
190        Req: ProtoJsonPayload,
191    {
192        let payload = protojson_conv::to_value(body).map_err(|e| {
193            A2AError::internal(format!("failed to serialize request as ProtoJSON: {e}"))
194        })?;
195        let resp = self
196            .send(
197                self.build_request(reqwest::Method::POST, path, params)
198                    .header("Accept", "text/event-stream")
199                    .json(&payload),
200            )
201            .await?;
202
203        if !resp.status().is_success() {
204            return Err(Self::into_rest_error(resp).await);
205        }
206
207        let stream = resp.bytes_stream();
208        Ok(crate::jsonrpc::parse_sse_stream_rest(stream))
209    }
210
211    async fn get_streaming(
212        &self,
213        path: &str,
214        params: &ServiceParams,
215    ) -> Result<BoxStream<'static, Result<StreamResponse, A2AError>>, A2AError> {
216        let resp = self
217            .send(
218                self.build_request(reqwest::Method::GET, path, params)
219                    .header("Accept", "text/event-stream"),
220            )
221            .await?;
222
223        if !resp.status().is_success() {
224            return Err(Self::into_rest_error(resp).await);
225        }
226
227        let stream = resp.bytes_stream();
228        Ok(crate::jsonrpc::parse_sse_stream_rest(stream))
229    }
230
231    async fn delete(&self, path: &str, params: &ServiceParams) -> Result<(), A2AError> {
232        let resp = self
233            .send(self.build_request(reqwest::Method::DELETE, path, params))
234            .await?;
235
236        if !resp.status().is_success() {
237            return Err(Self::into_rest_error(resp).await);
238        }
239        Ok(())
240    }
241}
242
243fn parse_rest_error(status: reqwest::StatusCode, body: &str) -> A2AError {
244    let Ok(envelope) = serde_json::from_str::<RestErrorEnvelope>(body) else {
245        return A2AError::internal(format!("HTTP {status}: {body}"));
246    };
247
248    crate::a2a_error_from_details(
249        error_code::INTERNAL_ERROR,
250        envelope.error.message,
251        envelope.error.details,
252    )
253}
254
255fn should_retry_subscribe_with_legacy_path(err: &A2AError) -> bool {
256    if matches!(
257        err.code,
258        error_code::METHOD_NOT_FOUND | error_code::UNSUPPORTED_OPERATION
259    ) {
260        return true;
261    }
262
263    if err.code != error_code::INTERNAL_ERROR {
264        return false;
265    }
266
267    let msg = err.message.to_ascii_lowercase();
268    msg.contains("http 404")
269        || msg.contains("http 405")
270        || msg.contains("not found")
271        || msg.contains("method not allowed")
272}
273
274#[async_trait]
275impl Transport for RestTransport {
276    async fn send_message(
277        &self,
278        params: &ServiceParams,
279        req: &SendMessageRequest,
280    ) -> Result<SendMessageResponse, A2AError> {
281        self.post_json(REST_SEND_MESSAGE_PATH, params, req).await
282    }
283
284    async fn send_streaming_message(
285        &self,
286        params: &ServiceParams,
287        req: &SendMessageRequest,
288    ) -> Result<BoxStream<'static, Result<StreamResponse, A2AError>>, A2AError> {
289        self.post_streaming(REST_STREAM_MESSAGE_PATH, params, req)
290            .await
291    }
292
293    async fn get_task(
294        &self,
295        params: &ServiceParams,
296        req: &GetTaskRequest,
297    ) -> Result<Task, A2AError> {
298        let path = format!("/tasks/{}", req.id);
299        let mut query_parts = Vec::new();
300        if let Some(hl) = req.history_length {
301            query_parts.push(("historyLength".to_string(), hl.to_string()));
302        }
303        self.get_json(&path, params, &query_parts).await
304    }
305
306    async fn list_tasks(
307        &self,
308        params: &ServiceParams,
309        req: &ListTasksRequest,
310    ) -> Result<ListTasksResponse, A2AError> {
311        let mut query_parts = Vec::new();
312        if let Some(ref cid) = req.context_id {
313            query_parts.push(("contextId".to_string(), cid.clone()));
314        }
315        if let Some(ref status) = req.status {
316            let s = serde_json::to_value(status)
317                .ok()
318                .and_then(|v| v.as_str().map(String::from))
319                .unwrap_or_default();
320            query_parts.push(("status".to_string(), s));
321        }
322        if let Some(ps) = req.page_size {
323            query_parts.push(("pageSize".to_string(), ps.to_string()));
324        }
325        if let Some(ref pt) = req.page_token {
326            query_parts.push(("pageToken".to_string(), pt.clone()));
327        }
328        if let Some(hl) = req.history_length {
329            query_parts.push(("historyLength".to_string(), hl.to_string()));
330        }
331        if let Some(ref ts) = req.status_timestamp_after {
332            query_parts.push(("statusTimestampAfter".to_string(), ts.to_rfc3339()));
333        }
334        if let Some(ia) = req.include_artifacts {
335            query_parts.push(("includeArtifacts".to_string(), ia.to_string()));
336        }
337        self.get_json("/tasks", params, &query_parts).await
338    }
339
340    async fn cancel_task(
341        &self,
342        params: &ServiceParams,
343        req: &CancelTaskRequest,
344    ) -> Result<Task, A2AError> {
345        self.post_json(&format!("/tasks/{}:cancel", req.id), params, req)
346            .await
347    }
348
349    async fn subscribe_to_task(
350        &self,
351        params: &ServiceParams,
352        req: &SubscribeToTaskRequest,
353    ) -> Result<BoxStream<'static, Result<StreamResponse, A2AError>>, A2AError> {
354        let canonical_path = format!("/tasks/{}:subscribe", req.id);
355        match self.get_streaming(&canonical_path, params).await {
356            Ok(stream) => Ok(stream),
357            Err(err) if should_retry_subscribe_with_legacy_path(&err) => {
358                self.get_streaming(&format!("/tasks/{}/subscribe", req.id), params)
359                    .await
360            }
361            Err(err) => Err(err),
362        }
363    }
364
365    async fn create_push_config(
366        &self,
367        params: &ServiceParams,
368        req: &TaskPushNotificationConfig,
369    ) -> Result<TaskPushNotificationConfig, A2AError> {
370        let payload = self
371            .post_value(
372                &format!("/tasks/{}/pushNotificationConfigs", req.task_id),
373                params,
374                req,
375            )
376            .await?;
377        deserialize_task_push_notification_config(payload)
378    }
379
380    async fn get_push_config(
381        &self,
382        params: &ServiceParams,
383        req: &GetTaskPushNotificationConfigRequest,
384    ) -> Result<TaskPushNotificationConfig, A2AError> {
385        let payload = self
386            .get_value(
387                &format!("/tasks/{}/pushNotificationConfigs/{}", req.task_id, req.id),
388                params,
389                &[],
390            )
391            .await?;
392        deserialize_task_push_notification_config(payload)
393    }
394
395    async fn list_push_configs(
396        &self,
397        params: &ServiceParams,
398        req: &ListTaskPushNotificationConfigsRequest,
399    ) -> Result<ListTaskPushNotificationConfigsResponse, A2AError> {
400        let mut query_parts = Vec::new();
401        if let Some(page_size) = req.page_size {
402            query_parts.push(("pageSize".to_string(), page_size.to_string()));
403        }
404        if let Some(ref page_token) = req.page_token {
405            query_parts.push(("pageToken".to_string(), page_token.clone()));
406        }
407
408        let payload = self
409            .get_value(
410                &format!("/tasks/{}/pushNotificationConfigs", req.task_id),
411                params,
412                &query_parts,
413            )
414            .await?;
415        deserialize_list_task_push_notification_configs_response(payload)
416    }
417
418    async fn delete_push_config(
419        &self,
420        params: &ServiceParams,
421        req: &DeleteTaskPushNotificationConfigRequest,
422    ) -> Result<(), A2AError> {
423        self.delete(
424            &format!("/tasks/{}/pushNotificationConfigs/{}", req.task_id, req.id),
425            params,
426        )
427        .await
428    }
429
430    async fn get_extended_agent_card(
431        &self,
432        params: &ServiceParams,
433        _req: &GetExtendedAgentCardRequest,
434    ) -> Result<AgentCard, A2AError> {
435        self.get_json(REST_EXTENDED_AGENT_CARD_PATH, params, &[])
436            .await
437    }
438
439    async fn destroy(&self) -> Result<(), A2AError> {
440        Ok(())
441    }
442}
443
444/// Factory for creating [`RestTransport`] instances.
445pub struct RestTransportFactory {
446    client: Client,
447}
448
449impl RestTransportFactory {
450    pub fn new(client: Option<Client>) -> Self {
451        RestTransportFactory {
452            client: client
453                .unwrap_or_else(|| crate::default_reqwest_client(None).expect("default client")),
454        }
455    }
456
457    #[cfg(any(
458        feature = "rustls-tls",
459        feature = "rustls-no-provider",
460        feature = "native-tls"
461    ))]
462    pub fn with_root_certificates_pem(pem: &[u8]) -> Result<Self, A2AError> {
463        Ok(Self {
464            client: crate::default_reqwest_client(Some(pem))?,
465        })
466    }
467}
468
469#[async_trait]
470impl TransportFactory for RestTransportFactory {
471    fn protocol(&self) -> &str {
472        TRANSPORT_PROTOCOL_HTTP_JSON
473    }
474
475    async fn create(
476        &self,
477        _card: &AgentCard,
478        iface: &AgentInterface,
479    ) -> Result<Box<dyn Transport>, A2AError> {
480        Ok(Box::new(RestTransport::new(
481            self.client.clone(),
482            iface.url.clone(),
483        )))
484    }
485}
486
487#[cfg(test)]
488mod tests {
489    use super::*;
490    use futures::StreamExt;
491    use serde_json::json;
492    use tokio::io::{AsyncReadExt, AsyncWriteExt};
493    use tokio::net::{TcpListener, TcpStream};
494
495    async fn read_http_request(socket: &mut TcpStream) -> String {
496        let mut buffer = Vec::new();
497        let mut chunk = [0_u8; 1024];
498
499        loop {
500            let read = socket.read(&mut chunk).await.unwrap();
501            if read == 0 {
502                break;
503            }
504            buffer.extend_from_slice(&chunk[..read]);
505            if buffer.windows(4).any(|window| window == b"\r\n\r\n") {
506                break;
507            }
508        }
509
510        String::from_utf8(buffer).unwrap()
511    }
512
513    async fn spawn_subscribe_fallback_server() -> String {
514        let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
515        let addr = listener.local_addr().unwrap();
516
517        tokio::spawn(async move {
518            let (mut first_socket, _) = listener.accept().await.unwrap();
519            let first_request = read_http_request(&mut first_socket).await;
520            assert!(
521                first_request.starts_with("GET /tasks/task-1:subscribe HTTP/1.1"),
522                "unexpected first request: {first_request}"
523            );
524
525            let first_body = json!({
526                "error": {
527                    "code": 404,
528                    "status": "NOT_FOUND",
529                    "message": "not found",
530                    "details": []
531                }
532            })
533            .to_string();
534            let first_response = format!(
535                "HTTP/1.1 404 Not Found\r\ncontent-type: application/json\r\ncontent-length: {}\r\nconnection: close\r\n\r\n{}",
536                first_body.len(),
537                first_body,
538            );
539            first_socket
540                .write_all(first_response.as_bytes())
541                .await
542                .unwrap();
543
544            let (mut second_socket, _) = listener.accept().await.unwrap();
545            let second_request = read_http_request(&mut second_socket).await;
546            assert!(
547                second_request.starts_with("GET /tasks/task-1/subscribe HTTP/1.1"),
548                "unexpected second request: {second_request}"
549            );
550
551            let status_update = StreamResponse::StatusUpdate(TaskStatusUpdateEvent {
552                task_id: "task-1".into(),
553                context_id: "ctx-1".into(),
554                status: TaskStatus {
555                    state: TaskState::Working,
556                    message: None,
557                    timestamp: None,
558                },
559                metadata: None,
560            });
561            let sse_payload =
562                serde_json::to_string(&protojson_conv::to_value(&status_update).unwrap()).unwrap();
563            let second_body = format!("data: {sse_payload}\n\n");
564            let second_response = format!(
565                "HTTP/1.1 200 OK\r\ncontent-type: text/event-stream\r\ncontent-length: {}\r\ncache-control: no-cache\r\nconnection: close\r\n\r\n{}",
566                second_body.len(),
567                second_body,
568            );
569            second_socket
570                .write_all(second_response.as_bytes())
571                .await
572                .unwrap();
573        });
574
575        format!("http://{addr}")
576    }
577
578    #[test]
579    fn test_rest_transport_new_strips_trailing_slash() {
580        let t = RestTransport::new(
581            crate::default_reqwest_client(None).unwrap(),
582            "http://localhost:8080/".into(),
583        );
584        assert_eq!(t.base_url, "http://localhost:8080");
585    }
586
587    #[test]
588    fn test_rest_transport_new_no_trailing_slash() {
589        let t = RestTransport::new(
590            crate::default_reqwest_client(None).unwrap(),
591            "http://localhost:8080".into(),
592        );
593        assert_eq!(t.base_url, "http://localhost:8080");
594    }
595
596    #[test]
597    fn test_rest_transport_factory_protocol() {
598        let f = RestTransportFactory::new(None);
599        assert_eq!(f.protocol(), "HTTP+JSON");
600    }
601
602    #[tokio::test]
603    async fn test_rest_transport_factory_create() {
604        let f = RestTransportFactory::new(None);
605        let card = AgentCard {
606            name: "Test".into(),
607            description: "Test".into(),
608            version: "1.0".into(),
609            supported_interfaces: vec![],
610            capabilities: AgentCapabilities::default(),
611            default_input_modes: vec!["text/plain".into()],
612            default_output_modes: vec!["text/plain".into()],
613            skills: vec![],
614            provider: None,
615            documentation_url: None,
616            icon_url: None,
617            security_schemes: None,
618            security_requirements: None,
619            signatures: None,
620        };
621        let iface = AgentInterface::new("http://localhost:8080/", "HTTP+JSON");
622        let transport = f.create(&card, &iface).await.unwrap();
623        transport.destroy().await.unwrap();
624    }
625
626    #[test]
627    fn test_build_request_adds_params() {
628        let t = RestTransport::new(
629            crate::default_reqwest_client(None).unwrap(),
630            "http://localhost:8080".into(),
631        );
632        let mut params = ServiceParams::new();
633        params.insert("X-Custom".into(), vec!["val1".into(), "val2".into()]);
634        let builder = t.build_request(reqwest::Method::GET, "/test", &params);
635        let req = builder.build().unwrap();
636        let vals: Vec<_> = req
637            .headers()
638            .get_all("X-Custom")
639            .iter()
640            .map(|v| v.to_str().unwrap().to_string())
641            .collect();
642        assert_eq!(vals, vec!["val1", "val2"]);
643    }
644
645    #[test]
646    fn test_parse_rest_error_preserves_a2a_error_code() {
647        let body = json!({
648            "error": {
649                "code": 404,
650                "status": "NOT_FOUND",
651                "message": "task not found: t1",
652                "details": [
653                    {
654                        "@type": errordetails::ERROR_INFO_TYPE,
655                        "reason": "TASK_NOT_FOUND",
656                        "domain": errordetails::PROTOCOL_DOMAIN,
657                        "metadata": {
658                            "taskId": "t1"
659                        }
660                    },
661                    {
662                        "resource": "task"
663                    }
664                ]
665            }
666        })
667        .to_string();
668
669        let err = parse_rest_error(reqwest::StatusCode::NOT_FOUND, &body);
670
671        assert_eq!(err.code, error_code::TASK_NOT_FOUND);
672        assert_eq!(err.message, "task not found: t1");
673        let details = err.details.expect("expected structured details");
674        assert_eq!(details.len(), 2);
675        assert_eq!(details[0].type_url, errordetails::ERROR_INFO_TYPE);
676        assert_eq!(
677            details[1].value.get("resource"),
678            Some(&Value::String("task".into()))
679        );
680    }
681
682    #[test]
683    fn test_parse_rest_error_accepts_go_reason_aliases() {
684        let body = json!({
685            "error": {
686                "code": 400,
687                "status": "INVALID_ARGUMENT",
688                "message": "incompatible content types",
689                "details": [
690                    {
691                        "@type": errordetails::ERROR_INFO_TYPE,
692                        "reason": "UNSUPPORTED_CONTENT_TYPE",
693                        "domain": errordetails::PROTOCOL_DOMAIN,
694                        "metadata": {}
695                    }
696                ]
697            }
698        })
699        .to_string();
700
701        let err = parse_rest_error(reqwest::StatusCode::BAD_REQUEST, &body);
702        assert_eq!(err.code, error_code::CONTENT_TYPE_NOT_SUPPORTED);
703    }
704
705    #[test]
706    fn test_parse_rest_error_bad_request_fallback() {
707        let body = json!({
708            "error": {
709                "code": 400,
710                "status": "INVALID_ARGUMENT",
711                "message": "invalid request parameters",
712                "details": [
713                    {
714                        "@type": errordetails::BAD_REQUEST_TYPE,
715                        "fieldViolations": [
716                            {
717                                "field": "message.parts",
718                                "description": "At least one part is required"
719                            }
720                        ]
721                    }
722                ]
723            }
724        })
725        .to_string();
726
727        let err = parse_rest_error(reqwest::StatusCode::BAD_REQUEST, &body);
728        assert_eq!(err.code, error_code::INVALID_PARAMS);
729        assert!(
730            err.message
731                .contains("message.parts: At least one part is required")
732        );
733        let details = err.details.expect("expected details");
734        assert_eq!(details.len(), 1);
735        assert_eq!(details[0].type_url, errordetails::BAD_REQUEST_TYPE);
736        let violations = details[0].value.get("fieldViolations").unwrap();
737        assert_eq!(violations[0]["field"], "message.parts");
738    }
739
740    #[test]
741    fn test_parse_rest_error_bad_request_with_error_info_uses_reason() {
742        let body = json!({
743            "error": {
744                "code": 400,
745                "status": "INVALID_ARGUMENT",
746                "message": "bad params",
747                "details": [
748                    {
749                        "@type": errordetails::BAD_REQUEST_TYPE,
750                        "fieldViolations": [
751                            {"field": "task.id", "description": "required"}
752                        ]
753                    },
754                    {
755                        "@type": errordetails::ERROR_INFO_TYPE,
756                        "reason": "INVALID_PARAMS",
757                        "domain": errordetails::PROTOCOL_DOMAIN,
758                        "metadata": {}
759                    }
760                ]
761            }
762        })
763        .to_string();
764
765        let err = parse_rest_error(reqwest::StatusCode::BAD_REQUEST, &body);
766        assert_eq!(err.code, error_code::INVALID_PARAMS);
767    }
768
769    #[test]
770    fn test_should_retry_subscribe_with_legacy_path() {
771        assert!(should_retry_subscribe_with_legacy_path(&A2AError {
772            code: error_code::METHOD_NOT_FOUND,
773            message: "method missing".to_string(),
774            details: None,
775        }));
776
777        assert!(should_retry_subscribe_with_legacy_path(&A2AError {
778            code: error_code::UNSUPPORTED_OPERATION,
779            message: "unsupported".to_string(),
780            details: None,
781        }));
782
783        assert!(should_retry_subscribe_with_legacy_path(&A2AError {
784            code: error_code::INTERNAL_ERROR,
785            message: "HTTP 404: not found".to_string(),
786            details: None,
787        }));
788
789        assert!(!should_retry_subscribe_with_legacy_path(&A2AError {
790            code: error_code::TASK_NOT_FOUND,
791            message: "task not found".to_string(),
792            details: None,
793        }));
794    }
795
796    #[tokio::test]
797    async fn test_subscribe_to_task_falls_back_to_legacy_rest_path() {
798        let base_url = spawn_subscribe_fallback_server().await;
799        let transport = RestTransport::new(crate::default_reqwest_client(None).unwrap(), base_url);
800
801        let mut stream = Transport::subscribe_to_task(
802            &transport,
803            &ServiceParams::new(),
804            &SubscribeToTaskRequest {
805                id: "task-1".into(),
806                tenant: None,
807            },
808        )
809        .await
810        .unwrap();
811
812        let item = stream.next().await.unwrap().unwrap();
813        assert!(matches!(item, StreamResponse::StatusUpdate(_)));
814    }
815
816    #[cfg(any(feature = "rustls-tls", feature = "native-tls"))]
817    #[test]
818    fn test_with_root_certificates_pem_valid() {
819        let pem = crate::test_utils::rcgen_self_signed_ca_pem();
820        let f = RestTransportFactory::with_root_certificates_pem(&pem).unwrap();
821        assert_eq!(f.protocol(), TRANSPORT_PROTOCOL_HTTP_JSON);
822    }
823
824    #[cfg(any(feature = "rustls-tls", feature = "native-tls"))]
825    #[tokio::test]
826    async fn test_with_root_certificates_pem_factory_create() {
827        let pem = crate::test_utils::rcgen_self_signed_ca_pem();
828        let f = RestTransportFactory::with_root_certificates_pem(&pem).unwrap();
829        let card = AgentCard {
830            name: "Test".into(),
831            description: "Test".into(),
832            version: "1.0".into(),
833            supported_interfaces: vec![],
834            capabilities: AgentCapabilities::default(),
835            default_input_modes: vec!["text/plain".into()],
836            default_output_modes: vec!["text/plain".into()],
837            skills: vec![],
838            provider: None,
839            documentation_url: None,
840            icon_url: None,
841            security_schemes: None,
842            security_requirements: None,
843            signatures: None,
844        };
845        let iface = AgentInterface::new("https://localhost:3443/rest", "HTTP+JSON");
846        let transport = f.create(&card, &iface).await.unwrap();
847        transport.destroy().await.unwrap();
848    }
849}
a2a_client/rest.rs

a2a_client/
rest.rs
