Skip to content

feat(rest): serve and send the /{tenant} path prefix #468

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
muscariello merged 3 commits into from
Nov 4, 2026
Merged

feat(rest): serve and send the /{tenant} path prefix #468

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
5f311f2
feat(rest): serve and send the /{tenant} path prefix
gulldan Nov 3, 2026
ed08239
Merge branch 'main' into fix/rest-tenant-prefix
muscariello Nov 3, 2026
0cffc6e
Merge branch 'main' into fix/rest-tenant-prefix
muscariello Nov 4, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
121 changes: 100 additions & 21 deletions a2a-client/src/rest.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -33,6 +33,15 @@ fn segment(id: &str) -> PercentEncode<'_> {
utf8_percent_encode(id, PATH_SEGMENT)
}

/// The proto's `/{tenant}/…` binding of `path` when the request names a
/// tenant (§8.3.2): on HTTP+JSON the tenant travels in the path.
fn tenant_path(tenant: Option<&str>, path: &str) -> String {
match tenant {
Some(tenant) if !tenant.is_empty() => format!("/{}{path}", segment(tenant)),
_ => path.to_owned(),
}
}

#[derive(Debug, Deserialize)]
struct RestErrorEnvelope {
error: RestErrorStatus,
Expand Down Expand Up @@ -291,24 +300,28 @@ impl Transport for RestTransport {
params: &ServiceParams,
req: &SendMessageRequest,
) -> Result<SendMessageResponse, A2AError> {
self.post_json(REST_SEND_MESSAGE_PATH, params, req).await
let path = tenant_path(req.tenant.as_deref(), REST_SEND_MESSAGE_PATH);
self.post_json(&path, params, req).await
}

async fn send_streaming_message(
&self,
params: &ServiceParams,
req: &SendMessageRequest,
) -> Result<BoxStream<'static, Result<StreamResponse, A2AError>>, A2AError> {
self.post_streaming(REST_STREAM_MESSAGE_PATH, params, req)
.await
let path = tenant_path(req.tenant.as_deref(), REST_STREAM_MESSAGE_PATH);
self.post_streaming(&path, params, req).await
}

async fn get_task(
&self,
params: &ServiceParams,
req: &GetTaskRequest,
) -> Result<Task, A2AError> {
let path = format!("/tasks/{}", segment(&req.id));
let path = tenant_path(
req.tenant.as_deref(),
&format!("/tasks/{}", segment(&req.id)),
);
let mut query_parts = Vec::new();
if let Some(hl) = req.history_length {
query_parts.push(("historyLength".to_string(), hl.to_string()));
Expand Down Expand Up @@ -347,15 +360,17 @@ impl Transport for RestTransport {
if let Some(ia) = req.include_artifacts {
query_parts.push(("includeArtifacts".to_string(), ia.to_string()));
}
self.get_json("/tasks", params, &query_parts).await
let path = tenant_path(req.tenant.as_deref(), "/tasks");
self.get_json(&path, params, &query_parts).await
}

async fn cancel_task(
&self,
params: &ServiceParams,
req: &CancelTaskRequest,
) -> Result<Task, A2AError> {
self.post_json(&format!("/tasks/{}:cancel", segment(&req.id)), params, req)
let path = format!("/tasks/{}:cancel", segment(&req.id));
self.post_json(&tenant_path(req.tenant.as_deref(), &path), params, req)
.await
}

Expand All @@ -364,11 +379,13 @@ impl Transport for RestTransport {
params: &ServiceParams,
req: &SubscribeToTaskRequest,
) -> Result<BoxStream<'static, Result<StreamResponse, A2AError>>, A2AError> {
let canonical_path = format!("/tasks/{}:subscribe", segment(&req.id));
let tenant = req.tenant.as_deref();
let canonical_path = tenant_path(tenant, &format!("/tasks/{}:subscribe", segment(&req.id)));
match self.get_streaming(&canonical_path, params).await {
Ok(stream) => Ok(stream),
Err(err) if should_retry_subscribe_with_legacy_path(&err) => {
self.get_streaming(&format!("/tasks/{}/subscribe", segment(&req.id)), params)
let legacy_path = format!("/tasks/{}/subscribe", segment(&req.id));
self.get_streaming(&tenant_path(tenant, &legacy_path), params)
.await
}
Err(err) => Err(err),
Expand All @@ -382,7 +399,10 @@ impl Transport for RestTransport {
) -> Result<TaskPushNotificationConfig, A2AError> {
let payload = self
.post_value(
&format!("/tasks/{}/pushNotificationConfigs", segment(&req.task_id)),
&tenant_path(
req.tenant.as_deref(),
&format!("/tasks/{}/pushNotificationConfigs", segment(&req.task_id)),
),
params,
req,
)
Expand All @@ -397,10 +417,13 @@ impl Transport for RestTransport {
) -> Result<TaskPushNotificationConfig, A2AError> {
let payload = self
.get_value(
&format!(
"/tasks/{}/pushNotificationConfigs/{}",
segment(&req.task_id),
segment(&req.id)
&tenant_path(
req.tenant.as_deref(),
&format!(
"/tasks/{}/pushNotificationConfigs/{}",
segment(&req.task_id),
segment(&req.id)
),
),
params,
&[],
Expand All @@ -424,7 +447,10 @@ impl Transport for RestTransport {

let payload = self
.get_value(
&format!("/tasks/{}/pushNotificationConfigs", segment(&req.task_id)),
&tenant_path(
req.tenant.as_deref(),
&format!("/tasks/{}/pushNotificationConfigs", segment(&req.task_id)),
),
params,
&query_parts,
)
Expand All @@ -438,10 +464,13 @@ impl Transport for RestTransport {
req: &DeleteTaskPushNotificationConfigRequest,
) -> Result<(), A2AError> {
self.delete(
&format!(
"/tasks/{}/pushNotificationConfigs/{}",
segment(&req.task_id),
segment(&req.id)
&tenant_path(
req.tenant.as_deref(),
&format!(
"/tasks/{}/pushNotificationConfigs/{}",
segment(&req.task_id),
segment(&req.id)
),
),
params,
)
Expand All @@ -451,10 +480,10 @@ impl Transport for RestTransport {
async fn get_extended_agent_card(
&self,
params: &ServiceParams,
_req: &GetExtendedAgentCardRequest,
req: &GetExtendedAgentCardRequest,
) -> Result<AgentCard, A2AError> {
self.get_json(REST_EXTENDED_AGENT_CARD_PATH, params, &[])
.await
let path = tenant_path(req.tenant.as_deref(), REST_EXTENDED_AGENT_CARD_PATH);
self.get_json(&path, params, &[]).await
}

async fn destroy(&self) -> Result<(), A2AError> {
Expand Down Expand Up @@ -657,6 +686,56 @@ mod tests {
);
}

/// A request that names a tenant goes to the proto's `/{tenant}/…`
/// binding, its tenant percent-encoded like an id.
#[tokio::test]
async fn test_requests_with_a_tenant_use_the_tenant_path() {
let params = ServiceParams::new();
let tenant = Some("acme eu".to_string());

let (url, line) = spawn_request_line_server().await;
let transport = RestTransport::new(crate::default_reqwest_client(None).unwrap(), url);
let req = GetTaskRequest {
id: "t1".into(),
history_length: None,
tenant: tenant.clone(),
};
let _ = transport.get_task(&params, &req).await;
assert_eq!(line.await.unwrap(), "GET /a2a/acme%20eu/tasks/t1 HTTP/1.1");

let (url, line) = spawn_request_line_server().await;
let transport = RestTransport::new(crate::default_reqwest_client(None).unwrap(), url);
let req = DeleteTaskPushNotificationConfigRequest {
task_id: "t1".into(),
id: "c1".into(),
tenant: tenant.clone(),
};
let _ = transport.delete_push_config(&params, &req).await;
assert_eq!(
line.await.unwrap(),
"DELETE /a2a/acme%20eu/tasks/t1/pushNotificationConfigs/c1 HTTP/1.1"
);

let (url, line) = spawn_request_line_server().await;
let transport = RestTransport::new(crate::default_reqwest_client(None).unwrap(), url);
let req = GetExtendedAgentCardRequest { tenant };
let _ = transport.get_extended_agent_card(&params, &req).await;
assert_eq!(
line.await.unwrap(),
"GET /a2a/acme%20eu/extendedAgentCard HTTP/1.1"
);

let (url, line) = spawn_request_line_server().await;
let transport = RestTransport::new(crate::default_reqwest_client(None).unwrap(), url);
let req = GetTaskRequest {
id: "t1".into(),
history_length: None,
tenant: Some(String::new()),
};
let _ = transport.get_task(&params, &req).await;
assert_eq!(line.await.unwrap(), "GET /a2a/tasks/t1 HTTP/1.1");
}

#[test]
fn test_rest_transport_new_strips_trailing_slash() {
let t = RestTransport::new(
Expand Down
55 changes: 33 additions & 22 deletions a2a-server/src/lib.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -85,19 +85,30 @@ pub(crate) mod test_util {
/// stub responses so tests don't depend on `DefaultRequestHandler` behavior.
pub(crate) struct CapturingHandler {
captured: CapturedParams,
tenants: Mutex<HashMap<&'static str, Option<String>>>,
}

impl CapturingHandler {
pub fn new() -> (Arc<Self>, CapturedParams) {
let captured = Arc::new(Mutex::new(HashMap::new()));
let handler = Arc::new(CapturingHandler {
captured: captured.clone(),
tenants: Mutex::new(HashMap::new()),
});
(handler, captured)
}

fn record(&self, method: &'static str, params: &ServiceParams) {
fn record(&self, method: &'static str, params: &ServiceParams, tenant: Option<&str>) {
self.captured.lock().unwrap().insert(method, params.clone());
self.tenants
.lock()
.unwrap()
.insert(method, tenant.map(str::to_owned));
}

/// The tenant of `method`'s last call; `None` if it was not called since.
pub fn take_tenant(&self, method: &str) -> Option<Option<String>> {
self.tenants.lock().unwrap().remove(method)
}
}

Expand All @@ -106,36 +117,36 @@ pub(crate) mod test_util {
async fn send_message(
&self,
params: &ServiceParams,
_req: SendMessageRequest,
req: SendMessageRequest,
) -> Result<SendMessageResponse, A2AError> {
self.record("send_message", params);
self.record("send_message", params, req.tenant.as_deref());
Err(A2AError::internal("stub"))
}

async fn send_streaming_message(
&self,
params: &ServiceParams,
_req: SendMessageRequest,
req: SendMessageRequest,
) -> Result<BoxStream<'static, Result<StreamResponse, A2AError>>, A2AError> {
self.record("send_streaming_message", params);
self.record("send_streaming_message", params, req.tenant.as_deref());
Ok(Box::pin(futures::stream::empty()))
}

async fn get_task(
&self,
params: &ServiceParams,
_req: GetTaskRequest,
req: GetTaskRequest,
) -> Result<Task, A2AError> {
self.record("get_task", params);
self.record("get_task", params, req.tenant.as_deref());
Err(A2AError::task_not_found("stub"))
}

async fn list_tasks(
&self,
params: &ServiceParams,
_req: ListTasksRequest,
req: ListTasksRequest,
) -> Result<ListTasksResponse, A2AError> {
self.record("list_tasks", params);
self.record("list_tasks", params, req.tenant.as_deref());
Ok(ListTasksResponse {
tasks: Vec::new(),
next_page_token: String::new(),
Expand All @@ -147,18 +158,18 @@ pub(crate) mod test_util {
async fn cancel_task(
&self,
params: &ServiceParams,
_req: CancelTaskRequest,
req: CancelTaskRequest,
) -> Result<Task, A2AError> {
self.record("cancel_task", params);
self.record("cancel_task", params, req.tenant.as_deref());
Err(A2AError::task_not_found("stub"))
}

async fn subscribe_to_task(
&self,
params: &ServiceParams,
_req: SubscribeToTaskRequest,
req: SubscribeToTaskRequest,
) -> Result<BoxStream<'static, Result<StreamResponse, A2AError>>, A2AError> {
self.record("subscribe_to_task", params);
self.record("subscribe_to_task", params, req.tenant.as_deref());
Ok(Box::pin(futures::stream::empty()))
}

Expand All @@ -167,25 +178,25 @@ pub(crate) mod test_util {
params: &ServiceParams,
req: TaskPushNotificationConfig,
) -> Result<TaskPushNotificationConfig, A2AError> {
self.record("create_push_config", params);
self.record("create_push_config", params, req.tenant.as_deref());
Ok(req)
}

async fn get_push_config(
&self,
params: &ServiceParams,
_req: GetTaskPushNotificationConfigRequest,
req: GetTaskPushNotificationConfigRequest,
) -> Result<TaskPushNotificationConfig, A2AError> {
self.record("get_push_config", params);
self.record("get_push_config", params, req.tenant.as_deref());
Err(A2AError::internal("stub"))
}

async fn list_push_configs(
&self,
params: &ServiceParams,
_req: ListTaskPushNotificationConfigsRequest,
req: ListTaskPushNotificationConfigsRequest,
) -> Result<ListTaskPushNotificationConfigsResponse, A2AError> {
self.record("list_push_configs", params);
self.record("list_push_configs", params, req.tenant.as_deref());
Ok(ListTaskPushNotificationConfigsResponse {
configs: Vec::new(),
next_page_token: None,
Expand All @@ -195,18 +206,18 @@ pub(crate) mod test_util {
async fn delete_push_config(
&self,
params: &ServiceParams,
_req: DeleteTaskPushNotificationConfigRequest,
req: DeleteTaskPushNotificationConfigRequest,
) -> Result<(), A2AError> {
self.record("delete_push_config", params);
self.record("delete_push_config", params, req.tenant.as_deref());
Ok(())
}

async fn get_extended_agent_card(
&self,
params: &ServiceParams,
_req: GetExtendedAgentCardRequest,
req: GetExtendedAgentCardRequest,
) -> Result<AgentCard, A2AError> {
self.record("get_extended_agent_card", params);
self.record("get_extended_agent_card", params, req.tenant.as_deref());
Err(A2AError::internal("stub"))
}
}
Expand Down
Loading
Loading