rmcp - crates.io Package Compare versions

+73

tests/test_meta_helpers.rs

		#![allow(deprecated)]

		use rmcp::model::{ClientCapabilities, Implementation, LoggingLevel, Meta, ProtocolVersion};
		use serde_json::json;

		const META_KEY_PROTOCOL_VERSION: &str = "io.modelcontextprotocol/protocolVersion";
		const META_KEY_CLIENT_INFO: &str = "io.modelcontextprotocol/clientInfo";
		const META_KEY_CLIENT_CAPABILITIES: &str = "io.modelcontextprotocol/clientCapabilities";
		const META_KEY_LOG_LEVEL: &str = "io.modelcontextprotocol/logLevel";

		#[test]
		fn meta_setters_store_sep_2575_values() {
		let mut meta = Meta::new();
		meta.set_protocol_version(ProtocolVersion::V_2026_07_28);
		meta.set_client_info(Implementation::new("test-client", "1.0.0"));
		meta.set_client_capabilities(ClientCapabilities::default());
		meta.set_log_level(LoggingLevel::Warning);

		assert_eq!(
		meta.get(META_KEY_PROTOCOL_VERSION),
		Some(&json!("2026-07-28"))
		);
		assert_eq!(
		meta.get(META_KEY_CLIENT_INFO),
		Some(&json!({ "name": "test-client", "version": "1.0.0" }))
		);
		assert_eq!(meta.get(META_KEY_CLIENT_CAPABILITIES), Some(&json!({})));
		assert_eq!(meta.get(META_KEY_LOG_LEVEL), Some(&json!("warning")));
		}

		#[test]
		fn meta_accessors_decode_wire_values() {
		let meta: Meta = serde_json::from_value(json!({
		"progressToken": "progress-1",
		"io.modelcontextprotocol/protocolVersion": "2026-07-28",
		"io.modelcontextprotocol/clientInfo": {
		"name": "wire-client",
		"version": "9.0.0"
		},
		"io.modelcontextprotocol/clientCapabilities": {
		"sampling": {}
		},
		"io.modelcontextprotocol/logLevel": "error"
		}))
		.unwrap();

		assert_eq!(meta.protocol_version(), Some(ProtocolVersion::V_2026_07_28));
		assert_eq!(
		meta.client_info(),
		Some(Implementation::new("wire-client", "9.0.0"))
		);
		assert!(
		meta.client_capabilities()
		.is_some_and(\|capabilities\| capabilities.sampling.is_some())
		);
		assert_eq!(meta.log_level(), Some(LoggingLevel::Error));
		}

		#[test]
		fn meta_accessors_ignore_missing_or_malformed_values() {
		let meta: Meta = serde_json::from_value(json!({
		"io.modelcontextprotocol/protocolVersion": 20260728,
		"io.modelcontextprotocol/clientInfo": "not an implementation",
		"io.modelcontextprotocol/clientCapabilities": "not capabilities",
		"io.modelcontextprotocol/logLevel": "loud"
		}))
		.unwrap();

		assert_eq!(meta.protocol_version(), None);
		assert_eq!(meta.client_info(), None);
		assert_eq!(meta.client_capabilities(), None);
		assert_eq!(meta.log_level(), None);
		}

+83

tests/test_protocol_version_negotiation.rs

		//! Tests for protocol version negotiation in the default ServerHandler::initialize impl.
		//!
		//! Known versions are echoed back; unknown versions fall back to LATEST.
		#![cfg(not(feature = "local"))]
		#![cfg(feature = "client")]

		use rmcp::{
		ClientHandler, ServerHandler, ServiceExt,
		model::{ClientInfo, ProtocolVersion, ServerInfo},
		};

		#[derive(Debug, Clone, Default)]
		struct EchoServer;

		impl ServerHandler for EchoServer {
		fn get_info(&self) -> ServerInfo {
		ServerInfo::default()
		}
		}

		#[derive(Debug, Clone)]
		struct VersionedClient {
		protocol_version: ProtocolVersion,
		}

		impl ClientHandler for VersionedClient {
		fn get_info(&self) -> ClientInfo {
		let mut info = ClientInfo::default();
		info.protocol_version = self.protocol_version.clone();
		info
		}
		}

		async fn negotiated_version(client_version: ProtocolVersion) -> ProtocolVersion {
		let (server_transport, client_transport) = tokio::io::duplex(4096);

		tokio::spawn(async move {
		let _ = EchoServer
		.serve(server_transport)
		.await
		.expect("server should start")
		.waiting()
		.await;
		});

		let client = VersionedClient {
		protocol_version: client_version,
		}
		.serve(client_transport)
		.await
		.expect("client should connect");

		let version = client
		.peer_info()
		.expect("peer_info should be set")
		.protocol_version
		.clone();

		client.cancel().await.expect("client should cancel");
		version
		}

		#[tokio::test]
		async fn known_version_echoed_back() {
		for version in ProtocolVersion::KNOWN_VERSIONS {
		let negotiated = negotiated_version(version.clone()).await;
		assert_eq!(
		negotiated, *version,
		"known version {version} should be echoed back"
		);
		}
		}

		#[tokio::test]
		async fn unknown_version_falls_back_to_latest() {
		let unknown: ProtocolVersion = serde_json::from_str(r#""1999-01-01""#).unwrap();
		let negotiated = negotiated_version(unknown).await;
		assert_eq!(
		negotiated,
		ProtocolVersion::LATEST,
		"unknown version should fall back to LATEST"
		);
		}

+99

tests/test_stateless_protocol_version.rs

		//! Tests for protocol version negotiation in stateless HTTP mode.
		//!
		//! Known versions are echoed back; unknown versions fall back to LATEST.
		#![cfg(not(feature = "local"))]

		use rmcp::{
		model::ProtocolVersion,
		transport::streamable_http_server::{
		StreamableHttpServerConfig, StreamableHttpService, session::local::LocalSessionManager,
		},
		};
		use tokio_util::sync::CancellationToken;

		mod common;
		use common::calculator::Calculator;

		fn stateless_json_config() -> StreamableHttpServerConfig {
		StreamableHttpServerConfig::default()
		.with_stateful_mode(false)
		.with_json_response(true)
		.with_sse_keep_alive(None)
		.with_cancellation_token(CancellationToken::new())
		}

		async fn spawn_server(
		config: StreamableHttpServerConfig,
		) -> (reqwest::Client, String, CancellationToken) {
		let ct = config.cancellation_token.clone();
		let service: StreamableHttpService<Calculator, LocalSessionManager> =
		StreamableHttpService::new(\|\| Ok(Calculator::new()), Default::default(), config);

		let router = axum::Router::new().nest_service("/mcp", service);
		let tcp_listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
		let addr = tcp_listener.local_addr().unwrap();

		tokio::spawn({
		let ct = ct.clone();
		async move {
		let _ = axum::serve(tcp_listener, router)
		.with_graceful_shutdown(async move { ct.cancelled_owned().await })
		.await;
		}
		});

		(reqwest::Client::new(), format!("http://{addr}/mcp"), ct)
		}

		async fn post_init(client: &reqwest::Client, url: &str, body_version: &str) -> serde_json::Value {
		let body = serde_json::json!({
		"jsonrpc": "2.0",
		"id": 1,
		"method": "initialize",
		"params": {
		"protocolVersion": body_version,
		"capabilities": {},
		"clientInfo": {"name": "test", "version": "0.0.1"}
		}
		});
		let resp = client
		.post(url)
		.header("Content-Type", "application/json")
		.header("Accept", "application/json, text/event-stream")
		.body(body.to_string())
		.send()
		.await
		.expect("send request");
		assert!(resp.status().is_success(), "HTTP {}", resp.status());
		resp.json().await.expect("parse JSON")
		}

		#[tokio::test]
		async fn stateless_init_echoes_known_version() {
		let (client, url, ct) = spawn_server(stateless_json_config()).await;

		for version in ProtocolVersion::KNOWN_VERSIONS {
		let resp = post_init(&client, &url, version.as_str()).await;
		assert_eq!(
		resp["result"]["protocolVersion"],
		version.as_str(),
		"known version {version} should be echoed back"
		);
		}

		ct.cancel();
		}

		#[tokio::test]
		async fn stateless_init_unknown_version_falls_back_to_latest() {
		let (client, url, ct) = spawn_server(stateless_json_config()).await;

		let resp = post_init(&client, &url, "1999-01-01").await;
		assert_eq!(
		resp["result"]["protocolVersion"],
		ProtocolVersion::LATEST.as_str(),
		"unknown version should fall back to LATEST"
		);

		ct.cancel();
		}

+198

tests/test_stdio_response_concurrency.rs

		#![cfg(not(feature = "local"))]

		use std::{collections::BTreeSet, process::Stdio, time::Duration};

		use rmcp::{
		ErrorData as McpError, ServerHandler, ServiceExt,
		model::{CallToolRequestParams, CallToolResult, ContentBlock, ServerCapabilities, ServerInfo},
		};
		use serde_json::{Value, json};
		use tokio::{
		io::{AsyncBufReadExt, AsyncWrite, AsyncWriteExt, BufReader},
		process::{Child, Command},
		};

		const HELPER_ENV: &str = "RMCP_STDIO_RESPONSE_CONCURRENCY_HELPER";
		const REQUESTS: usize = 200;
		const RESPONSE_BYTES: usize = 64 * 1024;
		const READ_TIMEOUT: Duration = Duration::from_secs(10);

		#[tokio::test(flavor = "multi_thread", worker_threads = 8)]
		async fn raw_client_concurrent_large_stdio_tool_responses_are_not_lost() -> anyhow::Result<()> {
		// Spawn the same test binary as a child process so the server uses real
		// stdio pipes, not an in-process transport.
		let mut child = spawn_helper();
		let mut writer = child.stdin.take().expect("helper stdin");
		let stdout = child.stdout.take().expect("helper stdout");
		let mut reader = BufReader::new(stdout);

		// Complete the normal MCP initialization flow before stressing tools/call.
		send_json(
		&mut writer,
		&json!({
		"jsonrpc": "2.0",
		"id": 1,
		"method": "initialize",
		"params": {
		"protocolVersion": "2024-11-05",
		"capabilities": {},
		"clientInfo": { "name": "raw-test-client", "version": "0.0.0" }
		}
		}),
		)
		.await?;
		read_response_for_id(&mut reader, 1).await?;

		send_json(
		&mut writer,
		&json!({ "jsonrpc": "2.0", "method": "notifications/initialized" }),
		)
		.await?;

		// Send the whole batch before reading responses. This creates concurrent
		// request handling and concurrent response production inside rmcp.
		for id in request_ids() {
		send_json(
		&mut writer,
		&json!({
		"jsonrpc": "2.0",
		"id": id,
		"method": "tools/call",
		"params": { "name": "large-response", "arguments": {} }
		}),
		)
		.await?;
		}

		let missing_ids = read_responses_for_ids(&mut reader, request_ids(), READ_TIMEOUT).await?;
		assert!(
		missing_ids.is_empty(),
		"missing response ids: {missing_ids:?}"
		);

		drop(writer);
		wait_for_child(&mut child).await;
		Ok(())
		}

		struct LargeResponseServer;

		impl ServerHandler for LargeResponseServer {
		#[allow(deprecated)]
		fn get_info(&self) -> ServerInfo {
		ServerInfo::new(ServerCapabilities::builder().enable_tools().build())
		}

		async fn call_tool(
		&self,
		request: CallToolRequestParams,
		_context: rmcp::service::RequestContext<rmcp::RoleServer>,
		) -> Result<CallToolResult, McpError> {
		assert_eq!("large-response", request.name.as_ref());
		Ok(CallToolResult::success(vec![ContentBlock::text(
		"x".repeat(RESPONSE_BYTES),
		)]))
		}
		}

		#[tokio::test]
		async fn stdio_response_concurrency_helper() -> anyhow::Result<()> {
		// The parent test starts this same binary with HELPER_ENV=1 so it can act
		// as a small MCP server connected over real stdin/stdout pipes.
		if std::env::var(HELPER_ENV).as_deref() != Ok("1") {
		return Ok(());
		}
		let server = LargeResponseServer.serve(rmcp::transport::stdio()).await?;
		server.waiting().await?;
		Ok(())
		}

		fn spawn_helper() -> Child {
		let exe = std::env::current_exe().expect("current test exe");
		Command::new(exe)
		.arg("--exact")
		.arg("stdio_response_concurrency_helper")
		.arg("--quiet")
		.arg("--nocapture")
		.arg("--test-threads")
		.arg("1")
		.env(HELPER_ENV, "1")
		.stdin(Stdio::piped())
		.stdout(Stdio::piped())
		.stderr(Stdio::null())
		.kill_on_drop(true)
		.spawn()
		.expect("spawn helper")
		}

		async fn wait_for_child(child: &mut Child) {
		let _ = tokio::time::timeout(Duration::from_secs(2), child.wait()).await;
		if child.id().is_some() {
		let _ = child.kill().await;
		}
		}

		fn request_ids() -> BTreeSet<u64> {
		(1000..1000 + REQUESTS as u64).collect()
		}

		async fn send_json<W>(writer: &mut W, message: &Value) -> anyhow::Result<()>
		where
		W: AsyncWrite + Unpin,
		{
		let serialized = serde_json::to_string(message)?;
		writer.write_all(serialized.as_bytes()).await?;
		writer.write_all(b"\n").await?;
		writer.flush().await?;
		Ok(())
		}

		async fn read_response_for_id<R>(reader: &mut BufReader<R>, expected_id: u64) -> anyhow::Result<()>
		where
		R: tokio::io::AsyncRead + Unpin,
		{
		let missing =
		read_responses_for_ids(reader, BTreeSet::from([expected_id]), READ_TIMEOUT).await?;
		if missing.is_empty() {
		Ok(())
		} else {
		anyhow::bail!("missing response id {expected_id}")
		}
		}

		async fn read_responses_for_ids<R>(
		reader: &mut BufReader<R>,
		mut pending_ids: BTreeSet<u64>,
		timeout: Duration,
		) -> anyhow::Result<BTreeSet<u64>>
		where
		R: tokio::io::AsyncRead + Unpin,
		{
		let deadline = tokio::time::Instant::now() + timeout;
		while !pending_ids.is_empty() {
		let remaining = deadline.saturating_duration_since(tokio::time::Instant::now());
		if remaining.is_zero() {
		break;
		}
		let mut line = String::new();
		let Ok(read_result) = tokio::time::timeout(remaining, reader.read_line(&mut line)).await
		else {
		break;
		};
		let read = read_result?;
		if read == 0 {
		break;
		}
		let trimmed = line.trim();
		if trimmed.is_empty() {
		continue;
		}
		let Ok(value) = serde_json::from_str::<Value>(trimmed) else {
		continue;
		};
		if let Some(id) = value.get("id").and_then(Value::as_u64) {
		pending_ids.remove(&id);
		}
		}
		Ok(pending_ids)
		}

+85

tests/test_trace_context.rs

		#![cfg(not(feature = "local"))]
		//! SEP-414: the reserved trace-context `_meta` keys survive a client→server round trip unchanged.
		use std::sync::Arc;

		use rmcp::{
		RoleServer, ServerHandler, ServiceExt,
		model::{ClientRequest, CustomRequest, CustomResult, Meta},
		service::{PeerRequestOptions, RequestContext},
		};
		use serde_json::json;
		use tokio::sync::{Mutex, Notify};

		const TRACEPARENT: &str = "00-0af7651916cd43dd8448eb211c80319c-00f067aa0ba902b7-01";
		const TRACESTATE: &str = "vendor1=value1,vendor2=value2";
		const BAGGAGE: &str = "userId=alice,region=us-east-1";

		/// Records the `_meta` it receives on the incoming request so the test can assert passthrough.
		struct TraceCapturingServer {
		receive_signal: Arc<Notify>,
		seen: Arc<Mutex<Option<Meta>>>,
		}

		impl ServerHandler for TraceCapturingServer {
		async fn on_custom_request(
		&self,
		_request: CustomRequest,
		context: RequestContext<RoleServer>,
		) -> Result<CustomResult, rmcp::ErrorData> {
		*self.seen.lock().await = Some(context.meta);
		self.receive_signal.notify_one();
		Ok(CustomResult::new(json!({ "status": "ok" })))
		}
		}

		#[tokio::test]
		async fn trace_context_meta_survives_round_trip() -> anyhow::Result<()> {
		let (server_transport, client_transport) = tokio::io::duplex(4096);
		let receive_signal = Arc::new(Notify::new());
		let seen = Arc::new(Mutex::new(None));

		{
		let receive_signal = receive_signal.clone();
		let seen = seen.clone();
		tokio::spawn(async move {
		let server = TraceCapturingServer {
		receive_signal,
		seen,
		}
		.serve(server_transport)
		.await?;
		server.waiting().await?;
		anyhow::Ok(())
		});
		}

		let client = ().serve(client_transport).await?;

		// Client attaches trace context to the outgoing request's `_meta`.
		let mut meta = Meta::new();
		meta.set_traceparent(TRACEPARENT);
		meta.set_tracestate(TRACESTATE);
		meta.set_baggage(BAGGAGE);

		let mut options = PeerRequestOptions::no_options();
		options.meta = Some(meta);
		client
		.send_cancellable_request(
		ClientRequest::CustomRequest(CustomRequest::new("requests/trace-test", None)),
		options,
		)
		.await?
		.await_response()
		.await?;

		tokio::time::timeout(std::time::Duration::from_secs(5), receive_signal.notified()).await?;

		// Server saw the reserved keys unchanged (alongside the injected progressToken).
		let seen = seen.lock().await.take().expect("server observed meta");
		assert_eq!(seen.get_traceparent(), Some(TRACEPARENT));
		assert_eq!(seen.get_tracestate(), Some(TRACESTATE));
		assert_eq!(seen.get_baggage(), Some(BAGGAGE));

		client.cancel().await?;
		Ok(())
		}

+1

-1

.cargo_vcs_info.json

		{
		"git": {
		"sha1": "67a30859443ab0fe79f2d50307c7d7bc9518f7e3"
		"sha1": "8e44af499bfa8d54f18fb475eb86a601a5a2e432"
		},
		"path_in_vcs": "crates/rmcp"
		}

+35

-2

Cargo.toml

		@@ -15,3 +15,3 @@ # THIS FILE IS AUTOMATICALLY GENERATED BY CARGO
		name = "rmcp"
		version = "2.0.0"
		version = "2.1.0"
		build = "build.rs"
		@@ -269,2 +269,6 @@ autolib = false
		[[test]]
		name = "test_meta_helpers"
		path = "tests/test_meta_helpers.rs"

		[[test]]
		name = "test_notification"
		@@ -307,2 +311,10 @@ path = "tests/test_notification.rs"
		[[test]]
		name = "test_protocol_version_negotiation"
		path = "tests/test_protocol_version_negotiation.rs"
		required-features = [
		"server",
		"client",
		]

		[[test]]
		name = "test_request_timeout_progress"
		@@ -352,2 +364,15 @@ path = "tests/test_request_timeout_progress.rs"
		[[test]]
		name = "test_stateless_protocol_version"
		path = "tests/test_stateless_protocol_version.rs"
		required-features = [
		"server",
		"transport-streamable-http-server",
		"reqwest",
		]

		[[test]]
		name = "test_stdio_response_concurrency"
		path = "tests/test_stdio_response_concurrency.rs"

		[[test]]
		name = "test_streamable_http_4xx_error_body"
		@@ -486,2 +511,10 @@ path = "tests/test_streamable_http_4xx_error_body.rs"
		[[test]]
		name = "test_trace_context"
		path = "tests/test_trace_context.rs"
		required-features = [
		"server",
		"client",
		]

		[[test]]
		name = "test_unix_socket_transport"
		@@ -591,3 +624,3 @@ path = "tests/test_unix_socket_transport.rs"
		[dependencies.rmcp-macros]
		version = "2.0.0"
		version = "2.1.0"
		optional = true
		@@ -594,0 +627,0 @@

+15

-0

CHANGELOG.md

		@@ -10,2 +10,17 @@ # Changelog

		## [2.1.0](https://github.com/modelcontextprotocol/rust-sdk/compare/rmcp-v2.0.0...rmcp-v2.1.0) - 2026-07-02

		### Added

		- add SEP-414 trace context meta accessors ([#910](https://github.com/modelcontextprotocol/rust-sdk/pull/910))
		- add SEP-2575 meta helpers ([#942](https://github.com/modelcontextprotocol/rust-sdk/pull/942))

		### Fixed

		- (transport) make AsyncRwTransport::receive cancel-safe ([#941](https://github.com/modelcontextprotocol/rust-sdk/pull/941)) ([#947](https://github.com/modelcontextprotocol/rust-sdk/pull/947))
		- (auth) preserve refresh_token when refresh response omits it ([#949](https://github.com/modelcontextprotocol/rust-sdk/pull/949))
		- block redirect header leaks ([#936](https://github.com/modelcontextprotocol/rust-sdk/pull/936))
		- don't respond to unparsable messages ([#940](https://github.com/modelcontextprotocol/rust-sdk/pull/940))
		- negotiate protocol version in handler ([#930](https://github.com/modelcontextprotocol/rust-sdk/pull/930))

		## [2.0.0](https://github.com/modelcontextprotocol/rust-sdk/compare/rmcp-v1.8.0...rmcp-v2.0.0) - 2026-06-27
		@@ -12,0 +27,0 @@

+8

-2

src/handler/server.rs

		@@ -10,2 +10,3 @@ // Sampling/Roots/Logging are SEP-2577-deprecated; internal references are expected.
		MaybeSendFuture, NotificationContext, RequestContext, RoleServer, Service, ServiceRole,
		negotiate_protocol_version,
		},
		@@ -206,4 +207,9 @@ };
		) -> impl Future<Output = Result<InitializeResult, McpError>> + MaybeSendFuture + '_ {
		context.peer.set_peer_info(request);
		std::future::ready(Ok(self.get_info()))
		context.peer.set_peer_info(request.clone());
		let mut info = self.get_info();
		info.protocol_version = negotiate_protocol_version(
		&request.protocol_version,
		info.protocol_version,
		);
		std::future::ready(Ok(info))
		}
		@@ -210,0 +216,0 @@ fn complete(

+233

-20

src/model/meta.rs

		@@ -7,4 +7,5 @@ use std::ops::{Deref, DerefMut};
		use super::{
		ClientNotification, ClientRequest, CustomNotification, CustomRequest, Extensions, JsonObject,
		JsonRpcMessage, NumberOrString, ProgressToken, ServerNotification, ServerRequest, TaskMetadata,
		ClientCapabilities, ClientNotification, ClientRequest, CustomNotification, CustomRequest,
		Extensions, Implementation, JsonObject, JsonRpcMessage, LoggingLevel, NumberOrString,
		ProgressToken, ProtocolVersion, ServerNotification, ServerRequest, TaskMetadata,
		};
		@@ -50,2 +51,30 @@
		}
		/// Get the W3C `traceparent` value from meta, if present (SEP-414)
		fn traceparent(&self) -> Option<&str> {
		self.meta().and_then(\|m\| m.get_traceparent())
		}
		/// Set the W3C `traceparent` value in meta (SEP-414)
		fn set_traceparent(&mut self, value: &str) {
		self.meta_or_default().set_traceparent(value);
		}
		/// Get the W3C `tracestate` value from meta, if present (SEP-414)
		fn tracestate(&self) -> Option<&str> {
		self.meta().and_then(\|m\| m.get_tracestate())
		}
		/// Set the W3C `tracestate` value in meta (SEP-414)
		fn set_tracestate(&mut self, value: &str) {
		self.meta_or_default().set_tracestate(value);
		}
		/// Get the W3C `baggage` value from meta, if present (SEP-414)
		fn baggage(&self) -> Option<&str> {
		self.meta().and_then(\|m\| m.get_baggage())
		}
		/// Set the W3C `baggage` value in meta (SEP-414)
		fn set_baggage(&mut self, value: &str) {
		self.meta_or_default().set_baggage(value);
		}
		/// Get a mutable reference to meta, inserting an empty one if absent.
		fn meta_or_default(&mut self) -> &mut Meta {
		self.meta_mut().get_or_insert_with(Meta::new)
		}
		}
		@@ -204,4 +233,16 @@
		pub struct Meta(pub JsonObject);
		const PROGRESS_TOKEN_FIELD: &str = "progressToken";

		impl Meta {
		const PROGRESS_TOKEN_FIELD: &str = "progressToken";
		const META_KEY_PROTOCOL_VERSION: &str = "io.modelcontextprotocol/protocolVersion";
		const META_KEY_CLIENT_INFO: &str = "io.modelcontextprotocol/clientInfo";
		const META_KEY_CLIENT_CAPABILITIES: &str = "io.modelcontextprotocol/clientCapabilities";
		const META_KEY_LOG_LEVEL: &str = "io.modelcontextprotocol/logLevel";
		/// Reserved `_meta` key for the W3C Trace Context `traceparent` value (SEP-414).
		const TRACEPARENT_FIELD: &str = "traceparent";
		/// Reserved `_meta` key for the W3C Trace Context `tracestate` value (SEP-414).
		const TRACESTATE_FIELD: &str = "tracestate";
		/// Reserved `_meta` key for the W3C Baggage value (SEP-414).
		const BAGGAGE_FIELD: &str = "baggage";

		pub fn new() -> Self {
		@@ -224,19 +265,23 @@ Self(JsonObject::new())
		pub fn get_progress_token(&self) -> Option<ProgressToken> {
		self.0.get(PROGRESS_TOKEN_FIELD).and_then(\|v\| match v {
		Value::String(s) => Some(ProgressToken(NumberOrString::String(s.to_string().into()))),
		Value::Number(n) => {
		if let Some(i) = n.as_i64() {
		Some(ProgressToken(NumberOrString::Number(i)))
		} else if let Some(u) = n.as_u64() {
		if u <= i64::MAX as u64 {
		Some(ProgressToken(NumberOrString::Number(u as i64)))
		self.0
		.get(Self::PROGRESS_TOKEN_FIELD)
		.and_then(\|v\| match v {
		Value::String(s) => {
		Some(ProgressToken(NumberOrString::String(s.to_string().into())))
		}
		Value::Number(n) => {
		if let Some(i) = n.as_i64() {
		Some(ProgressToken(NumberOrString::Number(i)))
		} else if let Some(u) = n.as_u64() {
		if u <= i64::MAX as u64 {
		Some(ProgressToken(NumberOrString::Number(u as i64)))
		} else {
		None
		}
		} else {
		None
		}
		} else {
		None
		}
		}
		_ => None,
		})
		_ => None,
		})
		}
		@@ -247,11 +292,107 @@
		NumberOrString::String(ref s) => self.0.insert(
		PROGRESS_TOKEN_FIELD.to_string(),
		Self::PROGRESS_TOKEN_FIELD.to_string(),
		Value::String(s.to_string()),
		),
		NumberOrString::Number(n) => self
		.0
		.insert(PROGRESS_TOKEN_FIELD.to_string(), Value::Number(n.into())),
		NumberOrString::Number(n) => self.0.insert(
		Self::PROGRESS_TOKEN_FIELD.to_string(),
		Value::Number(n.into()),
		),
		};
		}

		/// Get the MCP protocol version carried in `_meta`, if present and valid.
		pub fn protocol_version(&self) -> Option<ProtocolVersion> {
		self.decode_value(Self::META_KEY_PROTOCOL_VERSION)
		}

		/// Set the MCP protocol version carried in `_meta`.
		pub fn set_protocol_version(&mut self, protocol_version: ProtocolVersion) {
		self.0.insert(
		Self::META_KEY_PROTOCOL_VERSION.to_string(),
		Value::String(protocol_version.to_string()),
		);
		}

		/// Get the client implementation identity carried in `_meta`, if present and valid.
		pub fn client_info(&self) -> Option<Implementation> {
		self.decode_value(Self::META_KEY_CLIENT_INFO)
		}

		/// Set the client implementation identity carried in `_meta`.
		pub fn set_client_info(&mut self, client_info: Implementation) {
		self.insert_serialized(Self::META_KEY_CLIENT_INFO, client_info);
		}

		/// Get the client capabilities carried in `_meta`, if present and valid.
		pub fn client_capabilities(&self) -> Option<ClientCapabilities> {
		self.decode_value(Self::META_KEY_CLIENT_CAPABILITIES)
		}

		/// Set the client capabilities carried in `_meta`.
		pub fn set_client_capabilities(&mut self, client_capabilities: ClientCapabilities) {
		self.insert_serialized(Self::META_KEY_CLIENT_CAPABILITIES, client_capabilities);
		}

		/// Get the requested per-request log level carried in `_meta`, if present and valid.
		pub fn log_level(&self) -> Option<LoggingLevel> {
		self.decode_value(Self::META_KEY_LOG_LEVEL)
		}

		/// Set the requested per-request log level carried in `_meta`.
		pub fn set_log_level(&mut self, log_level: LoggingLevel) {
		self.insert_serialized(Self::META_KEY_LOG_LEVEL, log_level);
		}

		/// Read a string-valued `_meta` field, or `None` if absent or not a string.
		fn get_str(&self, field: &str) -> Option<&str> {
		self.0.get(field).and_then(Value::as_str)
		}

		/// Write a string-valued `_meta` field.
		fn set_str(&mut self, field: &str, value: impl Into<String>) {
		self.0
		.insert(field.to_string(), Value::String(value.into()));
		}

		/// Get the W3C `traceparent` value (SEP-414), if present.
		pub fn get_traceparent(&self) -> Option<&str> {
		self.get_str(Self::TRACEPARENT_FIELD)
		}

		/// Set the W3C `traceparent` value (SEP-414).
		///
		/// ```
		/// use rmcp::model::Meta;
		///
		/// let mut meta = Meta::new();
		/// meta.set_traceparent("00-0af7651916cd43dd8448eb211c80319c-00f067aa0ba902b7-01");
		/// assert_eq!(
		/// meta.get_traceparent(),
		/// Some("00-0af7651916cd43dd8448eb211c80319c-00f067aa0ba902b7-01"),
		/// );
		/// ```
		pub fn set_traceparent(&mut self, value: impl Into<String>) {
		self.set_str(Self::TRACEPARENT_FIELD, value);
		}

		/// Get the W3C `tracestate` value (SEP-414), if present.
		pub fn get_tracestate(&self) -> Option<&str> {
		self.get_str(Self::TRACESTATE_FIELD)
		}

		/// Set the W3C `tracestate` value (SEP-414).
		pub fn set_tracestate(&mut self, value: impl Into<String>) {
		self.set_str(Self::TRACESTATE_FIELD, value);
		}

		/// Get the W3C `baggage` value (SEP-414), if present.
		pub fn get_baggage(&self) -> Option<&str> {
		self.get_str(Self::BAGGAGE_FIELD)
		}

		/// Set the W3C `baggage` value (SEP-414).
		pub fn set_baggage(&mut self, value: impl Into<String>) {
		self.set_str(Self::BAGGAGE_FIELD, value);
		}

		pub fn extend(&mut self, other: Meta) {
		@@ -262,2 +403,18 @@ for (k, v) in other.0.into_iter() {
		}

		fn decode_value<T>(&self, key: &str) -> Option<T>
		where
		T: for<'de> Deserialize<'de>,
		{
		self.0.get(key).and_then(\|value\| T::deserialize(value).ok())
		}

		fn insert_serialized<T>(&mut self, key: &str, value: T)
		where
		T: Serialize,
		{
		let value = serde_json::to_value(value)
		.expect("MCP meta helper value should serialize to valid JSON");
		self.0.insert(key.to_string(), value);
		}
		}
		@@ -299,1 +456,57 @@
		}

		#[cfg(test)]
		mod tests {
		use super::*;

		#[derive(Default)]
		struct Params {
		meta: Option<Meta>,
		}

		impl RequestParamsMeta for Params {
		fn meta(&self) -> Option<&Meta> {
		self.meta.as_ref()
		}
		fn meta_mut(&mut self) -> &mut Option<Meta> {
		&mut self.meta
		}
		}

		const TRACEPARENT: &str = "00-0af7651916cd43dd8448eb211c80319c-00f067aa0ba902b7-01";

		#[test]
		fn trace_context_round_trip() {
		let mut meta = Meta::new();
		meta.set_traceparent(TRACEPARENT);
		meta.set_tracestate("vendor1=value1,vendor2=value2");
		meta.set_baggage("userId=alice,region=us-east-1");
		assert_eq!(meta.get_traceparent(), Some(TRACEPARENT));
		assert_eq!(meta.get_tracestate(), Some("vendor1=value1,vendor2=value2"));
		assert_eq!(meta.get_baggage(), Some("userId=alice,region=us-east-1"));
		}

		#[test]
		fn absent_field_is_none() {
		let meta = Meta::new();
		assert_eq!(meta.get_traceparent(), None);
		assert_eq!(meta.get_tracestate(), None);
		assert_eq!(meta.get_baggage(), None);
		}

		#[test]
		fn non_string_value_is_none() {
		let mut meta = Meta::new();
		meta.0
		.insert(Meta::TRACEPARENT_FIELD.to_string(), Value::from(42));
		assert_eq!(meta.get_traceparent(), None);
		}

		#[test]
		fn trait_setter_inserts_meta_when_absent() {
		let mut params = Params::default();
		assert_eq!(params.traceparent(), None);
		params.set_traceparent(TRACEPARENT);
		assert_eq!(params.traceparent(), Some(TRACEPARENT));
		}
		}

+6

-1

src/service/server.rs

		@@ -165,3 +165,3 @@ // Sampling/Roots/Logging are SEP-2577-deprecated; internal references are expected.
		/// Echoes the client-requested version if known; otherwise returns `server_fallback`.
		fn negotiate_protocol_version(
		pub(crate) fn negotiate_protocol_version(
		client_requested: &ProtocolVersion,
		@@ -258,2 +258,7 @@ server_fallback: ProtocolVersion,
		);
		// Update peer_info so context.protocol_version() reflects the negotiated
		// version in all subsequent request handlers.
		let mut negotiated_peer_info = peer_info.params.clone();
		negotiated_peer_info.protocol_version = init_response.protocol_version.clone();
		peer.set_peer_info(negotiated_peer_info);
		transport
		@@ -260,0 +265,0 @@ .send(ServerJsonRpcMessage::response(

+106

-30

src/transport/async_rw.rs

		@@ -127,4 +127,15 @@ use std::{marker::PhantomData, sync::Arc};
		loop {
		self.line_buf.clear();
		// `read_until` is not cancellation-safe on its own, and `receive` is
		// polled inside a `select!` in the service loop: an in-progress line
		// read is dropped whenever another branch (e.g. an outgoing response)
		// becomes ready. We rely on `read_until` appending into `self.line_buf`
		// and only returning at a delimiter or EOF, so a cancelled read leaves
		// its partial bytes in `self.line_buf`. Keeping that buffer across
		// calls lets the next read resume the same line; it is cleared only
		// after a whole line has been consumed. Clearing at the top of the
		// loop (the previous behaviour) discarded the partial read and so
		// dropped incoming requests under concurrent response load.
		match self.read.read_until(b'\n', &mut self.line_buf).await {
		// EOF. Any bytes still in `line_buf` are an incomplete trailing
		// message with no delimiter, so there is nothing to deliver.
		Ok(0) => return None,
		@@ -137,21 +148,44 @@ Ok(_) => {}
		}
		let line = without_carriage_return(
		self.line_buf.strip_suffix(b"\n").unwrap_or(&self.line_buf),
		);
		if line.is_empty() {
		continue;
		}
		match try_parse_with_compatibility::<RxJsonRpcMessage<Role>>(line, "receive") {
		// A returned `read_until` means a full line is buffered. Parse it
		// (borrowing `line_buf`), then clear the buffer — retaining its
		// capacity for the next read — before handling the parse result.
		let parsed = {
		let line = without_carriage_return(
		self.line_buf.strip_suffix(b"\n").unwrap_or(&self.line_buf),
		);
		if line.is_empty() {
		self.line_buf.clear();
		continue;
		}
		try_parse_with_compatibility::<RxJsonRpcMessage<Role>>(line, "receive")
		};
		self.line_buf.clear();
		match parsed {
		Ok(Some(msg)) => return Some(msg),
		Ok(None) => continue,
		Err(JsonRpcMessageCodecError::Serde(e)) => {
		tracing::debug!("Parse error on incoming message: {e}");
		let mut write = self.write.lock().await;
		let framed = write.as_mut()?;
		let response = TxJsonRpcMessage::<Role>::error(
		ErrorData::parse_error("Parse error", None),
		None,
		);
		if framed.send(response).await.is_err() {
		return None;
		match e.classify() {
		serde_json::error::Category::Syntax \| serde_json::error::Category::Eof => {
		// The input isn't valid JSON, so there's no message id to correlate a
		// response to, and replying to invalid data can trigger an error storm
		// if the peer echoes the response back as more invalid data. This
		// matches the other official MCP SDKs, which ignore unparsable input.
		// See https://github.com/modelcontextprotocol/rust-sdk/issues/938
		tracing::debug!("Ignoring unparsable incoming message: {e}");
		}
		serde_json::error::Category::Data \| serde_json::error::Category::Io => {
		// Well-formed JSON that doesn't match the expected message shape is a
		// real protocol error rather than unparsable input, so surface it with
		// an Invalid Request response instead of silently dropping it.
		tracing::debug!("Protocol error on incoming message: {e}");
		let mut write = self.write.lock().await;
		let framed = write.as_mut()?;
		let response = TxJsonRpcMessage::<Role>::error(
		ErrorData::invalid_request("Invalid request", None),
		None,
		);
		if framed.send(response).await.is_err() {
		return None;
		}
		}
		}
		@@ -623,4 +657,4 @@ }
		#[tokio::test]
		async fn receive_recovers_from_parse_error() {
		use tokio::io::AsyncWriteExt;
		async fn receive_ignores_parse_error() {
		use tokio::io::{AsyncReadExt, AsyncWriteExt};

		@@ -644,14 +678,60 @@ use crate::{RoleServer, transport::Transport};

		// The unparsable line is skipped and the next valid message is still yielded.
		let received = transport
		.receive()
		.await
		.expect("transport should recover and yield the next valid message");
		.expect("transport should skip the invalid line and yield the next valid message");
		assert_eq!(
		serde_json::to_value(&received).unwrap()["method"],
		"notifications/initialized",
		);

		// Read one line back from the peer side and parse as JSON.
		// No response is sent back for the unparsable message (issue #938). Dropping the
		// transport closes its write side, so the peer reads to EOF and should see no bytes.
		drop(transport);
		let mut reply_buf = Vec::new();
		let mut peer = tokio::io::BufReader::new(&mut client_r);
		client_r.read_to_end(&mut reply_buf).await.unwrap();
		assert!(
		reply_buf.is_empty(),
		"expected no response to an unparsable message, got: {}",
		String::from_utf8_lossy(&reply_buf),
		);
		}

		#[cfg(feature = "server")]
		#[tokio::test]
		async fn receive_responds_to_protocol_error() {
		use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader};

		use crate::{RoleServer, transport::Transport};

		let (server_io, client_io) = tokio::io::duplex(4096);
		let (server_r, server_w) = tokio::io::split(server_io);
		let (client_r, mut client_w) = tokio::io::split(client_io);

		let mut transport = AsyncRwTransport::<RoleServer, _, _>::new(server_r, server_w);

		// Well-formed JSON that does not match the JSON-RPC message shape, followed by a
		// valid notification. Unlike unparsable bytes, this is a protocol error: the
		// transport should reply to it and still yield the next valid message.
		client_w
		.write_all(
		b"{\"foo\":\"bar\"}\n{\"jsonrpc\":\"2.0\",\"method\":\"notifications/initialized\"}\n",
		)
		.await
		.unwrap();

		let received = transport.receive().await.expect(
		"transport should reply to the protocol error and yield the next valid message",
		);
		assert_eq!(
		serde_json::to_value(&received).unwrap()["method"],
		"notifications/initialized",
		);

		// A protocol error gets an error response back (id omitted since it can't be read).
		let mut reply_buf = Vec::new();
		let mut peer = BufReader::new(client_r);
		peer.read_until(b'\n', &mut reply_buf).await.unwrap();
		let reply: serde_json::Value = serde_json::from_slice(&reply_buf).unwrap();

		// Per MCP 2025-11-25: id is omitted when the server can't read the request id.
		assert_eq!(
		@@ -661,10 +741,6 @@ reply,
		"jsonrpc": "2.0",
		"error": {"code": -32700, "message": "Parse error"},
		})
		"error": {"code": -32600, "message": "Invalid request"},
		}),
		);
		assert_eq!(
		serde_json::to_value(&received).unwrap()["method"],
		"notifications/initialized",
		);
		}
		}

+136

-1

src/transport/common/reqwest/streamable_http_client.rs

		@@ -302,5 +302,9 @@ use std::{borrow::Cow, collections::HashMap, sync::Arc};
		/// fully consumed before the pool attempts to reuse the connection.
		///
		/// Automatic redirects are disabled so caller-supplied custom headers
		/// cannot be replayed to a redirect target.
		fn default_http_client() -> reqwest::Client {
		reqwest::Client::builder()
		.pool_max_idle_per_host(0)
		.redirect(reqwest::redirect::Policy::none())
		.build()
		@@ -317,3 +321,3 @@ .expect("failed to build default reqwest client")
		use crate::{
		model::JsonRpcMessage,
		model::{ClientJsonRpcMessage, ClientRequest, JsonRpcMessage, PingRequest, RequestId},
		transport::streamable_http_client::{AuthRequiredError, InsufficientScopeError},
		@@ -364,2 +368,133 @@ };
		}

		#[tokio::test]
		async fn default_http_client_does_not_leak_custom_headers_to_redirect_target()
		-> anyhow::Result<()> {
		use std::{collections::HashMap, net::SocketAddr, sync::Arc};

		use axum::{
		Router, extract::State, http::StatusCode, response::IntoResponse, routing::post,
		};
		use http::{HeaderMap, HeaderName, HeaderValue, header::LOCATION};
		use tokio::sync::Mutex;

		use super::StreamableHttpClientTransport;
		use crate::transport::streamable_http_client::{StreamableHttpClient, StreamableHttpError};

		const API_KEY_HEADER: &str = "x-api-key";
		const API_KEY_VALUE: &str = "secret";

		type CapturedHeader = Arc<Mutex<Option<String>>>;

		#[derive(Clone)]
		struct RedirectState {
		location: String,
		captured_header: CapturedHeader,
		}

		async fn capture_api_key_header(headers: &HeaderMap, captured_header: &CapturedHeader) {
		if let Some(value) = headers
		.get(API_KEY_HEADER)
		.and_then(\|value\| value.to_str().ok())
		{
		*captured_header.lock().await = Some(value.to_owned());
		}
		}

		async fn redirect_handler(
		State(state): State<RedirectState>,
		headers: HeaderMap,
		) -> impl IntoResponse {
		capture_api_key_header(&headers, &state.captured_header).await;

		(
		StatusCode::TEMPORARY_REDIRECT,
		[(LOCATION, state.location)],
		"",
		)
		}

		async fn redirected_handler(
		State(captured_header): State<CapturedHeader>,
		headers: HeaderMap,
		) -> impl IntoResponse {
		capture_api_key_header(&headers, &captured_header).await;

		(
		StatusCode::OK,
		[(http::header::CONTENT_TYPE, "application/json")],
		r#"{"jsonrpc":"2.0","id":1,"result":{}}"#,
		)
		}

		let redirected_header = Arc::new(Mutex::new(None));
		let redirected_listener =
		tokio::net::TcpListener::bind(SocketAddr::from(([127, 0, 0, 1], 0))).await?;
		let redirected_addr = redirected_listener.local_addr()?;
		let redirected_server = tokio::spawn({
		let redirected_header = redirected_header.clone();
		async move {
		let app = Router::new()
		.route("/capture", post(redirected_handler))
		.with_state(redirected_header);
		axum::serve(redirected_listener, app).await
		}
		});

		let original_header = Arc::new(Mutex::new(None));
		let redirect_listener =
		tokio::net::TcpListener::bind(SocketAddr::from(([127, 0, 0, 1], 0))).await?;
		let redirect_addr = redirect_listener.local_addr()?;
		let redirect_server = tokio::spawn({
		let state = RedirectState {
		location: format!("http://{redirected_addr}/capture"),
		captured_header: original_header.clone(),
		};
		async move {
		let app = Router::new()
		.route("/mcp", post(redirect_handler))
		.with_state(state);
		axum::serve(redirect_listener, app).await
		}
		});

		let mut custom_headers = HashMap::new();
		custom_headers.insert(
		HeaderName::from_static(API_KEY_HEADER),
		HeaderValue::from_static(API_KEY_VALUE),
		);
		let message = ClientJsonRpcMessage::request(
		ClientRequest::PingRequest(PingRequest::default()),
		RequestId::Number(1),
		);

		let client = StreamableHttpClientTransport::<reqwest::Client>::default_http_client();
		let result = client
		.post_message(
		Arc::<str>::from(format!("http://{redirect_addr}/mcp")),
		message,
		None,
		None,
		custom_headers,
		)
		.await;

		assert!(
		matches!(
		result,
		Err(StreamableHttpError::UnexpectedServerResponse(_))
		),
		"redirect response should be returned to the transport, got {result:?}"
		);
		assert_eq!(original_header.lock().await.as_deref(), Some(API_KEY_VALUE));
		assert!(
		redirected_header.lock().await.is_none(),
		"custom headers should not be sent to redirect targets"
		);

		redirect_server.abort();
		redirected_server.abort();

		Ok(())
		}
		}

+41

-3

src/transport/streamable_http_server/tower.rs

		@@ -19,4 +19,5 @@ use std::{
		model::{
		ClientJsonRpcMessage, ClientNotification, ClientRequest, ErrorData, GetExtensions,
		InitializeRequest, InitializedNotification, JsonRpcError, ProtocolVersion, RequestId,
		ClientCapabilities, ClientJsonRpcMessage, ClientNotification, ClientRequest, ErrorData,
		GetExtensions, Implementation, InitializeRequest, InitializeRequestParams,
		InitializedNotification, JsonRpcError, ProtocolVersion, RequestId,
		},
		@@ -1243,6 +1244,13 @@ serve_server,
		ClientJsonRpcMessage::Request(mut request) => {
		// Build a peer_info so context.protocol_version() works inside handlers.
		// serve_directly skips the handshake and receives None by default, making
		// protocol_version() always return None in stateless mode. We reconstruct it:
		// - initialize requests: version comes from the request body params
		// - all other requests: version comes from the MCP-Protocol-Version header
		// (already validated above; absent header defaults to 2025-03-26)
		let peer_info = Self::peer_info_for_stateless_request(&request, &part.headers);
		request.request.extensions_mut().insert(part);
		let (transport, mut receiver) =
		OneshotTransport::<RoleServer>::new(ClientJsonRpcMessage::Request(request));
		let service = serve_directly(service, transport, None);
		let service = serve_directly(service, transport, peer_info);
		tokio::spawn(async move {
		@@ -1336,2 +1344,32 @@ // on service created
		}

		/// Build a `ClientInfo` (peer_info) for a stateless request so that
		/// `context.protocol_version()` returns the correct value inside handlers.
		///
		/// `serve_directly` skips the MCP handshake and accepts `peer_info = None`,
		/// which means `context.protocol_version()` is always `None` in stateless mode.
		/// We reconstruct the protocol version from the available signal per request type:
		/// - initialize: version is in the request body params (authoritative)
		/// - all other requests: version is in the MCP-Protocol-Version header
		/// (validated before this point; absent header defaults to 2025-03-26)
		fn peer_info_for_stateless_request(
		request: &crate::model::JsonRpcRequest<ClientRequest>,
		headers: &HeaderMap,
		) -> Option<InitializeRequestParams> {
		let version = if let ClientRequest::InitializeRequest(ref init) = request.request {
		init.params.protocol_version.clone()
		} else {
		headers
		.get(HEADER_MCP_PROTOCOL_VERSION)
		.and_then(\|v\| v.to_str().ok())
		.and_then(\|s\| serde_json::from_value(serde_json::Value::String(s.to_owned())).ok())
		.unwrap_or(ProtocolVersion::V_2025_03_26)
		};
		Some(InitializeRequestParams {
		meta: None,
		protocol_version: version,
		capabilities: ClientCapabilities::default(),
		client_info: Implementation::default(),
		})
		}
		}

Cargo.lock

Sorry, the diff of this file is too big to display

Cargo.toml.orig

Sorry, the diff of this file is not supported yet

src/transport/auth.rs

Sorry, the diff of this file is too big to display

rmcp - cargo Package Compare versions