diff --git a/Cargo.lock b/Cargo.lock
index 012e93bb..0499dff9 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -107,7 +107,6 @@ name = "agent-client-protocol-polyfill"
version = "1.3.0"
dependencies = [
"agent-client-protocol",
- "anyhow",
"async-stream",
"axum",
"futures",
@@ -116,7 +115,6 @@ dependencies = [
"serde_json",
"thiserror",
"tokio",
- "tokio-util",
"tracing",
"uuid",
]
diff --git a/README.md b/README.md
index b5c0cd38..c90563a9 100644
--- a/README.md
+++ b/README.md
@@ -19,10 +19,15 @@ This repository is the official **Rust SDK** for ACP. It provides crates for bui
- [`agent-client-protocol-rmcp`](./src/agent-client-protocol-rmcp/) – Integration with the [`rmcp`](https://docs.rs/rmcp) MCP SDK.
- [`agent-client-protocol-derive`](./src/agent-client-protocol-derive/) – Derive macros used by the core crate.
+Native MCP-over-ACP support is currently opt-in through the core crate's
+`unstable_mcp_over_acp` feature. Standalone MCP servers need no ACP transport
+feature; the rmcp integration exposes a matching passthrough feature when those
+servers are attached to ACP.
+
**Proxy orchestration**
- [`agent-client-protocol-conductor`](./src/agent-client-protocol-conductor/) – Binary and library that manages chains of proxy components.
-- [`agent-client-protocol-polyfill`](./src/agent-client-protocol-polyfill/) – Compatibility proxies, including bridging legacy `acp:` MCP declarations for agents that cannot consume them directly.
+- [`agent-client-protocol-polyfill`](./src/agent-client-protocol-polyfill/) – Compatibility proxies, including adapting native MCP-over-ACP declarations to HTTP for agents that cannot consume them directly.
- [`agent-client-protocol-trace-viewer`](./src/agent-client-protocol-trace-viewer/) – Interactive sequence-diagram viewer for conductor trace files.
**Patterns, examples, and testing**
diff --git a/md/conductor.md b/md/conductor.md
index bfe44e6f..583beec9 100644
--- a/md/conductor.md
+++ b/md/conductor.md
@@ -98,8 +98,9 @@ agent-client-protocol-conductor --serve agent "proxy-one" "base-agent"
agent-client-protocol-conductor --trace ./trace.jsons --serve agent "proxy-one" "base-agent"
```
-There is no conductor `mcp` subcommand. MCP compatibility bridging lives in
-`agent-client-protocol-polyfill` and must be inserted explicitly when needed.
+There is no conductor `mcp` subcommand. Compatibility for HTTP-capable agents that lack the
+native ACP MCP transport lives in `agent-client-protocol-polyfill` and must
+be inserted explicitly when needed.
## Programmatic Usage
@@ -122,11 +123,13 @@ the chain depends on initialization data.
## MCP Compatibility
MCP-over-ACP adaptation is intentionally not built into `ConductorImpl`. Add
-`McpOverAcpPolyfill::http()` or `McpOverAcpPolyfill::stdio(...)` as a proxy in
-the chain when an agent cannot consume the legacy `McpServer::Http` `acp:`
-declaration directly. Keeping the polyfill explicit prevents instrumentation
-or orchestration from silently changing session MCP declarations. See
-[MCP Bridge](./mcp-bridge.md).
+`McpOverAcpPolyfill::http()` as a proxy in the chain immediately before a final
+agent that cannot consume native `McpServer::Acp` declarations. The
+provider-facing side continues to use the feature-gated `mcp/connect`,
+`mcp/message`, and `mcp/disconnect` methods; only the final-agent side is
+adapted to HTTP. Keeping the polyfill explicit prevents instrumentation or
+orchestration from silently changing session MCP declarations. See [MCP
+Bridge](./mcp-bridge.md).
## Tracing
diff --git a/md/design.md b/md/design.md
index 561a0503..06090e0b 100644
--- a/md/design.md
+++ b/md/design.md
@@ -15,7 +15,7 @@ The core SDK. Provides:
- **Message handling** (`on_receive_request`, `on_receive_notification`, `on_receive_dispatch`)
- **Protocol types** (`agent_client_protocol::schema::*`) - all ACP message types
- **Transports and process launching** (`Channel`, `Lines`, `ByteStreams`, `Stdio`, `AcpAgent`)
-- **MCP server attachment** - runtime-agnostic interfaces for wiring MCP servers into ACP sessions
+- **MCP server attachment** - runtime-agnostic interfaces for wiring MCP servers into ACP sessions through the opt-in `unstable_mcp_over_acp` transport
### agent-client-protocol-http
@@ -28,6 +28,11 @@ Integration with the [rmcp](https://docs.rs/rmcp) crate:
- **`McpServer::builder()`** - define MCP tools in Rust code
- **`McpServer::from_rmcp()`** - wrap an rmcp server as an ACP MCP server
+Standalone rmcp-backed servers need no ACP transport feature. Enable the
+integration crate's `unstable_mcp_over_acp` feature to advertise an attached
+server as `McpServer::Acp`. Agents limited to HTTP MCP transports require the
+separate compatibility polyfill.
+
## Role System
The type system is built around **roles** - the logical identity of an endpoint.
diff --git a/md/introduction.md b/md/introduction.md
index 25e142f6..1be2cf58 100644
--- a/md/introduction.md
+++ b/md/introduction.md
@@ -26,7 +26,7 @@ src/
├── agent-client-protocol-cookbook/ # Usage patterns (rendered as rustdoc)
├── agent-client-protocol-derive/ # Proc macros
├── agent-client-protocol-conductor/ # Conductor binary and library
-├── agent-client-protocol-polyfill/ # Compatibility proxy implementations
+├── agent-client-protocol-polyfill/ # MCP-over-ACP transport compatibility
├── agent-client-protocol-test/ # Test utilities and fixtures
├── agent-client-protocol-trace-viewer/ # Trace visualization tool
└── yopo/ # "You Only Prompt Once" example client
@@ -40,7 +40,7 @@ graph TD
http[agent-client-protocol-http
HTTP/SSE/WebSocket transport]
rmcp[agent-client-protocol-rmcp
rmcp integration]
conductor[agent-client-protocol-conductor
Proxy orchestration]
- polyfill[agent-client-protocol-polyfill
Compatibility proxies]
+ polyfill[agent-client-protocol-polyfill
MCP transport compatibility]
trace[agent-client-protocol-trace-viewer
Trace visualization]
cookbook[agent-client-protocol-cookbook
Usage patterns]
@@ -60,6 +60,7 @@ graph TD
- [Transport Architecture](./transport-architecture.md) - The frame-aware boundary shared by transports and in-process components
- [Conductor Design](./conductor.md) - How the conductor orchestrates proxy chains
- [Protocol Reference](./protocol.md) - Wire protocol details and extension methods
+- [MCP Bridge](./mcp-bridge.md) - Adapting native MCP-over-ACP for HTTP-capable agents
- [Original P/ACP Design Proposal](./proxying-acp.md) - Historical design context; not the current wire reference
- [Migrating to v2.0](./migration_v2.0.md) - Upgrade guide from 1.x to 2.0
- [Migrating to v0.11](./migration_v0.11.x.md) - Upgrade guide from 0.10.x to 0.11
diff --git a/md/mcp-bridge.md b/md/mcp-bridge.md
index eecaeafd..5572abee 100644
--- a/md/mcp-bridge.md
+++ b/md/mcp-bridge.md
@@ -1,27 +1,28 @@
# MCP-over-ACP Compatibility Bridge
-`agent-client-protocol-polyfill::mcp_over_acp::McpOverAcpPolyfill` is an
-explicit proxy for agents that cannot consume the SDK's legacy ACP-routed MCP
-server declarations directly. MCP adaptation is no longer built into the
-conductor.
+`agent-client-protocol-polyfill::mcp_over_acp::McpOverAcpPolyfill` adapts the
+native ACP MCP transport for a final agent that accepts HTTP MCP
+servers. MCP adaptation is explicit and is not built into the conductor.
-## Native and Legacy Transports
+The component-facing side of the bridge always uses the opt-in native protocol:
-Two similarly named mechanisms coexist and must not be mixed:
+- Servers are declared as `McpServer::Acp` with a `serverId`.
+- Connections use `mcp/connect`, `mcp/message`, and `mcp/disconnect`.
+- `mcp/disconnect` is a request with a response.
-- The draft **native** transport is enabled by `unstable_mcp_over_acp`. It uses
- `McpServer::Acp` plus `mcp/connect`, `mcp/message`, and `mcp/disconnect`.
-- The polyfill's **legacy compatibility** path recognizes
- `McpServer::Http` entries whose URL begins with `acp:`. It routes them with
- `_mcp/connect`, `_mcp/message`, and `_mcp/disconnect`.
+The SDK-local underscore-prefixed method family and HTTP declarations with a
+special URL scheme have been retired. The polyfill now translates native
+declarations to real localhost HTTP URLs only at
+the compatibility boundary.
-The polyfill currently implements the second path. New implementations that
-control both peers should prefer the draft native schema types; use the
-polyfill only where compatibility requires it.
+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.
## Placement
-Insert the polyfill as a proxy before the final agent:
+Insert the polyfill immediately before the final agent that lacks native
+MCP-over-ACP support:
```rust,ignore
use agent_client_protocol_conductor::{ConductorImpl, ProxiesAndAgent};
@@ -36,40 +37,56 @@ ConductorImpl::new_agent("conductor", components)
.await?;
```
-The application proxy can then supply a legacy declaration such as an HTTP MCP
-server whose URL is `acp:server-1`. The identifier is opaque and must route back
-to the component that provides the MCP server.
+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.
During initialization, the polyfill forwards the request to its successor and
-sets `agentCapabilities.mcpCapabilities.acp` in the response seen upstream. In
-this chain position that capability means the polyfill can handle the routed
-transport; it does not imply that the final agent implements it natively.
+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.
+
+If the successor already advertises native ACP MCP support, the polyfill leaves
+the capability, declarations, and `mcp/message` traffic unchanged. If it
+supports neither native nor HTTP MCP, the polyfill does not advertise ACP MCP
+support and rejects any native declaration that is nevertheless supplied.
## Transformation
-For each `McpServer::Http` entry with an `acp:` URL in `session/new`, the
-polyfill:
+For each `McpServer::Acp` entry in `session/new`, `session/load`,
+`session/resume`, or feature-gated `session/fork`, 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
+ final agent.
+2. Retains the native `serverId` so connections can be routed back to the
+ component that provided the server.
+3. Opens the endpoint's native connection by sending `mcp/connect` with that
+ server ID toward the provider.
+4. Relays requests and notifications through `mcp/message`, using the returned
+ `connectionId` for that active MCP connection.
+5. Sends an `mcp/disconnect` request when the local transport closes and removes
+ the connection from the bridge.
+
+Enable the polyfill crate's `unstable_session_fork` feature when adapting fork
+requests.
-1. Rejects non-empty HTTP headers, which have no defined meaning for this
- compatibility transport.
-2. Reuses an existing listener for the same ACP identifier or binds a new
- localhost TCP listener.
-3. Replaces the server declaration with a transport the final agent can open.
-4. When that transport connects, sends `_mcp/connect` toward the component that
- declared the identifier.
-5. Relays MCP requests and notifications through `_mcp/message`, keyed by the
- returned connection identifier, and sends `_mcp/disconnect` when the bridge
- closes.
+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.
-The exact underscore-prefixed envelopes are documented in the [Proxy Extension
-Protocol Reference](./protocol.md#legacy-mcp-polyfill-methods).
+The native wire envelopes are documented in the [SDK Protocol
+Reference](./protocol.md#native-mcp-over-acp).
## HTTP Mode
`McpOverAcpPolyfill::http()` is the default compatibility shape. It replaces
-the `acp:` declaration with an HTTP MCP URL on `localhost`. The embedded server
-accepts MCP POST requests and an SSE GET stream at `/`, retaining JSON-RPC batch
-frames and correlating each POST with its response.
+the native declaration with an HTTP MCP URL at `http://127.0.0.1:PORT`. The
+embedded server accepts MCP POST requests and an SSE GET stream at `/`, retaining
+JSON-RPC batch frames and correlating each POST with its response.
```rust,ignore
let bridge = McpOverAcpPolyfill::http();
@@ -78,30 +95,17 @@ let bridge = McpOverAcpPolyfill::http();
The listener is bound only on loopback and uses an ephemeral port. It does not
implement resumable SSE event IDs.
-## Stdio Mode
-
-`McpOverAcpPolyfill::stdio(command)` replaces the declaration with an MCP stdio
-server. It appends `mcp PORT` to the supplied command and expects that process
-to copy newline-delimited JSON-RPC between stdio and the designated localhost
-TCP port.
-
-```rust,ignore
-let bridge = McpOverAcpPolyfill::stdio(vec![
- "legacy-mcp-bridge".to_owned(),
-]);
-```
-
-The current `agent-client-protocol-conductor` binary has no `mcp` subcommand,
-so it is not itself a valid stdio bridge command. Use this mode only with a
-compatible bridge executable; otherwise use HTTP mode.
-
## Lifecycle and Failure Behavior
-Each accepted bridge connection receives a unique connection ID from
-`_mcp/connect`. The polyfill keeps a connection map until the local MCP
-transport closes, then removes the entry and sends `_mcp/disconnect`.
-Connection or relay failures propagate through the proxy connection rather than
-being encoded as notification responses.
+Each bridge endpoint receives a unique `connectionId` from `mcp/connect`. The
+polyfill keeps a connection map until the endpoint's transport task closes,
+then removes the entry, sends `mcp/disconnect`, and observes its response.
+Request failures use the corresponding request's error path; notifications are
+never answered with synthetic errors.
+
+A reverse `mcp/message` request for an unknown `connectionId` receives
+`Invalid params`. A reverse notification for an unknown connection is ignored,
+as required for JSON-RPC notifications.
The polyfill does not infer or store ACP session IDs. Association is carried by
-the MCP server identifier and the resulting MCP connection ID.
+the declared `serverId` and the resulting active `connectionId`.
diff --git a/md/migration_v2.0.md b/md/migration_v2.0.md
index d2071350..30b39d1e 100644
--- a/md/migration_v2.0.md
+++ b/md/migration_v2.0.md
@@ -4,7 +4,8 @@ Version 2.0 makes JSON-RPC notification semantics explicit, changes the low-leve
transport boundary so frames remain intact across components and adapters, clarifies the
distinction between responding to requests and routing responses, makes dynamic handler lifetimes
explicit, and gives `AcpAgent` an SDK-owned process-launch configuration instead of reusing an MCP
-wire-schema type.
+wire-schema type. It also replaces the SDK-local MCP-over-ACP wire extension with the shared
+schema's opt-in native transport.
## Notifications cannot receive error responses
@@ -167,16 +168,78 @@ or responses. Their method names now reflect that input:
## Connection and session accessors borrow
-`McpConnectionTo::acp_id` now returns `&str`. The deprecated `acp_url` alias was removed; use
-`acp_id` instead. `McpConnectionTo::connection_to` is now `connection` and returns
-`&ConnectionTo<_>`.
+`McpConnectionTo::acp_id` is now `server_id` and returns `Option<&McpServerAcpId>`. The new name
+matches the native `McpServer::Acp` declaration; the `Option` reflects that a server can also be
+connected directly without ACP. `connection_id` returns an `Option<&McpConnectionId>` for the
+distinct active connection created by `mcp/connect`. Use `context()` to match explicitly on
+`McpConnectionContext::Standalone` or `McpConnectionContext::Acp { server_id, connection_id }`.
+The deprecated `acp_url` alias was removed. `McpConnectionTo::connection_to` is now `connection`
+and returns `&ConnectionTo<_>`.
`ActiveSession::modes` and `ActiveSession::meta` now return `Option<&T>` instead of `&Option`,
and `ActiveSession::connection` returns `&ConnectionTo<_>`.
-These accessors avoid implicit allocation and handle cloning. Call `.to_owned()` on `acp_id`,
-`.cloned()` on `modes` or `meta`, and `.clone()` on either connection accessor when an owned value
-is required.
+These accessors avoid implicit allocation and handle cloning. Call `.cloned()` on `server_id()`,
+`connection_id()`, `modes`, or `meta`, and `.clone()` on either connection accessor when an owned
+value is required.
+
+## MCP servers use the native opt-in transport
+
+The runtime-agnostic `agent_client_protocol::mcp_server` module remains available without an
+unstable ACP feature, so standalone MCP servers do not allocate or retain schema transport IDs.
+Enable the feature when attaching a server to ACP with `Builder::with_mcp_server` or
+`SessionBuilder::with_mcp_server`:
+
+```toml
+agent-client-protocol = { version = "2", features = ["unstable_mcp_over_acp"] }
+```
+
+`agent-client-protocol-rmcp` no longer enables this feature merely to build or directly serve an
+MCP server. Applications that attach an rmcp-backed server to ACP should enable its matching
+`unstable_mcp_over_acp` passthrough feature. The transport remains unstable and may change
+independently of the stable ACP surface.
+
+In 1.x, the SDK represented an ACP-provided MCP server as `McpServer::Http` with an `acp:` URL and
+routed it through SDK-local underscore-prefixed methods. In 2.0, providers and native consumers
+use:
+
+- `McpServer::Acp(McpServerAcp { name, server_id, .. })` in session setup requests;
+- `mcp/connect` with `serverId`, returning a distinct `connectionId`;
+- `mcp/message` requests and notifications keyed by that connection ID; and
+- an `mcp/disconnect` request with an empty response.
+
+The low-level SDK-local `McpConnectRequest`, `McpConnectResponse`, `McpOverAcpMessage`, and
+`McpDisconnectNotification` types were removed. Use the feature-gated schema types instead:
+
+| 1.x SDK-local type | 2.0 schema type |
+| --- | --- |
+| `McpConnectRequest` | `schema::v1::ConnectMcpRequest` |
+| `McpConnectResponse` | `schema::v1::ConnectMcpResponse` |
+| `McpOverAcpMessage` request | `schema::v1::MessageMcpRequest` |
+| `McpOverAcpMessage` notification | `schema::v1::MessageMcpNotification` |
+| `McpDisconnectNotification` | `schema::v1::DisconnectMcpRequest` and `DisconnectMcpResponse` |
+
+The public method-name constants moved to the schema's generated method-name
+tables:
+
+| 1.x SDK-local constant | 2.0 schema constant |
+| --- | --- |
+| `METHOD_MCP_CONNECT_REQUEST` | `schema::v1::CLIENT_METHOD_NAMES.mcp_connect` |
+| `METHOD_MCP_MESSAGE` | `schema::v1::CLIENT_METHOD_NAMES.mcp_message` or `AGENT_METHOD_NAMES.mcp_message`, depending on direction |
+| `METHOD_MCP_DISCONNECT_NOTIFICATION` | `schema::v1::CLIENT_METHOD_NAMES.mcp_disconnect` |
+
+Code using `Builder::with_mcp_server` or `SessionBuilder::with_mcp_server` continues to attach the
+high-level server in the same place; the emitted declaration and wire methods change. Global
+builder attachment advertises the same server ID on `session/new`, `session/load`,
+`session/resume`, and feature-gated `session/fork`. Per-session attachment remains specific to
+`session/new`. Do not construct an HTTP server with an `acp:` URL. If the final agent accepts HTTP
+but not native ACP MCP servers, insert `McpOverAcpPolyfill` immediately before it. The polyfill now
+consumes native `McpServer::Acp` declarations and adapts only its final-agent-facing side.
+
+The polyfill's public `BridgeMode` enum and `McpOverAcpPolyfill::stdio` were removed because the
+required conductor `mcp` helper subcommand no longer exists. The polyfill has one supported mode;
+construct it with `McpOverAcpPolyfill::http()` or `Default`, or manage a standard MCP transport
+separately.
## Low-level helpers have a narrower surface
diff --git a/md/protocol.md b/md/protocol.md
index 7e0ffb1f..cb434938 100644
--- a/md/protocol.md
+++ b/md/protocol.md
@@ -1,9 +1,9 @@
-# Proxy Extension Protocol Reference
+# SDK Protocol Reference
-This chapter documents the extension methods implemented by the Rust SDK's
-conductor and MCP-over-ACP polyfill. These methods are provisional SDK
-extensions. They are separate from stable ACP methods and from the draft native
-MCP-over-ACP methods described below.
+This chapter documents the proxy extension implemented by the Rust SDK's
+conductor and the opt-in native MCP-over-ACP transport exposed by the shared ACP
+schema. The proxy methods are provisional SDK extensions. MCP-over-ACP is also
+unstable and is available only with the `unstable_mcp_over_acp` feature.
## Method Summary
@@ -11,13 +11,13 @@ MCP-over-ACP methods described below.
| --- | --- | --- |
| `_proxy/initialize` | request | Initialize a component as a proxy |
| `_proxy/successor` | request or notification | Forward one inner ACP message to the next component |
-| `_mcp/connect` | request | Open a legacy polyfill MCP connection |
-| `_mcp/message` | request or notification | Carry one inner MCP message through the legacy polyfill |
-| `_mcp/disconnect` | notification | Close a legacy polyfill MCP connection |
+| `mcp/connect` | request | Open a connection to an ACP-provided MCP server |
+| `mcp/message` | request or notification | Carry one inner MCP message over ACP |
+| `mcp/disconnect` | request | Close an MCP-over-ACP connection |
-There are no `_proxy/successor/request`, `_proxy/successor/notification`,
-`_mcp/request`, or `_mcp/notification` methods. The presence of an outer
-JSON-RPC `id` distinguishes requests from notifications.
+There are no separate request and notification method names for successor or
+MCP message forwarding. The presence of an outer JSON-RPC `id` distinguishes a
+request from a notification.
## Proxy Initialization
@@ -56,47 +56,71 @@ forward an inner notification, omit the outer `id`; no response is produced.
Optional extension metadata may be included as `_meta` alongside the flattened
inner message.
-## Legacy MCP Polyfill Methods
+## Native MCP-over-ACP
-The underscore-prefixed `_mcp/*` family is used by
-`agent-client-protocol-polyfill` for its legacy `acp:` URL compatibility path.
-It is not the draft native ACP MCP transport.
+Enable `unstable_mcp_over_acp` to use the draft native transport. A component
+providing an MCP server adds `McpServer::Acp` to session setup requests
+(`session/new`, `session/load`, `session/resume`, and the opt-in `session/fork`).
+Its wire shape contains a human-readable name and an opaque server identifier:
-### `_mcp/connect`
+```json
+{
+ "type": "acp",
+ "name": "project-tools",
+ "serverId": "mcp-server:01"
+}
+```
-The polyfill opens a connection to the component identified by `acp_id`:
+`serverId` identifies the declared server and is used to route `mcp/connect`
+back to the component that provided it. A provider must not reuse one server ID
+for multiple visible servers on the same ACP connection. The high-level
+`agent_client_protocol::mcp_server::McpServer` APIs create this declaration
+automatically.
+
+An agent that consumes this transport advertises
+`agentCapabilities.mcpCapabilities.acp`. If the final agent supports HTTP but
+not ACP-transport MCP servers, place the [MCP-over-ACP compatibility
+bridge](./mcp-bridge.md) immediately before it.
+
+### `mcp/connect`
+
+The MCP client opens a connection to the declared server ID:
```json
{
"jsonrpc": "2.0",
"id": 20,
- "method": "_mcp/connect",
- "params": { "acp_id": "acp:server-1" }
+ "method": "mcp/connect",
+ "params": { "serverId": "mcp-server:01" }
}
```
-The result contains a polyfill connection identifier:
+The provider creates one active MCP connection and returns a distinct
+connection ID:
```json
{
"jsonrpc": "2.0",
"id": 20,
- "result": { "connection_id": "connection-1" }
+ "result": { "connectionId": "mcp-connection:01" }
}
```
-### `_mcp/message`
+The server ID selects what to connect to; the connection ID selects that
+particular running connection. All subsequent messages use the connection ID.
+
+### `mcp/message`
-`_mcp/message` flattens one inner MCP method into its parameters. This method is
-bidirectional because MCP clients and servers can both issue requests:
+`mcp/message` carries one inner MCP method and its named parameters. The method
+is bidirectional because MCP clients and servers can both issue requests:
```json
{
"jsonrpc": "2.0",
"id": 21,
- "method": "_mcp/message",
+ "method": "mcp/message",
"params": {
- "connectionId": "connection-1",
+ "connectionId": "mcp-connection:01",
"method": "tools/call",
"params": {
"name": "example",
@@ -110,35 +134,29 @@ Use an outer request for an inner MCP request and an outer notification for an
inner MCP notification. The outer response carries the inner MCP result or
error.
-### `_mcp/disconnect`
+### `mcp/disconnect`
-The disconnect notification ends the named polyfill connection:
+Disconnect is a request so the caller knows that the provider has released the
+active connection:
```json
{
"jsonrpc": "2.0",
- "method": "_mcp/disconnect",
- "params": { "connection_id": "connection-1" }
+ "id": 22,
+ "method": "mcp/disconnect",
+ "params": { "connectionId": "mcp-connection:01" }
}
```
-The local extension types intentionally retain their existing serialized field
-names: `acp_id` and `connection_id` for connect/disconnect, but `connectionId`
-inside `_mcp/message`.
+A successful disconnect returns an empty result:
-## Draft Native MCP-over-ACP
-
-With the `unstable_mcp_over_acp` feature, the protocol schema also exposes the
-draft native transport. A server is declared as `McpServer::Acp`, serialized
-with `type: "acp"`, `name`, and `serverId`. Native messages use
-`mcp/connect`, `mcp/message`, and `mcp/disconnect` without a leading underscore
-and use the schema's camel-case fields.
-
-The native and legacy method families are not interchangeable. Prefer the
-native schema types for new implementations that explicitly opt into the draft
-feature. Use the [MCP Bridge](./mcp-bridge.md) only to adapt the legacy
-`McpServer::Http` `acp:` declaration for an agent that cannot consume that
-routed form directly.
+```json
+{
+ "jsonrpc": "2.0",
+ "id": 22,
+ "result": {}
+}
+```
## Related Documentation
diff --git a/md/proxying-acp.md b/md/proxying-acp.md
index 532a316e..970389e2 100644
--- a/md/proxying-acp.md
+++ b/md/proxying-acp.md
@@ -4,6 +4,12 @@
> historical context. It contains method names and capability shapes that were
> superseded during implementation. Do not use it as a wire-protocol
> specification; see the current [Protocol Reference](./protocol.md) instead.
+>
+> In particular, the SDK-local `_mcp/*` methods and `McpServer::Http` values
+> using an `acp:` URL were retired. Current opt-in implementations use
+> `McpServer::Acp` with `mcp/connect`, `mcp/message`, and `mcp/disconnect`; the
+> compatibility polyfill translates those native declarations only for
+> HTTP-capable agents. Do not copy the historical MCP examples below.
# Elevator pitch
diff --git a/md/trace-viewer.md b/md/trace-viewer.md
index 2d27e5f0..74fa3e99 100644
--- a/md/trace-viewer.md
+++ b/md/trace-viewer.md
@@ -89,7 +89,7 @@ The trace shows logical component-to-component traffic rather than conductor
plumbing:
- `_proxy/successor` is unwrapped and logged as its inner ACP method.
-- `_mcp/message` is unwrapped and its inner method is marked with protocol
+- `mcp/message` is unwrapped and its inner method is marked with protocol
`mcp`.
- Responses are correlated with the request details retained by the trace
writer.
diff --git a/src/agent-client-protocol-conductor/CHANGELOG.md b/src/agent-client-protocol-conductor/CHANGELOG.md
index dae815b3..b7c1bfa9 100644
--- a/src/agent-client-protocol-conductor/CHANGELOG.md
+++ b/src/agent-client-protocol-conductor/CHANGELOG.md
@@ -13,6 +13,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
handlers/types they expose must be migrated together.
- **Breaking change:** Rename the public `ConductorResponder` background task to
`ConductorRunner`, matching the core runner API it implements.
+- **Changed:** Enable the opt-in native MCP-over-ACP schema so tracing recognizes `mcp/message`.
+ HTTP adaptation remains an explicit `agent-client-protocol-polyfill` proxy rather than
+ conductor behavior.
- **Fixed:** Preserve JSON-RPC batch framing when conductor tracing is enabled.
- **Documentation:** Remove references to the retired MCP bridge CLI mode and
`serve()` API.
diff --git a/src/agent-client-protocol-conductor/Cargo.toml b/src/agent-client-protocol-conductor/Cargo.toml
index 9209a3c7..00ff0570 100644
--- a/src/agent-client-protocol-conductor/Cargo.toml
+++ b/src/agent-client-protocol-conductor/Cargo.toml
@@ -19,7 +19,7 @@ path = "src/main.rs"
default = []
[dependencies]
-agent-client-protocol = { workspace = true }
+agent-client-protocol = { workspace = true, features = ["unstable_mcp_over_acp"] }
agent-client-protocol-trace-viewer.workspace = true
anyhow.workspace = true
chrono.workspace = true
@@ -37,7 +37,6 @@ tracing-subscriber.workspace = true
uuid.workspace = true
[dev-dependencies]
-agent-client-protocol = { workspace = true, features = ["unstable_mcp_over_acp"] }
agent-client-protocol-rmcp.workspace = true
agent-client-protocol-test.workspace = true
yopo.workspace = true
diff --git a/src/agent-client-protocol-conductor/README.md b/src/agent-client-protocol-conductor/README.md
index f3111da8..6dcb30a1 100644
--- a/src/agent-client-protocol-conductor/README.md
+++ b/src/agent-client-protocol-conductor/README.md
@@ -54,7 +54,7 @@ Binary will be at `target/release/agent-client-protocol-conductor`.
## Related Crates
- **[agent-client-protocol](../agent-client-protocol/)** — Core ACP protocol types and traits
-- **[agent-client-protocol-polyfill](../agent-client-protocol-polyfill/)** — Compatibility proxies, including the legacy v1 MCP-over-ACP bridge
+- **[agent-client-protocol-polyfill](../agent-client-protocol-polyfill/)** — Compatibility proxies, including adapting MCP-over-ACP to HTTP
- **[agent-client-protocol-trace-viewer](../agent-client-protocol-trace-viewer/)** — Interactive trace visualization
## License
diff --git a/src/agent-client-protocol-conductor/src/lib.rs b/src/agent-client-protocol-conductor/src/lib.rs
index 53c9ad5f..2f379215 100644
--- a/src/agent-client-protocol-conductor/src/lib.rs
+++ b/src/agent-client-protocol-conductor/src/lib.rs
@@ -60,7 +60,7 @@
//! ## Related Crates
//!
//! - **[agent-client-protocol](https://crates.io/crates/agent-client-protocol)** - Core ACP SDK
-//! - **[agent-client-protocol-polyfill](https://crates.io/crates/agent-client-protocol-polyfill)** - Compatibility proxies, including the legacy v1 MCP-over-ACP bridge
+//! - **[agent-client-protocol-polyfill](https://crates.io/crates/agent-client-protocol-polyfill)** - Compatibility proxies, including the native MCP-over-ACP to HTTP adapter
//! - **[agent-client-protocol-trace-viewer](https://crates.io/crates/agent-client-protocol-trace-viewer)** - Interactive trace visualization
use std::path::PathBuf;
diff --git a/src/agent-client-protocol-conductor/src/trace.rs b/src/agent-client-protocol-conductor/src/trace.rs
index 9c5682d6..62763290 100644
--- a/src/agent-client-protocol-conductor/src/trace.rs
+++ b/src/agent-client-protocol-conductor/src/trace.rs
@@ -9,10 +9,11 @@ use std::io::{BufWriter, Write};
use std::path::Path;
use std::time::Instant;
+use agent_client_protocol::schema::SuccessorMessage;
use agent_client_protocol::schema::v1::{
- Notification as RpcNotification, Request as RpcRequest, RequestId, Response as RpcResponse,
+ MessageMcpNotification, MessageMcpRequest, Notification as RpcNotification,
+ Request as RpcRequest, RequestId, Response as RpcResponse,
};
-use agent_client_protocol::schema::{McpOverAcpMessage, SuccessorMessage};
use agent_client_protocol::{
DynConnectTo, JsonRpcMessage, RawJsonRpcMessage, RawJsonRpcParams, Role, UntypedMessage,
};
@@ -44,7 +45,7 @@ pub enum TraceEvent {
pub enum Protocol {
/// Standard ACP protocol messages.
Acp,
- /// Legacy v1 MCP-over-ACP messages (agent calling a proxy's MCP server).
+ /// MCP messages carried over ACP.
Mcp,
}
@@ -556,14 +557,14 @@ impl MessageInfo {
///
/// This unwraps protocol wrappers to get the "real" message:
/// - `_proxy/successor` messages are unwrapped to get the inner message
- /// - `_mcp/message` messages are detected and marked as MCP protocol
+ /// - `mcp/message` messages are detected and marked as MCP protocol
///
/// Returns (protocol, method, params).
fn from_request(req: RpcRequest) -> Self {
let untyped =
UntypedMessage::parse_message(&req.method, ¶ms_from_transport(req.params))
.expect("untyped message is infallible");
- Self::from_untyped(Successor(false), Some(req.id), Protocol::Acp, untyped)
+ Self::from_untyped_request(Successor(false), Some(req.id), Protocol::Acp, untyped)
}
fn from_notification(notification: RpcNotification) -> Self {
@@ -572,23 +573,69 @@ impl MessageInfo {
¶ms_from_transport(notification.params),
)
.expect("untyped message is infallible");
- Self::from_untyped(Successor(false), None, Protocol::Acp, untyped)
+ Self::from_untyped_notification(Successor(false), Protocol::Acp, untyped)
}
- fn from_untyped(
+ fn from_untyped_request(
successor: Successor,
id: Option,
protocol: Protocol,
untyped: UntypedMessage,
) -> Self {
if let Ok(m) = SuccessorMessage::parse_message(&untyped.method, &untyped.params) {
- return Self::from_untyped(Successor(true), id, protocol, m.message);
+ return Self::from_untyped_request(Successor(true), id, protocol, m.message);
}
- if let Ok(m) = McpOverAcpMessage::parse_message(&untyped.method, &untyped.params) {
- return Self::from_untyped(successor, id, Protocol::Mcp, m.message);
+ if let Ok(m) = MessageMcpRequest::parse_message(&untyped.method, &untyped.params) {
+ let params = m
+ .params
+ .map_or(serde_json::Value::Null, serde_json::Value::Object);
+ return Self::from_untyped_request(
+ successor,
+ id,
+ Protocol::Mcp,
+ UntypedMessage {
+ method: m.method,
+ params,
+ },
+ );
+ }
+
+ Self::new(successor, id, protocol, untyped)
+ }
+
+ fn from_untyped_notification(
+ successor: Successor,
+ protocol: Protocol,
+ untyped: UntypedMessage,
+ ) -> Self {
+ if let Ok(m) = SuccessorMessage::parse_message(&untyped.method, &untyped.params) {
+ return Self::from_untyped_notification(Successor(true), protocol, m.message);
}
+ if let Ok(m) = MessageMcpNotification::parse_message(&untyped.method, &untyped.params) {
+ let params = m
+ .params
+ .map_or(serde_json::Value::Null, serde_json::Value::Object);
+ return Self::from_untyped_notification(
+ successor,
+ Protocol::Mcp,
+ UntypedMessage {
+ method: m.method,
+ params,
+ },
+ );
+ }
+
+ Self::new(successor, None, protocol, untyped)
+ }
+
+ fn new(
+ successor: Successor,
+ id: Option,
+ protocol: Protocol,
+ untyped: UntypedMessage,
+ ) -> Self {
Self {
successor,
id,
@@ -598,3 +645,32 @@ impl MessageInfo {
}
}
}
+
+#[cfg(test)]
+mod tests {
+ use agent_client_protocol::RawJsonRpcMessage;
+ use serde_json::json;
+
+ use super::{MessageInfo, Protocol};
+
+ #[test]
+ fn tolerant_mcp_notification_params_are_traced_as_mcp() {
+ let RawJsonRpcMessage::Notification(notification) = RawJsonRpcMessage::notification(
+ "mcp/message".into(),
+ json!({
+ "connectionId": "connection-1",
+ "method": "notifications/progress",
+ "params": ["invalid named params"]
+ }),
+ )
+ .expect("notification is valid JSON-RPC") else {
+ unreachable!("notification constructor returned a different message kind")
+ };
+
+ let info = MessageInfo::from_notification(notification);
+
+ assert_eq!(info.protocol, Protocol::Mcp);
+ assert_eq!(info.method, "notifications/progress");
+ assert_eq!(info.params, serde_json::Value::Null);
+ }
+}
diff --git a/src/agent-client-protocol-conductor/tests/mcp-integration.rs b/src/agent-client-protocol-conductor/tests/mcp-integration.rs
index 21aa6173..d86f772a 100644
--- a/src/agent-client-protocol-conductor/tests/mcp-integration.rs
+++ b/src/agent-client-protocol-conductor/tests/mcp-integration.rs
@@ -34,128 +34,6 @@ async fn recv(
.map_err(|_| agent_client_protocol::Error::internal_error())?
}
-async fn run_test_with_mode(
- components: ProxiesAndAgent,
- editor_task: impl AsyncFnOnce(
- agent_client_protocol::ConnectionTo,
- ) -> Result<(), agent_client_protocol::Error>,
-) -> Result<(), agent_client_protocol::Error> {
- // Initialize tracing for debug output
- drop(
- tracing_subscriber::fmt()
- .with_env_filter(tracing_subscriber::EnvFilter::from_default_env())
- .with_test_writer()
- .try_init(),
- );
-
- // Set up editor <-> conductor communication
- let (editor_out, conductor_in) = duplex(1024);
- let (conductor_out, editor_in) = duplex(1024);
-
- let transport =
- agent_client_protocol::ByteStreams::new(editor_out.compat_write(), editor_in.compat());
-
- agent_client_protocol::Client
- .builder()
- .name("editor-to-connector")
- .with_spawned(|_cx| async move {
- ConductorImpl::new_agent("conductor".to_string(), components)
- .run(agent_client_protocol::ByteStreams::new(
- conductor_out.compat_write(),
- conductor_in.compat(),
- ))
- .await
- })
- .connect_with(transport, editor_task)
- .await
-}
-
-/// Test that proxy-provided MCP tools work with stdio bridge mode
-#[tokio::test]
-async fn test_proxy_provides_mcp_tools_stdio() -> Result<(), agent_client_protocol::Error> {
- run_test_with_mode(
- ProxiesAndAgent::new(Testy::new())
- .proxy(mcp_integration::proxy::ProxyComponent)
- .proxy(McpOverAcpPolyfill::http()),
- async |connection_to_editor| {
- // Send initialization request
- let init_response = recv(
- connection_to_editor.send_request(InitializeRequest::new(ProtocolVersion::V1)),
- )
- .await;
-
- assert!(
- init_response.is_ok(),
- "Initialize should succeed: {init_response:?}"
- );
-
- // Send session/new request
- let session_response = recv(
- connection_to_editor
- .send_request(NewSessionRequest::new(std::path::PathBuf::from("/"))),
- )
- .await;
-
- assert!(
- session_response.is_ok(),
- "Session/new should succeed: {session_response:?}"
- );
-
- let session = session_response.unwrap();
- // ElizACP generates UUID session IDs, just verify it's non-empty
- assert!(!session.session_id.0.is_empty());
-
- Ok(())
- },
- )
- .await?;
-
- Ok(())
-}
-
-/// Test that proxy-provided MCP tools work with HTTP bridge mode
-#[tokio::test]
-async fn test_proxy_provides_mcp_tools_http() -> Result<(), agent_client_protocol::Error> {
- run_test_with_mode(
- ProxiesAndAgent::new(Testy::new())
- .proxy(mcp_integration::proxy::ProxyComponent)
- .proxy(McpOverAcpPolyfill::http()),
- async |connection_to_editor| {
- // Send initialization request
- let init_response = recv(
- connection_to_editor.send_request(InitializeRequest::new(ProtocolVersion::V1)),
- )
- .await;
-
- assert!(
- init_response.is_ok(),
- "Initialize should succeed: {init_response:?}"
- );
-
- // Send session/new request
- let session_response = recv(
- connection_to_editor
- .send_request(NewSessionRequest::new(std::path::PathBuf::from("/"))),
- )
- .await;
-
- assert!(
- session_response.is_ok(),
- "Session/new should succeed: {session_response:?}"
- );
-
- let session = session_response.unwrap();
- // ElizACP generates UUID session IDs, just verify it's non-empty
- assert!(!session.session_id.0.is_empty());
-
- Ok(())
- },
- )
- .await?;
-
- Ok(())
-}
-
#[tokio::test]
async fn test_agent_handles_prompt() -> Result<(), agent_client_protocol::Error> {
// Initialize tracing for debug output
@@ -260,7 +138,7 @@ async fn test_agent_handles_prompt() -> Result<(), agent_client_protocol::Error>
}
// Verify we got a successful tool call response
- // The session ID is a UUID generated by ElizACP, so we check for the tool result pattern
+ // The session ID is opaque, so check the observable tool result pattern.
assert_eq!(log_entries.len(), 2, "Expected notification + response");
assert!(
log_entries[0].contains("OK: CallToolResult"),
diff --git a/src/agent-client-protocol-conductor/tests/mcp_over_acp_polyfill.rs b/src/agent-client-protocol-conductor/tests/mcp_over_acp_polyfill.rs
new file mode 100644
index 00000000..e2adc450
--- /dev/null
+++ b/src/agent-client-protocol-conductor/tests/mcp_over_acp_polyfill.rs
@@ -0,0 +1,358 @@
+//! Integration tests for the public MCP-over-ACP compatibility proxy.
+
+use std::path::PathBuf;
+use std::sync::atomic::{AtomicUsize, Ordering};
+use std::sync::{Arc, Mutex};
+
+use agent_client_protocol::schema::ProtocolVersion;
+use agent_client_protocol::schema::v1::{
+ AgentCapabilities, ConnectMcpRequest, ConnectMcpResponse, InitializeRequest,
+ InitializeResponse, LoadSessionRequest, LoadSessionResponse, McpCapabilities, McpServer,
+ McpServerAcp, NewSessionRequest, NewSessionResponse, ResumeSessionRequest,
+ ResumeSessionResponse, SessionCapabilities, SessionResumeCapabilities,
+};
+use agent_client_protocol::{Agent, Client, Conductor, ConnectTo, Proxy};
+use agent_client_protocol_conductor::{ConductorImpl, ProxiesAndAgent};
+use agent_client_protocol_polyfill::mcp_over_acp::McpOverAcpPolyfill;
+use tokio::io::duplex;
+use tokio_util::compat::{TokioAsyncReadCompatExt, TokioAsyncWriteCompatExt};
+
+const SERVER_NAME: &str = "shared-server";
+const SERVER_ID: &str = "shared-server-id";
+
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+enum SetupMethod {
+ New,
+ Load,
+ 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: AgentCapabilities,
+ observed: Arc,
+}
+
+struct NativeMcpProvider {
+ connect_count: Arc,
+}
+
+impl ConnectTo for NativeMcpProvider {
+ async fn connect_to(
+ self,
+ client: impl ConnectTo,
+ ) -> Result<(), agent_client_protocol::Error> {
+ Proxy
+ .builder()
+ .name("native-mcp-provider")
+ .on_receive_request_from(
+ Agent,
+ async move |request: ConnectMcpRequest, responder, _cx| {
+ assert_eq!(request.server_id.to_string(), SERVER_ID);
+ self.connect_count.fetch_add(1, Ordering::SeqCst);
+ responder.respond(ConnectMcpResponse::new("test-connection-id"))
+ },
+ agent_client_protocol::on_receive_request!(),
+ )
+ .connect_to(client)
+ .await
+ }
+}
+
+impl ConnectTo for RecordingAgent {
+ async fn connect_to(
+ self,
+ client: impl ConnectTo,
+ ) -> Result<(), agent_client_protocol::Error> {
+ let capabilities = self.capabilities;
+ let new_observed = self.observed.clone();
+ let load_observed = self.observed.clone();
+ let resume_observed = self.observed;
+
+ Agent
+ .builder()
+ .name("recording-agent")
+ .on_receive_request(
+ async move |request: InitializeRequest, responder, _cx| {
+ responder.respond(
+ InitializeResponse::new(request.protocol_version)
+ .agent_capabilities(capabilities.clone()),
+ )
+ },
+ agent_client_protocol::on_receive_request!(),
+ )
+ .on_receive_request(
+ async move |request: NewSessionRequest, responder, _cx| {
+ new_observed.record(SetupMethod::New, request.mcp_servers);
+ responder.respond(NewSessionResponse::new("session-id"))
+ },
+ agent_client_protocol::on_receive_request!(),
+ )
+ .on_receive_request(
+ async move |request: LoadSessionRequest, responder, _cx| {
+ load_observed.record(SetupMethod::Load, request.mcp_servers);
+ responder.respond(LoadSessionResponse::new())
+ },
+ agent_client_protocol::on_receive_request!(),
+ )
+ .on_receive_request(
+ async move |request: ResumeSessionRequest, responder, _cx| {
+ resume_observed.record(SetupMethod::Resume, request.mcp_servers);
+ responder.respond(ResumeSessionResponse::new())
+ },
+ agent_client_protocol::on_receive_request!(),
+ )
+ .connect_to(client)
+ .await
+ }
+}
+
+fn agent_capabilities(mcp_capabilities: McpCapabilities) -> AgentCapabilities {
+ AgentCapabilities::new()
+ .load_session(true)
+ .session_capabilities(SessionCapabilities::new().resume(SessionResumeCapabilities::new()))
+ .mcp_capabilities(mcp_capabilities)
+}
+
+fn native_server() -> McpServer {
+ let meta = serde_json::Map::from_iter([(
+ "source".to_string(),
+ serde_json::Value::String("integration-test".to_string()),
+ )]);
+ McpServer::Acp(McpServerAcp::new(SERVER_NAME, SERVER_ID).meta(meta))
+}
+
+async fn recv(
+ response: agent_client_protocol::SentRequest,
+) -> Result {
+ let (tx, rx) = tokio::sync::oneshot::channel();
+ response.on_receiving_result(async move |result| {
+ tx.send(result)
+ .map_err(|_| agent_client_protocol::Error::internal_error())
+ })?;
+ rx.await
+ .map_err(|_| agent_client_protocol::Error::internal_error())?
+}
+
+async fn run_with_polyfill(
+ agent: RecordingAgent,
+ provider_connect_count: Arc,
+ editor_task: impl AsyncFnOnce(
+ agent_client_protocol::ConnectionTo,
+ ) -> Result<(), agent_client_protocol::Error>,
+) -> Result<(), agent_client_protocol::Error> {
+ drop(
+ tracing_subscriber::fmt()
+ .with_env_filter(tracing_subscriber::EnvFilter::from_default_env())
+ .with_test_writer()
+ .try_init(),
+ );
+
+ 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
+ .builder()
+ .name("polyfill-test-client")
+ .with_spawned(|_cx| async move {
+ ConductorImpl::new_agent(
+ "polyfill-test-conductor".to_string(),
+ ProxiesAndAgent::new(agent)
+ .proxy(NativeMcpProvider {
+ connect_count: provider_connect_count,
+ })
+ .proxy(McpOverAcpPolyfill::http()),
+ )
+ .run(agent_client_protocol::ByteStreams::new(
+ conductor_out.compat_write(),
+ conductor_in.compat(),
+ ))
+ .await
+ })
+ .connect_with(transport, editor_task)
+ .await
+}
+
+#[tokio::test]
+async fn http_downstream_receives_stable_transformed_declarations_for_all_setup_methods()
+-> Result<(), agent_client_protocol::Error> {
+ let observed = Arc::new(ObservedRequests::default());
+ let agent = RecordingAgent {
+ capabilities: agent_capabilities(McpCapabilities::new().http(true)),
+ observed: observed.clone(),
+ };
+ let connect_count = Arc::new(AtomicUsize::new(0));
+
+ run_with_polyfill(agent, connect_count.clone(), async |connection| {
+ let initialize =
+ recv(connection.send_request(InitializeRequest::new(ProtocolVersion::V1))).await?;
+ assert!(initialize.agent_capabilities.mcp_capabilities.http);
+ assert!(
+ initialize.agent_capabilities.mcp_capabilities.acp,
+ "the HTTP adapter should advertise native MCP support upstream"
+ );
+
+ let cwd = PathBuf::from("/tmp");
+ let session =
+ recv(connection.send_request(
+ NewSessionRequest::new(cwd.clone()).mcp_servers(vec![native_server()]),
+ ))
+ .await?;
+ recv(
+ connection.send_request(
+ LoadSessionRequest::new(session.session_id.clone(), cwd.clone())
+ .mcp_servers(vec![native_server()]),
+ ),
+ )
+ .await?;
+ recv(connection.send_request(
+ ResumeSessionRequest::new(session.session_id, cwd).mcp_servers(vec![native_server()]),
+ ))
+ .await?;
+
+ 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 native MCP connection"
+ );
+ assert_eq!(setup.len(), 3);
+ assert_eq!(setup[0].method, SetupMethod::New);
+ assert_eq!(setup[1].method, SetupMethod::Load);
+ assert_eq!(setup[2].method, SetupMethod::Resume);
+
+ let expected_meta = serde_json::Map::from_iter([(
+ "source".to_string(),
+ serde_json::Value::String("integration-test".to_string()),
+ )]);
+ let mut endpoint = None;
+ for request in setup.iter() {
+ let [McpServer::Http(server)] = request.mcp_servers.as_slice() else {
+ panic!(
+ "expected one HTTP MCP declaration 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:"));
+ 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_downstream_keeps_capability_and_declaration_unchanged()
+-> Result<(), agent_client_protocol::Error> {
+ let observed = Arc::new(ObservedRequests::default());
+ let agent = RecordingAgent {
+ capabilities: agent_capabilities(McpCapabilities::new().acp(true)),
+ observed: observed.clone(),
+ };
+ let declaration = native_server();
+ let expected = declaration.clone();
+ let connect_count = Arc::new(AtomicUsize::new(0));
+
+ run_with_polyfill(agent, connect_count.clone(), async move |connection| {
+ let initialize =
+ recv(connection.send_request(InitializeRequest::new(ProtocolVersion::V1))).await?;
+ assert!(!initialize.agent_capabilities.mcp_capabilities.http);
+ assert!(initialize.agent_capabilities.mcp_capabilities.acp);
+
+ recv(connection.send_request(
+ NewSessionRequest::new(PathBuf::from("/tmp")).mcp_servers(vec![declaration]),
+ ))
+ .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, vec![expected]);
+ assert_eq!(
+ connect_count.load(Ordering::SeqCst),
+ 0,
+ "a native-capable downstream should not be routed through the HTTP adapter"
+ );
+
+ Ok(())
+}
+
+#[tokio::test]
+async fn unsupported_downstream_does_not_gain_native_capability()
+-> Result<(), agent_client_protocol::Error> {
+ let agent = RecordingAgent {
+ capabilities: agent_capabilities(McpCapabilities::new()),
+ observed: Arc::default(),
+ };
+
+ run_with_polyfill(agent, Arc::default(), async |connection| {
+ let initialize =
+ recv(connection.send_request(InitializeRequest::new(ProtocolVersion::V1))).await?;
+ assert!(!initialize.agent_capabilities.mcp_capabilities.http);
+ assert!(
+ !initialize.agent_capabilities.mcp_capabilities.acp,
+ "the adapter must not advertise native MCP without a usable downstream transport"
+ );
+
+ let error = recv(connection.send_request(
+ NewSessionRequest::new(PathBuf::from("/tmp")).mcp_servers(vec![native_server()]),
+ ))
+ .await
+ .expect_err("native declarations must not reach an unsupported downstream agent");
+ 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
+}
diff --git a/src/agent-client-protocol-conductor/tests/mcp_server_handler_chain.rs b/src/agent-client-protocol-conductor/tests/mcp_server_handler_chain.rs
index 51134dd2..e41bf32a 100644
--- a/src/agent-client-protocol-conductor/tests/mcp_server_handler_chain.rs
+++ b/src/agent-client-protocol-conductor/tests/mcp_server_handler_chain.rs
@@ -8,8 +8,9 @@
use agent_client_protocol::mcp_server::McpServer;
use agent_client_protocol::schema::ProtocolVersion;
use agent_client_protocol::schema::v1::{
- AgentCapabilities, InitializeRequest, InitializeResponse, NewSessionRequest,
- NewSessionResponse, SessionId,
+ AgentCapabilities, InitializeRequest, InitializeResponse, LoadSessionRequest,
+ LoadSessionResponse, McpServer as SchemaMcpServer, McpServerAcpId, NewSessionRequest,
+ NewSessionResponse, ResumeSessionRequest, ResumeSessionResponse, SessionId,
};
use agent_client_protocol::{Agent, Client, Conductor, ConnectTo, DynConnectTo, Proxy};
use agent_client_protocol_conductor::{ConductorImpl, ProxiesAndAgent};
@@ -17,8 +18,8 @@ use agent_client_protocol_rmcp::McpServerExt as _;
use schemars::JsonSchema;
use serde::{Deserialize, Serialize};
use std::path::PathBuf;
-use std::sync::Arc;
use std::sync::atomic::{AtomicBool, Ordering};
+use std::sync::{Arc, Mutex};
use tokio::io::duplex;
use tokio_util::compat::{TokioAsyncReadCompatExt, TokioAsyncWriteCompatExt};
@@ -53,6 +54,26 @@ struct HandlerConfig {
new_session_handler_called: AtomicBool,
}
+#[derive(Default)]
+struct SetupRequests {
+ server_ids: Mutex>,
+}
+
+impl SetupRequests {
+ fn record(&self, mcp_servers: &[SchemaMcpServer]) {
+ let server_id = mcp_servers.iter().find_map(|server| match server {
+ SchemaMcpServer::Acp(server) if server.name == "test-server" => {
+ Some(server.server_id.clone())
+ }
+ _ => None,
+ });
+
+ if let Some(server_id) = server_id {
+ self.server_ids.lock().unwrap().push(server_id);
+ }
+ }
+}
+
impl HandlerConfig {
fn new() -> Arc {
Arc::new(Self {
@@ -122,13 +143,19 @@ impl ConnectTo for ProxyWithMcpAndHandler {
}
/// A simple agent that responds to initialization and session requests
-struct SimpleAgent;
+struct SimpleAgent {
+ setup_requests: Arc,
+}
impl ConnectTo for SimpleAgent {
async fn connect_to(
self,
client: impl ConnectTo,
) -> Result<(), agent_client_protocol::Error> {
+ let new_requests = self.setup_requests.clone();
+ let load_requests = self.setup_requests.clone();
+ let resume_requests = self.setup_requests;
+
Agent
.builder()
.name("simple-agent")
@@ -142,13 +169,28 @@ impl ConnectTo for SimpleAgent {
agent_client_protocol::on_receive_request!(),
)
.on_receive_request(
- async |_request: NewSessionRequest, responder, _cx| {
+ async move |request: NewSessionRequest, responder, _cx| {
+ new_requests.record(&request.mcp_servers);
responder.respond(NewSessionResponse::new(SessionId::new(
uuid::Uuid::new_v4().to_string(),
)))
},
agent_client_protocol::on_receive_request!(),
)
+ .on_receive_request(
+ async move |request: LoadSessionRequest, responder, _cx| {
+ load_requests.record(&request.mcp_servers);
+ responder.respond(LoadSessionResponse::new())
+ },
+ agent_client_protocol::on_receive_request!(),
+ )
+ .on_receive_request(
+ async move |request: ResumeSessionRequest, responder, _cx| {
+ resume_requests.record(&request.mcp_servers);
+ responder.respond(ResumeSessionResponse::new())
+ },
+ agent_client_protocol::on_receive_request!(),
+ )
.connect_to(client)
.await
}
@@ -195,7 +237,9 @@ async fn test_new_session_handler_invoked_with_mcp_server()
let proxy = DynConnectTo::::new(ProxyWithMcpAndHandler {
config: handler_config,
});
- let agent = DynConnectTo::::new(SimpleAgent);
+ let agent = DynConnectTo::::new(SimpleAgent {
+ setup_requests: Arc::default(),
+ });
run_test(vec![proxy], agent, async |connection_to_editor| {
// Initialize first
@@ -228,3 +272,55 @@ async fn test_new_session_handler_invoked_with_mcp_server()
Ok(())
}
+
+/// Global MCP servers are advertised consistently on every stable session setup request.
+#[tokio::test]
+async fn test_mcp_server_injected_into_all_session_setup_requests()
+-> Result<(), agent_client_protocol::Error> {
+ let handler_config = HandlerConfig::new();
+ let setup_requests = Arc::new(SetupRequests::default());
+
+ let proxy = DynConnectTo::::new(ProxyWithMcpAndHandler {
+ config: handler_config,
+ });
+ let agent = DynConnectTo::::new(SimpleAgent {
+ setup_requests: setup_requests.clone(),
+ });
+
+ run_test(vec![proxy], agent, async |connection_to_editor| {
+ recv(connection_to_editor.send_request(InitializeRequest::new(ProtocolVersion::V1)))
+ .await?;
+
+ let cwd = PathBuf::from("/tmp");
+ let new_session =
+ recv(connection_to_editor.send_request(NewSessionRequest::new(cwd.clone()))).await?;
+
+ recv(connection_to_editor.send_request(LoadSessionRequest::new(
+ new_session.session_id.clone(),
+ cwd.clone(),
+ )))
+ .await?;
+
+ recv(
+ connection_to_editor
+ .send_request(ResumeSessionRequest::new(new_session.session_id, cwd)),
+ )
+ .await?;
+
+ Ok::<(), agent_client_protocol::Error>(())
+ })
+ .await?;
+
+ let server_ids = setup_requests.server_ids.lock().unwrap();
+ assert_eq!(
+ server_ids.len(),
+ 3,
+ "each setup request should advertise the server"
+ );
+ assert!(
+ server_ids.windows(2).all(|ids| ids[0] == ids[1]),
+ "a global MCP server should keep one advertised server ID"
+ );
+
+ Ok(())
+}
diff --git a/src/agent-client-protocol-conductor/tests/request_cancellation.rs b/src/agent-client-protocol-conductor/tests/request_cancellation.rs
index 2c93361d..34eac429 100644
--- a/src/agent-client-protocol-conductor/tests/request_cancellation.rs
+++ b/src/agent-client-protocol-conductor/tests/request_cancellation.rs
@@ -19,14 +19,15 @@ use std::sync::{Arc, Mutex};
use std::time::Duration;
use agent_client_protocol::DynConnectTo;
+use agent_client_protocol::schema::ProtocolVersion;
use agent_client_protocol::schema::v1::{
- CancelRequestNotification, ContentBlock, ContentChunk, InitializeRequest, InitializeResponse,
- McpServer as SchemaMcpServer, NewSessionRequest, NewSessionResponse, PermissionOption,
- PermissionOptionKind, PromptRequest, PromptResponse, RequestPermissionOutcome,
- RequestPermissionRequest, RequestPermissionResponse, SelectedPermissionOutcome, SessionId,
- SessionNotification, SessionUpdate, StopReason, ToolCallUpdate, ToolCallUpdateFields,
+ CancelRequestNotification, ConnectMcpRequest, ContentBlock, ContentChunk, InitializeRequest,
+ InitializeResponse, McpServer as SchemaMcpServer, McpServerAcpId, NewSessionRequest,
+ NewSessionResponse, PermissionOption, PermissionOptionKind, PromptRequest, PromptResponse,
+ RequestPermissionOutcome, RequestPermissionRequest, RequestPermissionResponse,
+ SelectedPermissionOutcome, SessionId, SessionNotification, SessionUpdate, StopReason,
+ ToolCallUpdate, ToolCallUpdateFields,
};
-use agent_client_protocol::schema::{McpConnectRequest, ProtocolVersion};
use agent_client_protocol::{
Agent, ByteStreams, Client, Conductor, ConnectTo, ConnectionTo, Error, JsonRpcRequest,
JsonRpcResponse, NullRun, Proxy, Responder, Role, SentRequest,
@@ -53,7 +54,7 @@ struct SimpleResponse {
#[derive(Clone)]
struct TrackingMcpServer {
- connect_tx: mpsc::UnboundedSender,
+ connect_tx: mpsc::UnboundedSender,
}
impl McpServerConnect for TrackingMcpServer {
@@ -63,7 +64,11 @@ impl McpServerConnect for TrackingMcpServer {
fn connect(&self, cx: McpConnectionTo) -> DynConnectTo {
self.connect_tx
- .unbounded_send(cx.acp_id().to_owned())
+ .unbounded_send(
+ cx.server_id()
+ .expect("cancellation test server is attached through ACP")
+ .clone(),
+ )
.unwrap();
DynConnectTo::new(EmptyMcpServerComponent)
}
@@ -102,10 +107,10 @@ fn assert_no_event(rx: &mut mpsc::UnboundedReceiver) {
}
}
-fn advertised_mcp_acp_id(request: &NewSessionRequest) -> String {
+fn advertised_mcp_server_id(request: &NewSessionRequest) -> McpServerAcpId {
match request.mcp_servers.as_slice() {
- [SchemaMcpServer::Http(http)] => http.url.clone(),
- servers => panic!("expected exactly one HTTP MCP server, got {servers:?}"),
+ [SchemaMcpServer::Acp(acp)] => acp.server_id.clone(),
+ servers => panic!("expected exactly one ACP MCP server, got {servers:?}"),
}
}
@@ -902,7 +907,7 @@ async fn proxy_session_helper_cleans_up_mcp_handlers_after_cancelled_session() -
let (parked_id_tx, mut parked_id_rx) = mpsc::unbounded();
let (mcp_connect_tx, mut mcp_connect_rx) = mpsc::unbounded();
let (probe_barrier_tx, mut probe_barrier_rx) = mpsc::unbounded();
- let cancelled_mcp_acp_id = Arc::new(Mutex::new(None::));
+ let cancelled_mcp_server_id = Arc::new(Mutex::new(None::));
let agent = Agent
.builder()
@@ -914,22 +919,22 @@ async fn proxy_session_helper_cleans_up_mcp_handlers_after_cancelled_session() -
)
.on_receive_request(
{
- let cancelled_mcp_acp_id = cancelled_mcp_acp_id.clone();
+ let cancelled_mcp_server_id = cancelled_mcp_server_id.clone();
let parked_id_tx = parked_id_tx.clone();
let probe_barrier_tx = probe_barrier_tx.clone();
async move |request: NewSessionRequest,
responder: Responder,
cx: ConnectionTo| {
- let cancelled_mcp_acp_id = cancelled_mcp_acp_id.clone();
+ let cancelled_mcp_server_id = cancelled_mcp_server_id.clone();
let parked_id_tx = parked_id_tx.clone();
let probe_barrier_tx = probe_barrier_tx.clone();
- let advertised_mcp_acp_id = advertised_mcp_acp_id(&request);
+ let advertised_mcp_server_id = advertised_mcp_server_id(&request);
if request.cwd.ends_with("park-session") {
- *cancelled_mcp_acp_id
+ *cancelled_mcp_server_id
.lock()
.expect("cancelled MCP ID mutex poisoned") =
- Some(advertised_mcp_acp_id);
+ Some(advertised_mcp_server_id);
parked_id_tx.unbounded_send(responder.id().clone()).unwrap();
let cancellation = responder.cancellation();
cx.spawn(async move {
@@ -945,7 +950,7 @@ async fn proxy_session_helper_cleans_up_mcp_handlers_after_cancelled_session() -
responder.respond(NewSessionResponse::new(SessionId::new("normal-session")))?;
- let stale_acp_id = cancelled_mcp_acp_id
+ let stale_server_id = cancelled_mcp_server_id
.lock()
.expect("cancelled MCP ID mutex poisoned")
.clone()
@@ -953,10 +958,7 @@ async fn proxy_session_helper_cleans_up_mcp_handlers_after_cancelled_session() -
let connection = cx.clone();
cx.spawn(async move {
connection
- .send_request(McpConnectRequest {
- acp_id: stale_acp_id,
- meta: None,
- })
+ .send_request(ConnectMcpRequest::new(stale_server_id))
.on_receiving_result(async |_| Ok(()))?;
let barrier = connection
diff --git a/src/agent-client-protocol-conductor/tests/standalone_mcp_server.rs b/src/agent-client-protocol-conductor/tests/standalone_mcp_server.rs
index 4ac52c2c..a07b5dcc 100644
--- a/src/agent-client-protocol-conductor/tests/standalone_mcp_server.rs
+++ b/src/agent-client-protocol-conductor/tests/standalone_mcp_server.rs
@@ -1,7 +1,7 @@
//! Tests for running McpServer as a standalone MCP server (not part of ACP).
//!
//! These tests verify that `McpServer` can be used directly with MCP clients
-//! via the `Component` implementation.
+//! via its `ConnectTo` implementation.
use agent_client_protocol::{
ByteStreams, ConnectTo, RunWithConnectionTo, mcp_server::McpServer, role::mcp, util::run_until,
@@ -38,7 +38,12 @@ fn create_test_server() -> McpServer DynConnectTo {
- // Create MCP server with an echo tool that returns the session_id
let mcp_server = McpServer::builder("echo_server".to_string())
- .instructions("Test MCP server with session_id echo tool")
+ .instructions("Test MCP server with a connection-context echo tool")
.tool_fn_mut(
"echo",
- "Returns the current session_id",
+ "Returns the current MCP connection context",
async |_input: EchoInput, context| {
Ok(EchoOutput {
- acp_id: context.acp_id().to_owned(),
+ server_id: context
+ .server_id()
+ .expect("tool is attached through ACP")
+ .to_string(),
+ connection_id: context
+ .connection_id()
+ .expect("tool is attached through ACP")
+ .to_string(),
})
},
agent_client_protocol::tool_fn_mut!(),
)
.build();
- // Create proxy component
DynConnectTo::new(ProxyWithEchoServer { mcp_server })
}
@@ -87,17 +86,17 @@ async fn test_list_tools_from_mcp_server() -> Result<(), agent_client_protocol::
)
.await?;
- // Check the response using expect_test
expect![[r"
Available tools:
- - echo: Returns the current session_id"]]
+ - echo: Returns the current MCP connection context"]]
.assert_eq(&result);
Ok(())
}
#[tokio::test]
-async fn test_session_id_delivered_to_mcp_tools() -> Result<(), agent_client_protocol::Error> {
+async fn test_acp_identifiers_are_delivered_to_mcp_tools()
+-> Result<(), agent_client_protocol::Error> {
let result = yopo::prompt(
ConductorImpl::new_agent(
"test-conductor".to_string(),
@@ -114,8 +113,16 @@ async fn test_session_id_delivered_to_mcp_tools() -> Result<(), agent_client_pro
)
.await?;
- let pattern = regex::Regex::new(r#""acp_id":\s*String\("acp:[0-9a-f-]+"\)"#).unwrap();
- assert!(pattern.is_match(&result), "unexpected result: {result}");
+ let server_id = regex::Regex::new(r#""server_id":\s*String\("mcp-server:[0-9a-f-]+"\)"#)
+ .expect("valid server ID regex");
+ let connection_id =
+ regex::Regex::new(r#""connection_id":\s*String\("mcp-over-acp-connection:[0-9a-f-]+"\)"#)
+ .expect("valid connection ID regex");
+ assert!(server_id.is_match(&result), "unexpected result: {result}");
+ assert!(
+ connection_id.is_match(&result),
+ "unexpected result: {result}"
+ );
Ok(())
}
diff --git a/src/agent-client-protocol-conductor/tests/trace_client_mcp_server.rs b/src/agent-client-protocol-conductor/tests/trace_client_mcp_server.rs
index 5b2b61d6..f350fee8 100644
--- a/src/agent-client-protocol-conductor/tests/trace_client_mcp_server.rs
+++ b/src/agent-client-protocol-conductor/tests/trace_client_mcp_server.rs
@@ -32,15 +32,17 @@ use tokio_util::compat::{TokioAsyncReadCompatExt, TokioAsyncWriteCompatExt};
/// - Strips timestamps (set to 0.0)
/// - Replaces UUIDs with sequential IDs (id:0, id:1, etc.)
/// - Replaces session IDs with "session:0", etc.
-/// - Replaces acp: URLs with "acp:url:0", etc.
-/// - Replaces connection_id with "connection:0", etc.
+/// - Replaces loopback HTTP endpoints with "http:endpoint:0", etc.
+/// - Replaces MCP server and connection IDs with stable sequential IDs
struct EventNormalizer {
id_map: HashMap,
next_id: usize,
session_map: HashMap,
next_session: usize,
- acp_url_map: HashMap,
- next_acp_url: usize,
+ endpoint_map: HashMap,
+ next_endpoint: usize,
+ server_map: HashMap,
+ next_server: usize,
connection_map: HashMap,
next_connection: usize,
}
@@ -52,8 +54,10 @@ impl EventNormalizer {
next_id: 0,
session_map: HashMap::new(),
next_session: 0,
- acp_url_map: HashMap::new(),
- next_acp_url: 0,
+ endpoint_map: HashMap::new(),
+ next_endpoint: 0,
+ server_map: HashMap::new(),
+ next_server: 0,
connection_map: HashMap::new(),
next_connection: 0,
}
@@ -90,12 +94,23 @@ impl EventNormalizer {
.clone()
}
- fn normalize_acp_url(&mut self, url: &str) -> String {
- self.acp_url_map
+ fn normalize_endpoint(&mut self, url: &str) -> String {
+ self.endpoint_map
.entry(url.to_string())
.or_insert_with(|| {
- let n = format!("acp:url:{}", self.next_acp_url);
- self.next_acp_url += 1;
+ let n = format!("http:endpoint:{}", self.next_endpoint);
+ self.next_endpoint += 1;
+ n
+ })
+ .clone()
+ }
+
+ fn normalize_server_id(&mut self, id: &str) -> String {
+ self.server_map
+ .entry(id.to_string())
+ .or_insert_with(|| {
+ let n = format!("server:{}", self.next_server);
+ self.next_server += 1;
n
})
.clone()
@@ -112,7 +127,7 @@ impl EventNormalizer {
.clone()
}
- /// Recursively normalize session IDs, acp: URLs, and connection IDs in JSON values.
+ /// Recursively normalize session IDs, MCP endpoints, and MCP IDs in JSON values.
fn normalize_json(&mut self, value: serde_json::Value) -> serde_json::Value {
match value {
serde_json::Value::Object(map) => {
@@ -125,17 +140,25 @@ impl EventNormalizer {
} else {
self.normalize_json(v)
}
- } else if k == "url" || k == "acp_id" {
+ } else if k == "url" {
if let serde_json::Value::String(s) = &v {
- if s.starts_with("acp:") || s.starts_with("http://localhost:") {
- serde_json::Value::String(self.normalize_acp_url(s))
+ if s.starts_with("http://127.0.0.1:")
+ || s.starts_with("http://localhost:")
+ {
+ serde_json::Value::String(self.normalize_endpoint(s))
} else {
v
}
} else {
self.normalize_json(v)
}
- } else if k == "connection_id" {
+ } else if k == "serverId" {
+ if let serde_json::Value::String(s) = &v {
+ serde_json::Value::String(self.normalize_server_id(s))
+ } else {
+ self.normalize_json(v)
+ }
+ } else if k == "connectionId" {
if let serde_json::Value::String(s) = &v {
serde_json::Value::String(self.normalize_connection_id(s))
} else {
@@ -227,7 +250,7 @@ async fn test_trace_client_mcp_server() -> Result<(), agent_client_protocol::Err
let (client_write, conductor_read) = duplex(8192);
let (conductor_write, client_read) = duplex(8192);
- // Spawn the conductor with ElizaAgent (no proxies - simple setup)
+ // Spawn the conductor with Testy (no application proxies; only the compatibility adapter).
let conductor_handle = tokio::spawn(async move {
ConductorImpl::new_agent(
"conductor".to_string(),
@@ -385,10 +408,9 @@ async fn test_trace_client_mcp_server() -> Result<(), agent_client_protocol::Err
"cwd": String("."),
"mcpServers": Array [
Object {
- "type": String("http"),
+ "type": String("acp"),
"name": String("echo-server"),
- "url": String("acp:url:0"),
- "headers": Array [],
+ "serverId": String("server:0"),
},
],
},
diff --git a/src/agent-client-protocol-conductor/tests/trace_mcp_tool_call.rs b/src/agent-client-protocol-conductor/tests/trace_mcp_tool_call.rs
index 21d4ed12..a777ddb7 100644
--- a/src/agent-client-protocol-conductor/tests/trace_mcp_tool_call.rs
+++ b/src/agent-client-protocol-conductor/tests/trace_mcp_tool_call.rs
@@ -31,15 +31,17 @@ use tokio_util::compat::{TokioAsyncReadCompatExt, TokioAsyncWriteCompatExt};
/// - Strips timestamps (set to 0.0)
/// - Replaces UUIDs with sequential IDs (id:0, id:1, etc.)
/// - Replaces session IDs with "session:0", etc.
-/// - Replaces acp: URLs with "acp:url:0", etc.
-/// - Replaces connection_id with "connection:0", etc.
+/// - Replaces loopback HTTP endpoints with "http:endpoint:0", etc.
+/// - Replaces MCP server and connection IDs with stable sequential IDs
struct EventNormalizer {
id_map: HashMap,
next_id: usize,
session_map: HashMap,
next_session: usize,
- acp_url_map: HashMap,
- next_acp_url: usize,
+ endpoint_map: HashMap,
+ next_endpoint: usize,
+ server_map: HashMap,
+ next_server: usize,
connection_map: HashMap,
next_connection: usize,
}
@@ -51,8 +53,10 @@ impl EventNormalizer {
next_id: 0,
session_map: HashMap::new(),
next_session: 0,
- acp_url_map: HashMap::new(),
- next_acp_url: 0,
+ endpoint_map: HashMap::new(),
+ next_endpoint: 0,
+ server_map: HashMap::new(),
+ next_server: 0,
connection_map: HashMap::new(),
next_connection: 0,
}
@@ -89,12 +93,23 @@ impl EventNormalizer {
.clone()
}
- fn normalize_acp_url(&mut self, url: &str) -> String {
- self.acp_url_map
+ fn normalize_endpoint(&mut self, url: &str) -> String {
+ self.endpoint_map
.entry(url.to_string())
.or_insert_with(|| {
- let n = format!("acp:url:{}", self.next_acp_url);
- self.next_acp_url += 1;
+ let n = format!("http:endpoint:{}", self.next_endpoint);
+ self.next_endpoint += 1;
+ n
+ })
+ .clone()
+ }
+
+ fn normalize_server_id(&mut self, id: &str) -> String {
+ self.server_map
+ .entry(id.to_string())
+ .or_insert_with(|| {
+ let n = format!("server:{}", self.next_server);
+ self.next_server += 1;
n
})
.clone()
@@ -111,7 +126,7 @@ impl EventNormalizer {
.clone()
}
- /// Recursively normalize session IDs, acp: URLs, and connection IDs in JSON values.
+ /// Recursively normalize session IDs, MCP endpoints, and MCP IDs in JSON values.
fn normalize_json(&mut self, value: serde_json::Value) -> serde_json::Value {
match value {
serde_json::Value::Object(map) => {
@@ -124,17 +139,25 @@ impl EventNormalizer {
} else {
self.normalize_json(v)
}
- } else if k == "url" || k == "acp_id" {
+ } else if k == "url" {
if let serde_json::Value::String(s) = &v {
- if s.starts_with("acp:") || s.starts_with("http://localhost:") {
- serde_json::Value::String(self.normalize_acp_url(s))
+ if s.starts_with("http://127.0.0.1:")
+ || s.starts_with("http://localhost:")
+ {
+ serde_json::Value::String(self.normalize_endpoint(s))
} else {
v
}
} else {
self.normalize_json(v)
}
- } else if k == "connection_id" {
+ } else if k == "serverId" {
+ if let serde_json::Value::String(s) = &v {
+ serde_json::Value::String(self.normalize_server_id(s))
+ } else {
+ self.normalize_json(v)
+ }
+ } else if k == "connectionId" {
if let serde_json::Value::String(s) = &v {
serde_json::Value::String(self.normalize_connection_id(s))
} else {
@@ -210,7 +233,7 @@ async fn test_trace_mcp_tool_call() -> Result<(), agent_client_protocol::Error>
let (conductor_write, client_read) = duplex(8192);
// Spawn the conductor with:
- // - ElizaAgent (deterministic mode) as the agent
+ // - Testy as the deterministic agent
// - ProxyComponent that provides the "test" MCP server with echo tool
// - Tracing enabled to capture events
let conductor_handle = tokio::spawn(async move {
@@ -459,10 +482,9 @@ async fn test_trace_mcp_tool_call() -> Result<(), agent_client_protocol::Error>
"cwd": String("/"),
"mcpServers": Array [
Object {
- "type": String("http"),
+ "type": String("acp"),
"name": String("test"),
- "url": String("acp:url:0"),
- "headers": Array [],
+ "serverId": String("server:0"),
},
],
},
@@ -475,10 +497,10 @@ async fn test_trace_mcp_tool_call() -> Result<(), agent_client_protocol::Error>
from: "Proxy(1)",
to: "Proxy(0)",
id: String("id:4"),
- method: "_mcp/connect",
+ method: "mcp/connect",
session: None,
params: Object {
- "acp_id": String("acp:url:0"),
+ "serverId": String("server:0"),
},
},
),
@@ -490,7 +512,7 @@ async fn test_trace_mcp_tool_call() -> Result<(), agent_client_protocol::Error>
id: String("id:4"),
is_error: false,
payload: Object {
- "connection_id": String("connection:0"),
+ "connectionId": String("connection:0"),
},
},
),
diff --git a/src/agent-client-protocol-cookbook/Cargo.toml b/src/agent-client-protocol-cookbook/Cargo.toml
index a43a289e..085a260f 100644
--- a/src/agent-client-protocol-cookbook/Cargo.toml
+++ b/src/agent-client-protocol-cookbook/Cargo.toml
@@ -14,7 +14,7 @@ categories = ["development-tools"]
[dependencies]
[dev-dependencies]
-agent-client-protocol.workspace = true
+agent-client-protocol = { workspace = true, features = ["unstable_mcp_over_acp"] }
agent-client-protocol-rmcp.workspace = true
rmcp.workspace = true
schemars.workspace = true
diff --git a/src/agent-client-protocol-cookbook/src/lib.rs b/src/agent-client-protocol-cookbook/src/lib.rs
index 2aeda1fc..56f65a63 100644
--- a/src/agent-client-protocol-cookbook/src/lib.rs
+++ b/src/agent-client-protocol-cookbook/src/lib.rs
@@ -163,7 +163,8 @@ pub mod connecting_as_client {
//! - [`read_update`] - Read the next update (text chunk, tool call, etc.)
//! - [`read_to_string`] - Read all text until the turn ends
//!
- //! The session builder also supports adding MCP servers with [`with_mcp_server`].
+ //! With the core SDK's `unstable_mcp_over_acp` feature, the session builder
+ //! also supports adding MCP servers with [`with_mcp_server`].
//!
//! # Handling Permission Requests
//!
@@ -434,7 +435,9 @@ pub mod global_mcp_server {
//!
//! Use this pattern when you want a single MCP server that handles tool calls
//! for all sessions. The server is added to the connection's handler chain and
- //! automatically injects itself into every `NewSessionRequest` that passes through.
+ //! automatically injects itself into every supported session setup request.
+ //! This pattern requires the core SDK's `unstable_mcp_over_acp` feature (or
+ //! the rmcp crate's matching passthrough feature).
//!
//! # When to use
//!
@@ -545,10 +548,12 @@ pub mod global_mcp_server {
//! When you call [`with_mcp_server`], the MCP server is added as a message
//! handler. It:
//!
- //! 1. Intercepts `NewSessionRequest` messages and adds its `acp:UUID` URL to the
- //! request's `mcp_servers` list
+ //! 1. Intercepts session setup requests and adds a schema-native
+ //! `McpServer::Acp` declaration with a unique server ID to each request's
+ //! `mcp_servers` list (`session/new`, `session/load`, `session/resume`, and
+ //! feature-gated `session/fork`)
//! 2. Passes the modified request through to the next handler
- //! 3. Handles incoming MCP protocol messages (tool calls, etc.) for its URL
+ //! 3. Handles `mcp/connect`, `mcp/message`, and `mcp/disconnect` for that server ID
//!
//! [`McpServer::builder`]: agent_client_protocol_rmcp::McpServerExt::builder
//! [`McpServer::from_rmcp`]: agent_client_protocol_rmcp::McpServerExt::from_rmcp
@@ -560,6 +565,8 @@ pub mod per_session_mcp_server {
//!
//! Use this pattern when each session needs its own MCP server instance
//! with access to session-specific context like the working directory.
+ //! It requires the core SDK's `unstable_mcp_over_acp` feature (or the rmcp
+ //! crate's matching passthrough feature).
//!
//! # When to use
//!
diff --git a/src/agent-client-protocol-polyfill/CHANGELOG.md b/src/agent-client-protocol-polyfill/CHANGELOG.md
index 061859a0..5484ebf5 100644
--- a/src/agent-client-protocol-polyfill/CHANGELOG.md
+++ b/src/agent-client-protocol-polyfill/CHANGELOG.md
@@ -11,10 +11,22 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
- **Breaking:** Upgrade to `agent-client-protocol` 2.x. Polyfill components and the core
handlers/types they connect must be migrated together.
+- **Breaking:** Consume native `McpServer::Acp` declarations and route `mcp/connect`,
+ `mcp/message`, and request/response `mcp/disconnect`. The bridge no longer recognizes
+ `McpServer::Http` values with `acp:` URLs or the SDK-local underscore-prefixed method family.
+- **Breaking:** Remove the public `BridgeMode` enum and `McpOverAcpPolyfill::stdio`; the repository
+ no longer ships the conductor helper subcommand that stdio mode required. The polyfill now has
+ one supported configuration, selected with `McpOverAcpPolyfill::http()` or `Default`; otherwise
+ use a separately managed MCP transport.
- **Changed:** Align the bridge background-task terminology with the core runner APIs.
-- **Documentation:** Update legacy v1 MCP-over-ACP conductor composition examples for the current
- constructor and distinguish it from the draft native MCP-over-ACP transport.
-- **Fixed:** Preserve JSON-RPC batch frames through the legacy MCP-over-ACP HTTP bridge, answer
+- **Changed:** Keep MCP-over-ACP native on the provider-facing side while translating declarations
+ to localhost HTTP only for a final agent that lacks native transport support.
+- **Changed:** Adapt native declarations in new, load, resume, and optionally fork session setup;
+ pass them through unchanged when the successor already supports native MCP-over-ACP, and
+ advertise the adapted capability only when the successor supports HTTP MCP.
+- **Documentation:** Update MCP-over-ACP conductor composition examples for the current native
+ declaration and compatibility boundary.
+- **Fixed:** Preserve JSON-RPC batch frames through the MCP-over-ACP HTTP bridge, answer
malformed calls on their originating POST, and ignore malformed response-shaped input.
- **Fixed:** Serialize overlapping request IDs (including `id: null`) and response-bearing batches
without identifiable request IDs, and retain active correlations after an HTTP caller
diff --git a/src/agent-client-protocol-polyfill/Cargo.toml b/src/agent-client-protocol-polyfill/Cargo.toml
index 370603d7..ccef2395 100644
--- a/src/agent-client-protocol-polyfill/Cargo.toml
+++ b/src/agent-client-protocol-polyfill/Cargo.toml
@@ -11,9 +11,12 @@ description = "Polyfill proxies for Agent Client Protocol backward compatibility
keywords = ["acp", "agent", "mcp", "polyfill"]
categories = ["development-tools"]
+[features]
+default = []
+unstable_session_fork = ["agent-client-protocol/unstable_session_fork"]
+
[dependencies]
agent-client-protocol = { workspace = true, features = ["unstable_mcp_over_acp"] }
-anyhow.workspace = true
async-stream.workspace = true
axum.workspace = true
futures.workspace = true
@@ -22,7 +25,6 @@ rustc-hash.workspace = true
serde_json.workspace = true
thiserror = "2.0"
tokio = { workspace = true, features = ["net"] }
-tokio-util.workspace = true
tracing.workspace = true
uuid.workspace = true
diff --git a/src/agent-client-protocol-polyfill/src/lib.rs b/src/agent-client-protocol-polyfill/src/lib.rs
index c7113723..46a7bf6c 100644
--- a/src/agent-client-protocol-polyfill/src/lib.rs
+++ b/src/agent-client-protocol-polyfill/src/lib.rs
@@ -1,14 +1,10 @@
//! # agent-client-protocol-polyfill
//!
-//! Polyfill proxies for backward compatibility with agents that don't support
-//! newer ACP features natively.
+//! Polyfill proxies for adapting newer ACP features to older agents.
//!
//! ## MCP-over-ACP Polyfill
//!
-//! The [`mcp_over_acp`] module implements the legacy v1 MCP-over-ACP extension for
-//! agents that don't support `mcpCapabilities.acp`. It transforms `McpServer::Http`
-//! entries with `acp:` URLs into localhost TCP bridges and routes the legacy
-//! `_mcp/connect`, `_mcp/message`, and `_mcp/disconnect` methods through those bridges.
-//! This is distinct from the draft native `McpServer::Acp` transport and its `mcp/*` methods.
+//! The [`mcp_over_acp`] module consumes schema-native `McpServer::Acp` declarations and
+//! `mcp/*` messages, exposing each server to an agent through a loopback HTTP bridge.
pub mod mcp_over_acp;
diff --git a/src/agent-client-protocol-polyfill/src/mcp_over_acp/actor.rs b/src/agent-client-protocol-polyfill/src/mcp_over_acp/actor.rs
index 77af1a02..908b6b87 100644
--- a/src/agent-client-protocol-polyfill/src/mcp_over_acp/actor.rs
+++ b/src/agent-client-protocol-polyfill/src/mcp_over_acp/actor.rs
@@ -1,6 +1,4 @@
-use agent_client_protocol::{
- ConnectTo, Dispatch, DynConnectTo, role::mcp, schema::McpDisconnectNotification,
-};
+use agent_client_protocol::{ConnectTo, Dispatch, DynConnectTo, role::mcp};
use futures::{SinkExt as _, StreamExt as _, channel::mpsc};
use tracing::info;
@@ -10,7 +8,7 @@ use super::BridgeMessage;
/// and the ACP proxy chain.
#[derive(Debug)]
pub(crate) struct BridgeConnectionActor {
- /// How to connect to the MCP server (e.g., stdio or HTTP transport).
+ /// The loopback HTTP transport accepted by the compatibility listener.
transport: DynConnectTo,
/// Sender for messages back to the polyfill's bridge runner loop.
@@ -71,12 +69,7 @@ impl BridgeConnectionActor {
.await;
bridge_tx
- .send(BridgeMessage::Disconnected {
- notification: McpDisconnectNotification {
- connection_id,
- meta: None,
- },
- })
+ .send(BridgeMessage::Disconnected { connection_id })
.await
.map_err(|_| agent_client_protocol::Error::internal_error())?;
diff --git a/src/agent-client-protocol-polyfill/src/mcp_over_acp/http.rs b/src/agent-client-protocol-polyfill/src/mcp_over_acp/http.rs
index 02a7df60..afed504b 100644
--- a/src/agent-client-protocol-polyfill/src/mcp_over_acp/http.rs
+++ b/src/agent-client-protocol-polyfill/src/mcp_over_acp/http.rs
@@ -31,14 +31,14 @@ use super::{BridgeConnection, BridgeMessage, actor::BridgeConnectionActor};
/// Runs an HTTP listener for MCP bridge connections.
pub async fn run_http_listener(
tcp_listener: TcpListener,
- acp_id: String,
+ server_id: String,
mut bridge_tx: mpsc::Sender,
) -> Result<(), agent_client_protocol::Error> {
let (to_mcp_client_tx, to_mcp_client_rx) = mpsc::channel(128);
bridge_tx
.send(BridgeMessage::ConnectionReceived {
- acp_id,
+ server_id,
actor: BridgeConnectionActor::new(
HttpMcpBridge::new(tcp_listener),
bridge_tx.clone(),
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 7aa96203..14b4b280 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,21 +1,14 @@
-//! Legacy v1 MCP-over-ACP polyfill proxy.
+//! MCP-over-ACP compatibility proxy.
//!
-//! This proxy bridges the legacy v1 MCP-over-ACP transport for agents that don't support
-//! `mcpCapabilities.acp` natively. It sits in the proxy chain and:
-//!
-//! - Intercepts `NewSessionRequest` to transform `McpServer::Http` entries with `acp:` URLs
-//! into localhost TCP bridges
-//! - Handles `_mcp/connect`, `_mcp/message`, `_mcp/disconnect` by routing through those bridges
-//!
-//! This extension is distinct from the draft native `McpServer::Acp` transport and
-//! its `mcp/*` methods.
+//! 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.
//!
//! # Usage
//!
//! ```rust,ignore
//! use agent_client_protocol_polyfill::mcp_over_acp::McpOverAcpPolyfill;
//!
-//! // Add to a conductor proxy chain
//! let conductor = ConductorImpl::new_agent(
//! "conductor",
//! ProxiesAndAgent::new(my_agent).proxy(McpOverAcpPolyfill::http()),
@@ -24,59 +17,105 @@
mod actor;
pub(crate) mod http;
-pub(crate) mod stdio;
use std::collections::HashMap;
-use std::path::PathBuf;
-use agent_client_protocol::schema::v1::{
- McpServer, McpServerHttp, McpServerStdio, NewSessionRequest,
-};
-use agent_client_protocol::schema::{
- InitializeProxyRequest, McpConnectRequest, McpConnectResponse, McpDisconnectNotification,
- McpOverAcpMessage,
-};
use agent_client_protocol::{
- Agent, Client, Conductor, ConnectTo, ConnectionTo, Dispatch, Proxy, Role,
+ 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,
+ },
+ },
};
-use futures::{SinkExt, channel::mpsc};
+use futures::{SinkExt, channel::mpsc, channel::oneshot};
use tokio::net::TcpListener;
-use tracing::info;
+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
+ }
+ }
+}
+
/// Internal messages for the polyfill's bridge management.
#[derive(Debug)]
pub(crate) enum BridgeMessage {
- /// A new TCP connection was accepted and needs an ACP connection ID.
+ /// Record which MCP transport the successor can consume.
+ SetDownstreamMode(DownstreamMcpMode),
+
+ /// Transform the MCP declarations for one session setup request.
+ TransformServers {
+ servers: Vec,
+ response_tx: oneshot::Sender, agent_client_protocol::Error>>,
+ },
+
+ /// A new TCP connection was accepted and needs a native MCP connection ID.
ConnectionReceived {
- acp_id: String,
+ server_id: String,
actor: BridgeConnectionActor,
connection: BridgeConnection,
},
- /// ACP connection ID received — spawn the actor and store the connection.
+ /// A native MCP connection ID was received; spawn the actor and store its sender.
ConnectionEstablished {
- response: McpConnectResponse,
+ server_id: String,
+ connection_id: McpConnectionId,
actor: BridgeConnectionActor,
connection: BridgeConnection,
},
- /// MCP message from a bridge client that needs to be forwarded over ACP.
+ /// Opening a native MCP connection failed.
+ ConnectionFailed { server_id: String },
+
+ /// An MCP message from the local agent that must be sent over ACP.
ClientToServer {
connection_id: String,
message: Dispatch,
},
- /// Bridge client disconnected.
- Disconnected {
- notification: McpDisconnectNotification,
+ /// An MCP server request received over ACP for the local agent's MCP client.
+ ServerToClientRequest {
+ request: MessageMcpRequest,
+ responder: Responder,
},
+
+ /// An MCP server notification received over ACP for the local agent's MCP client.
+ ServerToClientNotification {
+ notification: MessageMcpNotification,
+ },
+
+ /// The local MCP bridge disconnected.
+ Disconnected { connection_id: String },
}
/// Connection handle for sending messages to an MCP client via a bridge.
#[derive(Clone, Debug)]
-#[allow(dead_code)]
pub(crate) struct BridgeConnection {
to_mcp_client_tx: mpsc::Sender,
}
@@ -86,52 +125,23 @@ impl BridgeConnection {
Self { to_mcp_client_tx }
}
- #[allow(dead_code)]
- pub async fn send(&mut self, message: Dispatch) -> Result<(), agent_client_protocol::Error> {
+ fn try_send(&mut self, message: Dispatch) -> Option> {
self.to_mcp_client_tx
- .send(message)
- .await
- .map_err(|_| agent_client_protocol::Error::internal_error())
+ .try_send(message)
+ .err()
+ .map(|error| Box::new(error.into_inner()))
}
}
-/// Mode for the MCP bridge transport.
-#[derive(Debug, Clone, Default)]
-pub enum BridgeMode {
- /// Use stdio-based MCP bridge with a subprocess.
- Stdio {
- /// Command and args to spawn bridge processes.
- conductor_command: Vec,
- },
-
- /// Use HTTP-based MCP bridge (default).
- #[default]
- Http,
-}
-
-/// Legacy v1 MCP-over-ACP polyfill proxy.
-///
-/// Bridges the legacy transport for agents that don't support `mcpCapabilities.acp`.
-#[derive(Debug)]
-pub struct McpOverAcpPolyfill {
- mode: BridgeMode,
-}
+/// Adapts schema-native MCP-over-ACP declarations for agents that support HTTP MCP.
+#[derive(Debug, Default)]
+pub struct McpOverAcpPolyfill;
impl McpOverAcpPolyfill {
- /// Create a polyfill using HTTP bridge mode.
+ /// Create a polyfill that exposes each ACP MCP server through loopback HTTP.
#[must_use]
pub fn http() -> Self {
- Self {
- mode: BridgeMode::Http,
- }
- }
-
- /// Create a polyfill using stdio bridge mode.
- #[must_use]
- pub fn stdio(conductor_command: Vec) -> Self {
- Self {
- mode: BridgeMode::Stdio { conductor_command },
- }
+ Self
}
}
@@ -141,60 +151,167 @@ impl ConnectTo for McpOverAcpPolyfill {
client: impl ConnectTo,
) -> Result<(), agent_client_protocol::Error> {
let (bridge_tx, bridge_rx) = mpsc::channel(128);
- let mode = self.mode;
- Proxy
+ let builder = Proxy
.builder()
.name("mcp-over-acp-polyfill")
.with_runner(BridgeRunner {
bridge_tx: bridge_tx.clone(),
bridge_rx,
+ downstream_mode: DownstreamMcpMode::Unknown,
+ listeners: BridgeListeners::default(),
bridge_connections: HashMap::new(),
})
.on_receive_request_from(
Client,
- async move |request: InitializeProxyRequest,
- responder,
- cx: ConnectionTo| {
- // Forward initialize to successor, then set mcpCapabilities.acp = true
- // in the response to advertise that we handle MCP-over-ACP.
- cx.send_request_to(Agent, request.initialize)
- .on_receiving_result(async move |result| {
- responder.respond_with_result(result.map(|mut response| {
- response.agent_capabilities.mcp_capabilities.acp = true;
- response
- }))
- })
+ {
+ 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 bridge_tx = bridge_tx.clone();
+ let mut bridge_tx = bridge_tx.clone();
async move |mut request: NewSessionRequest,
responder,
cx: ConnectionTo| {
- // Transform acp: URLs in MCP servers
- let mut listeners = BridgeListeners::default();
- for mcp_server in &mut request.mcp_servers {
- listeners
- .transform_mcp_server(cx.clone(), mcp_server, &bridge_tx, &mode)
- .await?;
- }
- // Forward modified request to successor
+ 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!(),
+ );
+
+ #[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!(),
+ );
+
+ 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)
.await
}
}
-/// Manages active bridge listeners (TCP listeners for acp: URLs).
+async fn transform_session_servers(
+ servers: &mut Vec,
+ bridge_tx: &mut mpsc::Sender,
+) -> Result<(), agent_client_protocol::Error> {
+ let (response_tx, response_rx) = oneshot::channel();
+ bridge_tx
+ .send(BridgeMessage::TransformServers {
+ servers: std::mem::take(servers),
+ response_tx,
+ })
+ .await
+ .map_err(agent_client_protocol::Error::into_internal_error)?;
+ *servers = response_rx
+ .await
+ .map_err(agent_client_protocol::Error::into_internal_error)??;
+ Ok(())
+}
+
#[derive(Default, Debug)]
struct BridgeListeners {
listeners: HashMap,
@@ -202,125 +319,103 @@ struct BridgeListeners {
#[derive(Clone, Debug)]
struct BridgeListener {
- server: McpServer,
+ tcp_port: u16,
+}
+
+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),
+ )
+ }
}
impl BridgeListeners {
- /// Transform an MCP server with `acp:` URL into a bridged localhost server.
- async fn transform_mcp_server(
+ async fn transform_servers(
&mut self,
- connection: ConnectionTo,
- mcp_server: &mut McpServer,
+ connection: &ConnectionTo,
+ servers: Vec,
bridge_tx: &mpsc::Sender,
- mode: &BridgeMode,
- ) -> Result<(), agent_client_protocol::Error> {
- let McpServer::Http(http) = mcp_server else {
- return Ok(());
- };
-
- if !http.url.starts_with("acp:") {
- return Ok(());
+ ) -> 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?);
}
-
- if !http.headers.is_empty() {
- return Err(agent_client_protocol::Error::internal_error());
- }
-
- let name = http.name.clone();
- let url = http.url.clone();
-
- info!(
- server_name = %name,
- acp_id = %url,
- "Detected MCP server with ACP transport, spawning TCP bridge"
- );
-
- let transformed = self
- .spawn_bridge(connection, &name, &url, bridge_tx, mode)
- .await?;
- *mcp_server = transformed;
- Ok(())
+ Ok(transformed)
}
- async fn spawn_bridge(
+ async fn transform_server(
&mut self,
- connection: ConnectionTo,
- server_name: &str,
- acp_id: &str,
+ connection: &ConnectionTo,
+ server: McpServer,
bridge_tx: &mpsc::Sender,
- mode: &BridgeMode,
- ) -> anyhow::Result {
- if let Some(listener) = self.listeners.get(acp_id) {
- return Ok(listener.server.clone());
- }
-
- let tcp_listener = TcpListener::bind("127.0.0.1:0").await?;
- let tcp_port = tcp_listener.local_addr()?.port();
-
- info!(acp_id = acp_id, tcp_port, "Bound listener for MCP bridge");
-
- let new_server = match mode {
- BridgeMode::Stdio { conductor_command } => McpServer::Stdio(
- McpServerStdio::new(
- server_name.to_string(),
- PathBuf::from(&conductor_command[0]),
- )
- .args(
- conductor_command[1..]
- .iter()
- .cloned()
- .chain(vec!["mcp".to_string(), format!("{tcp_port}")])
- .collect::>(),
- ),
- ),
-
- BridgeMode::Http => McpServer::Http(McpServerHttp::new(
- server_name.to_string(),
- format!("http://localhost:{tcp_port}"),
- )),
+ ) -> Result {
+ let McpServer::Acp(acp_server) = server else {
+ return Ok(server);
};
+ let server_id = acp_server.server_id.to_string();
- self.listeners.insert(
- acp_id.to_string(),
- BridgeListener {
- server: new_server.clone(),
- },
+ info!(
+ server_name = %acp_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));
+ }
+
+ let tcp_listener = TcpListener::bind("127.0.0.1:0")
+ .await
+ .map_err(agent_client_protocol::Error::into_internal_error)?;
+ let tcp_port = tcp_listener
+ .local_addr()
+ .map_err(agent_client_protocol::Error::into_internal_error)?
+ .port();
+ let listener = BridgeListener { tcp_port };
+
connection.spawn({
- let acp_id = acp_id.to_string();
+ let server_id = server_id.clone();
let bridge_tx = bridge_tx.clone();
- let mode = mode.clone();
async move {
info!(
- acp_id = acp_id,
- tcp_port, "now accepting bridge connections"
+ server_id,
+ tcp_port, "accepting MCP compatibility connections"
);
- match mode {
- BridgeMode::Stdio {
- conductor_command: _,
- } => stdio::run_tcp_listener(tcp_listener, acp_id, bridge_tx).await,
- BridgeMode::Http => {
- http::run_http_listener(tcp_listener, acp_id, bridge_tx).await
- }
- }
+ http::run_http_listener(tcp_listener, server_id, bridge_tx).await
}
})?;
- Ok(new_server)
+ let declaration = listener.declaration(acp_server);
+ self.listeners.insert(server_id, listener);
+ Ok(declaration)
+ }
+
+ fn remove(&mut self, server_id: &str) {
+ self.listeners.remove(server_id);
}
}
-/// Runner that manages bridge state alongside the proxy.
+#[derive(Debug)]
+struct ActiveBridgeConnection {
+ server_id: String,
+ bridge: BridgeConnection,
+}
+
struct BridgeRunner {
bridge_tx: mpsc::Sender,
bridge_rx: mpsc::Receiver,
- bridge_connections: HashMap,
+ downstream_mode: DownstreamMcpMode,
+ listeners: BridgeListeners,
+ bridge_connections: HashMap,
}
impl std::fmt::Debug for BridgeRunner {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("BridgeRunner")
+ .field("downstream_mode", &self.downstream_mode)
+ .field("listeners", &self.listeners.listeners.len())
.field("bridge_connections", &self.bridge_connections.len())
.finish_non_exhaustive()
}
@@ -335,71 +430,532 @@ 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::TransformServers {
+ servers,
+ response_tx,
+ } => {
+ let result = match self.downstream_mode {
+ DownstreamMcpMode::Native => Ok(servers),
+ DownstreamMcpMode::HttpAdapter => {
+ self.listeners
+ .transform_servers(&connection, servers, &self.bridge_tx)
+ .await
+ }
+ DownstreamMcpMode::Unavailable => reject_native_servers(
+ servers,
+ "the downstream agent supports neither native nor HTTP MCP transport",
+ ),
+ DownstreamMcpMode::Unknown => reject_native_servers(
+ servers,
+ "MCP transport capabilities are unavailable before initialize",
+ ),
+ };
+ drop(response_tx.send(result));
+ }
+
BridgeMessage::ConnectionReceived {
- acp_id,
+ server_id,
actor,
- connection: bridge_conn,
+ connection: bridge,
} => {
- // Send _mcp/connect request back through the chain.
- // When the response arrives, send ConnectionEstablished back to ourselves.
- connection
- .send_request_to(Client, McpConnectRequest { acp_id, meta: None })
- .on_receiving_result({
- let mut bridge_tx = self.bridge_tx.clone();
- async move |result| match result {
- Ok(response) => bridge_tx
- .send(BridgeMessage::ConnectionEstablished {
- response,
- actor,
- connection: bridge_conn,
- })
- .await
- .map_err(|_| agent_client_protocol::Error::internal_error()),
- Err(_) => Ok(()),
- }
- })?;
+ let request =
+ AgentRequest::ConnectMcpRequest(ConnectMcpRequest::new(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 }
+ }
+ }
+ }
+ Err(error) => {
+ warn!(?error, "mcp/connect failed");
+ BridgeMessage::ConnectionFailed { server_id }
+ }
+ };
+ drop(bridge_tx.send(message).await);
+ Ok(())
+ });
+ if let Err(error) = scheduled {
+ warn!(?error, "could not schedule mcp/connect response handling");
+ }
}
BridgeMessage::ConnectionEstablished {
- response: McpConnectResponse { connection_id, .. },
+ server_id,
+ connection_id,
actor,
- connection: bridge_conn,
+ connection: bridge,
} => {
- self.bridge_connections
- .insert(connection_id.clone(), bridge_conn);
+ let connection_id = connection_id.to_string();
+ self.bridge_connections.insert(
+ connection_id.clone(),
+ ActiveBridgeConnection { server_id, bridge },
+ );
connection.spawn(actor.run(connection_id))?;
}
+ BridgeMessage::ConnectionFailed { server_id } => {
+ self.listeners.remove(&server_id);
+ }
+
BridgeMessage::ClientToServer {
connection_id,
message,
- } => {
- let wrapped = message.map(
- |request, responder| {
- (
- McpOverAcpMessage {
- connection_id: connection_id.clone(),
- message: request,
- meta: None,
- },
- responder,
- )
- },
- |notification| McpOverAcpMessage {
- connection_id: connection_id.clone(),
- message: notification,
- meta: None,
- },
- );
- connection.send_proxied_message_to(Client, wrapped)?;
+ } => 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");
+ }
+ }
+ 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),
+ ) {
+ 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");
+ }
+ }
+ },
+
+ BridgeMessage::ServerToClientRequest { request, responder } => {
+ match self.downstream_mode {
+ DownstreamMcpMode::Native => {
+ let pending = connection
+ .send_request_to(Agent, AgentRequest::MessageMcpRequest(request));
+ 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 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);
+ if let Some(message) = active
+ .bridge
+ .try_send(Dispatch::Request(message, responder))
+ {
+ let Dispatch::Request(_, responder) = *message else {
+ unreachable!("the failed bridge message was a request")
+ };
+ if let Err(send_error) = responder.respond_with_internal_error(
+ "the local MCP client is unavailable or backpressured",
+ ) {
+ debug!(
+ ?send_error,
+ "could not reject unavailable MCP connection"
+ );
+ }
+ }
+ }
+ DownstreamMcpMode::Unknown | DownstreamMcpMode::Unavailable => {
+ if let Err(error) =
+ responder.respond_with_error(
+ agent_client_protocol::Error::method_not_found(),
+ )
+ {
+ debug!(?error, "could not reject unsupported native MCP request");
+ }
+ }
+ }
}
- BridgeMessage::Disconnected { notification } => {
- self.bridge_connections.remove(¬ification.connection_id);
- connection.send_notification_to(Client, notification)?;
+ BridgeMessage::ServerToClientNotification { notification } => {
+ match self.downstream_mode {
+ DownstreamMcpMode::Native => {
+ if let Err(error) = connection.send_notification_to(
+ Agent,
+ AgentNotification::MessageMcpNotification(notification),
+ ) {
+ debug!(?error, "could not forward native MCP notification");
+ }
+ }
+ DownstreamMcpMode::HttpAdapter => {
+ let connection_id = notification.connection_id.to_string();
+ let Some(active) = self.bridge_connections.get_mut(&connection_id)
+ else {
+ debug!(
+ connection_id,
+ "ignoring notification for unknown MCP connection"
+ );
+ continue;
+ };
+ let message = message_mcp_notification_to_untyped(notification);
+ if active
+ .bridge
+ .try_send(Dispatch::Notification(message))
+ .is_some()
+ {
+ debug!("discarding MCP notification for unavailable local client");
+ }
+ }
+ DownstreamMcpMode::Unknown | DownstreamMcpMode::Unavailable => {
+ debug!("ignoring unsupported native MCP notification");
+ }
+ }
+ }
+
+ BridgeMessage::Disconnected { connection_id } => {
+ let Some(active) = self.bridge_connections.remove(&connection_id) else {
+ debug!(connection_id, "local MCP connection was already removed");
+ continue;
+ };
+ self.listeners.remove(&active.server_id);
+
+ let request = AgentRequest::DisconnectMcpRequest(DisconnectMcpRequest::new(
+ connection_id,
+ ));
+ let scheduled = connection
+ .send_request_to(Client, request)
+ .on_receiving_result(async |result| {
+ match result {
+ Ok(response) => {
+ if let Err(error) =
+ serde_json::from_value::(response)
+ {
+ warn!(?error, "invalid response to mcp/disconnect");
+ }
+ }
+ Err(error) => {
+ debug!(?error, "mcp/disconnect failed");
+ }
+ }
+ Ok(())
+ });
+ if let Err(error) = scheduled {
+ debug!(
+ ?error,
+ "could not schedule mcp/disconnect response handling"
+ );
+ }
}
}
}
Ok(())
}
}
+
+fn reject_native_servers(
+ servers: Vec,
+ reason: &'static str,
+) -> Result, agent_client_protocol::Error> {
+ if servers
+ .iter()
+ .any(|server| matches!(server, McpServer::Acp(_)))
+ {
+ Err(agent_client_protocol::Error::invalid_params().data(reason))
+ } else {
+ Ok(servers)
+ }
+}
+
+fn into_mcp_params(
+ params: serde_json::Value,
+) -> Result