diff --git a/src/client/api.rs b/src/client/api.rs index 01f4ae6..4fa7231 100644 --- a/src/client/api.rs +++ b/src/client/api.rs @@ -10,6 +10,7 @@ use serde::Serialize; use serde::de::DeserializeOwned; use crate::A2AError; +use crate::error::ProblemDetails; use crate::jsonrpc::{ CONTENT_TYPE_NOT_SUPPORTED, EXTENDED_AGENT_CARD_NOT_CONFIGURED, EXTENSION_SUPPORT_REQUIRED, INTERNAL_ERROR, INVALID_AGENT_RESPONSE, INVALID_PARAMS, INVALID_REQUEST, JSONRPC_VERSION, @@ -490,7 +491,11 @@ impl A2AClient { .map_err(|error| A2AError::InvalidAgentResponse(error.to_string())); } - if let Ok(error) = serde_json::from_slice::(&bytes) { + if let Ok(problem) = serde_json::from_slice::(&bytes) { + return Err(problem.to_a2a_error()); + } + + if let Ok(error) = serde_json::from_slice::(&bytes) { return Err(map_jsonrpc_error(error.error)); } @@ -507,7 +512,11 @@ impl A2AClient { let status = response.status(); if !status.is_success() { let bytes = response.bytes().await?; - if let Ok(error) = serde_json::from_slice::(&bytes) { + if let Ok(problem) = serde_json::from_slice::(&bytes) { + return Err(problem.to_a2a_error()); + } + + if let Ok(error) = serde_json::from_slice::(&bytes) { return Err(map_jsonrpc_error(error.error)); } @@ -534,7 +543,17 @@ fn select_transport( base_url: &Url, interfaces: &[AgentInterface], ) -> Result { + let mut advertised_versions = Vec::new(); + for interface in interfaces { + if !interface + .protocol_version + .eq_ignore_ascii_case(PROTOCOL_VERSION) + { + advertised_versions.push(interface.protocol_version.clone()); + continue; + } + if interface.protocol_binding.eq_ignore_ascii_case("JSONRPC") { return resolve_interface_url(base_url, &interface.url).map(TransportEndpoint::JsonRpc); } @@ -546,6 +565,16 @@ fn select_transport( } } + if !advertised_versions.is_empty() { + advertised_versions.sort(); + advertised_versions.dedup(); + return Err(A2AError::VersionNotSupported(format!( + "client supports A2A-Version {}, agent advertised {}", + PROTOCOL_VERSION, + advertised_versions.join(", ") + ))); + } + Err(A2AError::InvalidAgentResponse( "agent card does not advertise a supported interface".to_owned(), )) @@ -555,16 +584,37 @@ fn select_http_json_transport( base_url: &Url, interfaces: &[AgentInterface], ) -> Result { - interfaces - .iter() - .find(|interface| interface.protocol_binding.eq_ignore_ascii_case("HTTP+JSON")) - .ok_or_else(|| { - A2AError::InvalidAgentResponse( - "agent card does not advertise an HTTP+JSON interface".to_owned(), - ) - }) - .and_then(|interface| resolve_interface_url(base_url, &interface.url)) - .map(ensure_trailing_slash) + let mut advertised_versions = Vec::new(); + + for interface in interfaces { + if !interface.protocol_binding.eq_ignore_ascii_case("HTTP+JSON") { + continue; + } + + if !interface + .protocol_version + .eq_ignore_ascii_case(PROTOCOL_VERSION) + { + advertised_versions.push(interface.protocol_version.clone()); + continue; + } + + return resolve_interface_url(base_url, &interface.url).map(ensure_trailing_slash); + } + + if !advertised_versions.is_empty() { + advertised_versions.sort(); + advertised_versions.dedup(); + return Err(A2AError::VersionNotSupported(format!( + "client supports A2A-Version {}, agent advertised HTTP+JSON {}", + PROTOCOL_VERSION, + advertised_versions.join(", ") + ))); + } + + Err(A2AError::InvalidAgentResponse( + "agent card does not advertise an HTTP+JSON interface".to_owned(), + )) } fn rest_url(base_url: &Url, tenant: Option<&str>, segments: &[&str]) -> Result { @@ -586,6 +636,10 @@ fn rest_url(base_url: &Url, tenant: Option<&str>, segments: &[&str]) -> Result A2AError { + if let Some(info) = error.first_error_info() { + return A2AError::from_error_info(error.code, &error.message, Some(&info)); + } + let detail = error .data .as_ref() @@ -748,7 +802,7 @@ struct ListTaskPushNotificationConfigQuery { } #[derive(serde::Deserialize)] -struct RestErrorEnvelope { +struct LegacyRestErrorEnvelope { error: JsonRpcError, } @@ -856,4 +910,22 @@ mod tests { } } } + + #[test] + fn map_jsonrpc_error_prefers_structured_error_info() { + let mapped = map_jsonrpc_error(JsonRpcError { + code: TASK_NOT_FOUND, + message: "fallback message".to_owned(), + data: Some(serde_json::json!({ + "@type": "type.googleapis.com/google.rpc.ErrorInfo", + "reason": "TASK_NOT_FOUND", + "domain": "a2a-protocol.org", + "metadata": { + "taskId": "task-123" + } + })), + }); + + assert!(matches!(mapped, A2AError::TaskNotFound(value) if value == "task-123")); + } } diff --git a/src/error.rs b/src/error.rs index 6d05cd0..7993df2 100644 --- a/src/error.rs +++ b/src/error.rs @@ -1,10 +1,56 @@ +use std::collections::BTreeMap; + use http::StatusCode; +use serde::{Deserialize, Serialize}; use serde_json::Value; use thiserror::Error; use crate::jsonrpc; use crate::jsonrpc::JsonRpcError; +/// Type URL used for structured `ErrorInfo` entries. +pub const ERROR_INFO_TYPE_URL: &str = "type.googleapis.com/google.rpc.ErrorInfo"; +/// Domain used for SDK-generated structured error details. +pub const ERROR_INFO_DOMAIN: &str = "a2a-protocol.org"; + +/// Structured protocol error detail modeled after `google.rpc.ErrorInfo`. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +pub struct ErrorInfo { + /// Type URL identifying the detail payload. + #[serde(rename = "@type", default = "error_info_type_url")] + pub type_url: String, + /// Stable machine-readable reason string. + pub reason: String, + /// Domain that defined the reason. + pub domain: String, + /// Additional structured metadata for the error. + #[serde(default, skip_serializing_if = "BTreeMap::is_empty")] + pub metadata: BTreeMap, +} + +/// Structured HTTP error payload using RFC 9457-style problem details. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +pub struct ProblemDetails { + /// Stable type URI for the problem kind. + #[serde(rename = "type")] + pub type_url: String, + /// Short human-readable problem title. + pub title: String, + /// HTTP status code. + pub status: u16, + /// Human-readable error detail message. + pub detail: String, + /// Optional stable machine-readable reason string. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub reason: Option, + /// Optional domain associated with the reason. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub domain: Option, + /// Additional structured problem metadata. + #[serde(default, flatten, skip_serializing_if = "BTreeMap::is_empty")] + pub extensions: BTreeMap, +} + /// Unified error type for A2A protocol, HTTP, and serialization failures. #[derive(Debug, Error)] pub enum A2AError { @@ -60,6 +106,28 @@ pub enum A2AError { } impl A2AError { + /// Return the stable structured reason name for this error. + pub fn reason(&self) -> &'static str { + match self { + Self::TaskNotFound(_) => "TASK_NOT_FOUND", + Self::TaskNotCancelable(_) => "TASK_NOT_CANCELABLE", + Self::PushNotificationNotSupported(_) => "PUSH_NOTIFICATION_NOT_SUPPORTED", + Self::UnsupportedOperation(_) => "UNSUPPORTED_OPERATION", + Self::ContentTypeNotSupported(_) => "CONTENT_TYPE_NOT_SUPPORTED", + Self::InvalidAgentResponse(_) => "INVALID_AGENT_RESPONSE", + Self::ExtendedAgentCardNotConfigured(_) => "EXTENDED_AGENT_CARD_NOT_CONFIGURED", + Self::ExtensionSupportRequired(_) => "EXTENSION_SUPPORT_REQUIRED", + Self::VersionNotSupported(_) => "VERSION_NOT_SUPPORTED", + Self::ParseError(_) => "PARSE_ERROR", + Self::InvalidRequest(_) => "INVALID_REQUEST", + Self::MethodNotFound(_) => "METHOD_NOT_FOUND", + Self::InvalidParams(_) => "INVALID_PARAMS", + Self::Internal(_) | Self::Serialization(_) => "INTERNAL", + #[cfg(feature = "client")] + Self::Http(_) => "HTTP", + } + } + /// Return the JSON-RPC error code associated with this error. pub fn code(&self) -> i32 { match self { @@ -88,7 +156,27 @@ impl A2AError { JsonRpcError { code: self.code(), message: self.to_string(), - data: self.data(), + data: Some( + serde_json::to_value(self.to_error_info()).expect("error details should serialize"), + ), + } + } + + /// Convert this error into an RFC 9457-style HTTP problem payload. + pub fn to_problem_details(&self) -> ProblemDetails { + let status_code = self.status_code(); + ProblemDetails { + type_url: self.problem_type_url().to_owned(), + title: self.problem_title().to_owned(), + status: status_code.as_u16(), + detail: self.to_string(), + reason: Some(self.reason().to_owned()), + domain: Some(ERROR_INFO_DOMAIN.to_owned()), + extensions: self + .metadata() + .into_iter() + .map(|(key, value)| (key, Value::String(value))) + .collect(), } } @@ -115,10 +203,149 @@ impl A2AError { } } - fn data(&self) -> Option { + /// Convert this error into a structured `ErrorInfo` detail. + pub fn to_error_info(&self) -> ErrorInfo { + ErrorInfo { + type_url: error_info_type_url(), + reason: self.reason().to_owned(), + domain: ERROR_INFO_DOMAIN.to_owned(), + metadata: self.metadata(), + } + } + + /// Reconstruct an `A2AError` from HTTP problem details when possible. + pub fn from_problem_details(problem: &ProblemDetails) -> Self { + let reason = problem + .reason + .clone() + .unwrap_or_else(|| problem_reason(problem.type_url.as_str()).to_owned()); + let info = ErrorInfo { + type_url: error_info_type_url(), + reason: reason.clone(), + domain: problem + .domain + .clone() + .unwrap_or_else(|| ERROR_INFO_DOMAIN.to_owned()), + metadata: problem + .extensions + .iter() + .filter_map(|(key, value)| match value { + Value::String(value) => Some((key.clone(), value.clone())), + Value::Number(value) => Some((key.clone(), value.to_string())), + Value::Bool(value) => Some((key.clone(), value.to_string())), + _ => None, + }) + .collect(), + }; + + Self::from_error_info(reason_code(reason.as_str()), &problem.detail, Some(&info)) + } + + /// Reconstruct an `A2AError` from structured error details when possible. + pub fn from_error_info(error_code: i32, message: &str, info: Option<&ErrorInfo>) -> Self { + let fallback_detail = info + .and_then(|info| info.metadata.get("detail").cloned()) + .unwrap_or_else(|| message.to_owned()); + + let reason = info.map(|info| info.reason.as_str()).unwrap_or(""); + let metadata = info.map(|info| &info.metadata); + + match (error_code, reason) { + (jsonrpc::TASK_NOT_FOUND, "TASK_NOT_FOUND") => Self::TaskNotFound( + metadata + .and_then(|metadata| metadata.get("taskId").cloned()) + .unwrap_or(fallback_detail), + ), + (jsonrpc::TASK_NOT_CANCELABLE, "TASK_NOT_CANCELABLE") => Self::TaskNotCancelable( + metadata + .and_then(|metadata| metadata.get("taskId").cloned()) + .unwrap_or(fallback_detail), + ), + (jsonrpc::PUSH_NOTIFICATION_NOT_SUPPORTED, _) => { + Self::PushNotificationNotSupported(fallback_detail) + } + (jsonrpc::UNSUPPORTED_OPERATION, _) => Self::UnsupportedOperation(fallback_detail), + (jsonrpc::CONTENT_TYPE_NOT_SUPPORTED, _) => { + Self::ContentTypeNotSupported(fallback_detail) + } + (jsonrpc::INVALID_AGENT_RESPONSE, _) => Self::InvalidAgentResponse(fallback_detail), + (jsonrpc::EXTENDED_AGENT_CARD_NOT_CONFIGURED, _) => { + Self::ExtendedAgentCardNotConfigured(fallback_detail) + } + (jsonrpc::EXTENSION_SUPPORT_REQUIRED, _) => { + Self::ExtensionSupportRequired(fallback_detail) + } + (jsonrpc::VERSION_NOT_SUPPORTED, _) => Self::VersionNotSupported(fallback_detail), + (jsonrpc::PARSE_ERROR, _) => Self::ParseError(fallback_detail), + (jsonrpc::INVALID_REQUEST, _) => Self::InvalidRequest(fallback_detail), + (jsonrpc::METHOD_NOT_FOUND, _) => Self::MethodNotFound(fallback_detail), + (jsonrpc::INVALID_PARAMS, _) => Self::InvalidParams(fallback_detail), + (jsonrpc::INTERNAL_ERROR, _) => Self::Internal(fallback_detail), + _ => Self::Internal(fallback_detail), + } + } + + fn problem_type_url(&self) -> &'static str { match self { - Self::TaskNotFound(task_id) => Some(Value::String(task_id.clone())), - Self::TaskNotCancelable(task_id) => Some(Value::String(task_id.clone())), + Self::TaskNotFound(_) => "https://a2a-protocol.org/errors/task-not-found", + Self::TaskNotCancelable(_) => "https://a2a-protocol.org/errors/task-not-cancelable", + Self::PushNotificationNotSupported(_) => { + "https://a2a-protocol.org/errors/push-notification-not-supported" + } + Self::UnsupportedOperation(_) => { + "https://a2a-protocol.org/errors/unsupported-operation" + } + Self::ContentTypeNotSupported(_) => { + "https://a2a-protocol.org/errors/content-type-not-supported" + } + Self::InvalidAgentResponse(_) => { + "https://a2a-protocol.org/errors/invalid-agent-response" + } + Self::ExtendedAgentCardNotConfigured(_) => { + "https://a2a-protocol.org/errors/extended-agent-card-not-configured" + } + Self::ExtensionSupportRequired(_) => { + "https://a2a-protocol.org/errors/extension-support-required" + } + Self::VersionNotSupported(_) => "https://a2a-protocol.org/errors/version-not-supported", + Self::ParseError(_) => "about:blank", + Self::InvalidRequest(_) => "about:blank", + Self::MethodNotFound(_) => "about:blank", + Self::InvalidParams(_) => "about:blank", + Self::Internal(_) | Self::Serialization(_) => "about:blank", + #[cfg(feature = "client")] + Self::Http(_) => "about:blank", + } + } + + fn problem_title(&self) -> &'static str { + match self { + Self::TaskNotFound(_) => "Task not found", + Self::TaskNotCancelable(_) => "Task not cancelable", + Self::PushNotificationNotSupported(_) => "Push notifications not supported", + Self::UnsupportedOperation(_) => "Unsupported operation", + Self::ContentTypeNotSupported(_) => "Content type not supported", + Self::InvalidAgentResponse(_) => "Invalid agent response", + Self::ExtendedAgentCardNotConfigured(_) => "Extended agent card not configured", + Self::ExtensionSupportRequired(_) => "Extension support required", + Self::VersionNotSupported(_) => "Version not supported", + Self::ParseError(_) => "Bad Request", + Self::InvalidRequest(_) => "Bad Request", + Self::MethodNotFound(_) => "Not Found", + Self::InvalidParams(_) => "Bad Request", + Self::Internal(_) | Self::Serialization(_) => "Internal Server Error", + #[cfg(feature = "client")] + Self::Http(_) => "Bad Gateway", + } + } + + fn metadata(&self) -> BTreeMap { + let mut metadata = BTreeMap::new(); + + match self { + Self::TaskNotFound(task_id) | Self::TaskNotCancelable(task_id) => { + metadata.insert("taskId".to_owned(), task_id.clone()); + } Self::PushNotificationNotSupported(detail) | Self::UnsupportedOperation(detail) | Self::ContentTypeNotSupported(detail) @@ -130,10 +357,137 @@ impl A2AError { | Self::InvalidRequest(detail) | Self::MethodNotFound(detail) | Self::InvalidParams(detail) - | Self::Internal(detail) => Some(Value::String(detail.clone())), - Self::Serialization(_) => None, + | Self::Internal(detail) => { + metadata.insert("detail".to_owned(), detail.clone()); + } + Self::Serialization(error) => { + metadata.insert("detail".to_owned(), error.to_string()); + } #[cfg(feature = "client")] - Self::Http(_) => None, + Self::Http(error) => { + metadata.insert("detail".to_owned(), error.to_string()); + } + } + + metadata + } +} + +impl ProblemDetails { + /// Convert this HTTP error payload back into an `A2AError`. + pub fn to_a2a_error(&self) -> A2AError { + A2AError::from_problem_details(self) + } +} + +impl JsonRpcError { + /// Return the first structured `ErrorInfo` entry from `data`, when present. + pub fn first_error_info(&self) -> Option { + match self.data.as_ref()? { + Value::Array(details) => details + .iter() + .find_map(|detail| serde_json::from_value::(detail.clone()).ok()), + Value::Object(_) => serde_json::from_value::(self.data.clone()?).ok(), + _ => None, + } + } +} + +fn error_info_type_url() -> String { + ERROR_INFO_TYPE_URL.to_owned() +} + +fn problem_code(type_url: &str) -> i32 { + match type_url { + "https://a2a-protocol.org/errors/task-not-found" => jsonrpc::TASK_NOT_FOUND, + "https://a2a-protocol.org/errors/task-not-cancelable" => jsonrpc::TASK_NOT_CANCELABLE, + "https://a2a-protocol.org/errors/push-notification-not-supported" => { + jsonrpc::PUSH_NOTIFICATION_NOT_SUPPORTED + } + "https://a2a-protocol.org/errors/unsupported-operation" => jsonrpc::UNSUPPORTED_OPERATION, + "https://a2a-protocol.org/errors/content-type-not-supported" => { + jsonrpc::CONTENT_TYPE_NOT_SUPPORTED } + "https://a2a-protocol.org/errors/invalid-agent-response" => jsonrpc::INVALID_AGENT_RESPONSE, + "https://a2a-protocol.org/errors/extended-agent-card-not-configured" => { + jsonrpc::EXTENDED_AGENT_CARD_NOT_CONFIGURED + } + "https://a2a-protocol.org/errors/extension-support-required" => { + jsonrpc::EXTENSION_SUPPORT_REQUIRED + } + "https://a2a-protocol.org/errors/version-not-supported" => jsonrpc::VERSION_NOT_SUPPORTED, + _ => jsonrpc::INTERNAL_ERROR, + } +} + +fn problem_reason(type_url: &str) -> &'static str { + match problem_code(type_url) { + jsonrpc::TASK_NOT_FOUND => "TASK_NOT_FOUND", + jsonrpc::TASK_NOT_CANCELABLE => "TASK_NOT_CANCELABLE", + jsonrpc::PUSH_NOTIFICATION_NOT_SUPPORTED => "PUSH_NOTIFICATION_NOT_SUPPORTED", + jsonrpc::UNSUPPORTED_OPERATION => "UNSUPPORTED_OPERATION", + jsonrpc::CONTENT_TYPE_NOT_SUPPORTED => "CONTENT_TYPE_NOT_SUPPORTED", + jsonrpc::INVALID_AGENT_RESPONSE => "INVALID_AGENT_RESPONSE", + jsonrpc::EXTENDED_AGENT_CARD_NOT_CONFIGURED => "EXTENDED_AGENT_CARD_NOT_CONFIGURED", + jsonrpc::EXTENSION_SUPPORT_REQUIRED => "EXTENSION_SUPPORT_REQUIRED", + jsonrpc::VERSION_NOT_SUPPORTED => "VERSION_NOT_SUPPORTED", + jsonrpc::PARSE_ERROR => "PARSE_ERROR", + jsonrpc::INVALID_REQUEST => "INVALID_REQUEST", + jsonrpc::METHOD_NOT_FOUND => "METHOD_NOT_FOUND", + jsonrpc::INVALID_PARAMS => "INVALID_PARAMS", + _ => "INTERNAL", + } +} + +fn reason_code(reason: &str) -> i32 { + match reason { + "TASK_NOT_FOUND" => jsonrpc::TASK_NOT_FOUND, + "TASK_NOT_CANCELABLE" => jsonrpc::TASK_NOT_CANCELABLE, + "PUSH_NOTIFICATION_NOT_SUPPORTED" => jsonrpc::PUSH_NOTIFICATION_NOT_SUPPORTED, + "UNSUPPORTED_OPERATION" => jsonrpc::UNSUPPORTED_OPERATION, + "CONTENT_TYPE_NOT_SUPPORTED" => jsonrpc::CONTENT_TYPE_NOT_SUPPORTED, + "INVALID_AGENT_RESPONSE" => jsonrpc::INVALID_AGENT_RESPONSE, + "EXTENDED_AGENT_CARD_NOT_CONFIGURED" => jsonrpc::EXTENDED_AGENT_CARD_NOT_CONFIGURED, + "EXTENSION_SUPPORT_REQUIRED" => jsonrpc::EXTENSION_SUPPORT_REQUIRED, + "VERSION_NOT_SUPPORTED" => jsonrpc::VERSION_NOT_SUPPORTED, + "PARSE_ERROR" => jsonrpc::PARSE_ERROR, + "INVALID_REQUEST" => jsonrpc::INVALID_REQUEST, + "METHOD_NOT_FOUND" => jsonrpc::METHOD_NOT_FOUND, + "INVALID_PARAMS" => jsonrpc::INVALID_PARAMS, + _ => jsonrpc::INTERNAL_ERROR, + } +} + +#[cfg(test)] +mod tests { + use super::{A2AError, ERROR_INFO_DOMAIN, ERROR_INFO_TYPE_URL}; + + #[test] + fn jsonrpc_error_uses_structured_error_info_object() { + let error = A2AError::TaskNotFound("task-1".to_owned()).to_jsonrpc_error(); + + assert_eq!(error.code, crate::jsonrpc::TASK_NOT_FOUND); + assert_eq!( + error.data, + Some(serde_json::json!({ + "@type": ERROR_INFO_TYPE_URL, + "reason": "TASK_NOT_FOUND", + "domain": ERROR_INFO_DOMAIN, + "metadata": { + "taskId": "task-1", + } + })) + ); + } + + #[test] + fn problem_details_round_trip_to_a2a_error() { + let error = A2AError::ExtensionSupportRequired("missing extension".to_owned()); + let problem = error.to_problem_details(); + + assert_eq!( + A2AError::from_problem_details(&problem).to_string(), + error.to_string() + ); } } diff --git a/src/server/handler.rs b/src/server/handler.rs index b8037da..1307524 100644 --- a/src/server/handler.rs +++ b/src/server/handler.rs @@ -1,9 +1,12 @@ +use std::collections::BTreeSet; use std::pin::Pin; use async_trait::async_trait; +use axum::http::HeaderMap; use futures_core::Stream; use crate::A2AError; +use crate::jsonrpc::PROTOCOL_VERSION; use crate::types::{ AgentCard, CancelTaskRequest, CreateTaskPushNotificationConfigRequest, DeleteTaskPushNotificationConfigRequest, GetExtendedAgentCardRequest, @@ -171,4 +174,89 @@ pub trait A2AHandler: Send + Sync + 'static { "GetExtendedAgentCard".to_owned(), )) } + + /// Validate `A2A-Version` and `A2A-Extensions` request headers. + async fn validate_protocol_headers(&self, headers: &HeaderMap) -> Result<(), A2AError> { + let card = self.get_agent_card().await?; + validate_supported_version(&card, headers)?; + validate_required_extensions(&card, headers) + } + + /// Enforce that the request version is supported by the advertised interfaces. + async fn require_supported_version(&self, headers: &HeaderMap) -> Result<(), A2AError> { + let card = self.get_agent_card().await?; + validate_supported_version(&card, headers) + } + + /// Enforce that all required agent extensions are acknowledged by the caller. + async fn require_required_extensions(&self, headers: &HeaderMap) -> Result<(), A2AError> { + let card = self.get_agent_card().await?; + validate_required_extensions(&card, headers) + } +} + +fn header_value(headers: &HeaderMap, name: &str) -> Option { + headers + .get(name) + .and_then(|value| value.to_str().ok()) + .map(ToOwned::to_owned) +} + +fn validate_supported_version(card: &AgentCard, headers: &HeaderMap) -> Result<(), A2AError> { + let requested_version = match header_value(headers, "A2A-Version") { + Some(version) if version.trim().is_empty() => "0.3".to_owned(), + Some(version) => version, + None => PROTOCOL_VERSION.to_owned(), + }; + let supported_versions = card + .supported_interfaces + .iter() + .map(|interface| interface.protocol_version.as_str()) + .collect::>(); + + if supported_versions.is_empty() || supported_versions.contains(requested_version.as_str()) { + return Ok(()); + } + + Err(A2AError::VersionNotSupported(requested_version)) +} + +fn validate_required_extensions(card: &AgentCard, headers: &HeaderMap) -> Result<(), A2AError> { + let required_extensions = card + .capabilities + .extensions + .iter() + .filter(|extension| extension.required) + .map(|extension| extension.uri.as_str()) + .collect::>(); + + if required_extensions.is_empty() { + return Ok(()); + } + + let announced_extensions = header_value(headers, "A2A-Extensions") + .into_iter() + .flat_map(|value| { + value + .split(',') + .map(str::trim) + .filter(|value| !value.is_empty()) + .map(ToOwned::to_owned) + .collect::>() + }) + .collect::>(); + + let missing = required_extensions + .into_iter() + .filter(|extension| !announced_extensions.contains(*extension)) + .collect::>(); + + if missing.is_empty() { + return Ok(()); + } + + Err(A2AError::ExtensionSupportRequired(format!( + "missing required extensions: {}", + missing.join(", ") + ))) } diff --git a/src/server/jsonrpc.rs b/src/server/jsonrpc.rs index 4c4c292..bca1084 100644 --- a/src/server/jsonrpc.rs +++ b/src/server/jsonrpc.rs @@ -3,7 +3,7 @@ use std::sync::Arc; use axum::Json; use axum::body::Bytes; use axum::extract::State; -use axum::http::StatusCode; +use axum::http::{HeaderMap, StatusCode}; use crate::A2AError; use crate::jsonrpc::{ @@ -24,6 +24,7 @@ use super::handler::A2AHandler; pub(super) async fn handle( State(handler): State>, + headers: HeaderMap, body: Bytes, ) -> (StatusCode, Json) where @@ -55,6 +56,10 @@ where } let id = request.id.clone(); + if let Err(error) = handler.validate_protocol_headers(&headers).await { + return (StatusCode::OK, Json(error_response(id, error))); + } + let result = match request.method.as_str() { METHOD_SEND_MESSAGE => parse_params::(request.params) .and_then(|params| params.validate().map(|_| params)) diff --git a/src/server/rest.rs b/src/server/rest.rs index 4bd98d3..c39e433 100644 --- a/src/server/rest.rs +++ b/src/server/rest.rs @@ -2,11 +2,12 @@ use std::sync::Arc; use axum::Json; use axum::extract::{Path, Query, State}; -use axum::http::StatusCode; +use axum::http::{HeaderMap, StatusCode, header}; use axum::response::{IntoResponse, Response}; use serde::Deserialize; use crate::A2AError; +use crate::error::ProblemDetails; use crate::types::{ AgentCard, CancelTaskRequest, CreateTaskPushNotificationConfigRequest, DeleteTaskPushNotificationConfigRequest, GetExtendedAgentCardRequest, @@ -21,7 +22,7 @@ use super::streaming; pub(super) async fn get_agent_card( State(handler): State>, -) -> Result, (StatusCode, Json)> +) -> Result, RestErrorResponse> where H: A2AHandler, { @@ -30,11 +31,13 @@ where pub(super) async fn send_message( State(handler): State>, + headers: HeaderMap, Json(request): Json, -) -> Result, (StatusCode, Json)> +) -> Result, RestErrorResponse> where H: A2AHandler, { + handler.validate_protocol_headers(&headers).await?; request.validate()?; handler @@ -50,24 +53,30 @@ where pub(super) async fn tenant_send_message( State(handler): State>, + headers: HeaderMap, Path(tenant): Path, Json(mut request): Json, -) -> Result, (StatusCode, Json)> +) -> Result, RestErrorResponse> where H: A2AHandler, { request.tenant = Some(tenant); - send_message(State(handler), Json(request)).await + send_message(State(handler), headers, Json(request)).await } pub(super) async fn get_task_or_subscribe( State(handler): State>, + headers: HeaderMap, Path(id): Path, Query(query): Query, ) -> Response where H: A2AHandler, { + if let Err(error) = handler.validate_protocol_headers(&headers).await { + return rest_error(error).into_response(); + } + if let Err(error) = reject_query_tenant(&query.tenant) { return error.into_response(); } @@ -87,19 +96,24 @@ where }; } - get_task(State(handler), Path(id), Query(query)) + get_task(State(handler), headers, Path(id), Query(query)) .await .into_response() } pub(super) async fn tenant_get_task_or_subscribe( State(handler): State>, + headers: HeaderMap, Path((tenant, id)): Path<(String, String)>, Query(mut query): Query, ) -> Response where H: A2AHandler, { + if let Err(error) = handler.validate_protocol_headers(&headers).await { + return rest_error(error).into_response(); + } + query.tenant = Some(tenant); if let Some(id) = id.strip_suffix(":subscribe") { @@ -132,12 +146,14 @@ where pub(super) async fn get_task( State(handler): State>, + headers: HeaderMap, Path(id): Path, Query(query): Query, -) -> Result, (StatusCode, Json)> +) -> Result, RestErrorResponse> where H: A2AHandler, { + handler.validate_protocol_headers(&headers).await?; reject_query_tenant(&query.tenant)?; if id.ends_with(":cancel") || id.ends_with(":subscribe") { @@ -157,11 +173,13 @@ where pub(super) async fn list_tasks( State(handler): State>, + headers: HeaderMap, Query(request): Query, -) -> Result, (StatusCode, Json)> +) -> Result, RestErrorResponse> where H: A2AHandler, { + handler.validate_protocol_headers(&headers).await?; reject_query_tenant(&request.tenant)?; request.validate()?; @@ -174,12 +192,14 @@ where pub(super) async fn tenant_list_tasks( State(handler): State>, + headers: HeaderMap, Path(tenant): Path, Query(mut request): Query, -) -> Result, (StatusCode, Json)> +) -> Result, RestErrorResponse> where H: A2AHandler, { + handler.validate_protocol_headers(&headers).await?; request.tenant = Some(tenant); request.validate()?; @@ -193,12 +213,14 @@ where pub(super) async fn cancel_task( State(handler): State>, + headers: HeaderMap, Path(id): Path, Query(query): Query, -) -> Result, (StatusCode, Json)> +) -> Result, RestErrorResponse> where H: A2AHandler, { + handler.validate_protocol_headers(&headers).await?; reject_query_tenant(&query.tenant)?; let Some(id) = id.strip_suffix(":cancel") else { @@ -217,12 +239,14 @@ where pub(super) async fn tenant_cancel_task( State(handler): State>, + headers: HeaderMap, Path((tenant, id)): Path<(String, String)>, Query(mut query): Query, -) -> Result, (StatusCode, Json)> +) -> Result, RestErrorResponse> where H: A2AHandler, { + handler.validate_protocol_headers(&headers).await?; query.tenant = Some(tenant); let Some(id) = id.strip_suffix(":cancel") else { @@ -241,11 +265,13 @@ where pub(super) async fn get_extended_agent_card( State(handler): State>, + headers: HeaderMap, Query(query): Query, -) -> Result, (StatusCode, Json)> +) -> Result, RestErrorResponse> where H: A2AHandler, { + handler.validate_protocol_headers(&headers).await?; reject_query_tenant(&query.tenant)?; handler @@ -259,12 +285,14 @@ where pub(super) async fn tenant_get_extended_agent_card( State(handler): State>, + headers: HeaderMap, Path(tenant): Path, Query(mut query): Query, -) -> Result, (StatusCode, Json)> +) -> Result, RestErrorResponse> where H: A2AHandler, { + handler.validate_protocol_headers(&headers).await?; query.tenant = Some(tenant); handler @@ -278,13 +306,15 @@ where pub(super) async fn create_task_push_notification_config( State(handler): State>, + headers: HeaderMap, Path(task_id): Path, Query(query): Query, Json(config): Json, -) -> Result, (StatusCode, Json)> +) -> Result, RestErrorResponse> where H: A2AHandler, { + handler.validate_protocol_headers(&headers).await?; reject_query_tenant(&query.tenant)?; handler @@ -301,13 +331,15 @@ where pub(super) async fn tenant_create_task_push_notification_config( State(handler): State>, + headers: HeaderMap, Path((tenant, task_id)): Path<(String, String)>, Query(mut query): Query, Json(config): Json, -) -> Result, (StatusCode, Json)> +) -> Result, RestErrorResponse> where H: A2AHandler, { + handler.validate_protocol_headers(&headers).await?; query.tenant = Some(tenant); handler @@ -324,12 +356,14 @@ where pub(super) async fn get_task_push_notification_config( State(handler): State>, + headers: HeaderMap, Path((task_id, id)): Path<(String, String)>, Query(query): Query, -) -> Result, (StatusCode, Json)> +) -> Result, RestErrorResponse> where H: A2AHandler, { + handler.validate_protocol_headers(&headers).await?; reject_query_tenant(&query.tenant)?; handler @@ -345,12 +379,14 @@ where pub(super) async fn tenant_get_task_push_notification_config( State(handler): State>, + headers: HeaderMap, Path((tenant, task_id, id)): Path<(String, String, String)>, Query(mut query): Query, -) -> Result, (StatusCode, Json)> +) -> Result, RestErrorResponse> where H: A2AHandler, { + handler.validate_protocol_headers(&headers).await?; query.tenant = Some(tenant); handler @@ -366,12 +402,14 @@ where pub(super) async fn list_task_push_notification_config( State(handler): State>, + headers: HeaderMap, Path(task_id): Path, Query(query): Query, -) -> Result, (StatusCode, Json)> +) -> Result, RestErrorResponse> where H: A2AHandler, { + handler.validate_protocol_headers(&headers).await?; reject_query_tenant(&query.tenant)?; let request = ListTaskPushNotificationConfigRequest { @@ -391,12 +429,14 @@ where pub(super) async fn tenant_list_task_push_notification_config( State(handler): State>, + headers: HeaderMap, Path((tenant, task_id)): Path<(String, String)>, Query(mut query): Query, -) -> Result, (StatusCode, Json)> +) -> Result, RestErrorResponse> where H: A2AHandler, { + handler.validate_protocol_headers(&headers).await?; query.tenant = Some(tenant); let request = ListTaskPushNotificationConfigRequest { @@ -416,12 +456,14 @@ where pub(super) async fn delete_task_push_notification_config( State(handler): State>, + headers: HeaderMap, Path((task_id, id)): Path<(String, String)>, Query(query): Query, -) -> Result, (StatusCode, Json)> +) -> Result, RestErrorResponse> where H: A2AHandler, { + handler.validate_protocol_headers(&headers).await?; reject_query_tenant(&query.tenant)?; handler @@ -437,12 +479,14 @@ where pub(super) async fn tenant_delete_task_push_notification_config( State(handler): State>, + headers: HeaderMap, Path((tenant, task_id, id)): Path<(String, String, String)>, Query(mut query): Query, -) -> Result, (StatusCode, Json)> +) -> Result, RestErrorResponse> where H: A2AHandler, { + handler.validate_protocol_headers(&headers).await?; query.tenant = Some(tenant); handler @@ -491,28 +535,20 @@ pub(super) struct TenantQuery { pub tenant: Option, } -pub(super) fn rest_error(error: A2AError) -> (StatusCode, Json) { - let status = error.status_code(); - let body = serde_json::json!({ - "error": { - "code": error.code(), - "message": error.to_string(), - "data": error.to_jsonrpc_error().data, - } - }); - - (status, Json(body)) +pub(super) fn rest_error(error: A2AError) -> RestErrorResponse { + RestErrorResponse { + status: error.status_code(), + body: Box::new(error.to_problem_details()), + } } -impl From for (StatusCode, Json) { +impl From for RestErrorResponse { fn from(value: A2AError) -> Self { rest_error(value) } } -fn reject_query_tenant( - tenant: &Option, -) -> Result<(), (StatusCode, Json)> { +fn reject_query_tenant(tenant: &Option) -> Result<(), RestErrorResponse> { if tenant.is_some() { return Err(rest_error(A2AError::InvalidRequest( "tenant must be supplied via tenant-prefixed routes".to_owned(), @@ -521,3 +557,19 @@ fn reject_query_tenant( Ok(()) } + +pub(super) struct RestErrorResponse { + status: StatusCode, + body: Box, +} + +impl IntoResponse for RestErrorResponse { + fn into_response(self) -> Response { + let mut response = (self.status, Json(*self.body)).into_response(); + response.headers_mut().insert( + header::CONTENT_TYPE, + header::HeaderValue::from_static("application/problem+json"), + ); + response + } +} diff --git a/src/server/streaming.rs b/src/server/streaming.rs index 9e856e1..ec40e74 100644 --- a/src/server/streaming.rs +++ b/src/server/streaming.rs @@ -4,26 +4,26 @@ use std::time::Duration; use axum::Json; use axum::extract::{Path, State}; -use axum::http::StatusCode; +use axum::http::HeaderMap; use axum::response::sse::{Event, KeepAlive, Sse}; use futures_util::stream::StreamExt; use crate::types::{SendMessageRequest, StreamResponse, SubscribeToTaskRequest}; use super::handler::A2AHandler; +use super::rest::RestErrorResponse; const SSE_KEEP_ALIVE_INTERVAL: Duration = Duration::from_secs(15); pub(super) async fn send_message( State(handler): State>, + headers: HeaderMap, Json(request): Json, -) -> Result< - Sse>>, - (StatusCode, Json), -> +) -> Result>>, RestErrorResponse> where H: A2AHandler, { + handler.validate_protocol_headers(&headers).await?; request.validate()?; let stream = handler.send_streaming_message(request).await?; Ok(sse_response(stream)) @@ -31,26 +31,21 @@ where pub(super) async fn tenant_send_message( State(handler): State>, + headers: HeaderMap, Path(tenant): Path, Json(mut request): Json, -) -> Result< - Sse>>, - (StatusCode, Json), -> +) -> Result>>, RestErrorResponse> where H: A2AHandler, { request.tenant = Some(tenant); - send_message(State(handler), Json(request)).await + send_message(State(handler), headers, Json(request)).await } pub(super) async fn subscribe_to_task_response( handler: Arc, request: SubscribeToTaskRequest, -) -> Result< - Sse>>, - (StatusCode, Json), -> +) -> Result>>, RestErrorResponse> where H: A2AHandler, { diff --git a/src/types/agent_card.rs b/src/types/agent_card.rs index 2db3848..4b2fee7 100644 --- a/src/types/agent_card.rs +++ b/src/types/agent_card.rs @@ -1,6 +1,9 @@ use std::collections::BTreeMap; +use base64::Engine as _; use serde::{Deserialize, Serialize}; +use serde_json::Value; +use thiserror::Error; use crate::types::JsonObject; @@ -149,11 +152,211 @@ pub struct AgentCardSignature { pub header: Option, } +/// Decoded protected header for an agent-card detached JWS signature. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct JwsProtectedHeader { + /// JWS algorithm identifier such as `ES256`. + pub alg: String, + /// Key identifier used to resolve the verification key. + pub kid: String, + #[serde(default, skip_serializing_if = "Option::is_none")] + /// Optional JOSE type, typically `JOSE`. + pub typ: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + /// Optional JWK Set URL that can help callers resolve keys. + pub jku: Option, + #[serde(flatten, default, skip_serializing_if = "BTreeMap::is_empty")] + /// Additional JOSE header parameters. + pub extra: BTreeMap, +} + +/// Prepared detached-JWS verification input for an agent-card signature. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct AgentCardSignatureVerificationInput { + /// Parsed protected JOSE header. + pub protected_header: JwsProtectedHeader, + /// Original base64url-encoded protected segment. + pub protected_segment: String, + /// Decoded signature bytes. + pub signature: Vec, + /// Detached JWS signing input: `protected + "." + base64url(payload)`. + pub signing_input: Vec, + /// Optional unprotected JOSE header. + pub unprotected_header: Option, +} + +/// Errors produced by agent-card signature helper APIs. +#[derive(Debug, Error)] +pub enum AgentCardSignatureError { + /// The agent card does not contain any signatures. + #[error("agent card does not contain any signatures")] + MissingSignatures, + /// The signature protected header could not be base64url-decoded. + #[error("invalid protected header encoding: {0}")] + InvalidProtectedEncoding(String), + /// The signature bytes could not be base64url-decoded. + #[error("invalid signature encoding: {0}")] + InvalidSignatureEncoding(String), + /// The protected header JSON is malformed. + #[error("invalid protected header JSON: {0}")] + InvalidProtectedHeader(#[source] serde_json::Error), + /// No signature used a caller-supported algorithm. + #[error("no agent-card signature matched the supported algorithms")] + UnsupportedAlgorithm, + /// All candidate signatures failed caller-supplied verification. + #[error("agent-card signature verification failed")] + VerificationFailed, + /// JSON serialization needed for canonicalization failed. + #[error("agent-card serialization failed: {0}")] + Serialization(#[from] serde_json::Error), +} + +impl AgentCard { + /// Return a clone of the card with signature blocks removed. + pub fn unsigned_clone(&self) -> Self { + let mut card = self.clone(); + card.signatures.clear(); + card + } + + /// Canonicalize the unsigned agent card for detached-JWS verification. + pub fn canonical_signing_payload(&self) -> Result { + canonicalize_json(&serde_json::to_value(self.unsigned_clone())?) + } + + /// Verify any advertised signature using caller-supplied crypto. + /// + /// The caller controls key lookup and cryptographic verification. The SDK + /// prepares detached-JWS inputs and filters signatures by supported + /// algorithm identifiers. + pub fn verify_signatures( + &self, + supported_algorithms: &[&str], + mut verifier: F, + ) -> Result<(), AgentCardSignatureError> + where + F: FnMut(&AgentCardSignatureVerificationInput) -> Result, + { + if self.signatures.is_empty() { + return Err(AgentCardSignatureError::MissingSignatures); + } + + let mut matched_algorithm = false; + for signature in &self.signatures { + let input = signature.verification_input(self)?; + if !supported_algorithms.is_empty() + && !supported_algorithms.iter().any(|algorithm| { + algorithm.eq_ignore_ascii_case(input.protected_header.alg.as_str()) + }) + { + continue; + } + + matched_algorithm = true; + if verifier(&input)? { + return Ok(()); + } + } + + if !matched_algorithm { + return Err(AgentCardSignatureError::UnsupportedAlgorithm); + } + + Err(AgentCardSignatureError::VerificationFailed) + } +} + +impl AgentCardSignature { + /// Decode the protected JOSE header from its base64url segment. + pub fn protected_header(&self) -> Result { + let bytes = base64_url_engine() + .decode(self.protected.as_bytes()) + .map_err(|error| { + AgentCardSignatureError::InvalidProtectedEncoding(error.to_string()) + })?; + + serde_json::from_slice(&bytes).map_err(AgentCardSignatureError::InvalidProtectedHeader) + } + + /// Decode the raw signature bytes from their base64url representation. + pub fn signature_bytes(&self) -> Result, AgentCardSignatureError> { + base64_url_engine() + .decode(self.signature.as_bytes()) + .map_err(|error| AgentCardSignatureError::InvalidSignatureEncoding(error.to_string())) + } + + /// Build the detached-JWS verification input for this signature. + pub fn verification_input( + &self, + card: &AgentCard, + ) -> Result { + let protected_header = self.protected_header()?; + let signature = self.signature_bytes()?; + let payload = card.canonical_signing_payload()?; + let payload_segment = base64_url_engine().encode(payload.as_bytes()); + let signing_input = format!("{}.{}", self.protected, payload_segment).into_bytes(); + + Ok(AgentCardSignatureVerificationInput { + protected_header, + protected_segment: self.protected.clone(), + signature, + signing_input, + unprotected_header: self.header.clone(), + }) + } +} + +fn canonicalize_json(value: &Value) -> Result { + match value { + Value::Null => Ok("null".to_owned()), + Value::Bool(value) => Ok(if *value { "true" } else { "false" }.to_owned()), + Value::Number(value) => Ok(value.to_string()), + Value::String(value) => serde_json::to_string(value).map_err(AgentCardSignatureError::from), + Value::Array(values) => { + let mut json = String::from("["); + for (index, value) in values.iter().enumerate() { + if index > 0 { + json.push(','); + } + json.push_str(&canonicalize_json(value)?); + } + json.push(']'); + Ok(json) + } + Value::Object(values) => { + let mut keys = values.keys().collect::>(); + keys.sort_unstable(); + + let mut json = String::from("{"); + for (index, key) in keys.into_iter().enumerate() { + if index > 0 { + json.push(','); + } + json.push_str(&serde_json::to_string(key)?); + json.push(':'); + json.push_str(&canonicalize_json(&values[key])?); + } + json.push('}'); + Ok(json) + } + } +} + +fn base64_url_engine() -> &'static base64::engine::GeneralPurpose { + &base64::engine::general_purpose::URL_SAFE_NO_PAD +} + #[cfg(test)] mod tests { use std::collections::BTreeMap; - use super::{AgentCapabilities, AgentCard, AgentExtension, AgentInterface, AgentSkill}; + use super::{ + AgentCapabilities, AgentCard, AgentCardSignature, AgentCardSignatureError, AgentExtension, + AgentInterface, AgentSkill, JwsProtectedHeader, + }; + use base64::Engine as _; + use serde_json::json; #[test] fn agent_card_round_trip_serialization() { @@ -213,4 +416,144 @@ mod tests { assert!(!round_trip.capabilities.extensions[0].required); assert_eq!(round_trip.skills[0].id, "echo"); } + + #[test] + fn signature_helper_decodes_protected_header() { + let protected = base64::engine::general_purpose::URL_SAFE_NO_PAD.encode( + serde_json::to_vec(&json!({ + "alg": "ES256", + "kid": "key-1", + "typ": "JOSE", + })) + .expect("header should serialize"), + ); + let signature = AgentCardSignature { + protected, + signature: base64::engine::general_purpose::URL_SAFE_NO_PAD.encode([1_u8, 2, 3]), + header: None, + }; + + let header = signature + .protected_header() + .expect("protected header should decode"); + assert_eq!( + header, + JwsProtectedHeader { + alg: "ES256".to_owned(), + kid: "key-1".to_owned(), + typ: Some("JOSE".to_owned()), + jku: None, + extra: BTreeMap::new(), + } + ); + } + + #[test] + fn canonical_signing_payload_omits_signatures() { + let mut card = sample_card(); + card.signatures.push(sample_signature()); + + let payload = card + .canonical_signing_payload() + .expect("payload should canonicalize"); + + assert!(!payload.contains("\"signatures\"")); + assert!(payload.starts_with("{\"capabilities\"")); + } + + #[test] + fn verify_signatures_builds_detached_jws_input() { + let mut card = sample_card(); + let signature = sample_signature(); + let protected = signature.protected.clone(); + card.signatures.push(signature); + + let payload_segment = base64::engine::general_purpose::URL_SAFE_NO_PAD.encode( + card.canonical_signing_payload() + .expect("payload should canonicalize"), + ); + let expected_input = format!("{protected}.{payload_segment}"); + + card.verify_signatures(&["ES256"], |input| { + assert_eq!(input.protected_header.alg, "ES256"); + assert_eq!(input.protected_header.kid, "key-1"); + assert_eq!(input.signature, vec![1_u8, 2, 3]); + assert_eq!(input.signing_input, expected_input.as_bytes()); + Ok(true) + }) + .expect("verification should succeed"); + } + + #[test] + fn verify_signatures_rejects_cards_without_supported_algorithms() { + let mut card = sample_card(); + card.signatures.push(sample_signature()); + + let error = card + .verify_signatures(&["RS256"], |_input| Ok(true)) + .expect_err("unsupported algorithms should fail"); + + assert!(matches!( + error, + AgentCardSignatureError::UnsupportedAlgorithm + )); + } + + fn sample_card() -> AgentCard { + AgentCard { + name: "Echo Agent".to_owned(), + description: "Replies with the same text".to_owned(), + supported_interfaces: vec![AgentInterface { + url: "https://example.com/rpc".to_owned(), + protocol_binding: "JSONRPC".to_owned(), + tenant: None, + protocol_version: "1.0".to_owned(), + }], + provider: None, + version: "0.1.0".to_owned(), + documentation_url: None, + capabilities: AgentCapabilities { + streaming: Some(true), + push_notifications: Some(false), + extensions: vec![AgentExtension { + uri: "https://example.com/ext/streaming".to_owned(), + description: "Streaming support".to_owned(), + required: false, + params: None, + }], + extended_agent_card: Some(false), + }, + security_schemes: BTreeMap::new(), + security_requirements: Vec::new(), + default_input_modes: vec!["text/plain".to_owned()], + default_output_modes: vec!["text/plain".to_owned()], + skills: vec![AgentSkill { + id: "echo".to_owned(), + name: "Echo".to_owned(), + description: "Echo back user input".to_owned(), + tags: vec!["utility".to_owned()], + examples: vec!["echo hello".to_owned()], + input_modes: vec!["text/plain".to_owned()], + output_modes: vec!["text/plain".to_owned()], + security_requirements: Vec::new(), + }], + signatures: Vec::new(), + icon_url: None, + } + } + + fn sample_signature() -> AgentCardSignature { + AgentCardSignature { + protected: base64::engine::general_purpose::URL_SAFE_NO_PAD.encode( + serde_json::to_vec(&json!({ + "alg": "ES256", + "kid": "key-1", + "typ": "JOSE", + })) + .expect("header should serialize"), + ), + signature: base64::engine::general_purpose::URL_SAFE_NO_PAD.encode([1_u8, 2, 3]), + header: None, + } + } } diff --git a/tests/client_wiremock.rs b/tests/client_wiremock.rs index c6b4273..9a3a2fb 100644 --- a/tests/client_wiremock.rs +++ b/tests/client_wiremock.rs @@ -254,6 +254,71 @@ async fn client_parses_sse_streams_with_crlf_frame_delimiters() { } } +#[tokio::test] +async fn client_maps_http_problem_details_to_a2a_error() { + let server = MockServer::start().await; + Mock::given(method("GET")) + .and(path("/.well-known/agent-card.json")) + .respond_with(ResponseTemplate::new(200).set_body_json(agent_card( + vec![interface("/", "HTTP+JSON")], + capabilities(false, false), + ))) + .mount(&server) + .await; + Mock::given(method("POST")) + .and(path("/message:send")) + .respond_with(ResponseTemplate::new(400).set_body_json(json!({ + "type": "https://a2a-protocol.org/errors/extension-support-required", + "title": "Extension support required", + "status": 400, + "detail": "missing required extensions: https://example.com/ext/required", + "reason": "EXTENSION_SUPPORT_REQUIRED", + "domain": "a2a-protocol.org", + }))) + .mount(&server) + .await; + + let client = A2AClient::new(&server.uri()).expect("client should build"); + let error = client + .send_message(user_message_request(None)) + .await + .expect_err("problem details should map to an A2A error"); + + match error { + A2AError::ExtensionSupportRequired(detail) => { + assert!(detail.contains("missing required extensions")); + } + other => panic!("expected extension-support-required, got {other}"), + } +} + +#[tokio::test] +async fn client_rejects_agent_cards_without_a_supported_protocol_version() { + let server = MockServer::start().await; + Mock::given(method("GET")) + .and(path("/.well-known/agent-card.json")) + .respond_with(ResponseTemplate::new(200).set_body_json(agent_card( + vec![interface_with_version("/rpc", "JSONRPC", "0.9")], + capabilities(false, false), + ))) + .mount(&server) + .await; + + let client = A2AClient::new(&server.uri()).expect("client should build"); + let error = client + .send_message(user_message_request(None)) + .await + .expect_err("unsupported interface versions should fail"); + + match error { + A2AError::VersionNotSupported(detail) => { + assert!(detail.contains("1.0")); + assert!(detail.contains("0.9")); + } + other => panic!("expected version-not-supported, got {other}"), + } +} + async fn mount_jsonrpc_discovery(server: &MockServer) { Mock::given(method("GET")) .and(path("/.well-known/agent-card.json")) @@ -296,11 +361,19 @@ fn agent_card(interfaces: Vec, capabilities: AgentCapabilities) } fn interface(url: &str, protocol_binding: &str) -> AgentInterface { + interface_with_version(url, protocol_binding, "1.0") +} + +fn interface_with_version( + url: &str, + protocol_binding: &str, + protocol_version: &str, +) -> AgentInterface { AgentInterface { url: url.to_owned(), protocol_binding: protocol_binding.to_owned(), tenant: None, - protocol_version: "1.0".to_owned(), + protocol_version: protocol_version.to_owned(), } } diff --git a/tests/server_integration.rs b/tests/server_integration.rs index dd03bab..ef3ecd2 100644 --- a/tests/server_integration.rs +++ b/tests/server_integration.rs @@ -23,6 +23,9 @@ struct StreamingHandler; #[derive(Clone)] struct TenantEchoHandler; +#[derive(Clone)] +struct RequiredExtensionHandler; + fn tenant_metadata(tenant: Option) -> Option> { tenant.map(|tenant| { let mut metadata = serde_json::Map::new(); @@ -391,6 +394,50 @@ impl A2AHandler for TenantEchoHandler { } } +#[async_trait] +impl A2AHandler for RequiredExtensionHandler { + async fn get_agent_card(&self) -> Result { + Ok(AgentCard { + name: "Extension Agent".to_owned(), + description: "Requires an extension".to_owned(), + supported_interfaces: vec![AgentInterface { + url: "https://example.com/rpc".to_owned(), + protocol_binding: "JSONRPC".to_owned(), + tenant: None, + protocol_version: "1.0".to_owned(), + }], + provider: None, + version: "0.1.0".to_owned(), + documentation_url: None, + capabilities: AgentCapabilities { + streaming: Some(false), + push_notifications: Some(false), + extensions: vec![a2a_rust::types::AgentExtension { + uri: "https://example.com/extensions/required".to_owned(), + description: "Required extension".to_owned(), + required: true, + params: None, + }], + extended_agent_card: Some(false), + }, + security_schemes: Default::default(), + security_requirements: Vec::new(), + default_input_modes: vec!["text/plain".to_owned()], + default_output_modes: vec!["text/plain".to_owned()], + skills: Vec::new(), + signatures: Vec::new(), + icon_url: None, + }) + } + + async fn send_message( + &self, + request: SendMessageRequest, + ) -> Result { + TestHandler.send_message(request).await + } +} + #[tokio::test] async fn well_known_endpoint_serves_agent_card() { let response = router(TestHandler) @@ -470,6 +517,86 @@ async fn tenant_message_route_uses_path_tenant() { assert_eq!(json["message"]["metadata"]["tenant"], "tenant-a"); } +#[tokio::test] +async fn rest_version_mismatch_returns_problem_details() { + let body = serde_json::json!({ + "message": { + "messageId": "msg-1", + "role": "ROLE_USER", + "parts": [{"text": "ping"}] + } + }); + + let response = router(TestHandler) + .oneshot( + Request::builder() + .method("POST") + .uri("/message:send") + .header("content-type", "application/json") + .header("A2A-Version", "9.9") + .body(Body::from(body.to_string())) + .expect("request should build"), + ) + .await + .expect("response should succeed"); + + assert_eq!(response.status(), StatusCode::BAD_REQUEST); + assert_eq!( + response.headers()["content-type"], + "application/problem+json" + ); + + let bytes = to_bytes(response.into_body(), usize::MAX) + .await + .expect("body should read"); + let json: serde_json::Value = serde_json::from_slice(&bytes).expect("body should deserialize"); + assert_eq!( + json["type"], + "https://a2a-protocol.org/errors/version-not-supported" + ); + assert_eq!(json["reason"], "VERSION_NOT_SUPPORTED"); + assert_eq!(json["status"], 400); +} + +#[tokio::test] +async fn rest_missing_required_extension_returns_problem_details() { + let body = serde_json::json!({ + "message": { + "messageId": "msg-1", + "role": "ROLE_USER", + "parts": [{"text": "ping"}] + } + }); + + let response = router(RequiredExtensionHandler) + .oneshot( + Request::builder() + .method("POST") + .uri("/message:send") + .header("content-type", "application/json") + .body(Body::from(body.to_string())) + .expect("request should build"), + ) + .await + .expect("response should succeed"); + + assert_eq!(response.status(), StatusCode::BAD_REQUEST); + assert_eq!( + response.headers()["content-type"], + "application/problem+json" + ); + + let bytes = to_bytes(response.into_body(), usize::MAX) + .await + .expect("body should read"); + let json: serde_json::Value = serde_json::from_slice(&bytes).expect("body should deserialize"); + assert_eq!( + json["type"], + "https://a2a-protocol.org/errors/extension-support-required" + ); + assert_eq!(json["reason"], "EXTENSION_SUPPORT_REQUIRED"); +} + #[tokio::test] async fn jsonrpc_send_message_dispatches_pascal_case_method() { let body = serde_json::json!({ @@ -507,6 +634,47 @@ async fn jsonrpc_send_message_dispatches_pascal_case_method() { assert_eq!(json["result"]["message"]["parts"][0]["text"], "pong"); } +#[tokio::test] +async fn jsonrpc_missing_required_extension_returns_structured_error_info() { + let body = serde_json::json!({ + "jsonrpc": "2.0", + "id": "req-ext", + "method": "SendMessage", + "params": { + "message": { + "messageId": "msg-1", + "role": "ROLE_USER", + "parts": [{"text": "ping"}] + } + } + }); + + let response = router(RequiredExtensionHandler) + .oneshot( + Request::builder() + .method("POST") + .uri("/rpc") + .header("content-type", "application/json") + .body(Body::from(body.to_string())) + .expect("request should build"), + ) + .await + .expect("response should succeed"); + + assert_eq!(response.status(), StatusCode::OK); + + let bytes = to_bytes(response.into_body(), usize::MAX) + .await + .expect("body should read"); + let json: serde_json::Value = serde_json::from_slice(&bytes).expect("body should deserialize"); + assert_eq!(json["error"]["code"], -32008); + assert_eq!( + json["error"]["data"]["reason"], + "EXTENSION_SUPPORT_REQUIRED" + ); + assert_eq!(json["error"]["data"]["domain"], "a2a-protocol.org"); +} + #[tokio::test] async fn jsonrpc_unknown_method_returns_jsonrpc_error() { let body = serde_json::json!({ @@ -683,7 +851,8 @@ async fn non_tenant_list_tasks_rejects_query_tenant() { .await .expect("body should read"); let json: serde_json::Value = serde_json::from_slice(&bytes).expect("body should deserialize"); - assert_eq!(json["error"]["code"], -32600); + assert_eq!(json["reason"], "INVALID_REQUEST"); + assert_eq!(json["status"], 400); } #[tokio::test] @@ -734,7 +903,8 @@ async fn rest_get_extended_agent_card_returns_default_error() { .await .expect("body should read"); let json: serde_json::Value = serde_json::from_slice(&bytes).expect("body should deserialize"); - assert_eq!(json["error"]["code"], -32007); + assert_eq!(json["reason"], "EXTENDED_AGENT_CARD_NOT_CONFIGURED"); + assert_eq!(json["status"], 400); } #[tokio::test] @@ -755,7 +925,8 @@ async fn get_cancel_path_returns_not_found() { .await .expect("body should read"); let json: serde_json::Value = serde_json::from_slice(&bytes).expect("body should deserialize"); - assert_eq!(json["error"]["code"], -32601); + assert_eq!(json["reason"], "METHOD_NOT_FOUND"); + assert_eq!(json["status"], 404); } #[tokio::test] @@ -786,7 +957,8 @@ async fn streaming_route_returns_unsupported_when_capability_is_disabled() { .await .expect("body should read"); let json: serde_json::Value = serde_json::from_slice(&bytes).expect("body should deserialize"); - assert_eq!(json["error"]["code"], -32004); + assert_eq!(json["reason"], "UNSUPPORTED_OPERATION"); + assert_eq!(json["status"], 400); } #[tokio::test] @@ -807,7 +979,8 @@ async fn subscribe_route_returns_unsupported_when_capability_is_disabled() { .await .expect("body should read"); let json: serde_json::Value = serde_json::from_slice(&bytes).expect("body should deserialize"); - assert_eq!(json["error"]["code"], -32004); + assert_eq!(json["reason"], "UNSUPPORTED_OPERATION"); + assert_eq!(json["status"], 400); } #[tokio::test] @@ -828,7 +1001,8 @@ async fn push_config_route_returns_not_supported_when_capability_is_disabled() { .await .expect("body should read"); let json: serde_json::Value = serde_json::from_slice(&bytes).expect("body should deserialize"); - assert_eq!(json["error"]["code"], -32003); + assert_eq!(json["reason"], "PUSH_NOTIFICATION_NOT_SUPPORTED"); + assert_eq!(json["status"], 400); } #[tokio::test] @@ -1001,7 +1175,8 @@ async fn non_tenant_subscribe_rejects_query_tenant() { .await .expect("body should read"); let json: serde_json::Value = serde_json::from_slice(&bytes).expect("body should deserialize"); - assert_eq!(json["error"]["code"], -32600); + assert_eq!(json["reason"], "INVALID_REQUEST"); + assert_eq!(json["status"], 400); } #[tokio::test]