a2a_client/
client.rs

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

a2a_client/
client.rs
