Skip to content

feat(mcp)!: support acknowledged notification subscriptions #34

New issue

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

Sign up for GitHub

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

Already on GitHub? Sign in to your account

Merged
danielkov merged 3 commits into from
Oct 15, 2026
Merged

feat(mcp)!: support acknowledged notification subscriptions #34

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
f1f87d7
feat(mcp)!: support acknowledged notification subscriptions
danielkov Oct 15, 2026
b6af4a0
fix(mcp): preserve subscriptions across replay and recovery
danielkov Oct 15, 2026
b3e46cf
docs(mcp): explain subscription ownership in harness design
danielkov Oct 15, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
17 changes: 9 additions & 8 deletions Cargo.lock
View file
Open in desktop

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

26 changes: 23 additions & 3 deletions book/src/ch17-mcp.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -32,6 +32,26 @@ The MCP specification moves quickly. Because agentkit-mcp re-exports [`rmcp::mod

The same logic applies to transports: stdio and Streamable HTTP are rmcp implementations driven declaratively here; future rmcp transports are reachable through `McpConnection::from_running_service_with_events` without touching `McpTransportBinding`.

## Modern subscriptions and catalog refresh

Discovery gives a harness a snapshot of a server's tools, resources, and prompts.
A subscription does not replace that snapshot: its notifications are
invalidations, telling the harness that its view may be stale.

The host owns reconciliation. It decides when to fetch and publish a fresh
snapshot, coalescing repeated invalidations instead of refreshing once per event.
After a notification stream is restored, reconcile again to cover changes missed
during the gap. If a refresh fails, keep the last usable snapshot and leave the
refresh pending.

Bind refresh results to the connection that produced them. Authentication changes
or server replacement can make in-flight work obsolete; a result from an old
connection must never overwrite the catalog for its replacement.

Subscription lifecycle belongs to the connection, not to an individual agent
turn. Close subscriptions with their connection and rebuild them when it is
replaced, so background streams cannot outlive the server they describe.

## MCP in the capability model

MCP servers expose three capability types, which map directly to agentkit's capability layer:
Expand Down Expand Up @@ -132,7 +152,7 @@ while let Ok(event) = events.recv().await {

`set_logging_level(LoggingLevel::Info)` negotiates the minimum severity the server emits. `subscribe_resource` / `unsubscribe_resource` toggle per-URI watch on resources that support it.

List-changed events are _also_ delivered through the legacy `McpServerNotification` mpsc receiver that `McpServerManager::refresh_changed_catalogs` drains to re-run discovery — the two channels coexist: events for live UI/observability, mpsc for catalog re-sync.
List-changed events also advance the coalescing catalog revision watch used by `McpServerManager::refresh_changed_catalogs`. The compatibility `McpServerNotification` receiver remains available; catalog reconciliation does not require draining the event broadcast.

## Auth handled outside the loop

Expand Down Expand Up @@ -163,11 +183,11 @@ The manager owns server lifecycle:
| `connect_server(id)` / `connect_all()` | Open a connection and run discovery |
| `connect_all_settled()` | Attempt all registered servers concurrently, install successes, and return per-server failures |
| `refresh_server(id)` | Re-run discovery and emit per-tool/resource/prompt diff events |
| `refresh_changed_catalogs()` | Drain pending `*list_changed` notifications and refresh affected catalogs |
| `refresh_changed_catalogs()` | Reconcile catalogs marked dirty by coalesced invalidations |
| `disconnect_server(id)` | Close the connection, drop tools from the federated catalog, emit `ServerDisconnected` |
| `subscribe_catalog_events()` | Broadcast receiver for `McpCatalogEvent` (server connect / disconnect / tool added / removed / changed / refresh failed / auth changed) |

Register a server with `with_server_options(config, McpServerOptions::new().with_timeout(duration))` to bound connecting to that server. The timeout covers connection establishment — transport setup and the MCP initialize handshake — plus initial discovery, and applies to refresh discovery on its own; failures surface as `McpError::Timeout`.
Register a server with `with_server_options(config, McpServerOptions::new().with_timeout(duration))` to bound connecting to that server. The timeout covers connection establishment — transport setup and MCP lifecycle negotiation — plus initial discovery, and applies to refresh discovery on its own; failures surface as `McpError::Timeout`.

## Federating MCP into the agent

Expand Down
4 changes: 2 additions & 2 deletions crates/agentkit-integration-tests/Cargo.toml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -14,7 +14,7 @@ doctest = false
[dependencies]
agentkit-core = { version = "0.10.5", path = "../agentkit-core" }
agentkit-loop = { version = "0.10.5", path = "../agentkit-loop" }
agentkit-mcp = { version = "0.10.5", path = "../agentkit-mcp" }
agentkit-mcp = { version = "0.11.0", path = "../agentkit-mcp" }
agentkit-task-manager = { version = "0.10.5", path = "../agentkit-task-manager" }
agentkit-tool-compose = { version = "0.10.5", path = "../agentkit-tool-compose" }
agentkit-tools-core = { version = "0.10.5", path = "../agentkit-tools-core", features = ["schemars"] }
Expand All @@ -23,7 +23,7 @@ async-trait.workspace = true
schemars = { workspace = true, features = ["derive"] }
axum = { version = "0.8", default-features = false, features = ["http1", "tokio"] }
pretty_assertions = "1.4"
rmcp = { version = "3.1.2", default-features = false, features = ["client", "server", "transport-async-rw", "transport-streamable-http-server", "macros"] }
rmcp = { version = "=3.5.1", default-features = false, features = ["client", "server", "transport-async-rw", "transport-streamable-http-server", "macros"] }
ron = "0.8"
serde.workspace = true
serde_json.workspace = true
Expand Down
20 changes: 16 additions & 4 deletions crates/agentkit-integration-tests/src/http_mcp_server.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -14,8 +14,9 @@ use rmcp::{
RoleServer, ServerHandler,
model::{
CallToolRequestParams, CallToolResponse, CallToolResult, ContentBlock as Content,
ErrorData as RmcpError, InitializeRequestParams, InitializeResult, ListToolsResult,
PaginatedRequestParams, ServerCapabilities, ServerInfo, Tool,
DiscoverResult, ErrorData as RmcpError, InitializeRequestParams, InitializeResult,
ListToolsResult, PaginatedRequestParams, ProtocolVersion, ServerCapabilities, ServerConfig,
Tool,
},
service::{NotificationContext, Peer, RequestContext},
transport::streamable_http_server::{
Expand Down Expand Up @@ -49,13 +50,24 @@ impl MutableMcpServer {
}

impl ServerHandler for MutableMcpServer {
fn get_info(&self) -> ServerInfo {
ServerInfo::new(
fn get_info(&self) -> ServerConfig {
ServerConfig::new(
ServerCapabilities::builder()
.enable_tools()
.enable_tool_list_changed()
.build(),
)
.with_protocol_version(ProtocolVersion::V_2025_11_25)
}

async fn discover(
&self,
_context: RequestContext<RoleServer>,
) -> Result<DiscoverResult, RmcpError> {
// This fixture deliberately exercises the initialize-era protocol.
Err(RmcpError::method_not_found::<
rmcp::model::DiscoverRequestMethod,
>())
}

async fn initialize(
Expand Down
6 changes: 3 additions & 3 deletions crates/agentkit-mcp/Cargo.toml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -4,7 +4,7 @@ homepage.workspace = true
name = "agentkit-mcp"
readme = "README.md"
repository.workspace = true
version = "0.10.7"
version = "0.11.0"
edition.workspace = true
license.workspace = true
rust-version.workspace = true
Expand All @@ -20,7 +20,7 @@ async-trait.workspace = true
futures-util.workspace = true
http = "1"
reqwest.workspace = true
rmcp = { version = "3.1.2", default-features = false, features = ["client", "transport-child-process", "transport-streamable-http-client-reqwest", "reqwest", "auth"] }
rmcp = { version = "=3.5.1", default-features = false, features = ["client", "transport-child-process", "transport-streamable-http-client-reqwest", "reqwest", "auth"] }
serde.workspace = true
serde_json.workspace = true
sse-stream = "0.2"
Expand All @@ -30,5 +30,5 @@ tokio = { workspace = true, features = ["sync", "time"] }

[dev-dependencies]
axum = "0.8"
rmcp = { version = "3.1.2", default-features = false, features = ["client", "server", "transport-async-rw", "macros", "elicitation"] }
rmcp = { version = "=3.5.1", default-features = false, features = ["client", "server", "transport-async-rw", "macros", "elicitation"] }
tokio = { workspace = true, features = ["macros", "rt-multi-thread", "io-util", "time", "net"] }
45 changes: 45 additions & 0 deletions crates/agentkit-mcp/README.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -69,6 +69,51 @@ Use `connect_all_settled().await` when startup should be best effort: it attempt

For slow or unresponsive servers, register with `with_server_options(config, McpServerOptions::new().with_timeout(duration))`. The timeout bounds connection establishment — transport setup and the MCP initialize handshake — together with initial discovery, and bounds refresh discovery on its own; exceeding it returns `McpError::Timeout`.

## Negotiation and notification subscriptions

Built-in stdio and Streamable HTTP connections use RMCP's `Auto` lifecycle: probe
`server/discover`, then fall back to legacy initialization when needed. Inspect
`connection.protocol_version()` and the cloned `connection.capabilities()` for
the negotiated peer, rather than assuming a protocol from the transport.

For modern peers, discovery establishes a connection-owned `subscriptions/listen`
stream and waits for its acknowledgment. The requested filter covers advertised
tool, resource, and prompt catalogs. `subscription_status().await` exposes both
the requested and acknowledged filters: a server may accept only part of the
catalog filter. `subscription_changes().await` returns a watch receiver once a
modern stream exists; legacy peers return `None`. Adopting an existing RMCP
service with `from_running_service_with_events` is synchronous: call
`discover().await` to establish its modern subscription lazily.

`subscribe_resource(uri)` and `unsubscribe_resource(uri)` replace the modern
stream's complete filter. A resource update is committed only after the server
acknowledges the requested URI set; rejection preserves the previous active
filter and stream. Legacy peers retain `resources/subscribe` and
`resources/unsubscribe`. Streamed notifications feed the same
`subscribe_events()` broadcast used by legacy notifications.

Catalog invalidations also advance `catalog_revision()` and wake
`catalog_changes()`, a coalescing watch channel that does not require draining
broadcast events. Watch wakeups are invalidations, not newly published catalogs.
Hosts should call `McpServerManager::refresh_changed_catalogs()` to reconcile
snapshots, or use `prepare_changed_catalogs()`, fetch work outside their manager
lock, and `publish_catalog_refresh(...)` to publish it. Revision and connection
epoch checks fence obsolete results; canceled or failed fetches remain dirty.
A host must schedule this refresh work itself; event delivery does not silently
refresh the host's tool registry.

On transport failure, the modern subscription worker retries with bounded
backoff and marks catalogs dirty for reconciliation. Graceful server completion
is observable as `Closed { graceful: true, .. }` and is not retried. Authentication
reconnect replaces the subscription on the new peer, retaining committed resource
URIs. `connection_epoch()` changes when the transport is replaced or closed.
Call `close().await` to stop owned streams before transport cleanup; dropping the
connection also cancels its workers.

**Migration (agentkit-mcp 0.11.0):** `capabilities()` now returns an owned snapshot,
not a borrowed reference. Call it again after authentication reconnect to obtain
the current peer capabilities; no capability lock guard needs to span an await.

## Tool response timeouts

`McpConnection::call_tool_with_timeout(name, arguments, Duration)` is an opt-in,
Expand Down
Loading
Loading