> ## Documentation Index
> Fetch the complete documentation index at: https://docs.agentcat.com/llms.txt
> Use this file to discover all available pages before exploring further.

# Rust: rmcp

> Track a Rust MCP server built on rmcp: stdio and streamable HTTP servers, stateful and stateless, the call context, custom events, tests and shutdown.

This page covers a server built with the official Rust MCP SDK, [`rmcp`](https://github.com/modelcontextprotocol/rust-sdk), and the `agentcat-sdk` crate that tracks it: a stdio server, and a streamable HTTP server that is stateful or stateless. [Rust runtimes](https://docs.agentcat.com/sdk/rust/runtimes.md) says what the SDK needs from your tokio runtime and how it delivers the events. [Rust: behavior and limits](https://docs.agentcat.com/sdk/rust/limits.md) lists the supported versions and the known limits. The [crate guide](https://docs.rs/agentcat-sdk) on docs.rs is the full reference.

Each Rust snippet on this page is an excerpt of a compiled source in the crate: one of its examples or one of the doctests of its documentation. A line that the excerpt leaves out is marked `// ...`. `Calculator` is the server of the [quickstart](https://docs.agentcat.com/quickstart.md), and `Demo` is the server of the examples.

## Requirements

* Rust 1.88 or later.
* `rmcp` 3.0.0 or a later 3.x release. CI tests the latest patch of each 3.x minor release.
* A tokio runtime with the time driver. `#[tokio::main]` and `#[tokio::test]` turn it on (see [The tokio runtime](https://docs.agentcat.com/sdk/rust/runtimes.md#the-tokio-runtime)).
* One TLS feature of `agentcat-sdk`: `rustls` (on by default) or `native-tls`, for the connection to the AgentCat API.

```bash theme={null}
cargo add agentcat-sdk
```

To use `native-tls`, turn off the default features:

```toml theme={null}
[dependencies]
agentcat-sdk = { version = "1", default-features = false, features = ["native-tls"] }
```

[Rust: behavior and limits](https://docs.agentcat.com/sdk/rust/limits.md#features) lists the features, and the `test-util` feature that the [Testing](#testing) section uses.

## Track the server

`agentcat_sdk::track` wraps your `ServerHandler` with a `Tracked` server. Serve the `Tracked` server where you served your handler:

With `use agentcat_sdk::Options;` and `rmcp::ServiceExt` in scope:

```rust theme={null}
#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    // One line: wrap the server with AgentCat.
    let server = agentcat_sdk::track(Calculator, "proj_...", Options::default());
    let running = server.serve(rmcp::transport::stdio()).await?;
    running.waiting().await?;
    Ok(())
}
```

Replace `proj_...` with the ID of your project. To read the project ID from the `AGENTCAT_PROJECT_ID` environment variable instead, build the handle with `AgentCat::from_env`, and wrap the server with `AgentCat::track`:

```rust theme={null}
use agentcat_sdk::{AgentCat, Options};

// AGENTCAT_PROJECT_ID=proj_123 in the environment.
let agentcat = AgentCat::from_env(Options::default());
```

* **`AgentCat` is the handle.** `AgentCat::new(project_id, options)` and `AgentCat::from_env(options)` build it. It starts no thread until the first event or log line. A clone of the handle is cheap and shares everything with the original.
* **An empty project ID is telemetry only.** The handle builds each event and runs your hooks, but sends nothing to AgentCat. `from_env` does the same when the variable is unset or empty.
* **`Options` sets the hooks and the switches.** `Options::default()` has the defaults that the Options section of the [crate guide](https://docs.rs/agentcat-sdk) lists, and each setter consumes the options and returns them changed:

```rust theme={null}
use agentcat_sdk::{HookError, Options, ToolCall, UserIdentity};

let options = Options::default()
    .enable_agent_tracking(true)
    .identify(|call: ToolCall| async move {
        // Read your own data about the caller, for example an HTTP header.
        let user = call
            .headers()
            .and_then(|headers| headers.get("x-user-id"))
            .and_then(|value| value.as_str())
            .map(UserIdentity::new);
        Ok::<_, HookError>(user)
    });
```

The [SDK pages](https://docs.agentcat.com/sdk/identifying-users.md) show each option with a Rust tab: [Identifying Users](https://docs.agentcat.com/sdk/identifying-users.md), [Event Tags & Properties](https://docs.agentcat.com/sdk/event-tags-properties.md), [Client-Side Redaction](https://docs.agentcat.com/privacy-security/client-side-redaction.md) and [Tool Call Context](https://docs.agentcat.com/sdk/tool-call-context.md). The SDK runs the four tool-call hooks on the tokio runtime of the call, and it waits at most 30 seconds for each before it sends the event without that result.

## stdio

A stdio server serves the `Tracked` server on `rmcp`'s stdio transport, which needs the `transport-io` feature of `rmcp`:

```rust theme={null}
use agentcat_sdk::{AgentCat, Options};
use rmcp::handler::server::wrapper::Parameters;
use rmcp::model::{Implementation, InitializeResult, ServerCapabilities};
use rmcp::transport::stdio;
use rmcp::{ServerHandler, ServiceExt, tool, tool_handler, tool_router};
use schemars::JsonSchema;
use serde::Deserialize;

// ...

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    // Reads AGENTCAT_PROJECT_ID, and AGENTCAT_API_URL when set. Without a project ID, the handle
    // builds events but sends none.
    let server = AgentCat::from_env(Options::default()).track(Demo);

    // Serve the wrapper where you served `Demo`.
    let running = server.serve(stdio()).await?;
    running.waiting().await?;
    Ok(())
}
```

* **stdout is for JSON-RPC.** The SDK writes nothing to stdout or stderr. Send your own logs, and your `tracing` output, to stderr (see [Logging on stdio servers](https://docs.agentcat.com/sdk/rust/runtimes.md#logging-on-stdio-servers)).
* **No `shutdown` call.** The server ends when the client closes stdin. Then the last clone of the handle drops, and the drop delivers the queued events, for up to 5 seconds (see [When the process stops](#when-the-process-stops)).

## Streamable HTTP

A streamable HTTP server builds a new server handler for each MCP session, or for each request when it is stateless. Build one `AgentCat` handle for the process, and wrap each new handler with `AgentCat::track` in the closure that `rmcp` calls. This server needs the `transport-streamable-http-server` feature of `rmcp`, the `macros`, `rt-multi-thread`, `net` and `signal` features of tokio, and `axum`.

### Stateful, with server-sent events

The default `StreamableHttpServerConfig` of `rmcp` keeps one MCP session per client and answers with server-sent events (SSE):

```rust theme={null}
use std::sync::Arc;
use std::time::Duration;

use agentcat_sdk::{AgentCat, Options};
use rmcp::transport::streamable_http_server::session::local::LocalSessionManager;
use rmcp::transport::streamable_http_server::{StreamableHttpServerConfig, StreamableHttpService};

// ...

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    // One handle for the whole process. It reads AGENTCAT_PROJECT_ID, and AGENTCAT_API_URL when set.
    let agentcat = AgentCat::from_env(Options::default());

    // rmcp calls this closure once for each new MCP session.
    let handle = agentcat.clone();
    let service = StreamableHttpService::new(
        move || Ok(handle.track(Demo)),
        Arc::new(LocalSessionManager::default()),
        StreamableHttpServerConfig::default(),
    );
    let router = axum::Router::new().route_service("/mcp", service);

    // ...
    tokio::select! {
        served = axum::serve(listener, router) => served?,
        () = stop_signal() => {}
    }

    // Deliver the events that are still queued before the process ends.
    let delivered = agentcat.shutdown(Duration::from_secs(5)).await;
    // ...
    Ok(())
}
```

The SDK never groups calls by the `mcp-session-id` header of the transport. The `session_id` parameter that it adds to each tool groups them on every transport (see [Stateless Servers](https://docs.agentcat.com/sdk/stateless-servers.md)).

### Stateless, with JSON responses

In stateless mode, `rmcp` builds a new server handler for each request. The setup is the same. Only the configuration of the service changes:

```rust theme={null}
// One handle for the whole process. It reads AGENTCAT_PROJECT_ID, and AGENTCAT_API_URL when set.
let agentcat = AgentCat::from_env(Options::default());

// rmcp calls this closure once for each request.
let handle = agentcat.clone();
let service = StreamableHttpService::new(
    move || Ok(handle.track(Demo)),
    Arc::new(LocalSessionManager::default()),
    // No sessions, and JSON responses instead of SSE.
    StreamableHttpServerConfig::default()
        .with_legacy_session_mode(false)
        .with_json_response(true),
);
```

A stateless server loses no sessions, because the `session_id` parameter groups the calls. On a stateless server, an event names the client only when the request names it in its `_meta`. [Stateless Servers](https://docs.agentcat.com/sdk/stateless-servers.md) has the details.

### Stop on SIGTERM and Ctrl-C

An HTTP server does not end when a client goes away, and the SDK installs no signal handler. So the server waits for a stop signal, and calls `AgentCat::shutdown` after that: the queued events then go out before the process ends. Containers and process managers stop a server with SIGTERM, and a server that does not catch it skips `shutdown` and loses the queued events. This `stop_signal` function of the `streamable_http` and `stateless_http` examples resolves on SIGTERM (Unix) or Ctrl-C:

```rust theme={null}
/// Resolves on SIGTERM (Unix) or Ctrl-C.
async fn stop_signal() {
    #[cfg(unix)]
    {
        use tokio::signal::unix::{SignalKind, signal};
        if let Ok(mut terminate) = signal(SignalKind::terminate()) {
            tokio::select! {
                _ = terminate.recv() => {}
                _ = tokio::signal::ctrl_c() => {}
            }
            return;
        }
    }
    let _ = tokio::signal::ctrl_c().await;
}
```

## The call context in a tool

The SDK gives each tool call a `CallContext`. A tool reads it as an `Extension<CallContext>` parameter, next to its `Parameters<..>`:

* `session_id()` returns the session of the call: the ID that the agent sent, or the one the SDK issued for this call. It returns `None` when the call has no session, for example when the agent sent a value that this server did not issue, when the tool declares its own `session_id` parameter, when the options set a `resolve_session_id` hook, or when the SDK does not trace the call.
* `agent_id()` returns the `agent_id` that the agent sent, when agent tracking is on.
* `agentcat()` returns the handle that tracks the call, so a tool needs no clone of it.
* `publish_custom_event` publishes a [custom event](https://docs.agentcat.com/sdk/custom-events.md) with the client and the server of this call.

A custom event has no session unless its `CustomEvent` names one. Pass `call.session_id()` to `CustomEvent::session_id` to put the event in the session of the call:

```rust theme={null}
/// Sends the text back, and publishes one custom event for each call. The SDK passes the
/// `CallContext` of the call to the tool as an `Extension`.
#[tool(description = "Sends the text back")]
async fn echo(
    &self,
    Parameters(args): Parameters<EchoArgs>,
    Extension(call): Extension<CallContext>,
) -> String {
    let mut event = CustomEvent::new()
        .resource_name("echo-called")
        .message("the echo tool ran");
    // Attach the event to the session of the call.
    if let Some(session) = call.session_id() {
        event = event.session_id(session);
    }
    // Queues the event and returns at once. An event that cannot be queued is not an error
    // for the tool.
    let _ = call.publish_custom_event(&self.project_id, event);
    args.text
}
```

The event goes to the project of the handle: the project ID argument only has to be non-empty. The `redact_event` hook runs on the event, and `redact_sensitive_information` does not.

`rmcp` turns an error that a tool returns into the text of an error result, so the event gets only that text. Call `capture_error` before the tool returns the error, and the event also gets the type of the error and the texts of its `source()` chain:

```rust theme={null}
#[tool(description = "Always fails")]
async fn fail(&self) -> Result<String, String> {
    let error = std::io::Error::new(std::io::ErrorKind::NotFound, "the record does not exist");
    capture_error(&error);
    Err(error.to_string())
}
```

## Custom events from outside a tool

Code that runs outside a tool call has no `CallContext`. Publish from the handle with the session ID, the project ID and the event. The string is the session ID of the event, unchanged, unless the `CustomEvent` has a `session_id` that is not empty. This call needs no tracked server, and it runs no hook of a server: the event has no server or client fields, and no redaction hook runs for it.

```rust theme={null}
use agentcat_sdk::{AgentCat, CustomEvent, Options};

let agentcat = AgentCat::new("proj_123", Options::default());
let event = CustomEvent::new().resource_name("cache-warmed").message("40 keys");
agentcat.publish_custom_event("ses_1", "proj_123", event)?;
```

For an event with the client and the server of a tracked server, call `Tracked::publish_custom_event` on the `Tracked` server:

```rust theme={null}
tracked.publish_custom_event("proj_123", CustomEvent::new().resource_name("startup"))?;
```

Both calls only queue the event and return at once. An empty project ID returns `Error::ProjectIdRequired`, unless `AGENTCAT_ENABLED` turned the SDK off. [Custom Events](https://docs.agentcat.com/sdk/custom-events.md) covers the fields of the event.

## Testing

AgentCat stays on in tests unless you set `AGENTCAT_ENABLED` to `false`. To assert on what a tracked server publishes, turn on the `test-util` feature of the SDK. It adds the `testing` module with `testing::Fake`. The fake builds handles whose events stay in the process, and it records each event as a real handle sends it, after the hooks and the redaction. The test below serves a tracked server to an `rmcp` client in memory, calls a tool, and makes sure that the fake recorded the call. It needs these dev-dependencies:

```toml theme={null}
[dev-dependencies]
agentcat-sdk = { version = "1", features = ["test-util"] }
rmcp = { version = "3", features = ["client"] }
tokio = { version = "1", features = ["io-util", "macros", "rt"] }
```

In a `#[tokio::test]`:

```rust theme={null}
use agentcat_sdk::Options;
use agentcat_sdk::testing::Fake;
use rmcp::ServiceExt;
use rmcp::model::{
    CallToolRequestParams, ClientCapabilities, Implementation, InitializeRequestParams,
};

let fake = Fake::new();
let server = fake.agentcat("proj_test", Options::default()).track(Calculator);

// The server on one end of an in-memory stream, an rmcp client on the other.
let (server_end, client_end) = tokio::io::duplex(64 * 1024);
tokio::spawn(async move {
    if let Ok(running) = server.serve(server_end).await {
        let _ = running.waiting().await;
    }
});
let client = InitializeRequestParams::new(
    ClientCapabilities::default(),
    Implementation::new("test", "1.0.0"),
)
.serve(client_end)
.await?;

let mut arguments = rmcp::model::JsonObject::new();
arguments.insert("a".into(), 1.into());
arguments.insert("b".into(), 2.into());
client
    .call_tool(CallToolRequestParams::new("add").with_arguments(arguments))
    .await?;

fake.assert_tracked("add").await;
```

`Fake::events` returns the recorded events, for checks of your own. `Fake` also has `assert_tracked_count`, `assert_nothing_tracked` and `assert_custom_event_published`.

To unit-test a `redact_event` hook, call it with an event from `Event::default()`, and set the fields that the hook reads. Run this in an async test that returns a `Result<(), HookError>`:

```rust theme={null}
use agentcat_sdk::{Event, HookError};

// The closure that you pass to `Options::redact_event`.
let hook = |mut event: Event| async move {
    event.response = None;
    Ok::<_, HookError>(Some(event))
};

let mut event = Event::default();
event.response = Some(serde_json::json!({"token": "fake-secret-42"}));
let redacted = hook(event).await?;
assert_eq!(redacted.and_then(|event| event.response), None);
```

A client or a test that reads the first content block of the first call of a task gets the text block that begins `[session_id issued`, not the content of the tool (see [Stateless Servers](https://docs.agentcat.com/sdk/stateless-servers.md)).

## When the process stops

The SDK delivers the events from a thread of its own. What you do so that the last events go out depends on the server:

| Server | When the last events go out | What you do |
| - | - | - |
| stdio | When the client closes stdin, the server ends, and the drop of the last handle delivers the queued events, for up to 5 seconds | Nothing |
| Streamable HTTP, stateful or stateless | When you call `AgentCat::shutdown` after the server stops | Call `shutdown` after `axum::serve` returns, as in [Streamable HTTP](#streamable-http) |

* **`flush` and `shutdown`.** `AgentCat::flush(timeout)` waits until the handle delivered every event that it took, for at most `timeout`. `AgentCat::shutdown(timeout)` does the same, then stops the handle for good: the handle drops every later event. Both return `true` when the events went out in time:

```rust theme={null}
use std::time::Duration;

// ... serve until the server stops ...
let delivered = agentcat.shutdown(Duration::from_secs(5)).await;
if !delivered {
    // Some events did not go out within 5 s.
}
```

* **The drain of the last drop.** When the last clone of a handle drops, the SDK runs `shutdown` with a timeout of 5 seconds. The drop blocks the dropping thread until the shutdown ends. The `Tracked` servers hold clones of the handle too. The drain does not run after `std::process::exit`, or after a panic with `panic = "abort"`.
* **A current-thread runtime.** A drop on the thread of the runtime blocks its only thread, so no hook can run during the wait. Call `shutdown` before the runtime ends.
* **A hook that holds a clone of its handle** makes a reference cycle, and the handle never drops. Let the hook capture only what it needs, or call `shutdown` before the process ends.

[Flush and shutdown](https://docs.agentcat.com/sdk/rust/runtimes.md#flush-and-shutdown) and [The drain of the last drop](https://docs.agentcat.com/sdk/rust/runtimes.md#the-drain-of-the-last-drop) have the details.


This documentation is built and hosted on [Mintlify](https://mintlify.com), a developer documentation platform.