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!(err.code, error_code::METHOD_NOT_FOUND) {
257        return true;
258    }
259
260    if err.code != error_code::INTERNAL_ERROR {
261        return false;
262    }
263
264    let msg = err.message.to_ascii_lowercase();
265    msg.contains("http 404")
266        || msg.contains("http 405")
267        || msg.contains("not found")
268        || msg.contains("method not allowed")
269}
270
271#[async_trait]
272impl Transport for RestTransport {
273    async fn send_message(
274        &self,
275        params: &ServiceParams,
276        req: &SendMessageRequest,
277    ) -> Result<SendMessageResponse, A2AError> {
278        self.post_json(REST_SEND_MESSAGE_PATH, params, req).await
279    }
280
281    async fn send_streaming_message(
282        &self,
283        params: &ServiceParams,
284        req: &SendMessageRequest,
285    ) -> Result<BoxStream<'static, Result<StreamResponse, A2AError>>, A2AError> {
286        self.post_streaming(REST_STREAM_MESSAGE_PATH, params, req)
287            .await
288    }
289
290    async fn get_task(
291        &self,
292        params: &ServiceParams,
293        req: &GetTaskRequest,
294    ) -> Result<Task, A2AError> {
295        let path = format!("/tasks/{}", req.id);
296        let mut query_parts = Vec::new();
297        if let Some(hl) = req.history_length {
298            query_parts.push(("historyLength".to_string(), hl.to_string()));
299        }
300        self.get_json(&path, params, &query_parts).await
301    }
302
303    async fn list_tasks(
304        &self,
305        params: &ServiceParams,
306        req: &ListTasksRequest,
307    ) -> Result<ListTasksResponse, A2AError> {
308        let mut query_parts = Vec::new();
309        if let Some(ref cid) = req.context_id {
310            query_parts.push(("contextId".to_string(), cid.clone()));
311        }
312        if let Some(ref status) = req.status {
313            let s = serde_json::to_value(status)
314                .ok()
315                .and_then(|v| v.as_str().map(String::from))
316                .unwrap_or_default();
317            query_parts.push(("status".to_string(), s));
318        }
319        if let Some(ps) = req.page_size {
320            query_parts.push(("pageSize".to_string(), ps.to_string()));
321        }
322        if let Some(ref pt) = req.page_token {
323            query_parts.push(("pageToken".to_string(), pt.clone()));
324        }
325        if let Some(hl) = req.history_length {
326            query_parts.push(("historyLength".to_string(), hl.to_string()));
327        }
328        if let Some(ref ts) = req.status_timestamp_after {
329            query_parts.push(("statusTimestampAfter".to_string(), ts.to_rfc3339()));
330        }
331        if let Some(ia) = req.include_artifacts {
332            query_parts.push(("includeArtifacts".to_string(), ia.to_string()));
333        }
334        self.get_json("/tasks", params, &query_parts).await
335    }
336
337    async fn cancel_task(
338        &self,
339        params: &ServiceParams,
340        req: &CancelTaskRequest,
341    ) -> Result<Task, A2AError> {
342        self.post_json(&format!("/tasks/{}:cancel", req.id), params, req)
343            .await
344    }
345
346    async fn subscribe_to_task(
347        &self,
348        params: &ServiceParams,
349        req: &SubscribeToTaskRequest,
350    ) -> Result<BoxStream<'static, Result<StreamResponse, A2AError>>, A2AError> {
351        let canonical_path = format!("/tasks/{}:subscribe", req.id);
352        match self.get_streaming(&canonical_path, params).await {
353            Ok(stream) => Ok(stream),
354            Err(err) if should_retry_subscribe_with_legacy_path(&err) => {
355                self.get_streaming(&format!("/tasks/{}/subscribe", req.id), params)
356                    .await
357            }
358            Err(err) => Err(err),
359        }
360    }
361
362    async fn create_push_config(
363        &self,
364        params: &ServiceParams,
365        req: &TaskPushNotificationConfig,
366    ) -> Result<TaskPushNotificationConfig, A2AError> {
367        let payload = self
368            .post_value(
369                &format!("/tasks/{}/pushNotificationConfigs", req.task_id),
370                params,
371                req,
372            )
373            .await?;
374        deserialize_task_push_notification_config(payload)
375    }
376
377    async fn get_push_config(
378        &self,
379        params: &ServiceParams,
380        req: &GetTaskPushNotificationConfigRequest,
381    ) -> Result<TaskPushNotificationConfig, A2AError> {
382        let payload = self
383            .get_value(
384                &format!("/tasks/{}/pushNotificationConfigs/{}", req.task_id, req.id),
385                params,
386                &[],
387            )
388            .await?;
389        deserialize_task_push_notification_config(payload)
390    }
391
392    async fn list_push_configs(
393        &self,
394        params: &ServiceParams,
395        req: &ListTaskPushNotificationConfigsRequest,
396    ) -> Result<ListTaskPushNotificationConfigsResponse, A2AError> {
397        let mut query_parts = Vec::new();
398        if let Some(page_size) = req.page_size {
399            query_parts.push(("pageSize".to_string(), page_size.to_string()));
400        }
401        if let Some(ref page_token) = req.page_token {
402            query_parts.push(("pageToken".to_string(), page_token.clone()));
403        }
404
405        let payload = self
406            .get_value(
407                &format!("/tasks/{}/pushNotificationConfigs", req.task_id),
408                params,
409                &query_parts,
410            )
411            .await?;
412        deserialize_list_task_push_notification_configs_response(payload)
413    }
414
415    async fn delete_push_config(
416        &self,
417        params: &ServiceParams,
418        req: &DeleteTaskPushNotificationConfigRequest,
419    ) -> Result<(), A2AError> {
420        self.delete(
421            &format!("/tasks/{}/pushNotificationConfigs/{}", req.task_id, req.id),
422            params,
423        )
424        .await
425    }
426
427    async fn get_extended_agent_card(
428        &self,
429        params: &ServiceParams,
430        _req: &GetExtendedAgentCardRequest,
431    ) -> Result<AgentCard, A2AError> {
432        self.get_json(REST_EXTENDED_AGENT_CARD_PATH, params, &[])
433            .await
434    }
435
436    async fn destroy(&self) -> Result<(), A2AError> {
437        Ok(())
438    }
439}
440
441/// Factory for creating [`RestTransport`] instances.
442pub struct RestTransportFactory {
443    client: Client,
444}
445
446impl RestTransportFactory {
447    pub fn new(client: Option<Client>) -> Self {
448        RestTransportFactory {
449            client: client
450                .unwrap_or_else(|| crate::default_reqwest_client(None).expect("default client")),
451        }
452    }
453
454    #[cfg(any(
455        feature = "rustls-tls",
456        feature = "rustls-no-provider",
457        feature = "native-tls"
458    ))]
459    pub fn with_root_certificates_pem(pem: &[u8]) -> Result<Self, A2AError> {
460        Ok(Self {
461            client: crate::default_reqwest_client(Some(pem))?,
462        })
463    }
464}
465
466#[async_trait]
467impl TransportFactory for RestTransportFactory {
468    fn protocol(&self) -> &str {
469        TRANSPORT_PROTOCOL_HTTP_JSON
470    }
471
472    async fn create(
473        &self,
474        _card: &AgentCard,
475        iface: &AgentInterface,
476    ) -> Result<Box<dyn Transport>, A2AError> {
477        Ok(Box::new(RestTransport::new(
478            self.client.clone(),
479            iface.url.clone(),
480        )))
481    }
482}
483
484#[cfg(test)]
485mod tests {
486    use super::*;
487    use futures::StreamExt;
488    use serde_json::json;
489    use tokio::io::{AsyncReadExt, AsyncWriteExt};
490    use tokio::net::{TcpListener, TcpStream};
491
492    async fn read_http_request(socket: &mut TcpStream) -> String {
493        let mut buffer = Vec::new();
494        let mut chunk = [0_u8; 1024];
495
496        loop {
497            let read = socket.read(&mut chunk).await.unwrap();
498            if read == 0 {
499                break;
500            }
501            buffer.extend_from_slice(&chunk[..read]);
502            if buffer.windows(4).any(|window| window == b"\r\n\r\n") {
503                break;
504            }
505        }
506
507        String::from_utf8(buffer).unwrap()
508    }
509
510    async fn spawn_subscribe_fallback_server() -> String {
511        let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
512        let addr = listener.local_addr().unwrap();
513
514        tokio::spawn(async move {
515            let (mut first_socket, _) = listener.accept().await.unwrap();
516            let first_request = read_http_request(&mut first_socket).await;
517            assert!(
518                first_request.starts_with("GET /tasks/task-1:subscribe HTTP/1.1"),
519                "unexpected first request: {first_request}"
520            );
521
522            let first_body = json!({
523                "error": {
524                    "code": 404,
525                    "status": "NOT_FOUND",
526                    "message": "not found",
527                    "details": []
528                }
529            })
530            .to_string();
531            let first_response = format!(
532                "HTTP/1.1 404 Not Found\r\ncontent-type: application/json\r\ncontent-length: {}\r\nconnection: close\r\n\r\n{}",
533                first_body.len(),
534                first_body,
535            );
536            first_socket
537                .write_all(first_response.as_bytes())
538                .await
539                .unwrap();
540
541            let (mut second_socket, _) = listener.accept().await.unwrap();
542            let second_request = read_http_request(&mut second_socket).await;
543            assert!(
544                second_request.starts_with("GET /tasks/task-1/subscribe HTTP/1.1"),
545                "unexpected second request: {second_request}"
546            );
547
548            let status_update = StreamResponse::StatusUpdate(TaskStatusUpdateEvent {
549                task_id: "task-1".into(),
550                context_id: "ctx-1".into(),
551                status: TaskStatus {
552                    state: TaskState::Working,
553                    message: None,
554                    timestamp: None,
555                },
556                metadata: None,
557            });
558            let sse_payload =
559                serde_json::to_string(&protojson_conv::to_value(&status_update).unwrap()).unwrap();
560            let second_body = format!("data: {sse_payload}\n\n");
561            let second_response = format!(
562                "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{}",
563                second_body.len(),
564                second_body,
565            );
566            second_socket
567                .write_all(second_response.as_bytes())
568                .await
569                .unwrap();
570        });
571
572        format!("http://{addr}")
573    }
574
575    #[test]
576    fn test_rest_transport_new_strips_trailing_slash() {
577        let t = RestTransport::new(
578            crate::default_reqwest_client(None).unwrap(),
579            "http://localhost:8080/".into(),
580        );
581        assert_eq!(t.base_url, "http://localhost:8080");
582    }
583
584    #[test]
585    fn test_rest_transport_new_no_trailing_slash() {
586        let t = RestTransport::new(
587            crate::default_reqwest_client(None).unwrap(),
588            "http://localhost:8080".into(),
589        );
590        assert_eq!(t.base_url, "http://localhost:8080");
591    }
592
593    #[test]
594    fn test_rest_transport_factory_protocol() {
595        let f = RestTransportFactory::new(None);
596        assert_eq!(f.protocol(), "HTTP+JSON");
597    }
598
599    #[tokio::test]
600    async fn test_rest_transport_factory_create() {
601        let f = RestTransportFactory::new(None);
602        let card = AgentCard {
603            name: "Test".into(),
604            description: "Test".into(),
605            version: "1.0".into(),
606            supported_interfaces: vec![],
607            capabilities: AgentCapabilities::default(),
608            default_input_modes: vec!["text/plain".into()],
609            default_output_modes: vec!["text/plain".into()],
610            skills: vec![],
611            provider: None,
612            documentation_url: None,
613            icon_url: None,
614            security_schemes: None,
615            security_requirements: None,
616            signatures: None,
617        };
618        let iface = AgentInterface::new("http://localhost:8080/", "HTTP+JSON");
619        let transport = f.create(&card, &iface).await.unwrap();
620        transport.destroy().await.unwrap();
621    }
622
623    #[test]
624    fn test_build_request_adds_params() {
625        let t = RestTransport::new(
626            crate::default_reqwest_client(None).unwrap(),
627            "http://localhost:8080".into(),
628        );
629        let mut params = ServiceParams::new();
630        params.insert("X-Custom".into(), vec!["val1".into(), "val2".into()]);
631        let builder = t.build_request(reqwest::Method::GET, "/test", &params);
632        let req = builder.build().unwrap();
633        let vals: Vec<_> = req
634            .headers()
635            .get_all("X-Custom")
636            .iter()
637            .map(|v| v.to_str().unwrap().to_string())
638            .collect();
639        assert_eq!(vals, vec!["val1", "val2"]);
640    }
641
642    #[test]
643    fn test_parse_rest_error_preserves_a2a_error_code() {
644        let body = json!({
645            "error": {
646                "code": 404,
647                "status": "NOT_FOUND",
648                "message": "task not found: t1",
649                "details": [
650                    {
651                        "@type": errordetails::ERROR_INFO_TYPE,
652                        "reason": "TASK_NOT_FOUND",
653                        "domain": errordetails::PROTOCOL_DOMAIN,
654                        "metadata": {
655                            "taskId": "t1"
656                        }
657                    },
658                    {
659                        "resource": "task"
660                    }
661                ]
662            }
663        })
664        .to_string();
665
666        let err = parse_rest_error(reqwest::StatusCode::NOT_FOUND, &body);
667
668        assert_eq!(err.code, error_code::TASK_NOT_FOUND);
669        assert_eq!(err.message, "task not found: t1");
670        let details = err.details.expect("expected structured details");
671        assert_eq!(details.len(), 2);
672        assert_eq!(details[0].type_url, errordetails::ERROR_INFO_TYPE);
673        assert_eq!(
674            details[1].value.get("resource"),
675            Some(&Value::String("task".into()))
676        );
677    }
678
679    #[test]
680    fn test_parse_rest_error_accepts_go_reason_aliases() {
681        let body = json!({
682            "error": {
683                "code": 400,
684                "status": "INVALID_ARGUMENT",
685                "message": "incompatible content types",
686                "details": [
687                    {
688                        "@type": errordetails::ERROR_INFO_TYPE,
689                        "reason": "UNSUPPORTED_CONTENT_TYPE",
690                        "domain": errordetails::PROTOCOL_DOMAIN,
691                        "metadata": {}
692                    }
693                ]
694            }
695        })
696        .to_string();
697
698        let err = parse_rest_error(reqwest::StatusCode::BAD_REQUEST, &body);
699        assert_eq!(err.code, error_code::CONTENT_TYPE_NOT_SUPPORTED);
700    }
701
702    #[test]
703    fn test_parse_rest_error_bad_request_fallback() {
704        let body = json!({
705            "error": {
706                "code": 400,
707                "status": "INVALID_ARGUMENT",
708                "message": "invalid request parameters",
709                "details": [
710                    {
711                        "@type": errordetails::BAD_REQUEST_TYPE,
712                        "fieldViolations": [
713                            {
714                                "field": "message.parts",
715                                "description": "At least one part is required"
716                            }
717                        ]
718                    }
719                ]
720            }
721        })
722        .to_string();
723
724        let err = parse_rest_error(reqwest::StatusCode::BAD_REQUEST, &body);
725        assert_eq!(err.code, error_code::INVALID_PARAMS);
726        assert!(
727            err.message
728                .contains("message.parts: At least one part is required")
729        );
730        let details = err.details.expect("expected details");
731        assert_eq!(details.len(), 1);
732        assert_eq!(details[0].type_url, errordetails::BAD_REQUEST_TYPE);
733        let violations = details[0].value.get("fieldViolations").unwrap();
734        assert_eq!(violations[0]["field"], "message.parts");
735    }
736
737    #[test]
738    fn test_parse_rest_error_bad_request_with_error_info_uses_reason() {
739        let body = json!({
740            "error": {
741                "code": 400,
742                "status": "INVALID_ARGUMENT",
743                "message": "bad params",
744                "details": [
745                    {
746                        "@type": errordetails::BAD_REQUEST_TYPE,
747                        "fieldViolations": [
748                            {"field": "task.id", "description": "required"}
749                        ]
750                    },
751                    {
752                        "@type": errordetails::ERROR_INFO_TYPE,
753                        "reason": "INVALID_PARAMS",
754                        "domain": errordetails::PROTOCOL_DOMAIN,
755                        "metadata": {}
756                    }
757                ]
758            }
759        })
760        .to_string();
761
762        let err = parse_rest_error(reqwest::StatusCode::BAD_REQUEST, &body);
763        assert_eq!(err.code, error_code::INVALID_PARAMS);
764    }
765
766    #[test]
767    fn test_should_retry_subscribe_with_legacy_path() {
768        assert!(should_retry_subscribe_with_legacy_path(&A2AError {
769            code: error_code::METHOD_NOT_FOUND,
770            message: "method missing".to_string(),
771            details: None,
772        }));
773
774        assert!(!should_retry_subscribe_with_legacy_path(&A2AError {
775            code: error_code::UNSUPPORTED_OPERATION,
776            message: "unsupported".to_string(),
777            details: None,
778        }));
779
780        assert!(should_retry_subscribe_with_legacy_path(&A2AError {
781            code: error_code::INTERNAL_ERROR,
782            message: "HTTP 404: not found".to_string(),
783            details: None,
784        }));
785
786        assert!(!should_retry_subscribe_with_legacy_path(&A2AError {
787            code: error_code::TASK_NOT_FOUND,
788            message: "task not found".to_string(),
789            details: None,
790        }));
791    }
792
793    #[tokio::test]
794    async fn test_subscribe_to_task_falls_back_to_legacy_rest_path() {
795        let base_url = spawn_subscribe_fallback_server().await;
796        let transport = RestTransport::new(crate::default_reqwest_client(None).unwrap(), base_url);
797
798        let mut stream = Transport::subscribe_to_task(
799            &transport,
800            &ServiceParams::new(),
801            &SubscribeToTaskRequest {
802                id: "task-1".into(),
803                tenant: None,
804            },
805        )
806        .await
807        .unwrap();
808
809        let item = stream.next().await.unwrap().unwrap();
810        assert!(matches!(item, StreamResponse::StatusUpdate(_)));
811    }
812
813    #[cfg(any(feature = "rustls-tls", feature = "native-tls"))]
814    #[test]
815    fn test_with_root_certificates_pem_valid() {
816        let pem = crate::test_utils::rcgen_self_signed_ca_pem();
817        let f = RestTransportFactory::with_root_certificates_pem(&pem).unwrap();
818        assert_eq!(f.protocol(), TRANSPORT_PROTOCOL_HTTP_JSON);
819    }
820
821    #[cfg(any(feature = "rustls-tls", feature = "native-tls"))]
822    #[tokio::test]
823    async fn test_with_root_certificates_pem_factory_create() {
824        let pem = crate::test_utils::rcgen_self_signed_ca_pem();
825        let f = RestTransportFactory::with_root_certificates_pem(&pem).unwrap();
826        let card = AgentCard {
827            name: "Test".into(),
828            description: "Test".into(),
829            version: "1.0".into(),
830            supported_interfaces: vec![],
831            capabilities: AgentCapabilities::default(),
832            default_input_modes: vec!["text/plain".into()],
833            default_output_modes: vec!["text/plain".into()],
834            skills: vec![],
835            provider: None,
836            documentation_url: None,
837            icon_url: None,
838            security_schemes: None,
839            security_requirements: None,
840            signatures: None,
841        };
842        let iface = AgentInterface::new("https://localhost:3443/rest", "HTTP+JSON");
843        let transport = f.create(&card, &iface).await.unwrap();
844        transport.destroy().await.unwrap();
845    }
846}
a2a_client/rest.rs

a2a_client/
rest.rs
