From ebdc0aeb0e8aec19e55729a880bab8a9a6131bba Mon Sep 17 00:00:00 2001 From: Ben Brandt Date: Tue, 28 Jul 2026 23:06:09 +0200 Subject: [PATCH] feat(polyfill): Support v2 MCP-over-ACP bridge --- md/conductor.md | 5 + md/mcp-bridge.md | 54 +- md/protocol-v2.md | 15 + .../Cargo.toml | 10 +- .../tests/mcp_over_acp_polyfill_v2.rs | 566 +++++++++++++ .../CHANGELOG.md | 13 + src/agent-client-protocol-polyfill/Cargo.toml | 1 + .../src/mcp_over_acp/mod.rs | 782 ++++++++++-------- .../src/mcp_over_acp/protocol.rs | 697 ++++++++++++++++ 9 files changed, 1768 insertions(+), 375 deletions(-) create mode 100644 src/agent-client-protocol-conductor/tests/mcp_over_acp_polyfill_v2.rs create mode 100644 src/agent-client-protocol-polyfill/src/mcp_over_acp/protocol.rs diff --git a/md/conductor.md b/md/conductor.md index ce41535..0f1c83a 100644 --- a/md/conductor.md +++ b/md/conductor.md @@ -155,6 +155,11 @@ adapted to HTTP. Keeping the polyfill explicit prevents instrumentation or orchestration from silently changing session MCP declarations. See [MCP Bridge](./mcp-bridge.md). +The polyfill supports v1 by default. For a draft-v2 chain, enable +`unstable_protocol_v2` on both the conductor and polyfill crates; without the +polyfill feature, it rejects v2 initialization instead of interpreting v2 +traffic as v1. + ## Tracing The conductor can record an idealized logical sequence of ACP and MCP messages. diff --git a/md/mcp-bridge.md b/md/mcp-bridge.md index 5572abe..53bb994 100644 --- a/md/mcp-bridge.md +++ b/md/mcp-bridge.md @@ -19,6 +19,20 @@ Native MCP-over-ACP requires the core SDK's `unstable_mcp_over_acp` feature. The polyfill enables that feature on its core dependency, so applications using the polyfill receive it through Cargo feature unification. +The polyfill supports stable protocol v1 by default. To place it in a draft-v2 +conductor chain, enable `unstable_protocol_v2` on both the conductor and +polyfill dependencies: + +```toml +agent-client-protocol-conductor = { version = "...", features = ["unstable_protocol_v2"] } +agent-client-protocol-polyfill = { version = "...", features = ["unstable_protocol_v2"] } +``` + +The feature makes this concrete compatibility proxy recognize v2 +initialization, capability, session setup, and `mcp/*` wire types. It does not +add high-level v2 global MCP attachment or proxy-session helpers to the core +SDK; those remain v1-only. + ## Placement Insert the polyfill immediately before the final agent that lacks native @@ -37,17 +51,23 @@ ConductorImpl::new_agent("conductor", components) .await?; ``` -The application proxy can attach a high-level -`agent_client_protocol::mcp_server::McpServer`. The SDK advertises it in -session setup requests as `McpServer::Acp`; callers do not need to construct a -transport placeholder themselves. +For v1, the application proxy can attach a high-level +`agent_client_protocol::mcp_server::McpServer`. The SDK advertises it in session +setup requests as `McpServer::Acp`; callers do not need to construct a transport +placeholder themselves. In a v2 chain, a version-aware proxy currently supplies +the `schema::v2::McpServer::Acp` declaration directly because the high-level +global proxy attachment helpers remain v1-only. -During initialization, the polyfill forwards the request to its successor and -sets `agentCapabilities.mcpCapabilities.acp` in the response seen upstream when -the successor advertises HTTP MCP support. In this chain position that -capability means the chain can consume native MCP-over-ACP declarations through -the adapter; it does not imply that the final agent implements the transport -itself. +During initialization, the polyfill forwards the request to its successor. When +the successor advertises HTTP MCP support, the polyfill advertises native ACP +MCP support in the response seen upstream: + +- v1 sets `agentCapabilities.mcpCapabilities.acp` to `true`. +- v2 adds the `capabilities.session.mcp.acp` marker. + +In this chain position that capability means the chain can consume native +MCP-over-ACP declarations through the adapter; it does not imply that the final +agent implements the transport itself. If the successor already advertises native ACP MCP support, the polyfill leaves the capability, declarations, and `mcp/message` traffic unchanged. If it @@ -56,8 +76,8 @@ support and rejects any native declaration that is nevertheless supplied. ## Transformation -For each `McpServer::Acp` entry in `session/new`, `session/load`, -`session/resume`, or feature-gated `session/fork`, the polyfill: +For each schema-selected `McpServer::Acp` entry in a session setup request, the +polyfill: 1. Creates or reuses a connection-scoped localhost bridge endpoint for the `serverId` and replaces the declaration with the HTTP transport for the @@ -72,11 +92,17 @@ For each `McpServer::Acp` entry in `session/new`, `session/load`, the connection from the bridge. Enable the polyfill crate's `unstable_session_fork` feature when adapting fork -requests. +requests. Stable v1 setup includes `session/new`, `session/load`, and +`session/resume`; draft v2 includes `session/new` and `session/resume`. Both +versions include `session/fork` when `unstable_session_fork` is enabled. + +Declarations using another transport are left unchanged, including extension +transports represented by v2's `McpServer::Other`. Endpoints are cached by `serverId` across session setup requests on the ACP connection. The output declaration is rebuilt for each occurrence, preserving -that occurrence's `name` and `_meta` even when its endpoint is reused. +that occurrence's `name`, `_meta`, and other unmodified extension fields even +when its endpoint is reused. The native wire envelopes are documented in the [SDK Protocol Reference](./protocol.md#native-mcp-over-acp). diff --git a/md/protocol-v2.md b/md/protocol-v2.md index 6a2ac17..88b520f 100644 --- a/md/protocol-v2.md +++ b/md/protocol-v2.md @@ -251,6 +251,21 @@ validates the final agent's initialize response against that selection. The proxy connection also routes v2 `session/new` requests and responses without interpreting them as v1 payloads. +### MCP compatibility polyfill + +The concrete +`agent_client_protocol_polyfill::mcp_over_acp::McpOverAcpPolyfill` can +participate in a v2 conductor chain when its `unstable_protocol_v2` feature is +enabled. It selects v1 or v2 from `_proxy/initialize`, uses that version's MCP +capability and wire types, and adapts native `McpServer::Acp` declarations in +v2 `session/new`, `session/resume`, and feature-gated `session/fork` requests. +Other declarations and unrelated request fields remain unchanged. See +[MCP-over-ACP Compatibility Bridge](./mcp-bridge.md) for placement and feature +configuration. + +This feature extends the concrete compatibility proxy only. Global MCP +attachment and proxy-session helpers in the core SDK remain v1-only. + The SDK handles the `initialize` negotiation at the JSON-RPC boundary: - A v2 client advertises protocol v2 as its latest supported version. diff --git a/src/agent-client-protocol-conductor/Cargo.toml b/src/agent-client-protocol-conductor/Cargo.toml index c7afeff..9c9b21c 100644 --- a/src/agent-client-protocol-conductor/Cargo.toml +++ b/src/agent-client-protocol-conductor/Cargo.toml @@ -43,9 +43,15 @@ agent-client-protocol-test.workspace = true yopo.workspace = true expect-test.workspace = true regex.workspace = true -rmcp = { workspace = true, features = ["client", "server", "transport-io", "transport-child-process"] } +rmcp = { workspace = true, features = [ + "client", + "server", + "transport-io", + "transport-child-process", + "transport-streamable-http-client-reqwest", +] } schemars.workspace = true -agent-client-protocol-polyfill.workspace = true +agent-client-protocol-polyfill = { workspace = true, features = ["unstable_protocol_v2"] } [lints] workspace = true diff --git a/src/agent-client-protocol-conductor/tests/mcp_over_acp_polyfill_v2.rs b/src/agent-client-protocol-conductor/tests/mcp_over_acp_polyfill_v2.rs new file mode 100644 index 0000000..bf1548b --- /dev/null +++ b/src/agent-client-protocol-conductor/tests/mcp_over_acp_polyfill_v2.rs @@ -0,0 +1,566 @@ +#![cfg(feature = "unstable_protocol_v2")] + +//! V2 integration coverage for the public MCP-over-ACP compatibility proxy. + +use std::{ + collections::BTreeMap, + path::PathBuf, + sync::{ + Arc, Mutex, + atomic::{AtomicUsize, Ordering}, + }, +}; + +use agent_client_protocol::{ + Agent, Client, Conductor, ConnectTo, Proxy, V2ConnectionTo, + schema::{ProtocolVersion, v2}, +}; +use agent_client_protocol_conductor::{ConductorImpl, ProxiesAndAgent}; +use agent_client_protocol_polyfill::mcp_over_acp::McpOverAcpPolyfill; +use rmcp::{ + ServiceExt as _, + transport::{ + StreamableHttpClientTransport, streamable_http_client::StreamableHttpClientTransportConfig, + }, +}; +use tokio::io::duplex; +use tokio_util::compat::{TokioAsyncReadCompatExt, TokioAsyncWriteCompatExt}; + +const SERVER_NAME: &str = "shared-v2-server"; +const SERVER_ID: &str = "shared-v2-server-id"; + +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +enum SetupMethod { + New, + Resume, +} + +#[derive(Debug)] +struct SetupRequest { + method: SetupMethod, + mcp_servers: Vec, +} + +#[derive(Default)] +struct ObservedRequests { + setup: Mutex>, +} + +impl ObservedRequests { + fn record(&self, method: SetupMethod, mcp_servers: Vec) { + self.setup + .lock() + .expect("setup request mutex should not be poisoned") + .push(SetupRequest { + method, + mcp_servers, + }); + } +} + +struct RecordingAgent { + capabilities: v2::AgentCapabilities, + observed: Arc, +} + +impl ConnectTo for RecordingAgent { + async fn connect_to( + self, + client: impl ConnectTo, + ) -> Result<(), agent_client_protocol::Error> { + let capabilities = self.capabilities; + let new_observed = Arc::clone(&self.observed); + let resume_observed = self.observed; + + Agent + .v2() + .name("recording-v2-agent") + .on_receive_request( + async move |request: v2::InitializeRequest, responder, _cx| { + assert_eq!(request.protocol_version, ProtocolVersion::V2); + responder.respond( + v2::InitializeResponse::new( + request.protocol_version, + implementation("recording-v2-agent"), + ) + .capabilities(capabilities.clone()), + ) + }, + agent_client_protocol::on_receive_request!(), + ) + .on_receive_request( + async move |request: v2::NewSessionRequest, responder, _cx| { + new_observed.record(SetupMethod::New, request.mcp_servers); + responder.respond(v2::NewSessionResponse::new("v2-session-id")) + }, + agent_client_protocol::on_receive_request!(), + ) + .on_receive_request( + async move |request: v2::ResumeSessionRequest, responder, _cx| { + resume_observed.record(SetupMethod::Resume, request.mcp_servers); + responder.respond(v2::ResumeSessionResponse::new()) + }, + agent_client_protocol::on_receive_request!(), + ) + .connect_to(client) + .await + } +} + +struct NativeMcpProvider { + connect_count: Arc, + request_methods: Arc>>, + notification_methods: Arc>>, + disconnect_count: Arc, +} + +impl ConnectTo for NativeMcpProvider { + async fn connect_to( + self, + client: impl ConnectTo, + ) -> Result<(), agent_client_protocol::Error> { + let request_methods = Arc::clone(&self.request_methods); + let notification_methods = Arc::clone(&self.notification_methods); + let disconnect_count = Arc::clone(&self.disconnect_count); + + Proxy + .builder() + .name("native-v2-mcp-provider") + .on_receive_request_from( + Agent, + async move |request: v2::ConnectMcpRequest, responder, _cx| { + assert_eq!(request.server_id.to_string(), SERVER_ID); + self.connect_count.fetch_add(1, Ordering::SeqCst); + responder.respond(v2::ConnectMcpResponse::new("v2-test-connection-id")) + }, + agent_client_protocol::on_receive_request!(), + ) + .on_receive_request_from( + Agent, + async move |request: v2::MessageMcpRequest, responder, _cx| { + request_methods + .lock() + .expect("request method mutex should not be poisoned") + .push(request.method.clone()); + match request.method.as_str() { + "initialize" => { + let protocol_version = request + .params + .as_ref() + .and_then(|params| params.get("protocolVersion")) + .cloned() + .unwrap_or_else(|| serde_json::json!("2025-06-18")); + responder.respond(serde_json::from_value(serde_json::json!({ + "protocolVersion": protocol_version, + "capabilities": { + "tools": {} + }, + "serverInfo": { + "name": "v2-polyfill-test-mcp-server", + "version": env!("CARGO_PKG_VERSION") + } + }))?) + } + "tools/list" => responder + .respond(serde_json::from_value(serde_json::json!({ "tools": [] }))?), + method => responder.respond_with_error( + agent_client_protocol::Error::method_not_found().data(method), + ), + } + }, + agent_client_protocol::on_receive_request!(), + ) + .on_receive_notification_from( + Agent, + async move |notification: v2::MessageMcpNotification, _cx| { + notification_methods + .lock() + .expect("notification method mutex should not be poisoned") + .push(notification.method); + Ok(()) + }, + agent_client_protocol::on_receive_notification!(), + ) + .on_receive_request_from( + Agent, + async move |_request: v2::DisconnectMcpRequest, responder, _cx| { + disconnect_count.fetch_add(1, Ordering::SeqCst); + responder.respond(v2::DisconnectMcpResponse::new()) + }, + agent_client_protocol::on_receive_request!(), + ) + .connect_to(client) + .await + } +} + +fn implementation(name: &str) -> v2::Implementation { + v2::Implementation::new(name, env!("CARGO_PKG_VERSION")) +} + +fn agent_capabilities(mcp: v2::McpCapabilities) -> v2::AgentCapabilities { + v2::AgentCapabilities::new().session(v2::SessionCapabilities::new().mcp(mcp)) +} + +fn initialize_request() -> v2::InitializeRequest { + v2::InitializeRequest::new( + ProtocolVersion::V2, + implementation("v2-polyfill-test-client"), + ) +} + +fn server_meta() -> v2::Meta { + let mut meta = v2::Meta::new(); + meta.insert( + "source".to_owned(), + serde_json::Value::String("v2-integration-test".to_owned()), + ); + meta +} + +fn native_server() -> v2::McpServer { + v2::McpServer::Acp(v2::McpServerAcp::new(SERVER_NAME, SERVER_ID).meta(server_meta())) +} + +fn future_server() -> v2::McpServer { + v2::McpServer::Other(v2::OtherMcpServer::new( + "_future_transport", + BTreeMap::from([ + ("name".to_owned(), serde_json::json!("future-v2-server")), + ( + "configuration".to_owned(), + serde_json::json!({ "preserve": true }), + ), + ]), + )) +} + +fn test_servers() -> Vec { + vec![native_server(), future_server()] +} + +async fn run_with_polyfill( + agent: RecordingAgent, + provider_connect_count: Arc, + provider_request_methods: Arc>>, + provider_notification_methods: Arc>>, + provider_disconnect_count: Arc, + editor_task: impl AsyncFnOnce(V2ConnectionTo) -> Result<(), agent_client_protocol::Error>, +) -> Result<(), agent_client_protocol::Error> { + let (editor_out, conductor_in) = duplex(4096); + let (conductor_out, editor_in) = duplex(4096); + let transport = + agent_client_protocol::ByteStreams::new(editor_out.compat_write(), editor_in.compat()); + + Client + .v2() + .name("v2-polyfill-test-client") + .with_spawned(|_cx| async move { + ConductorImpl::new_agent( + "v2-polyfill-test-conductor", + ProxiesAndAgent::new(agent) + .proxy(NativeMcpProvider { + connect_count: provider_connect_count, + request_methods: provider_request_methods, + notification_methods: provider_notification_methods, + disconnect_count: provider_disconnect_count, + }) + .proxy(McpOverAcpPolyfill::http()), + ) + .run(agent_client_protocol::ByteStreams::new( + conductor_out.compat_write(), + conductor_in.compat(), + )) + .await + }) + .connect_with(transport, editor_task) + .await +} + +fn negotiated_mcp_capabilities(response: &v2::InitializeResponse) -> &v2::McpCapabilities { + response + .capabilities + .session + .as_ref() + .expect("the test agent should advertise session support") + .mcp + .as_ref() + .expect("the test agent should advertise MCP support") +} + +#[tokio::test] +async fn http_downstream_adapts_v2_capabilities_and_only_transforms_native_servers() +-> Result<(), agent_client_protocol::Error> { + let observed = Arc::new(ObservedRequests::default()); + let agent = RecordingAgent { + capabilities: agent_capabilities( + v2::McpCapabilities::new().http(v2::McpHttpCapabilities::new()), + ), + observed: Arc::clone(&observed), + }; + let connect_count = Arc::new(AtomicUsize::new(0)); + let request_methods = Arc::new(Mutex::new(Vec::new())); + let notification_methods = Arc::new(Mutex::new(Vec::new())); + let disconnect_count = Arc::new(AtomicUsize::new(0)); + + run_with_polyfill( + agent, + Arc::clone(&connect_count), + Arc::clone(&request_methods), + Arc::clone(¬ification_methods), + Arc::clone(&disconnect_count), + async |connection| { + let initialize = connection + .send_request(initialize_request()) + .block_task() + .await?; + let mcp = negotiated_mcp_capabilities(&initialize); + assert!(mcp.http.is_some()); + assert!( + mcp.acp.is_some(), + "the HTTP adapter should advertise v2 native MCP support upstream" + ); + + let cwd = PathBuf::from("/tmp"); + let session = connection + .send_request(v2::NewSessionRequest::new(cwd.clone()).mcp_servers(test_servers())) + .block_task() + .await?; + connection + .send_request( + v2::ResumeSessionRequest::new(session.session_id, cwd) + .mcp_servers(test_servers()), + ) + .block_task() + .await?; + + let endpoint = { + let setup = observed + .setup + .lock() + .expect("setup request mutex should not be poisoned"); + let v2::McpServer::Http(server) = &setup[0].mcp_servers[0] else { + panic!("expected the native declaration to be adapted to HTTP") + }; + server.url.clone() + }; + let mcp_client = () + .serve(StreamableHttpClientTransport::from_config( + StreamableHttpClientTransportConfig::with_uri(endpoint), + )) + .await + .map_err(agent_client_protocol::Error::into_internal_error)?; + let tools = mcp_client + .list_tools(None) + .await + .map_err(agent_client_protocol::Error::into_internal_error)?; + assert!(tools.tools.is_empty()); + mcp_client + .cancel() + .await + .map_err(agent_client_protocol::Error::into_internal_error)?; + Ok(()) + }, + ) + .await?; + + let setup = observed + .setup + .lock() + .expect("setup request mutex should not be poisoned"); + assert_eq!( + connect_count.load(Ordering::SeqCst), + 1, + "one reused listener should create one v2 native MCP connection" + ); + assert_eq!( + *request_methods + .lock() + .expect("request method mutex should not be poisoned"), + ["initialize", "tools/list"] + ); + assert_eq!( + *notification_methods + .lock() + .expect("notification method mutex should not be poisoned"), + ["notifications/initialized"] + ); + assert_eq!(setup.len(), 2); + assert_eq!(setup[0].method, SetupMethod::New); + assert_eq!(setup[1].method, SetupMethod::Resume); + + let expected_future_server = future_server(); + let expected_meta = server_meta(); + let mut endpoint = None; + for request in setup.iter() { + assert_eq!(request.mcp_servers.len(), 2); + let v2::McpServer::Http(server) = &request.mcp_servers[0] else { + panic!( + "expected the ACP declaration to become HTTP for {:?}, got {:?}", + request.method, request.mcp_servers + ); + }; + assert_eq!(server.name, SERVER_NAME); + assert_eq!(server.meta.as_ref(), Some(&expected_meta)); + assert!(server.headers.is_empty()); + assert!(server.url.starts_with("http://127.0.0.1:")); + assert_eq!( + request.mcp_servers[1], expected_future_server, + "the polyfill must preserve custom v2 MCP transports" + ); + if let Some(endpoint) = &endpoint { + assert_eq!( + &server.url, endpoint, + "the same ACP server ID should reuse one listener" + ); + } else { + endpoint = Some(server.url.clone()); + } + } + + Ok(()) +} + +#[tokio::test] +async fn native_v2_downstream_keeps_capability_and_declarations_unchanged() +-> Result<(), agent_client_protocol::Error> { + let observed = Arc::new(ObservedRequests::default()); + let agent = RecordingAgent { + capabilities: agent_capabilities( + v2::McpCapabilities::new().acp(v2::McpAcpCapabilities::new()), + ), + observed: Arc::clone(&observed), + }; + let expected = test_servers(); + let connect_count = Arc::new(AtomicUsize::new(0)); + let request_methods = Arc::new(Mutex::new(Vec::new())); + let notification_methods = Arc::new(Mutex::new(Vec::new())); + let disconnect_count = Arc::new(AtomicUsize::new(0)); + + run_with_polyfill( + agent, + Arc::clone(&connect_count), + Arc::clone(&request_methods), + Arc::clone(¬ification_methods), + Arc::clone(&disconnect_count), + async move |connection| { + let initialize = connection + .send_request(initialize_request()) + .block_task() + .await?; + let mcp = negotiated_mcp_capabilities(&initialize); + assert!(mcp.http.is_none()); + assert!(mcp.acp.is_some()); + + connection + .send_request( + v2::NewSessionRequest::new(PathBuf::from("/tmp")).mcp_servers(expected.clone()), + ) + .block_task() + .await?; + Ok(()) + }, + ) + .await?; + + let setup = observed + .setup + .lock() + .expect("setup request mutex should not be poisoned"); + assert_eq!(setup.len(), 1); + assert_eq!(setup[0].mcp_servers, test_servers()); + assert_eq!( + connect_count.load(Ordering::SeqCst), + 0, + "a native-capable v2 downstream should bypass the HTTP adapter" + ); + assert!( + request_methods + .lock() + .expect("request method mutex should not be poisoned") + .is_empty() + ); + assert!( + notification_methods + .lock() + .expect("notification method mutex should not be poisoned") + .is_empty() + ); + assert_eq!(disconnect_count.load(Ordering::SeqCst), 0); + + Ok(()) +} + +#[tokio::test] +async fn unavailable_v2_downstream_rejects_native_declarations() +-> Result<(), agent_client_protocol::Error> { + let observed = Arc::new(ObservedRequests::default()); + let agent = RecordingAgent { + capabilities: agent_capabilities(v2::McpCapabilities::new()), + observed: Arc::clone(&observed), + }; + let connect_count = Arc::new(AtomicUsize::new(0)); + let request_methods = Arc::new(Mutex::new(Vec::new())); + let notification_methods = Arc::new(Mutex::new(Vec::new())); + let disconnect_count = Arc::new(AtomicUsize::new(0)); + + run_with_polyfill( + agent, + Arc::clone(&connect_count), + Arc::clone(&request_methods), + Arc::clone(¬ification_methods), + Arc::clone(&disconnect_count), + async move |connection| { + let initialize = connection + .send_request(initialize_request()) + .block_task() + .await?; + let mcp = negotiated_mcp_capabilities(&initialize); + assert!(mcp.http.is_none()); + assert!(mcp.acp.is_none()); + + let error = connection + .send_request( + v2::NewSessionRequest::new(PathBuf::from("/tmp")) + .mcp_servers(vec![native_server()]), + ) + .block_task() + .await + .expect_err("native MCP should require a downstream transport"); + assert_eq!(error.code, agent_client_protocol::ErrorCode::InvalidParams); + assert_eq!( + error.data, + Some(serde_json::json!( + "the downstream agent supports neither native nor HTTP MCP transport" + )) + ); + Ok(()) + }, + ) + .await?; + + assert!( + observed + .setup + .lock() + .expect("setup request mutex should not be poisoned") + .is_empty(), + "the rejected request must not reach the downstream agent" + ); + assert_eq!(connect_count.load(Ordering::SeqCst), 0); + assert!( + request_methods + .lock() + .expect("request method mutex should not be poisoned") + .is_empty() + ); + assert!( + notification_methods + .lock() + .expect("notification method mutex should not be poisoned") + .is_empty() + ); + assert_eq!(disconnect_count.load(Ordering::SeqCst), 0); + + Ok(()) +} diff --git a/src/agent-client-protocol-polyfill/CHANGELOG.md b/src/agent-client-protocol-polyfill/CHANGELOG.md index ac15766..8098b69 100644 --- a/src/agent-client-protocol-polyfill/CHANGELOG.md +++ b/src/agent-client-protocol-polyfill/CHANGELOG.md @@ -7,6 +7,19 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +### Added + +- Add an `unstable_protocol_v2` feature for using `McpOverAcpPolyfill` in a + draft-v2 conductor chain. The polyfill selects the schema during proxy + initialization, advertises the v2 ACP MCP capability when adapting an + HTTP-capable agent, and handles v2 session setup and `mcp/*` messages without + interpreting them as v1. + +### Fixed + +- Forward MCP request cancellation hop by hop instead of tunneling the + loopback connection's `$/cancel_request` ID inside `mcp/message`. + ## [2.0.0](https://github.com/agentclientprotocol/rust-sdk/compare/agent-client-protocol-polyfill-v1.3.0...agent-client-protocol-polyfill-v2.0.0) - 2026-07-23 ### Breaking changes diff --git a/src/agent-client-protocol-polyfill/Cargo.toml b/src/agent-client-protocol-polyfill/Cargo.toml index 07094b8..7ee9c33 100644 --- a/src/agent-client-protocol-polyfill/Cargo.toml +++ b/src/agent-client-protocol-polyfill/Cargo.toml @@ -13,6 +13,7 @@ categories = ["development-tools"] [features] default = [] +unstable_protocol_v2 = ["agent-client-protocol/unstable_protocol_v2"] unstable_session_fork = ["agent-client-protocol/unstable_session_fork"] [dependencies] diff --git a/src/agent-client-protocol-polyfill/src/mcp_over_acp/mod.rs b/src/agent-client-protocol-polyfill/src/mcp_over_acp/mod.rs index 14b4b28..8b15edf 100644 --- a/src/agent-client-protocol-polyfill/src/mcp_over_acp/mod.rs +++ b/src/agent-client-protocol-polyfill/src/mcp_over_acp/mod.rs @@ -1,9 +1,13 @@ //! MCP-over-ACP compatibility proxy. //! -//! This proxy adapts schema-native [`McpServer::Acp`] declarations for agents that do not +//! This proxy adapts schema-native `McpServer::Acp` declarations for agents that do not //! support the ACP MCP transport. It replaces those declarations with loopback HTTP bridges and //! relays `mcp/connect`, `mcp/message`, and `mcp/disconnect` over ACP. //! +//! Stable protocol v1 is supported by default. Enable the crate's +//! `unstable_protocol_v2` feature to use the same proxy in a draft-v2 conductor +//! chain. +//! //! # Usage //! //! ```rust,ignore @@ -17,62 +21,37 @@ mod actor; pub(crate) mod http; +mod protocol; use std::collections::HashMap; use agent_client_protocol::{ - Agent, Client, Conductor, ConnectTo, ConnectionTo, Dispatch, Handled, Proxy, Responder, - UntypedMessage, - schema::{ - InitializeProxyRequest, - v1::{ - AgentNotification, AgentRequest, ConnectMcpRequest, ConnectMcpResponse, - DisconnectMcpRequest, DisconnectMcpResponse, LoadSessionRequest, McpConnectionId, - McpServer, McpServerAcp, McpServerHttp, MessageMcpNotification, MessageMcpRequest, - NewSessionRequest, ResumeSessionRequest, - }, - }, + Agent, Client, Conductor, ConnectTo, ConnectionTo, Dispatch, HandleDispatchFrom, Handled, + Proxy, Responder, UntypedMessage, is_cancel_request_notification, util::MatchDispatchFrom, }; use futures::{SinkExt, channel::mpsc, channel::oneshot}; +use serde_json::Value; use tokio::net::TcpListener; use tracing::{debug, info, warn}; use self::actor::BridgeConnectionActor; - -#[cfg(feature = "unstable_session_fork")] -use agent_client_protocol::schema::v1::ForkSessionRequest; - -#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)] -pub(crate) enum DownstreamMcpMode { - #[default] - Unknown, - Native, - HttpAdapter, - Unavailable, -} - -impl DownstreamMcpMode { - fn from_capabilities(http: bool, acp: bool) -> Self { - if acp { - Self::Native - } else if http { - Self::HttpAdapter - } else { - Self::Unavailable - } - } -} +use self::protocol::{ + DownstreamMcpMode, NativeMcpMessage, NativeServer, PolyfillProtocol, native_params_into_value, +}; /// Internal messages for the polyfill's bridge management. #[derive(Debug)] pub(crate) enum BridgeMessage { - /// Record which MCP transport the successor can consume. - SetDownstreamMode(DownstreamMcpMode), + /// Record the selected ACP schema and which MCP transport the successor can consume. + SetProtocol { + protocol: PolyfillProtocol, + downstream_mode: DownstreamMcpMode, + }, /// Transform the MCP declarations for one session setup request. TransformServers { - servers: Vec, - response_tx: oneshot::Sender, agent_client_protocol::Error>>, + servers: Vec, + response_tx: oneshot::Sender, agent_client_protocol::Error>>, }, /// A new TCP connection was accepted and needs a native MCP connection ID. @@ -85,7 +64,7 @@ pub(crate) enum BridgeMessage { /// A native MCP connection ID was received; spawn the actor and store its sender. ConnectionEstablished { server_id: String, - connection_id: McpConnectionId, + connection_id: String, actor: BridgeConnectionActor, connection: BridgeConnection, }, @@ -101,14 +80,12 @@ pub(crate) enum BridgeMessage { /// An MCP server request received over ACP for the local agent's MCP client. ServerToClientRequest { - request: MessageMcpRequest, + request: NativeMcpMessage, responder: Responder, }, /// An MCP server notification received over ACP for the local agent's MCP client. - ServerToClientNotification { - notification: MessageMcpNotification, - }, + ServerToClientNotification { notification: NativeMcpMessage }, /// The local MCP bridge disconnected. Disconnected { connection_id: String }, @@ -152,152 +129,187 @@ impl ConnectTo for McpOverAcpPolyfill { ) -> Result<(), agent_client_protocol::Error> { let (bridge_tx, bridge_rx) = mpsc::channel(128); - let builder = Proxy + Proxy .builder() .name("mcp-over-acp-polyfill") .with_runner(BridgeRunner { bridge_tx: bridge_tx.clone(), bridge_rx, + protocol: None, downstream_mode: DownstreamMcpMode::Unknown, listeners: BridgeListeners::default(), bridge_connections: HashMap::new(), }) - .on_receive_request_from( - Client, - { - let bridge_tx = bridge_tx.clone(); - async move |request: InitializeProxyRequest, - responder, - cx: ConnectionTo| { - let mut response_bridge_tx = bridge_tx.clone(); - cx.send_request_to(Agent, request.initialize) - .on_receiving_result(async move |result| { - let result = match result { - Ok(mut response) => { - let capabilities = - &mut response.agent_capabilities.mcp_capabilities; - let mode = DownstreamMcpMode::from_capabilities( - capabilities.http, - capabilities.acp, - ); - response_bridge_tx - .send(BridgeMessage::SetDownstreamMode(mode)) - .await - .map_err( - agent_client_protocol::Error::into_internal_error, - )?; - if mode == DownstreamMcpMode::HttpAdapter { - capabilities.acp = true; - } - Ok(response) - } - Err(error) => Err(error), - }; - responder.respond_with_result(result) - }) - } - }, - agent_client_protocol::on_receive_request!(), - ) - .on_receive_request_from( - Client, - { - let mut bridge_tx = bridge_tx.clone(); - async move |mut request: NewSessionRequest, - responder, - cx: ConnectionTo| { - transform_session_servers(&mut request.mcp_servers, &mut bridge_tx).await?; - cx.send_request_to(Agent, request) - .forward_response_to(responder) - } - }, - agent_client_protocol::on_receive_request!(), - ) - .on_receive_request_from( - Client, - { - let mut bridge_tx = bridge_tx.clone(); - async move |mut request: LoadSessionRequest, - responder, - cx: ConnectionTo| { - transform_session_servers(&mut request.mcp_servers, &mut bridge_tx).await?; - cx.send_request_to(Agent, request) - .forward_response_to(responder) - } - }, - agent_client_protocol::on_receive_request!(), - ) - .on_receive_request_from( - Client, - { - let mut bridge_tx = bridge_tx.clone(); - async move |mut request: ResumeSessionRequest, - responder, - cx: ConnectionTo| { - transform_session_servers(&mut request.mcp_servers, &mut bridge_tx).await?; - cx.send_request_to(Agent, request) - .forward_response_to(responder) - } - }, - agent_client_protocol::on_receive_request!(), - ); + .with_handler(PolyfillHandler { + protocol: None, + bridge_tx, + }) + .connect_to(client) + .await + } +} - #[cfg(feature = "unstable_session_fork")] - let builder = builder.on_receive_request_from( - Client, - { - let mut bridge_tx = bridge_tx.clone(); - async move |mut request: ForkSessionRequest, - responder, - cx: ConnectionTo| { - transform_session_servers(&mut request.mcp_servers, &mut bridge_tx).await?; - cx.send_request_to(Agent, request) - .forward_response_to(responder) - } - }, - agent_client_protocol::on_receive_request!(), - ); +#[derive(Debug)] +struct PolyfillHandler { + protocol: Option, + bridge_tx: mpsc::Sender, +} - builder - .on_receive_request_from( - Client, - { - let mut bridge_tx = bridge_tx.clone(); - async move |request: MessageMcpRequest, responder, _cx| { - bridge_tx - .send(BridgeMessage::ServerToClientRequest { - request, - responder: responder.erase_to_json(), - }) - .await - .map_err(agent_client_protocol::Error::into_internal_error)?; - Ok(Handled::Yes) - } - }, - agent_client_protocol::on_receive_request!(), - ) - .on_receive_notification_from( - Client, - { - let mut bridge_tx = bridge_tx.clone(); - async move |notification: MessageMcpNotification, _cx| { - bridge_tx - .send(BridgeMessage::ServerToClientNotification { notification }) - .await - .map_err(agent_client_protocol::Error::into_internal_error) - } - }, - agent_client_protocol::on_receive_notification!(), - ) - .connect_to(client) +impl HandleDispatchFrom for PolyfillHandler { + async fn handle_dispatch_from( + &mut self, + message: Dispatch, + cx: ConnectionTo, + ) -> Result, agent_client_protocol::Error> { + MatchDispatchFrom::new(message, &cx) + .if_dispatch_from(Client, async |message: Dispatch| { + self.handle_client_dispatch(message, &cx).await + }) .await + .done() + } + + fn describe_chain(&self) -> impl std::fmt::Debug { + self + } +} + +impl PolyfillHandler { + async fn handle_client_dispatch( + &mut self, + message: Dispatch, + cx: &ConnectionTo, + ) -> Result, agent_client_protocol::Error> { + match message { + Dispatch::Request(request, responder) => { + self.handle_client_request(request, responder, cx).await + } + Dispatch::Notification(notification) => { + self.handle_client_notification(notification).await + } + message @ Dispatch::Response(_, _) => Ok(Handled::No { + message, + retry: false, + }), + } } + + async fn handle_client_request( + &mut self, + mut request: UntypedMessage, + responder: Responder, + cx: &ConnectionTo, + ) -> Result, agent_client_protocol::Error> { + if request.method() == agent_client_protocol::schema::METHOD_INITIALIZE_PROXY { + if self.protocol.is_some() { + return Err(agent_client_protocol::Error::invalid_request() + .data("MCP-over-ACP polyfill was already initialized")); + } + let protocol = PolyfillProtocol::from_initialize_request(&request)?; + self.protocol = Some(protocol); + request.method = "initialize".to_string(); + + let sent = cx.send_request_to(Agent, request); + let sent = sent.forward_cancellation_from(responder.cancellation()); + let mut bridge_tx = self.bridge_tx.clone(); + sent.on_receiving_result(async move |result| { + let result = match result { + Ok(response) => { + adapt_initialize_response(protocol, response, &mut bridge_tx).await + } + Err(error) => Err(error), + }; + responder.respond_with_result(result) + })?; + return Ok(Handled::Yes); + } + + let Some(protocol) = self.protocol else { + return Ok(Handled::No { + message: Dispatch::Request(request, responder), + retry: false, + }); + }; + + if protocol.is_session_setup_method(request.method()) { + protocol.validate_session_setup_request(&request)?; + transform_session_servers(&mut request, &mut self.bridge_tx).await?; + cx.send_request_to(Agent, request) + .forward_response_to(responder)?; + return Ok(Handled::Yes); + } + + if request.method() == "mcp/message" { + let request = protocol.parse_message_request(request)?; + self.bridge_tx + .send(BridgeMessage::ServerToClientRequest { request, responder }) + .await + .map_err(agent_client_protocol::Error::into_internal_error)?; + return Ok(Handled::Yes); + } + + Ok(Handled::No { + message: Dispatch::Request(request, responder), + retry: false, + }) + } + + async fn handle_client_notification( + &mut self, + notification: UntypedMessage, + ) -> Result, agent_client_protocol::Error> { + let Some(protocol) = self.protocol else { + return Ok(Handled::No { + message: Dispatch::Notification(notification), + retry: false, + }); + }; + + if notification.method() == "mcp/message" { + let notification = protocol.parse_message_notification(notification)?; + self.bridge_tx + .send(BridgeMessage::ServerToClientNotification { notification }) + .await + .map_err(agent_client_protocol::Error::into_internal_error)?; + return Ok(Handled::Yes); + } + + Ok(Handled::No { + message: Dispatch::Notification(notification), + retry: false, + }) + } +} + +async fn adapt_initialize_response( + protocol: PolyfillProtocol, + mut response: Value, + bridge_tx: &mut mpsc::Sender, +) -> Result { + let downstream_mode = protocol.transform_initialize_response(&mut response)?; + bridge_tx + .send(BridgeMessage::SetProtocol { + protocol, + downstream_mode, + }) + .await + .map_err(agent_client_protocol::Error::into_internal_error)?; + Ok(response) } async fn transform_session_servers( - servers: &mut Vec, + request: &mut UntypedMessage, bridge_tx: &mut mpsc::Sender, ) -> Result<(), agent_client_protocol::Error> { + let Some(servers) = request + .params + .as_object_mut() + .and_then(|params| params.get_mut("mcpServers")) + .and_then(Value::as_array_mut) + else { + return Ok(()); + }; + let (response_tx, response_rx) = oneshot::channel(); bridge_tx .send(BridgeMessage::TransformServers { @@ -323,11 +335,12 @@ struct BridgeListener { } impl BridgeListener { - fn declaration(&self, server: McpServerAcp) -> McpServer { - McpServer::Http( - McpServerHttp::new(server.name, format!("http://127.0.0.1:{}", self.tcp_port)) - .meta(server.meta), - ) + fn declaration( + &self, + protocol: PolyfillProtocol, + server: NativeServer, + ) -> Result { + server.http_declaration(protocol, format!("http://127.0.0.1:{}", self.tcp_port)) } } @@ -335,12 +348,16 @@ impl BridgeListeners { async fn transform_servers( &mut self, connection: &ConnectionTo, - servers: Vec, + protocol: PolyfillProtocol, + servers: Vec, bridge_tx: &mpsc::Sender, - ) -> Result, agent_client_protocol::Error> { + ) -> Result, agent_client_protocol::Error> { let mut transformed = Vec::with_capacity(servers.len()); for server in servers { - transformed.push(self.transform_server(connection, server, bridge_tx).await?); + transformed.push( + self.transform_server(connection, protocol, server, bridge_tx) + .await?, + ); } Ok(transformed) } @@ -348,22 +365,23 @@ impl BridgeListeners { async fn transform_server( &mut self, connection: &ConnectionTo, - server: McpServer, + protocol: PolyfillProtocol, + server: Value, bridge_tx: &mpsc::Sender, - ) -> Result { - let McpServer::Acp(acp_server) = server else { + ) -> Result { + let Some(native_server) = protocol.native_server(server.clone()) else { return Ok(server); }; - let server_id = acp_server.server_id.to_string(); + let server_id = native_server.server_id.clone(); info!( - server_name = %acp_server.name, + server_name = %native_server.name, server_id, "detected native MCP-over-ACP server; creating compatibility bridge" ); if let Some(listener) = self.listeners.get(&server_id) { - return Ok(listener.declaration(acp_server)); + return listener.declaration(protocol, native_server); } let tcp_listener = TcpListener::bind("127.0.0.1:0") @@ -387,7 +405,7 @@ impl BridgeListeners { } })?; - let declaration = listener.declaration(acp_server); + let declaration = listener.declaration(protocol, native_server)?; self.listeners.insert(server_id, listener); Ok(declaration) } @@ -406,6 +424,7 @@ struct ActiveBridgeConnection { struct BridgeRunner { bridge_tx: mpsc::Sender, bridge_rx: mpsc::Receiver, + protocol: Option, downstream_mode: DownstreamMcpMode, listeners: BridgeListeners, bridge_connections: HashMap, @@ -414,6 +433,7 @@ struct BridgeRunner { impl std::fmt::Debug for BridgeRunner { fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { f.debug_struct("BridgeRunner") + .field("protocol", &self.protocol) .field("downstream_mode", &self.downstream_mode) .field("listeners", &self.listeners.listeners.len()) .field("bridge_connections", &self.bridge_connections.len()) @@ -430,29 +450,37 @@ impl agent_client_protocol::RunWithConnectionTo for BridgeRunner { while let Some(message) = self.bridge_rx.next().await { match message { - BridgeMessage::SetDownstreamMode(mode) => { - self.downstream_mode = mode; + BridgeMessage::SetProtocol { + protocol, + downstream_mode, + } => { + self.protocol = Some(protocol); + self.downstream_mode = downstream_mode; } BridgeMessage::TransformServers { servers, response_tx, } => { - let result = match self.downstream_mode { - DownstreamMcpMode::Native => Ok(servers), - DownstreamMcpMode::HttpAdapter => { + let result = match (self.protocol, self.downstream_mode) { + (Some(_), DownstreamMcpMode::Native) => Ok(servers), + (Some(protocol), DownstreamMcpMode::HttpAdapter) => { self.listeners - .transform_servers(&connection, servers, &self.bridge_tx) + .transform_servers(&connection, protocol, servers, &self.bridge_tx) .await } - DownstreamMcpMode::Unavailable => reject_native_servers( + (Some(protocol), DownstreamMcpMode::Unavailable) => reject_native_servers( + protocol, servers, "the downstream agent supports neither native nor HTTP MCP transport", ), - DownstreamMcpMode::Unknown => reject_native_servers( + (Some(protocol), DownstreamMcpMode::Unknown) => reject_native_servers( + protocol, servers, "MCP transport capabilities are unavailable before initialize", ), + (None, _) => Err(agent_client_protocol::Error::invalid_request() + .data("MCP transport capabilities are unavailable before initialize")), }; drop(response_tx.send(result)); } @@ -462,29 +490,32 @@ impl agent_client_protocol::RunWithConnectionTo for BridgeRunner { actor, connection: bridge, } => { - let request = - AgentRequest::ConnectMcpRequest(ConnectMcpRequest::new(server_id.clone())); + let Some(protocol) = self.protocol else { + warn!( + server_id, + "cannot open MCP bridge before ACP initialization" + ); + self.listeners.remove(&server_id); + continue; + }; + let request = protocol.connect_request(server_id.clone())?; let mut bridge_tx = self.bridge_tx.clone(); let scheduled = connection .send_request_to(Client, request) .on_receiving_result(async move |result| { let message = match result { - Ok(response) => { - match serde_json::from_value::(response) { - Ok(ConnectMcpResponse { connection_id, .. }) => { - BridgeMessage::ConnectionEstablished { - server_id, - connection_id, - actor, - connection: bridge, - } - } - Err(error) => { - warn!(?error, "invalid response to mcp/connect"); - BridgeMessage::ConnectionFailed { server_id } - } + Ok(response) => match protocol.connect_response_id(response) { + Ok(connection_id) => BridgeMessage::ConnectionEstablished { + server_id, + connection_id, + actor, + connection: bridge, + }, + Err(error) => { + warn!(?error, "invalid response to mcp/connect"); + BridgeMessage::ConnectionFailed { server_id } } - } + }, Err(error) => { warn!(?error, "mcp/connect failed"); BridgeMessage::ConnectionFailed { server_id } @@ -504,7 +535,6 @@ impl agent_client_protocol::RunWithConnectionTo for BridgeRunner { actor, connection: bridge, } => { - let connection_id = connection_id.to_string(); self.bridge_connections.insert( connection_id.clone(), ActiveBridgeConnection { server_id, bridge }, @@ -519,64 +549,89 @@ impl agent_client_protocol::RunWithConnectionTo for BridgeRunner { BridgeMessage::ClientToServer { connection_id, message, - } => match message { - Dispatch::Request(message, responder) => { - match message_mcp_request(connection_id, message) { - Ok(request) => { - let pending = connection.send_request_to( - Client, - AgentRequest::MessageMcpRequest(request), - ); - if let Err(error) = pending.forward_response_to(responder) { - warn!(?error, "could not forward local MCP request response"); + } => { + let Some(protocol) = self.protocol else { + let rejection = match message { + Dispatch::Request(_, responder) => responder + .respond_with_internal_error( + "ACP protocol is unavailable before initialize", + ), + Dispatch::Notification(_) | Dispatch::Response(_, _) => Ok(()), + }; + if let Err(error) = rejection { + debug!(?error, "could not reject MCP request before initialize"); + } + continue; + }; + + match message { + Dispatch::Request(message, responder) => { + match protocol.message_request(connection_id, message) { + Ok(request) => { + let pending = connection.send_request_to(Client, request); + if let Err(error) = pending.forward_response_to(responder) { + warn!( + ?error, + "could not forward local MCP request response" + ); + } } - } - Err(error) => { - if let Err(send_error) = responder.respond_with_error(error) { - debug!(?send_error, "could not reject malformed MCP request"); + Err(error) => { + if let Err(send_error) = responder.respond_with_error(error) { + debug!( + ?send_error, + "could not reject malformed MCP request" + ); + } } } } - } - Dispatch::Notification(message) => { - match message_mcp_notification(connection_id, message) { - Ok(notification) => { - if let Err(error) = connection.send_notification_to( - Client, - AgentNotification::MessageMcpNotification(notification), - ) { + Dispatch::Notification(message) => { + match local_mcp_notification(protocol, connection_id, message) { + Ok(Some(notification)) => { + if let Err(error) = + connection.send_notification_to(Client, notification) + { + warn!(?error, "could not forward local MCP notification"); + } + } + Ok(None) => { + debug!( + "not tunneling hop-scoped MCP cancellation through mcp/message" + ); + } + Err(error) => { warn!(?error, "could not forward local MCP notification"); } } - Err(error) => { - warn!(?error, "discarding malformed local MCP notification"); - } } - } - Dispatch::Response(result, router) => { - if let Err(error) = router.route_with_result(result) { - debug!(?error, "could not route MCP client response"); + Dispatch::Response(result, router) => { + if let Err(error) = router.route_with_result(result) { + debug!(?error, "could not route MCP client response"); + } } } - }, + } BridgeMessage::ServerToClientRequest { request, responder } => { match self.downstream_mode { DownstreamMcpMode::Native => { - let pending = connection - .send_request_to(Agent, AgentRequest::MessageMcpRequest(request)); + let pending = connection.send_request_to(Agent, request.raw); if let Err(error) = pending.forward_response_to(responder) { debug!(?error, "could not forward native MCP request"); } } DownstreamMcpMode::HttpAdapter => { - let connection_id = request.connection_id.to_string(); + let connection_id = request.connection_id; let Some(active) = self.bridge_connections.get_mut(&connection_id) else { respond_unknown_connection(responder, &connection_id); continue; }; - let message = message_mcp_request_to_untyped(request); + let message = UntypedMessage { + method: request.method, + params: native_params_into_value(request.params), + }; if let Some(message) = active .bridge .try_send(Dispatch::Request(message, responder)) @@ -609,15 +664,14 @@ impl agent_client_protocol::RunWithConnectionTo for BridgeRunner { BridgeMessage::ServerToClientNotification { notification } => { match self.downstream_mode { DownstreamMcpMode::Native => { - if let Err(error) = connection.send_notification_to( - Agent, - AgentNotification::MessageMcpNotification(notification), - ) { + if let Err(error) = + connection.send_notification_to(Agent, notification.raw) + { debug!(?error, "could not forward native MCP notification"); } } DownstreamMcpMode::HttpAdapter => { - let connection_id = notification.connection_id.to_string(); + let connection_id = notification.connection_id; let Some(active) = self.bridge_connections.get_mut(&connection_id) else { debug!( @@ -626,7 +680,10 @@ impl agent_client_protocol::RunWithConnectionTo for BridgeRunner { ); continue; }; - let message = message_mcp_notification_to_untyped(notification); + let message = UntypedMessage { + method: notification.method, + params: native_params_into_value(notification.params), + }; if active .bridge .try_send(Dispatch::Notification(message)) @@ -648,16 +705,18 @@ impl agent_client_protocol::RunWithConnectionTo for BridgeRunner { }; self.listeners.remove(&active.server_id); - let request = AgentRequest::DisconnectMcpRequest(DisconnectMcpRequest::new( - connection_id, - )); + let Some(protocol) = self.protocol else { + debug!("could not disconnect MCP bridge before ACP initialization"); + continue; + }; + let request = protocol.disconnect_request(connection_id)?; let scheduled = connection .send_request_to(Client, request) - .on_receiving_result(async |result| { + .on_receiving_result(async move |result| { match result { Ok(response) => { if let Err(error) = - serde_json::from_value::(response) + protocol.validate_disconnect_response(response) { warn!(?error, "invalid response to mcp/disconnect"); } @@ -681,13 +740,27 @@ impl agent_client_protocol::RunWithConnectionTo for BridgeRunner { } } +fn local_mcp_notification( + protocol: PolyfillProtocol, + connection_id: String, + message: UntypedMessage, +) -> Result, agent_client_protocol::Error> { + if is_cancel_request_notification(&message) { + return Ok(None); + } + protocol + .message_notification(connection_id, message) + .map(Some) +} + fn reject_native_servers( - servers: Vec, + protocol: PolyfillProtocol, + servers: Vec, reason: &'static str, -) -> Result, agent_client_protocol::Error> { +) -> Result, agent_client_protocol::Error> { if servers .iter() - .any(|server| matches!(server, McpServer::Acp(_))) + .any(|server| protocol.native_server(server.clone()).is_some()) { Err(agent_client_protocol::Error::invalid_params().data(reason)) } else { @@ -695,59 +768,6 @@ fn reject_native_servers( } } -fn into_mcp_params( - params: serde_json::Value, -) -> Result>, agent_client_protocol::Error> { - match params { - serde_json::Value::Null => Ok(None), - serde_json::Value::Object(params) => Ok(Some(params)), - params => Err( - agent_client_protocol::Error::invalid_params().data(serde_json::json!({ - "reason": "MCP message params must be an object or null", - "params": params, - })), - ), - } -} - -fn message_mcp_request( - connection_id: String, - message: UntypedMessage, -) -> Result { - let (method, params) = message.into_parts(); - let mut request = MessageMcpRequest::new(connection_id, method); - request.params = into_mcp_params(params)?; - Ok(request) -} - -fn message_mcp_notification( - connection_id: String, - message: UntypedMessage, -) -> Result { - let (method, params) = message.into_parts(); - let mut notification = MessageMcpNotification::new(connection_id, method); - notification.params = into_mcp_params(params)?; - Ok(notification) -} - -fn message_mcp_request_to_untyped(request: MessageMcpRequest) -> UntypedMessage { - UntypedMessage { - method: request.method, - params: request - .params - .map_or(serde_json::Value::Null, serde_json::Value::Object), - } -} - -fn message_mcp_notification_to_untyped(notification: MessageMcpNotification) -> UntypedMessage { - UntypedMessage { - method: notification.method, - params: notification - .params - .map_or(serde_json::Value::Null, serde_json::Value::Object), - } -} - fn respond_unknown_connection(responder: Responder, connection_id: &str) { let error = agent_client_protocol::Error::invalid_params().data(serde_json::json!({ "reason": "unknown MCP connection", @@ -766,16 +786,17 @@ mod tests { use std::collections::HashMap; use agent_client_protocol::{ - Client, Conductor, Dispatch, ErrorCode, Handled, Proxy, + Conductor, Dispatch, ErrorCode, Proxy, UntypedMessage, schema::v1::{ McpServer, McpServerAcp, McpServerHttp, MessageMcpNotification, MessageMcpRequest, }, }; - use futures::{SinkExt, StreamExt, channel::mpsc}; + use futures::{StreamExt, channel::mpsc}; use super::{ - ActiveBridgeConnection, BridgeConnection, BridgeListener, BridgeListeners, BridgeMessage, - BridgeRunner, DownstreamMcpMode, reject_native_servers, + ActiveBridgeConnection, BridgeConnection, BridgeListener, BridgeListeners, BridgeRunner, + DownstreamMcpMode, PolyfillHandler, PolyfillProtocol, local_mcp_notification, + reject_native_servers, }; #[test] @@ -784,10 +805,28 @@ mod tests { let first_meta = serde_json::Map::from_iter([("source".into(), "first".into())]); let second_meta = serde_json::Map::from_iter([("source".into(), "second".into())]); - let first = - listener.declaration(McpServerAcp::new("first", "shared").meta(first_meta.clone())); - let second = - listener.declaration(McpServerAcp::new("second", "shared").meta(second_meta.clone())); + let first = PolyfillProtocol::V1 + .native_server( + serde_json::to_value(McpServer::Acp( + McpServerAcp::new("first", "shared").meta(first_meta.clone()), + )) + .unwrap(), + ) + .unwrap(); + let second = PolyfillProtocol::V1 + .native_server( + serde_json::to_value(McpServer::Acp( + McpServerAcp::new("second", "shared").meta(second_meta.clone()), + )) + .unwrap(), + ) + .unwrap(); + let first: McpServer = + serde_json::from_value(listener.declaration(PolyfillProtocol::V1, first).unwrap()) + .unwrap(); + let second: McpServer = + serde_json::from_value(listener.declaration(PolyfillProtocol::V1, second).unwrap()) + .unwrap(); let McpServer::Http(first) = first else { panic!("expected HTTP declaration") @@ -823,19 +862,69 @@ mod tests { ); } + #[test] + fn local_cancellation_is_not_tunneled_as_an_mcp_message() { + let cancellation = UntypedMessage { + method: "$/cancel_request".to_string(), + params: serde_json::json!({ + "requestId": "loopback-request" + }), + }; + assert_eq!( + local_mcp_notification( + PolyfillProtocol::V1, + "native-connection".to_string(), + cancellation, + ) + .expect("cancellation filtering should not fail"), + None + ); + + let notification = UntypedMessage { + method: "notifications/progress".to_string(), + params: serde_json::json!({ + "progressToken": "token", + "progress": 0.5 + }), + }; + let wrapped = local_mcp_notification( + PolyfillProtocol::V1, + "native-connection".to_string(), + notification, + ) + .expect("the notification should serialize") + .expect("ordinary MCP notifications should be forwarded"); + assert_eq!(wrapped.method, "mcp/message"); + assert_eq!( + wrapped.params["connectionId"], + serde_json::json!("native-connection") + ); + assert_eq!( + wrapped.params["method"], + serde_json::json!("notifications/progress") + ); + } + #[test] fn unavailable_mode_rejects_only_native_declarations() { - let standard = vec![McpServer::Http(McpServerHttp::new( - "remote", - "https://example.com/mcp", - ))]; + let standard = vec![ + serde_json::to_value(McpServer::Http(McpServerHttp::new( + "remote", + "https://example.com/mcp", + ))) + .unwrap(), + ]; assert_eq!( - reject_native_servers(standard.clone(), "unsupported").unwrap(), + reject_native_servers(PolyfillProtocol::V1, standard.clone(), "unsupported").unwrap(), standard ); let error = reject_native_servers( - vec![McpServer::Acp(McpServerAcp::new("native", "server-1"))], + PolyfillProtocol::V1, + vec![ + serde_json::to_value(McpServer::Acp(McpServerAcp::new("native", "server-1"))) + .unwrap(), + ], "unsupported", ) .expect_err("native declarations require a downstream transport"); @@ -862,40 +951,15 @@ mod tests { .with_runner(BridgeRunner { bridge_tx: bridge_tx.clone(), bridge_rx, + protocol: Some(PolyfillProtocol::V1), downstream_mode: DownstreamMcpMode::HttpAdapter, listeners: BridgeListeners::default(), bridge_connections, }) - .on_receive_request_from( - Client, - { - let mut bridge_tx = bridge_tx.clone(); - async move |request: MessageMcpRequest, responder, _cx| { - bridge_tx - .send(BridgeMessage::ServerToClientRequest { - request, - responder: responder.erase_to_json(), - }) - .await - .map_err(agent_client_protocol::Error::into_internal_error)?; - Ok(Handled::Yes) - } - }, - agent_client_protocol::on_receive_request!(), - ) - .on_receive_notification_from( - Client, - { - let mut bridge_tx = bridge_tx; - async move |notification: MessageMcpNotification, _cx| { - bridge_tx - .send(BridgeMessage::ServerToClientNotification { notification }) - .await - .map_err(agent_client_protocol::Error::into_internal_error) - } - }, - agent_client_protocol::on_receive_notification!(), - ); + .with_handler(PolyfillHandler { + protocol: Some(PolyfillProtocol::V1), + bridge_tx, + }); Conductor .builder() diff --git a/src/agent-client-protocol-polyfill/src/mcp_over_acp/protocol.rs b/src/agent-client-protocol-polyfill/src/mcp_over_acp/protocol.rs new file mode 100644 index 0000000..b3cedc1 --- /dev/null +++ b/src/agent-client-protocol-polyfill/src/mcp_over_acp/protocol.rs @@ -0,0 +1,697 @@ +use agent_client_protocol::{ + Error, JsonRpcMessage, JsonRpcResponse, UntypedMessage, + schema::{ + InitializeProxyRequest, METHOD_INITIALIZE_PROXY, ProtocolVersion, + v1::{ + ConnectMcpRequest, ConnectMcpResponse, DisconnectMcpRequest, DisconnectMcpResponse, + LoadSessionRequest, McpServer, MessageMcpNotification, MessageMcpRequest, + NewSessionRequest, ResumeSessionRequest, + }, + }, +}; +use serde_json::{Map, Value}; + +#[cfg(feature = "unstable_protocol_v2")] +use agent_client_protocol::schema::v2; + +#[cfg(feature = "unstable_session_fork")] +use agent_client_protocol::schema::v1::ForkSessionRequest; + +/// ACP schema selected by the conductor's proxy initialization request. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub(crate) enum PolyfillProtocol { + V1, + #[cfg(feature = "unstable_protocol_v2")] + V2, +} + +impl PolyfillProtocol { + pub(crate) fn from_initialize_request(request: &UntypedMessage) -> Result { + if request.method() != METHOD_INITIALIZE_PROXY { + return Err(Error::invalid_request() + .data(format!("expected `{METHOD_INITIALIZE_PROXY}` request"))); + } + + let requested = request + .params() + .get("protocolVersion") + .cloned() + .ok_or_else(invalid_initialize_protocol_version) + .and_then(|version| { + serde_json::from_value::(version) + .map_err(|_| invalid_initialize_protocol_version()) + })?; + + let protocol = if requested == ProtocolVersion::V1 { + Self::V1 + } else { + #[cfg(feature = "unstable_protocol_v2")] + { + if requested == ProtocolVersion::V2 { + Self::V2 + } else { + return Err(unsupported_protocol_version(requested)); + } + } + + #[cfg(not(feature = "unstable_protocol_v2"))] + { + return Err(unsupported_protocol_version(requested)); + } + }; + + protocol.validate_initialize_request(request)?; + Ok(protocol) + } + + fn validate_initialize_request(self, request: &UntypedMessage) -> Result<(), Error> { + match self { + Self::V1 => { + InitializeProxyRequest::parse_message(request.method(), request.params())?; + } + #[cfg(feature = "unstable_protocol_v2")] + Self::V2 => { + v2::InitializeProxyRequest::parse_message(request.method(), request.params())?; + } + } + Ok(()) + } + + pub(crate) fn transform_initialize_response( + self, + response: &mut Value, + ) -> Result { + let mode = match self { + Self::V1 => { + let response = agent_client_protocol::schema::v1::InitializeResponse::from_value( + "initialize", + response.clone(), + )?; + DownstreamMcpMode::from_capabilities( + response.agent_capabilities.mcp_capabilities.http, + response.agent_capabilities.mcp_capabilities.acp, + ) + } + #[cfg(feature = "unstable_protocol_v2")] + Self::V2 => { + let response = v2::InitializeResponse::from_value("initialize", response.clone())?; + let mcp = response + .capabilities + .session + .as_ref() + .and_then(|session| session.mcp.as_ref()); + DownstreamMcpMode::from_capabilities( + mcp.is_some_and(|mcp| mcp.http.is_some()), + mcp.is_some_and(|mcp| mcp.acp.is_some()), + ) + } + }; + + if mode == DownstreamMcpMode::HttpAdapter { + self.advertise_native_mcp(response)?; + } + Ok(mode) + } + + fn advertise_native_mcp(self, response: &mut Value) -> Result<(), Error> { + let response = response + .as_object_mut() + .ok_or_else(|| invalid_initialize_response("result must be an object"))?; + match self { + Self::V1 => { + let mcp = response + .get_mut("agentCapabilities") + .and_then(Value::as_object_mut) + .and_then(|capabilities| capabilities.get_mut("mcpCapabilities")) + .and_then(Value::as_object_mut) + .ok_or_else(|| { + invalid_initialize_response( + "HTTP MCP support did not have an object capability container", + ) + })?; + mcp.insert("acp".into(), Value::Bool(true)); + } + #[cfg(feature = "unstable_protocol_v2")] + Self::V2 => { + let mcp = response + .get_mut("capabilities") + .and_then(Value::as_object_mut) + .and_then(|capabilities| capabilities.get_mut("session")) + .and_then(Value::as_object_mut) + .and_then(|session| session.get_mut("mcp")) + .and_then(Value::as_object_mut) + .ok_or_else(|| { + invalid_initialize_response( + "HTTP MCP support did not have an object capability container", + ) + })?; + mcp.insert("acp".into(), Value::Object(Map::new())); + } + } + Ok(()) + } + + pub(crate) fn is_session_setup_method(self, method: &str) -> bool { + match self { + Self::V1 => { + matches!(method, "session/new" | "session/load" | "session/resume") + || cfg!(feature = "unstable_session_fork") && method == "session/fork" + } + #[cfg(feature = "unstable_protocol_v2")] + Self::V2 => { + matches!(method, "session/new" | "session/resume") + || cfg!(feature = "unstable_session_fork") && method == "session/fork" + } + } + } + + pub(crate) fn validate_session_setup_request( + self, + request: &UntypedMessage, + ) -> Result<(), Error> { + match self { + Self::V1 => match request.method() { + "session/new" => { + NewSessionRequest::parse_message(request.method(), request.params())?; + } + "session/load" => { + LoadSessionRequest::parse_message(request.method(), request.params())?; + } + "session/resume" => { + ResumeSessionRequest::parse_message(request.method(), request.params())?; + } + #[cfg(feature = "unstable_session_fork")] + "session/fork" => { + ForkSessionRequest::parse_message(request.method(), request.params())?; + } + method => return Err(unexpected_session_setup_method(method)), + }, + #[cfg(feature = "unstable_protocol_v2")] + Self::V2 => match request.method() { + "session/new" => { + v2::NewSessionRequest::parse_message(request.method(), request.params())?; + } + "session/resume" => { + v2::ResumeSessionRequest::parse_message(request.method(), request.params())?; + } + #[cfg(feature = "unstable_session_fork")] + "session/fork" => { + v2::ForkSessionRequest::parse_message(request.method(), request.params())?; + } + method => return Err(unexpected_session_setup_method(method)), + }, + } + Ok(()) + } + + pub(crate) fn native_server(self, value: Value) -> Option { + let raw = value.as_object()?.clone(); + match self { + Self::V1 => { + let McpServer::Acp(server) = serde_json::from_value(value).ok()? else { + return None; + }; + Some(NativeServer { + raw, + name: server.name, + server_id: server.server_id.to_string(), + }) + } + #[cfg(feature = "unstable_protocol_v2")] + Self::V2 => { + let v2::McpServer::Acp(server) = serde_json::from_value(value).ok()? else { + return None; + }; + Some(NativeServer { + raw, + name: server.name, + server_id: server.server_id.to_string(), + }) + } + } + } + + pub(crate) fn connect_request(self, server_id: String) -> Result { + match self { + Self::V1 => ConnectMcpRequest::new(server_id).to_untyped_message(), + #[cfg(feature = "unstable_protocol_v2")] + Self::V2 => v2::ConnectMcpRequest::new(server_id).to_untyped_message(), + } + } + + pub(crate) fn connect_response_id(self, response: Value) -> Result { + match self { + Self::V1 => ConnectMcpResponse::from_value("mcp/connect", response) + .map(|response| response.connection_id.to_string()), + #[cfg(feature = "unstable_protocol_v2")] + Self::V2 => v2::ConnectMcpResponse::from_value("mcp/connect", response) + .map(|response| response.connection_id.to_string()), + } + } + + pub(crate) fn message_request( + self, + connection_id: String, + message: UntypedMessage, + ) -> Result { + let (method, params) = message.into_parts(); + let params = into_mcp_params(params)?; + match self { + Self::V1 => MessageMcpRequest::new(connection_id, method) + .params(params) + .to_untyped_message(), + #[cfg(feature = "unstable_protocol_v2")] + Self::V2 => v2::MessageMcpRequest::new(connection_id, method) + .params(params) + .to_untyped_message(), + } + } + + pub(crate) fn message_notification( + self, + connection_id: String, + message: UntypedMessage, + ) -> Result { + let (method, params) = message.into_parts(); + let params = into_mcp_params(params)?; + match self { + Self::V1 => MessageMcpNotification::new(connection_id, method) + .params(params) + .to_untyped_message(), + #[cfg(feature = "unstable_protocol_v2")] + Self::V2 => v2::MessageMcpNotification::new(connection_id, method) + .params(params) + .to_untyped_message(), + } + } + + pub(crate) fn parse_message_request( + self, + request: UntypedMessage, + ) -> Result { + match self { + Self::V1 => { + let parsed = MessageMcpRequest::parse_message(request.method(), request.params())?; + Ok(NativeMcpMessage { + raw: request, + connection_id: parsed.connection_id.to_string(), + method: parsed.method, + params: parsed.params, + }) + } + #[cfg(feature = "unstable_protocol_v2")] + Self::V2 => { + let parsed = + v2::MessageMcpRequest::parse_message(request.method(), request.params())?; + Ok(NativeMcpMessage { + raw: request, + connection_id: parsed.connection_id.to_string(), + method: parsed.method, + params: parsed.params, + }) + } + } + } + + pub(crate) fn parse_message_notification( + self, + notification: UntypedMessage, + ) -> Result { + match self { + Self::V1 => { + let parsed = MessageMcpNotification::parse_message( + notification.method(), + notification.params(), + )?; + Ok(NativeMcpMessage { + raw: notification, + connection_id: parsed.connection_id.to_string(), + method: parsed.method, + params: parsed.params, + }) + } + #[cfg(feature = "unstable_protocol_v2")] + Self::V2 => { + let parsed = v2::MessageMcpNotification::parse_message( + notification.method(), + notification.params(), + )?; + Ok(NativeMcpMessage { + raw: notification, + connection_id: parsed.connection_id.to_string(), + method: parsed.method, + params: parsed.params, + }) + } + } + } + + pub(crate) fn disconnect_request(self, connection_id: String) -> Result { + match self { + Self::V1 => DisconnectMcpRequest::new(connection_id).to_untyped_message(), + #[cfg(feature = "unstable_protocol_v2")] + Self::V2 => v2::DisconnectMcpRequest::new(connection_id).to_untyped_message(), + } + } + + pub(crate) fn validate_disconnect_response(self, response: Value) -> Result<(), Error> { + match self { + Self::V1 => { + DisconnectMcpResponse::from_value("mcp/disconnect", response)?; + } + #[cfg(feature = "unstable_protocol_v2")] + Self::V2 => { + v2::DisconnectMcpResponse::from_value("mcp/disconnect", response)?; + } + } + Ok(()) + } +} + +#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)] +pub(crate) enum DownstreamMcpMode { + #[default] + Unknown, + Native, + HttpAdapter, + Unavailable, +} + +impl DownstreamMcpMode { + pub(crate) fn from_capabilities(http: bool, acp: bool) -> Self { + if acp { + Self::Native + } else if http { + Self::HttpAdapter + } else { + Self::Unavailable + } + } +} + +#[derive(Debug)] +pub(crate) struct NativeServer { + raw: Map, + pub(crate) name: String, + pub(crate) server_id: String, +} + +impl NativeServer { + pub(crate) fn http_declaration( + mut self, + protocol: PolyfillProtocol, + url: String, + ) -> Result { + self.raw.remove("serverId"); + self.raw.insert("type".into(), Value::String("http".into())); + self.raw.insert("name".into(), Value::String(self.name)); + self.raw.insert("url".into(), Value::String(url)); + // V1 requires the field and v2 accepts it. Keeping the explicit empty + // list gives both versions one stable raw compatibility shape. + self.raw.insert("headers".into(), Value::Array(Vec::new())); + let declaration = Value::Object(self.raw); + + match protocol { + PolyfillProtocol::V1 => { + serde_json::from_value::(declaration.clone()) + .map_err(Error::into_internal_error)?; + } + #[cfg(feature = "unstable_protocol_v2")] + PolyfillProtocol::V2 => { + serde_json::from_value::(declaration.clone()) + .map_err(Error::into_internal_error)?; + } + } + Ok(declaration) + } +} + +#[derive(Debug)] +pub(crate) struct NativeMcpMessage { + pub(crate) raw: UntypedMessage, + pub(crate) connection_id: String, + pub(crate) method: String, + pub(crate) params: Option>, +} + +pub(crate) fn native_params_into_value(params: Option>) -> Value { + params.map_or(Value::Null, Value::Object) +} + +fn into_mcp_params(params: Value) -> Result>, Error> { + match params { + Value::Null => Ok(None), + Value::Object(params) => Ok(Some(params)), + params => Err(Error::invalid_params().data(serde_json::json!({ + "reason": "MCP message params must be an object or null", + "params": params, + }))), + } +} + +fn invalid_initialize_protocol_version() -> Error { + Error::invalid_params().data("initialize.protocolVersion must be a valid ACP protocol version") +} + +fn unsupported_protocol_version(version: ProtocolVersion) -> Error { + Error::invalid_request().data(format!( + "MCP-over-ACP polyfill does not support ACP protocol version {version}" + )) +} + +fn unexpected_session_setup_method(method: &str) -> Error { + Error::invalid_request().data(format!( + "`{method}` is not a session setup method for the selected ACP version" + )) +} + +fn invalid_initialize_response(reason: &'static str) -> Error { + Error::invalid_params().data(format!("invalid initialize response: {reason}")) +} + +#[cfg(test)] +mod tests { + use agent_client_protocol::{ + JsonRpcMessage, + schema::{ProtocolVersion, v1}, + }; + + #[cfg(feature = "unstable_protocol_v2")] + use agent_client_protocol::{ErrorCode, JsonRpcResponse}; + + use super::PolyfillProtocol; + + #[test] + fn http_declaration_preserves_extension_fields() { + let declaration = serde_json::json!({ + "type": "acp", + "name": "native", + "serverId": "native-id", + "_meta": { + "source": "test" + }, + "futureField": { + "preserve": true + } + }); + let native = PolyfillProtocol::V1 + .native_server(declaration) + .expect("the declaration should be recognized as native MCP"); + + let transformed = native + .http_declaration(PolyfillProtocol::V1, "http://127.0.0.1:4321".to_string()) + .expect("the transformed declaration should be valid v1 MCP"); + + assert_eq!( + transformed, + serde_json::json!({ + "type": "http", + "name": "native", + "url": "http://127.0.0.1:4321", + "headers": [], + "_meta": { + "source": "test" + }, + "futureField": { + "preserve": true + } + }) + ); + } + + #[test] + fn native_message_keeps_the_original_wrapper() { + let request = agent_client_protocol::UntypedMessage { + method: "mcp/message".to_string(), + params: serde_json::json!({ + "connectionId": "connection", + "method": "tools/list", + "params": { + "cursor": "next" + }, + "_meta": { + "trace": "preserve" + }, + "futureField": true + }), + }; + + let parsed = PolyfillProtocol::V1 + .parse_message_request(request.clone()) + .expect("the native wrapper should parse"); + + assert_eq!(parsed.raw, request); + assert_eq!(parsed.connection_id, "connection"); + assert_eq!(parsed.method, "tools/list"); + assert_eq!( + parsed.params, + Some(serde_json::Map::from_iter([( + "cursor".to_string(), + serde_json::json!("next") + )])) + ); + } + + #[test] + fn v1_session_setup_methods_match_the_stable_schema() { + assert!(PolyfillProtocol::V1.is_session_setup_method("session/new")); + assert!(PolyfillProtocol::V1.is_session_setup_method("session/load")); + assert!(PolyfillProtocol::V1.is_session_setup_method("session/resume")); + assert_eq!( + PolyfillProtocol::V1.is_session_setup_method("session/fork"), + cfg!(feature = "unstable_session_fork") + ); + assert!(!PolyfillProtocol::V1.is_session_setup_method("session/prompt")); + } + + #[test] + fn session_setup_validation_allows_extensions_but_rejects_invalid_fields() { + let mut request = v1::NewSessionRequest::new(std::path::PathBuf::from("/tmp")) + .to_untyped_message() + .expect("the session request should serialize"); + request.params["futureField"] = serde_json::json!({ + "preserve": true + }); + PolyfillProtocol::V1 + .validate_session_setup_request(&request) + .expect("extension fields should remain forward-compatible"); + + request.params["cwd"] = serde_json::json!(42); + let error = PolyfillProtocol::V1 + .validate_session_setup_request(&request) + .expect_err("invalid selected-schema fields must be rejected"); + assert_eq!(error.code, agent_client_protocol::ErrorCode::InvalidParams); + } + + #[cfg(feature = "unstable_protocol_v2")] + #[test] + fn v2_session_setup_methods_exclude_v1_load() { + assert!(PolyfillProtocol::V2.is_session_setup_method("session/new")); + assert!(!PolyfillProtocol::V2.is_session_setup_method("session/load")); + assert!(PolyfillProtocol::V2.is_session_setup_method("session/resume")); + assert_eq!( + PolyfillProtocol::V2.is_session_setup_method("session/fork"), + cfg!(feature = "unstable_session_fork") + ); + assert!(!PolyfillProtocol::V2.is_session_setup_method("session/prompt")); + } + + #[cfg(feature = "unstable_protocol_v2")] + #[test] + fn future_protocol_version_is_not_assumed_to_be_v2() { + let initialize = agent_client_protocol::schema::v2::InitializeRequest::new( + ProtocolVersion::V2, + agent_client_protocol::schema::v2::Implementation::new("test", "1.0.0"), + ); + let mut request = + agent_client_protocol::schema::v2::InitializeProxyRequest::new(initialize) + .to_untyped_message() + .expect("the initialize request should serialize"); + request.params["protocolVersion"] = serde_json::json!(3); + + let error = PolyfillProtocol::from_initialize_request(&request) + .expect_err("an unselected future schema must not be interpreted as v2"); + + assert_eq!(error.code, ErrorCode::InvalidRequest); + assert_eq!( + error.data, + Some(serde_json::json!( + "MCP-over-ACP polyfill does not support ACP protocol version 3" + )) + ); + } + + #[cfg(feature = "unstable_protocol_v2")] + #[test] + fn v2_initialize_adaptation_preserves_the_raw_response() { + use agent_client_protocol::schema::v2; + + let response = v2::InitializeResponse::new( + ProtocolVersion::V2, + v2::Implementation::new("test", "1.0.0"), + ) + .capabilities( + v2::AgentCapabilities::new().session( + v2::SessionCapabilities::new() + .mcp(v2::McpCapabilities::new().http(v2::McpHttpCapabilities::new())), + ), + ); + let mut response = serde_json::to_value(response).expect("the response should serialize"); + response["futureField"] = serde_json::json!({ + "preserve": true + }); + + let mode = PolyfillProtocol::V2 + .transform_initialize_response(&mut response) + .expect("the v2 HTTP capability should be adaptable"); + + assert_eq!(mode, super::DownstreamMcpMode::HttpAdapter); + assert_eq!( + response["capabilities"]["session"]["mcp"]["acp"], + serde_json::json!({}) + ); + assert_eq!( + response["futureField"], + serde_json::json!({ + "preserve": true + }) + ); + + v2::InitializeResponse::from_value("initialize", response) + .expect("the adapted response should remain valid v2"); + } + + #[cfg(feature = "unstable_protocol_v2")] + #[test] + fn v2_disconnect_uses_and_validates_the_selected_schema() { + use agent_client_protocol::schema::v2; + + let request = PolyfillProtocol::V2 + .disconnect_request("connection".to_string()) + .expect("the v2 disconnect request should serialize"); + let parsed = v2::DisconnectMcpRequest::parse_message(request.method(), request.params()) + .expect("the disconnect request should be valid v2"); + assert_eq!(parsed.connection_id.to_string(), "connection"); + + let response = serde_json::to_value(v2::DisconnectMcpResponse::new()) + .expect("the v2 disconnect response should serialize"); + PolyfillProtocol::V2 + .validate_disconnect_response(response) + .expect("the v2 disconnect response should validate"); + } + + #[test] + fn v1_initialize_request_selects_v1() { + let request = agent_client_protocol::schema::InitializeProxyRequest { + initialize: v1::InitializeRequest::new(ProtocolVersion::V1), + } + .to_untyped_message() + .expect("the initialize request should serialize"); + + assert_eq!( + PolyfillProtocol::from_initialize_request(&request) + .expect("the request should select v1"), + PolyfillProtocol::V1 + ); + } +}