a2a_client/
client.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::jsonrpc::methods;
5use a2a::*;
6use async_trait::async_trait;
7use futures::stream::BoxStream;
8use std::borrow::Cow;
9use std::sync::Arc;
10
11use crate::middleware::CallInterceptor;
12use crate::transport::{ServiceParams, Transport};
13
14/// High-level A2A client wrapping a transport with middleware.
15pub struct A2AClient<T: Transport> {
16    transport: T,
17    interceptors: Vec<Arc<dyn CallInterceptor>>,
18    default_params: ServiceParams,
19    tenant: Option<String>,
20}
21
22impl<T: Transport> A2AClient<T> {
23    pub fn new(transport: T) -> Self {
24        let mut default_params = ServiceParams::new();
25        default_params.insert(SVC_PARAM_VERSION.to_string(), vec![VERSION.to_string()]);
26        A2AClient {
27            transport,
28            interceptors: Vec::new(),
29            default_params,
30            tenant: None,
31        }
32    }
33
34    pub fn with_interceptors(mut self, interceptors: Vec<Arc<dyn CallInterceptor>>) -> Self {
35        self.interceptors = interceptors;
36        self
37    }
38
39    /// Set the routing tenant declared by the selected [`AgentInterface`].
40    ///
41    /// A2A §8.3.2 rule 4 requires the `tenant` field of every request to be
42    /// "exactly the value declared in the selected `AgentInterface` entry",
43    /// so this value **replaces** whatever a request carried. The declared
44    /// tenant is the routing identifier for that interface; a caller does
45    /// not get to substitute another one.
46    pub fn with_tenant(mut self, tenant: impl Into<String>) -> Self {
47        self.tenant = Some(tenant.into());
48        self
49    }
50
51    /// The tenant applied to outgoing requests, if the selected interface
52    /// declared one.
53    pub fn tenant(&self) -> Option<&str> {
54        self.tenant.as_deref()
55    }
56
57    /// Apply the declared tenant to a request (§8.3.2 rule 4).
58    ///
59    /// With no declared tenant the request is passed through untouched and
60    /// unclonned — including any tenant the caller set, which is the only
61    /// case where a caller-supplied value is meaningful (connecting to an
62    /// interface directly, with no card to declare one).
63    fn apply_tenant<'a, R: Clone>(
64        &self,
65        req: &'a R,
66        tenant_field: impl FnOnce(&mut R) -> &mut Option<String>,
67    ) -> Cow<'a, R> {
68        let Some(tenant) = &self.tenant else {
69            return Cow::Borrowed(req);
70        };
71        let mut filled = req.clone();
72        *tenant_field(&mut filled) = Some(tenant.clone());
73        Cow::Owned(filled)
74    }
75
76    fn params(&self) -> ServiceParams {
77        self.default_params.clone()
78    }
79
80    async fn apply_before(&self, method: &str) -> Result<ServiceParams, A2AError> {
81        let mut params = self.params();
82        for interceptor in &self.interceptors {
83            interceptor.before(method, &mut params).await?;
84        }
85        Ok(params)
86    }
87
88    async fn apply_after(
89        &self,
90        method: &str,
91        result: &Result<(), A2AError>,
92    ) -> Result<(), A2AError> {
93        for interceptor in self.interceptors.iter().rev() {
94            interceptor.after(method, result).await?;
95        }
96        Ok(())
97    }
98
99    async fn finish_call<R>(
100        &self,
101        method: &str,
102        result: Result<R, A2AError>,
103    ) -> Result<R, A2AError> {
104        let status = result.as_ref().map(|_| ()).map_err(Clone::clone);
105        let after_result = self.apply_after(method, &status).await;
106
107        match (result, after_result) {
108            (Ok(value), Ok(())) => Ok(value),
109            (Err(error), _) => Err(error),
110            (Ok(_), Err(error)) => Err(error),
111        }
112    }
113
114    pub async fn send_message(
115        &self,
116        req: &SendMessageRequest,
117    ) -> Result<SendMessageResponse, A2AError> {
118        let params = self.apply_before(methods::SEND_MESSAGE).await?;
119        let req = self.apply_tenant(req, |r| &mut r.tenant);
120        let result = self.transport.send_message(&params, &req).await;
121        self.finish_call(methods::SEND_MESSAGE, result).await
122    }
123
124    pub async fn send_streaming_message(
125        &self,
126        req: &SendMessageRequest,
127    ) -> Result<BoxStream<'static, Result<StreamResponse, A2AError>>, A2AError> {
128        let params = self.apply_before(methods::SEND_STREAMING_MESSAGE).await?;
129        let req = self.apply_tenant(req, |r| &mut r.tenant);
130        let result = self.transport.send_streaming_message(&params, &req).await;
131        self.finish_call(methods::SEND_STREAMING_MESSAGE, result)
132            .await
133    }
134
135    pub async fn get_task(&self, req: &GetTaskRequest) -> Result<Task, A2AError> {
136        let params = self.apply_before(methods::GET_TASK).await?;
137        let req = self.apply_tenant(req, |r| &mut r.tenant);
138        let result = self.transport.get_task(&params, &req).await;
139        self.finish_call(methods::GET_TASK, result).await
140    }
141
142    pub async fn list_tasks(&self, req: &ListTasksRequest) -> Result<ListTasksResponse, A2AError> {
143        let params = self.apply_before(methods::LIST_TASKS).await?;
144        let req = self.apply_tenant(req, |r| &mut r.tenant);
145        let result = self.transport.list_tasks(&params, &req).await;
146        self.finish_call(methods::LIST_TASKS, result).await
147    }
148
149    pub async fn cancel_task(&self, req: &CancelTaskRequest) -> Result<Task, A2AError> {
150        let params = self.apply_before(methods::CANCEL_TASK).await?;
151        let req = self.apply_tenant(req, |r| &mut r.tenant);
152        let result = self.transport.cancel_task(&params, &req).await;
153        self.finish_call(methods::CANCEL_TASK, result).await
154    }
155
156    pub async fn subscribe_to_task(
157        &self,
158        req: &SubscribeToTaskRequest,
159    ) -> Result<BoxStream<'static, Result<StreamResponse, A2AError>>, A2AError> {
160        let params = self.apply_before(methods::SUBSCRIBE_TO_TASK).await?;
161        let req = self.apply_tenant(req, |r| &mut r.tenant);
162        let result = self.transport.subscribe_to_task(&params, &req).await;
163        self.finish_call(methods::SUBSCRIBE_TO_TASK, result).await
164    }
165
166    pub async fn create_push_config(
167        &self,
168        req: &TaskPushNotificationConfig,
169    ) -> Result<TaskPushNotificationConfig, A2AError> {
170        let params = self.apply_before(methods::CREATE_PUSH_CONFIG).await?;
171        let req = self.apply_tenant(req, |r| &mut r.tenant);
172        let result = self.transport.create_push_config(&params, &req).await;
173        self.finish_call(methods::CREATE_PUSH_CONFIG, result).await
174    }
175
176    pub async fn get_push_config(
177        &self,
178        req: &GetTaskPushNotificationConfigRequest,
179    ) -> Result<TaskPushNotificationConfig, A2AError> {
180        let params = self.apply_before(methods::GET_PUSH_CONFIG).await?;
181        let req = self.apply_tenant(req, |r| &mut r.tenant);
182        let result = self.transport.get_push_config(&params, &req).await;
183        self.finish_call(methods::GET_PUSH_CONFIG, result).await
184    }
185
186    pub async fn list_push_configs(
187        &self,
188        req: &ListTaskPushNotificationConfigsRequest,
189    ) -> Result<ListTaskPushNotificationConfigsResponse, A2AError> {
190        let params = self.apply_before(methods::LIST_PUSH_CONFIGS).await?;
191        let req = self.apply_tenant(req, |r| &mut r.tenant);
192        let result = self.transport.list_push_configs(&params, &req).await;
193        self.finish_call(methods::LIST_PUSH_CONFIGS, result).await
194    }
195
196    pub async fn delete_push_config(
197        &self,
198        req: &DeleteTaskPushNotificationConfigRequest,
199    ) -> Result<(), A2AError> {
200        let params = self.apply_before(methods::DELETE_PUSH_CONFIG).await?;
201        let req = self.apply_tenant(req, |r| &mut r.tenant);
202        let result = self.transport.delete_push_config(&params, &req).await;
203        self.finish_call(methods::DELETE_PUSH_CONFIG, result).await
204    }
205
206    pub async fn get_extended_agent_card(
207        &self,
208        req: &GetExtendedAgentCardRequest,
209    ) -> Result<AgentCard, A2AError> {
210        let params = self.apply_before(methods::GET_EXTENDED_AGENT_CARD).await?;
211        let req = self.apply_tenant(req, |r| &mut r.tenant);
212        let result = self.transport.get_extended_agent_card(&params, &req).await;
213        self.finish_call(methods::GET_EXTENDED_AGENT_CARD, result)
214            .await
215    }
216
217    pub async fn destroy(&self) -> Result<(), A2AError> {
218        self.transport.destroy().await
219    }
220}
221
222/// Convenience trait to extract client results.
223#[async_trait]
224pub trait SendMessageExt {
225    async fn send_text(
226        &self,
227        text: impl Into<String> + Send,
228    ) -> Result<SendMessageResponse, A2AError>;
229}
230
231#[async_trait]
232impl<T: Transport> SendMessageExt for A2AClient<T> {
233    async fn send_text(
234        &self,
235        text: impl Into<String> + Send,
236    ) -> Result<SendMessageResponse, A2AError> {
237        let msg = Message::new(Role::User, vec![Part::text(text)]);
238        let req = SendMessageRequest {
239            message: msg,
240            configuration: None,
241            metadata: None,
242            tenant: None,
243        };
244        self.send_message(&req).await
245    }
246}
247
248#[cfg(test)]
249mod tests {
250    use super::*;
251    use a2a::event::StreamResponse;
252    use futures::stream;
253    use std::sync::Mutex;
254
255    #[derive(Default)]
256    struct MockTransportState {
257        calls: Mutex<Vec<(String, ServiceParams)>>,
258        send_message_error: Mutex<Option<A2AError>>,
259        /// The `tenant` field of each request received, so a test can check
260        /// what actually went out rather than what was passed in.
261        request_tenants: Mutex<Vec<Option<String>>>,
262    }
263
264    /// Mock transport that returns canned responses.
265    struct MockTransport {
266        state: Arc<MockTransportState>,
267    }
268
269    impl MockTransport {
270        fn new() -> (Self, Arc<MockTransportState>) {
271            let state = Arc::new(MockTransportState::default());
272            (
273                MockTransport {
274                    state: state.clone(),
275                },
276                state,
277            )
278        }
279
280        fn record(&self, method: &str, params: &ServiceParams) {
281            self.state
282                .calls
283                .lock()
284                .unwrap()
285                .push((method.to_string(), params.clone()));
286        }
287    }
288
289    #[async_trait]
290    impl Transport for MockTransport {
291        async fn send_message(
292            &self,
293            params: &ServiceParams,
294            req: &SendMessageRequest,
295        ) -> Result<SendMessageResponse, A2AError> {
296            self.record(methods::SEND_MESSAGE, params);
297            self.state
298                .request_tenants
299                .lock()
300                .unwrap()
301                .push(req.tenant.clone());
302            if let Some(error) = self.state.send_message_error.lock().unwrap().clone() {
303                return Err(error);
304            }
305            Ok(SendMessageResponse::Task(Task {
306                id: "t1".into(),
307                context_id: "c1".into(),
308                status: TaskStatus {
309                    state: TaskState::Completed,
310                    message: None,
311                    timestamp: None,
312                },
313                artifacts: None,
314                history: None,
315                metadata: None,
316            }))
317        }
318
319        async fn send_streaming_message(
320            &self,
321            params: &ServiceParams,
322            _req: &SendMessageRequest,
323        ) -> Result<BoxStream<'static, Result<StreamResponse, A2AError>>, A2AError> {
324            self.record(methods::SEND_STREAMING_MESSAGE, params);
325            Ok(Box::pin(stream::once(async {
326                Ok(StreamResponse::StatusUpdate(
327                    a2a::event::TaskStatusUpdateEvent {
328                        task_id: "t1".into(),
329                        context_id: "c1".into(),
330                        status: TaskStatus {
331                            state: TaskState::Working,
332                            message: None,
333                            timestamp: None,
334                        },
335                        metadata: None,
336                    },
337                ))
338            })))
339        }
340
341        async fn get_task(
342            &self,
343            params: &ServiceParams,
344            req: &GetTaskRequest,
345        ) -> Result<Task, A2AError> {
346            self.record(methods::GET_TASK, params);
347            self.state
348                .request_tenants
349                .lock()
350                .unwrap()
351                .push(req.tenant.clone());
352            Ok(Task {
353                id: req.id.clone(),
354                context_id: "c1".into(),
355                status: TaskStatus {
356                    state: TaskState::Completed,
357                    message: None,
358                    timestamp: None,
359                },
360                artifacts: None,
361                history: None,
362                metadata: None,
363            })
364        }
365
366        async fn list_tasks(
367            &self,
368            params: &ServiceParams,
369            _req: &ListTasksRequest,
370        ) -> Result<ListTasksResponse, A2AError> {
371            self.record(methods::LIST_TASKS, params);
372            Ok(ListTasksResponse {
373                tasks: vec![],
374                next_page_token: String::new(),
375                page_size: 0,
376                total_size: 0,
377            })
378        }
379
380        async fn cancel_task(
381            &self,
382            params: &ServiceParams,
383            req: &CancelTaskRequest,
384        ) -> Result<Task, A2AError> {
385            self.record(methods::CANCEL_TASK, params);
386            Ok(Task {
387                id: req.id.clone(),
388                context_id: "c1".into(),
389                status: TaskStatus {
390                    state: TaskState::Canceled,
391                    message: None,
392                    timestamp: None,
393                },
394                artifacts: None,
395                history: None,
396                metadata: None,
397            })
398        }
399
400        async fn subscribe_to_task(
401            &self,
402            params: &ServiceParams,
403            _req: &SubscribeToTaskRequest,
404        ) -> Result<BoxStream<'static, Result<StreamResponse, A2AError>>, A2AError> {
405            self.record(methods::SUBSCRIBE_TO_TASK, params);
406            Ok(Box::pin(stream::empty()))
407        }
408
409        async fn create_push_config(
410            &self,
411            params: &ServiceParams,
412            req: &TaskPushNotificationConfig,
413        ) -> Result<TaskPushNotificationConfig, A2AError> {
414            self.record(methods::CREATE_PUSH_CONFIG, params);
415            Ok(req.clone())
416        }
417
418        async fn get_push_config(
419            &self,
420            params: &ServiceParams,
421            req: &GetTaskPushNotificationConfigRequest,
422        ) -> Result<TaskPushNotificationConfig, A2AError> {
423            self.record(methods::GET_PUSH_CONFIG, params);
424            Ok(TaskPushNotificationConfig {
425                task_id: req.task_id.clone(),
426                url: "http://example.com".into(),
427                id: Some(req.id.clone()),
428                token: None,
429                authentication: None,
430                tenant: None,
431            })
432        }
433
434        async fn list_push_configs(
435            &self,
436            params: &ServiceParams,
437            _req: &ListTaskPushNotificationConfigsRequest,
438        ) -> Result<ListTaskPushNotificationConfigsResponse, A2AError> {
439            self.record(methods::LIST_PUSH_CONFIGS, params);
440            Ok(ListTaskPushNotificationConfigsResponse {
441                configs: vec![],
442                next_page_token: None,
443            })
444        }
445
446        async fn delete_push_config(
447            &self,
448            params: &ServiceParams,
449            _req: &DeleteTaskPushNotificationConfigRequest,
450        ) -> Result<(), A2AError> {
451            self.record(methods::DELETE_PUSH_CONFIG, params);
452            Ok(())
453        }
454
455        async fn get_extended_agent_card(
456            &self,
457            params: &ServiceParams,
458            _req: &GetExtendedAgentCardRequest,
459        ) -> Result<AgentCard, A2AError> {
460            self.record(methods::GET_EXTENDED_AGENT_CARD, params);
461            Ok(AgentCard {
462                name: "Test".into(),
463                description: "Test agent".into(),
464                version: "1.0".into(),
465                supported_interfaces: vec![],
466                capabilities: AgentCapabilities::default(),
467                default_input_modes: vec!["text/plain".into()],
468                default_output_modes: vec!["text/plain".into()],
469                skills: vec![],
470                provider: None,
471                documentation_url: None,
472                icon_url: None,
473                security_schemes: None,
474                security_requirements: None,
475                signatures: None,
476            })
477        }
478
479        async fn destroy(&self) -> Result<(), A2AError> {
480            Ok(())
481        }
482    }
483
484    fn make_client() -> A2AClient<MockTransport> {
485        let (transport, _) = MockTransport::new();
486        A2AClient::new(transport)
487    }
488
489    struct RecordingInterceptor {
490        name: &'static str,
491        events: Arc<Mutex<Vec<String>>>,
492    }
493
494    #[async_trait]
495    impl CallInterceptor for RecordingInterceptor {
496        async fn before(&self, _method: &str, params: &mut ServiceParams) -> Result<(), A2AError> {
497            self.events
498                .lock()
499                .unwrap()
500                .push(format!("before:{}", self.name));
501            params
502                .entry("X-Interceptor".to_string())
503                .or_default()
504                .push(self.name.to_string());
505            Ok(())
506        }
507
508        async fn after(
509            &self,
510            _method: &str,
511            result: &Result<(), A2AError>,
512        ) -> Result<(), A2AError> {
513            let status = if result.is_ok() { "ok" } else { "err" };
514            self.events
515                .lock()
516                .unwrap()
517                .push(format!("after:{}:{status}", self.name));
518            Ok(())
519        }
520    }
521
522    #[test]
523    fn test_new_sets_default_params() {
524        let client = make_client();
525        let params = client.params();
526        assert!(params.contains_key(SVC_PARAM_VERSION));
527    }
528
529    #[test]
530    fn test_with_interceptors() {
531        let client = make_client().with_interceptors(vec![]);
532        assert!(client.interceptors.is_empty());
533    }
534
535    #[tokio::test]
536    async fn test_send_message() {
537        let client = make_client();
538        let req = SendMessageRequest {
539            message: Message::new(Role::User, vec![Part::text("hi")]),
540            configuration: None,
541            metadata: None,
542            tenant: None,
543        };
544        let resp = client.send_message(&req).await.unwrap();
545        assert!(matches!(resp, SendMessageResponse::Task(_)));
546    }
547
548    #[tokio::test]
549    async fn test_send_message_applies_interceptors_and_reverses_after_order() {
550        let (transport, state) = MockTransport::new();
551        let events = Arc::new(Mutex::new(Vec::new()));
552        let client = A2AClient::new(transport).with_interceptors(vec![
553            Arc::new(RecordingInterceptor {
554                name: "first",
555                events: events.clone(),
556            }),
557            Arc::new(RecordingInterceptor {
558                name: "second",
559                events: events.clone(),
560            }),
561        ]);
562
563        let req = SendMessageRequest {
564            message: Message::new(Role::User, vec![Part::text("hi")]),
565            configuration: None,
566            metadata: None,
567            tenant: None,
568        };
569
570        client.send_message(&req).await.unwrap();
571
572        let calls = state.calls.lock().unwrap();
573        let params = &calls[0].1;
574        assert_eq!(
575            params.get("X-Interceptor").unwrap(),
576            &vec!["first".to_string(), "second".to_string()]
577        );
578
579        let events = events.lock().unwrap().clone();
580        assert_eq!(
581            events,
582            vec![
583                "before:first".to_string(),
584                "before:second".to_string(),
585                "after:second:ok".to_string(),
586                "after:first:ok".to_string(),
587            ]
588        );
589    }
590
591    #[tokio::test]
592    async fn test_send_message_preserves_transport_error_after_after_hooks() {
593        let (transport, state) = MockTransport::new();
594        *state.send_message_error.lock().unwrap() = Some(A2AError::internal("boom"));
595        let events = Arc::new(Mutex::new(Vec::new()));
596        let client =
597            A2AClient::new(transport).with_interceptors(vec![Arc::new(RecordingInterceptor {
598                name: "only",
599                events: events.clone(),
600            })]);
601
602        let req = SendMessageRequest {
603            message: Message::new(Role::User, vec![Part::text("hi")]),
604            configuration: None,
605            metadata: None,
606            tenant: None,
607        };
608
609        let err = client.send_message(&req).await.unwrap_err();
610        assert_eq!(err.message, "boom");
611
612        let events = events.lock().unwrap().clone();
613        assert_eq!(
614            events,
615            vec!["before:only".to_string(), "after:only:err".to_string(),]
616        );
617    }
618
619    #[tokio::test]
620    async fn test_send_streaming_message() {
621        use futures::StreamExt;
622        let client = make_client();
623        let req = SendMessageRequest {
624            message: Message::new(Role::User, vec![Part::text("hi")]),
625            configuration: None,
626            metadata: None,
627            tenant: None,
628        };
629        let mut stream = client.send_streaming_message(&req).await.unwrap();
630        let item = stream.next().await.unwrap().unwrap();
631        assert!(matches!(item, StreamResponse::StatusUpdate(_)));
632    }
633
634    #[tokio::test]
635    async fn test_get_task() {
636        let client = make_client();
637        let req = GetTaskRequest {
638            id: "t1".into(),
639            history_length: None,
640            tenant: None,
641        };
642        let task = client.get_task(&req).await.unwrap();
643        assert_eq!(task.id, "t1");
644    }
645
646    #[tokio::test]
647    async fn test_list_tasks() {
648        let client = make_client();
649        let req = ListTasksRequest {
650            context_id: None,
651            status: None,
652            page_size: None,
653            page_token: None,
654            history_length: None,
655            status_timestamp_after: None,
656            include_artifacts: None,
657            tenant: None,
658        };
659        let resp = client.list_tasks(&req).await.unwrap();
660        assert!(resp.tasks.is_empty());
661    }
662
663    #[tokio::test]
664    async fn test_cancel_task() {
665        let client = make_client();
666        let req = CancelTaskRequest {
667            id: "t1".into(),
668            metadata: None,
669            tenant: None,
670        };
671        let task = client.cancel_task(&req).await.unwrap();
672        assert_eq!(task.status.state, TaskState::Canceled);
673    }
674
675    #[tokio::test]
676    async fn test_subscribe_to_task() {
677        let client = make_client();
678        let req = SubscribeToTaskRequest {
679            id: "t1".into(),
680            tenant: None,
681        };
682        let _stream = client.subscribe_to_task(&req).await.unwrap();
683    }
684
685    #[tokio::test]
686    async fn test_create_push_config() {
687        let client = make_client();
688        let req = TaskPushNotificationConfig {
689            task_id: "t1".into(),
690            url: "http://example.com".into(),
691            id: None,
692            token: None,
693            authentication: None,
694            tenant: None,
695        };
696        let resp = client.create_push_config(&req).await.unwrap();
697        assert_eq!(resp.task_id, "t1");
698    }
699
700    #[tokio::test]
701    async fn test_get_push_config() {
702        let client = make_client();
703        let req = GetTaskPushNotificationConfigRequest {
704            task_id: "t1".into(),
705            id: "cfg1".into(),
706            tenant: None,
707        };
708        let resp = client.get_push_config(&req).await.unwrap();
709        assert_eq!(resp.id, Some("cfg1".into()));
710    }
711
712    #[tokio::test]
713    async fn test_list_push_configs() {
714        let client = make_client();
715        let req = ListTaskPushNotificationConfigsRequest {
716            task_id: "t1".into(),
717            page_size: None,
718            page_token: None,
719            tenant: None,
720        };
721        let resp = client.list_push_configs(&req).await.unwrap();
722        assert!(resp.configs.is_empty());
723    }
724
725    #[tokio::test]
726    async fn test_delete_push_config() {
727        let client = make_client();
728        let req = DeleteTaskPushNotificationConfigRequest {
729            task_id: "t1".into(),
730            id: "cfg1".into(),
731            tenant: None,
732        };
733        client.delete_push_config(&req).await.unwrap();
734    }
735
736    #[tokio::test]
737    async fn test_get_extended_agent_card() {
738        let client = make_client();
739        let req = GetExtendedAgentCardRequest { tenant: None };
740        let card = client.get_extended_agent_card(&req).await.unwrap();
741        assert_eq!(card.name, "Test");
742    }
743
744    #[tokio::test]
745    async fn test_destroy() {
746        let client = make_client();
747        client.destroy().await.unwrap();
748    }
749
750    fn send_request() -> SendMessageRequest {
751        SendMessageRequest {
752            message: Message::new(Role::User, vec![Part::text("hi")]),
753            configuration: None,
754            metadata: None,
755            tenant: None,
756        }
757    }
758
759    /// A2A §8.3.2 rule 4: the declared tenant is echoed on every request.
760    #[tokio::test]
761    async fn test_declared_tenant_is_echoed_on_requests() {
762        let (transport, state) = MockTransport::new();
763        let client = A2AClient::new(transport).with_tenant("acme");
764
765        client.send_message(&send_request()).await.unwrap();
766        client
767            .get_task(&GetTaskRequest {
768                id: "t1".into(),
769                history_length: None,
770                tenant: None,
771            })
772            .await
773            .unwrap();
774
775        let tenants = state.request_tenants.lock().unwrap().clone();
776        assert_eq!(
777            tenants,
778            vec![Some("acme".to_string()), Some("acme".to_string())]
779        );
780    }
781
782    /// The rule says "exactly the value declared", so a caller-supplied
783    /// tenant does not displace it — the declared value is the routing
784    /// identifier for that interface.
785    #[tokio::test]
786    async fn test_declared_tenant_replaces_a_caller_supplied_one() {
787        let (transport, state) = MockTransport::new();
788        let client = A2AClient::new(transport).with_tenant("declared");
789
790        let mut req = send_request();
791        req.tenant = Some("caller-chosen".to_string());
792        client.send_message(&req).await.unwrap();
793
794        assert_eq!(
795            state.request_tenants.lock().unwrap().clone(),
796            vec![Some("declared".to_string())]
797        );
798        // The caller's own request value is untouched.
799        assert_eq!(req.tenant.as_deref(), Some("caller-chosen"));
800    }
801
802    /// With no declared tenant the request passes through as written —
803    /// the only case where a caller-supplied tenant is meaningful, such as
804    /// connecting to an interface directly with no card to consult.
805    #[tokio::test]
806    async fn test_without_a_declared_tenant_the_request_is_untouched() {
807        let (transport, state) = MockTransport::new();
808        let client = A2AClient::new(transport);
809        assert_eq!(client.tenant(), None);
810
811        let mut req = send_request();
812        req.tenant = Some("caller-chosen".to_string());
813        client.send_message(&req).await.unwrap();
814        client.send_message(&send_request()).await.unwrap();
815
816        assert_eq!(
817            state.request_tenants.lock().unwrap().clone(),
818            vec![Some("caller-chosen".to_string()), None]
819        );
820    }
821}
a2a_client/client.rs

a2a_client/
client.rs
