a2a_client/
rest.rs

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

a2a_client/
rest.rs
