From 1942f9bfe95400ed1ccc85161cb3c70301e54a42 Mon Sep 17 00:00:00 2001 From: Paul Flynn Date: Wed, 10 Jun 2026 19:22:31 -0400 Subject: [PATCH] =?UTF-8?q?fix(a2a-client):=20echo=20selected=20interface?= =?UTF-8?q?=20tenant=20on=20requests=20(spec=20=C2=A78.3.2)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Spec §8.3.2 rule 4 requires clients to set the tenant field in every request message to the value declared in the selected AgentInterface (omitted when not declared). The client had no mechanism for this: all request types carry tenant: Option, but nothing filled it from the agent card, so the tenant was never transmitted. Give A2AClient an optional tenant configured via with_tenant(); every request method now fills the request's tenant field from it when the caller left it unset, cloning only when a fill is needed. An explicitly set request tenant is never overridden. A2AClientFactory:: create_from_card propagates the selected interface's declared tenant to the constructed client, treating an empty string as "not declared" (proto3 default). Found by cross-SDK interop testing (a2a-conformance). --- a2a-client/src/client.rs | 272 +++++++++++++++++++++++++++++++++----- a2a-client/src/factory.rs | 68 ++++++++-- 2 files changed, 295 insertions(+), 45 deletions(-) diff --git a/a2a-client/src/client.rs b/a2a-client/src/client.rs index 43d7b10..bb6b7c7 100644 --- a/a2a-client/src/client.rs +++ b/a2a-client/src/client.rs @@ -4,6 +4,7 @@ use a2a::jsonrpc::methods; use a2a::*; use async_trait::async_trait; use futures::stream::BoxStream; +use std::borrow::Cow; use std::sync::Arc; use crate::middleware::CallInterceptor; @@ -14,6 +15,7 @@ pub struct A2AClient { transport: T, interceptors: Vec>, default_params: ServiceParams, + tenant: Option, } impl A2AClient { @@ -24,6 +26,7 @@ impl A2AClient { transport, interceptors: Vec::new(), default_params, + tenant: None, } } @@ -32,6 +35,40 @@ impl A2AClient { self } + /// Set the tenant declared by the selected [`AgentInterface`]. + /// + /// When set, the tenant is filled into every outgoing request whose + /// `tenant` field is unset, as required by spec §8.3.2 (rule 4). A + /// tenant explicitly set on a request is never overridden. + pub fn with_tenant(mut self, tenant: impl Into) -> Self { + self.tenant = Some(tenant.into()); + self + } + + /// The tenant filled into outgoing requests, if configured. + pub fn tenant(&self) -> Option<&str> { + self.tenant.as_deref() + } + + /// Fill the request's `tenant` field from the client's configured tenant + /// (spec §8.3.2 rule 4), without overriding an explicitly set tenant. + fn apply_tenant<'a, R: Clone>( + &self, + req: &'a R, + tenant_field: impl FnOnce(&mut R) -> &mut Option, + ) -> Cow<'a, R> { + let Some(tenant) = &self.tenant else { + return Cow::Borrowed(req); + }; + let mut filled = req.clone(); + let field = tenant_field(&mut filled); + if field.is_some() { + return Cow::Borrowed(req); + } + *field = Some(tenant.clone()); + Cow::Owned(filled) + } + fn params(&self) -> ServiceParams { self.default_params.clone() } @@ -75,7 +112,8 @@ impl A2AClient { req: &SendMessageRequest, ) -> Result { let params = self.apply_before(methods::SEND_MESSAGE).await?; - let result = self.transport.send_message(¶ms, req).await; + let req = self.apply_tenant(req, |r| &mut r.tenant); + let result = self.transport.send_message(¶ms, &req).await; self.finish_call(methods::SEND_MESSAGE, result).await } @@ -84,26 +122,30 @@ impl A2AClient { req: &SendMessageRequest, ) -> Result>, A2AError> { let params = self.apply_before(methods::SEND_STREAMING_MESSAGE).await?; - let result = self.transport.send_streaming_message(¶ms, req).await; + let req = self.apply_tenant(req, |r| &mut r.tenant); + let result = self.transport.send_streaming_message(¶ms, &req).await; self.finish_call(methods::SEND_STREAMING_MESSAGE, result) .await } pub async fn get_task(&self, req: &GetTaskRequest) -> Result { let params = self.apply_before(methods::GET_TASK).await?; - let result = self.transport.get_task(¶ms, req).await; + let req = self.apply_tenant(req, |r| &mut r.tenant); + let result = self.transport.get_task(¶ms, &req).await; self.finish_call(methods::GET_TASK, result).await } pub async fn list_tasks(&self, req: &ListTasksRequest) -> Result { let params = self.apply_before(methods::LIST_TASKS).await?; - let result = self.transport.list_tasks(¶ms, req).await; + let req = self.apply_tenant(req, |r| &mut r.tenant); + let result = self.transport.list_tasks(¶ms, &req).await; self.finish_call(methods::LIST_TASKS, result).await } pub async fn cancel_task(&self, req: &CancelTaskRequest) -> Result { let params = self.apply_before(methods::CANCEL_TASK).await?; - let result = self.transport.cancel_task(¶ms, req).await; + let req = self.apply_tenant(req, |r| &mut r.tenant); + let result = self.transport.cancel_task(¶ms, &req).await; self.finish_call(methods::CANCEL_TASK, result).await } @@ -112,7 +154,8 @@ impl A2AClient { req: &SubscribeToTaskRequest, ) -> Result>, A2AError> { let params = self.apply_before(methods::SUBSCRIBE_TO_TASK).await?; - let result = self.transport.subscribe_to_task(¶ms, req).await; + let req = self.apply_tenant(req, |r| &mut r.tenant); + let result = self.transport.subscribe_to_task(¶ms, &req).await; self.finish_call(methods::SUBSCRIBE_TO_TASK, result).await } @@ -121,7 +164,8 @@ impl A2AClient { req: &TaskPushNotificationConfig, ) -> Result { let params = self.apply_before(methods::CREATE_PUSH_CONFIG).await?; - let result = self.transport.create_push_config(¶ms, req).await; + let req = self.apply_tenant(req, |r| &mut r.tenant); + let result = self.transport.create_push_config(¶ms, &req).await; self.finish_call(methods::CREATE_PUSH_CONFIG, result).await } @@ -130,7 +174,8 @@ impl A2AClient { req: &GetTaskPushNotificationConfigRequest, ) -> Result { let params = self.apply_before(methods::GET_PUSH_CONFIG).await?; - let result = self.transport.get_push_config(¶ms, req).await; + let req = self.apply_tenant(req, |r| &mut r.tenant); + let result = self.transport.get_push_config(¶ms, &req).await; self.finish_call(methods::GET_PUSH_CONFIG, result).await } @@ -139,7 +184,8 @@ impl A2AClient { req: &ListTaskPushNotificationConfigsRequest, ) -> Result { let params = self.apply_before(methods::LIST_PUSH_CONFIGS).await?; - let result = self.transport.list_push_configs(¶ms, req).await; + let req = self.apply_tenant(req, |r| &mut r.tenant); + let result = self.transport.list_push_configs(¶ms, &req).await; self.finish_call(methods::LIST_PUSH_CONFIGS, result).await } @@ -148,7 +194,8 @@ impl A2AClient { req: &DeleteTaskPushNotificationConfigRequest, ) -> Result<(), A2AError> { let params = self.apply_before(methods::DELETE_PUSH_CONFIG).await?; - let result = self.transport.delete_push_config(¶ms, req).await; + let req = self.apply_tenant(req, |r| &mut r.tenant); + let result = self.transport.delete_push_config(¶ms, &req).await; self.finish_call(methods::DELETE_PUSH_CONFIG, result).await } @@ -157,7 +204,8 @@ impl A2AClient { req: &GetExtendedAgentCardRequest, ) -> Result { let params = self.apply_before(methods::GET_EXTENDED_AGENT_CARD).await?; - let result = self.transport.get_extended_agent_card(¶ms, req).await; + let req = self.apply_tenant(req, |r| &mut r.tenant); + let result = self.transport.get_extended_agent_card(¶ms, &req).await; self.finish_call(methods::GET_EXTENDED_AGENT_CARD, result) .await } @@ -202,7 +250,7 @@ mod tests { #[derive(Default)] struct MockTransportState { - calls: Mutex>, + calls: Mutex)>>, send_message_error: Mutex>, } @@ -222,12 +270,12 @@ mod tests { ) } - fn record(&self, method: &str, params: &ServiceParams) { - self.state - .calls - .lock() - .unwrap() - .push((method.to_string(), params.clone())); + fn record(&self, method: &str, params: &ServiceParams, tenant: &Option) { + self.state.calls.lock().unwrap().push(( + method.to_string(), + params.clone(), + tenant.clone(), + )); } } @@ -236,9 +284,9 @@ mod tests { async fn send_message( &self, params: &ServiceParams, - _req: &SendMessageRequest, + req: &SendMessageRequest, ) -> Result { - self.record(methods::SEND_MESSAGE, params); + self.record(methods::SEND_MESSAGE, params, &req.tenant); if let Some(error) = self.state.send_message_error.lock().unwrap().clone() { return Err(error); } @@ -259,9 +307,9 @@ mod tests { async fn send_streaming_message( &self, params: &ServiceParams, - _req: &SendMessageRequest, + req: &SendMessageRequest, ) -> Result>, A2AError> { - self.record(methods::SEND_STREAMING_MESSAGE, params); + self.record(methods::SEND_STREAMING_MESSAGE, params, &req.tenant); Ok(Box::pin(stream::once(async { Ok(StreamResponse::StatusUpdate( a2a::event::TaskStatusUpdateEvent { @@ -283,7 +331,7 @@ mod tests { params: &ServiceParams, req: &GetTaskRequest, ) -> Result { - self.record(methods::GET_TASK, params); + self.record(methods::GET_TASK, params, &req.tenant); Ok(Task { id: req.id.clone(), context_id: "c1".into(), @@ -301,9 +349,9 @@ mod tests { async fn list_tasks( &self, params: &ServiceParams, - _req: &ListTasksRequest, + req: &ListTasksRequest, ) -> Result { - self.record(methods::LIST_TASKS, params); + self.record(methods::LIST_TASKS, params, &req.tenant); Ok(ListTasksResponse { tasks: vec![], next_page_token: String::new(), @@ -317,7 +365,7 @@ mod tests { params: &ServiceParams, req: &CancelTaskRequest, ) -> Result { - self.record(methods::CANCEL_TASK, params); + self.record(methods::CANCEL_TASK, params, &req.tenant); Ok(Task { id: req.id.clone(), context_id: "c1".into(), @@ -335,9 +383,9 @@ mod tests { async fn subscribe_to_task( &self, params: &ServiceParams, - _req: &SubscribeToTaskRequest, + req: &SubscribeToTaskRequest, ) -> Result>, A2AError> { - self.record(methods::SUBSCRIBE_TO_TASK, params); + self.record(methods::SUBSCRIBE_TO_TASK, params, &req.tenant); Ok(Box::pin(stream::empty())) } @@ -346,7 +394,7 @@ mod tests { params: &ServiceParams, req: &TaskPushNotificationConfig, ) -> Result { - self.record(methods::CREATE_PUSH_CONFIG, params); + self.record(methods::CREATE_PUSH_CONFIG, params, &req.tenant); Ok(req.clone()) } @@ -355,7 +403,7 @@ mod tests { params: &ServiceParams, req: &GetTaskPushNotificationConfigRequest, ) -> Result { - self.record(methods::GET_PUSH_CONFIG, params); + self.record(methods::GET_PUSH_CONFIG, params, &req.tenant); Ok(TaskPushNotificationConfig { task_id: req.task_id.clone(), url: "http://example.com".into(), @@ -369,9 +417,9 @@ mod tests { async fn list_push_configs( &self, params: &ServiceParams, - _req: &ListTaskPushNotificationConfigsRequest, + req: &ListTaskPushNotificationConfigsRequest, ) -> Result { - self.record(methods::LIST_PUSH_CONFIGS, params); + self.record(methods::LIST_PUSH_CONFIGS, params, &req.tenant); Ok(ListTaskPushNotificationConfigsResponse { configs: vec![], next_page_token: None, @@ -381,18 +429,18 @@ mod tests { async fn delete_push_config( &self, params: &ServiceParams, - _req: &DeleteTaskPushNotificationConfigRequest, + req: &DeleteTaskPushNotificationConfigRequest, ) -> Result<(), A2AError> { - self.record(methods::DELETE_PUSH_CONFIG, params); + self.record(methods::DELETE_PUSH_CONFIG, params, &req.tenant); Ok(()) } async fn get_extended_agent_card( &self, params: &ServiceParams, - _req: &GetExtendedAgentCardRequest, + req: &GetExtendedAgentCardRequest, ) -> Result { - self.record(methods::GET_EXTENDED_AGENT_CARD, params); + self.record(methods::GET_EXTENDED_AGENT_CARD, params, &req.tenant); Ok(AgentCard { name: "Test".into(), description: "Test agent".into(), @@ -681,4 +729,156 @@ mod tests { let client = make_client(); client.destroy().await.unwrap(); } + + /// Spec §8.3.2 rule 4: a client configured with the selected interface's + /// tenant must fill it into every outgoing request whose tenant is unset. + #[tokio::test] + async fn test_with_tenant_fills_unset_request_tenant_on_all_methods() { + let (transport, state) = MockTransport::new(); + let client = A2AClient::new(transport).with_tenant("tenant-1"); + assert_eq!(client.tenant(), Some("tenant-1")); + + client + .send_message(&SendMessageRequest { + message: Message::new(Role::User, vec![Part::text("hi")]), + configuration: None, + metadata: None, + tenant: None, + }) + .await + .unwrap(); + let _stream = client + .send_streaming_message(&SendMessageRequest { + message: Message::new(Role::User, vec![Part::text("hi")]), + configuration: None, + metadata: None, + tenant: None, + }) + .await + .unwrap(); + client + .get_task(&GetTaskRequest { + id: "t1".into(), + history_length: None, + tenant: None, + }) + .await + .unwrap(); + client + .list_tasks(&ListTasksRequest { + context_id: None, + status: None, + page_size: None, + page_token: None, + history_length: None, + status_timestamp_after: None, + include_artifacts: None, + tenant: None, + }) + .await + .unwrap(); + client + .cancel_task(&CancelTaskRequest { + id: "t1".into(), + metadata: None, + tenant: None, + }) + .await + .unwrap(); + let _stream = client + .subscribe_to_task(&SubscribeToTaskRequest { + id: "t1".into(), + tenant: None, + }) + .await + .unwrap(); + client + .create_push_config(&TaskPushNotificationConfig { + task_id: "t1".into(), + url: "http://example.com".into(), + id: None, + token: None, + authentication: None, + tenant: None, + }) + .await + .unwrap(); + client + .get_push_config(&GetTaskPushNotificationConfigRequest { + task_id: "t1".into(), + id: "cfg1".into(), + tenant: None, + }) + .await + .unwrap(); + client + .list_push_configs(&ListTaskPushNotificationConfigsRequest { + task_id: "t1".into(), + page_size: None, + page_token: None, + tenant: None, + }) + .await + .unwrap(); + client + .delete_push_config(&DeleteTaskPushNotificationConfigRequest { + task_id: "t1".into(), + id: "cfg1".into(), + tenant: None, + }) + .await + .unwrap(); + client + .get_extended_agent_card(&GetExtendedAgentCardRequest { tenant: None }) + .await + .unwrap(); + + let calls = state.calls.lock().unwrap(); + assert_eq!(calls.len(), 11); + for (method, _, tenant) in calls.iter() { + assert_eq!( + tenant.as_deref(), + Some("tenant-1"), + "request for {method} must carry the client tenant", + ); + } + } + + /// A tenant explicitly set on a request must never be overridden by the + /// client's configured tenant. + #[tokio::test] + async fn test_explicit_request_tenant_is_not_overridden() { + let (transport, state) = MockTransport::new(); + let client = A2AClient::new(transport).with_tenant("tenant-1"); + + let req = SendMessageRequest { + message: Message::new(Role::User, vec![Part::text("hi")]), + configuration: None, + metadata: None, + tenant: Some("explicit-tenant".into()), + }; + client.send_message(&req).await.unwrap(); + + let calls = state.calls.lock().unwrap(); + assert_eq!(calls[0].2.as_deref(), Some("explicit-tenant")); + } + + /// Without a configured client tenant, requests are forwarded unchanged. + #[tokio::test] + async fn test_no_client_tenant_leaves_request_tenant_unset() { + let (transport, state) = MockTransport::new(); + let client = A2AClient::new(transport); + assert_eq!(client.tenant(), None); + + let req = SendMessageRequest { + message: Message::new(Role::User, vec![Part::text("hi")]), + configuration: None, + metadata: None, + tenant: None, + }; + client.send_message(&req).await.unwrap(); + + let calls = state.calls.lock().unwrap(); + assert_eq!(calls[0].2, None); + } } diff --git a/a2a-client/src/factory.rs b/a2a-client/src/factory.rs index 146cd53..dd4fd9f 100644 --- a/a2a-client/src/factory.rs +++ b/a2a-client/src/factory.rs @@ -100,9 +100,15 @@ impl A2AClientFactory { for (_prio, iface, factory) in &candidates { match factory.create(card, iface).await { Ok(transport) => { - return Ok( - A2AClient::new(transport).with_interceptors(self.interceptors.clone()) - ); + let mut client = + A2AClient::new(transport).with_interceptors(self.interceptors.clone()); + // Spec §8.3.2 rule 4: echo the tenant declared by the + // selected interface on every request. An empty string + // means "not declared" (proto3 default). + if let Some(tenant) = iface.tenant.as_deref().filter(|t| !t.is_empty()) { + client = client.with_tenant(tenant); + } + return Ok(client); } Err(e) => { tracing::debug!( @@ -244,14 +250,12 @@ mod tests { assert_eq!(factory.interceptors.len(), 1); } - #[tokio::test] - async fn test_create_from_card_no_matching_transport() { - let factory = A2AClientFactory::builder().no_defaults().build(); - let card = AgentCard { + fn card_with_interface(iface: AgentInterface) -> AgentCard { + AgentCard { name: "test".into(), description: "test agent".into(), version: "1.0".into(), - supported_interfaces: vec![AgentInterface::new("http://localhost", "unknown")], + supported_interfaces: vec![iface], capabilities: AgentCapabilities::default(), default_input_modes: vec!["text".into()], default_output_modes: vec!["text".into()], @@ -262,8 +266,54 @@ mod tests { security_schemes: None, security_requirements: None, signatures: None, - }; + } + } + + #[tokio::test] + async fn test_create_from_card_no_matching_transport() { + let factory = A2AClientFactory::builder().no_defaults().build(); + let card = card_with_interface(AgentInterface::new("http://localhost", "unknown")); let result = factory.create_from_card(&card).await; assert!(result.is_err()); } + + /// Spec §8.3.2 rule 4: the tenant declared by the selected interface is + /// propagated to the constructed client. + #[tokio::test] + async fn test_create_from_card_propagates_interface_tenant() { + let factory = A2AClientFactory::builder().build(); + let mut iface = AgentInterface::new("http://localhost:8080", TRANSPORT_PROTOCOL_JSONRPC); + iface.tenant = Some("tenant-1".into()); + let client = factory + .create_from_card(&card_with_interface(iface)) + .await + .unwrap(); + assert_eq!(client.tenant(), Some("tenant-1")); + } + + /// An interface without a declared tenant leaves the client tenant unset. + #[tokio::test] + async fn test_create_from_card_without_interface_tenant() { + let factory = A2AClientFactory::builder().build(); + let iface = AgentInterface::new("http://localhost:8080", TRANSPORT_PROTOCOL_JSONRPC); + let client = factory + .create_from_card(&card_with_interface(iface)) + .await + .unwrap(); + assert_eq!(client.tenant(), None); + } + + /// An empty-string tenant means "not declared" (proto3 default) and must + /// not be echoed. + #[tokio::test] + async fn test_create_from_card_treats_empty_tenant_as_unset() { + let factory = A2AClientFactory::builder().build(); + let mut iface = AgentInterface::new("http://localhost:8080", TRANSPORT_PROTOCOL_JSONRPC); + iface.tenant = Some(String::new()); + let client = factory + .create_from_card(&card_with_interface(iface)) + .await + .unwrap(); + assert_eq!(client.tenant(), None); + } }