modelcontextprotocol/rust-sdk
Watch
1
0
Fork
You've already forked rust-sdk
 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions 1

chore: declare and check MSRV (#1034)

Browse source 
* chore: declare and check MSRV

* fix: clippy fixes for collapsing if-statements

* ci: add explicit contents read permission to MSRV job
This commit is contained in:
Jack Amadeo 2026-07-23 15:59:55 -04:00 • committed by GitHub
commit 76988023b2
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
22 changed files with 322 additions and 315 deletions
Download patch file Download diff file Expand all files Collapse all files

24
.github/workflows/ci.yml vendored
View file

@ -205,6 +205,30 @@ jobs:
- name: Spell Check Repo
uses: crate-ci/typos@master
msrv:
name: Check MSRV
runs-on: ubuntu-latest
permissions:
contents: read
steps:
- uses: actions/checkout@v7
- name: Read MSRV from Cargo.toml
run: |
MSRV=$(cargo metadata --no-deps --format-version 1 \
| jq -r '.packages[] | select(.name == "rmcp") | .rust_version')
echo "MSRV=$MSRV" >> "$GITHUB_ENV"
- name: Install MSRV Rust
uses: dtolnay/rust-toolchain@master
with:
toolchain: ${{ env.MSRV }}
- uses: Swatinem/rust-cache@v2
- name: Check default workspace members with MSRV
run: cargo +${{ env.MSRV }} check --all-targets --all-features
test:
name: Run Tests
runs-on: ubuntu-latest

1
Cargo.toml
View file

@ -9,6 +9,7 @@ rmcp-macros = { version = "3.0.0-beta.1", path = "./crates/rmcp-macros" }
[workspace.package]
edition = "2024"
rust-version = "1.88"
version = "3.0.0-beta.1"
authors = ["4t145 <u4t145@163.com>"]
license = "Apache-2.0"

1
clippy.toml
View file

@ -1,3 +1,2 @@
msrv = "1.85"
too-many-arguments-threshold = 10
check-private-items = false

1
crates/rmcp-macros/Cargo.toml
View file

@ -5,6 +5,7 @@ name = "rmcp-macros"
license = { workspace = true }
version = { workspace = true }
edition = { workspace = true }
rust-version = { workspace = true }
repository = { workspace = true }
homepage = { workspace = true }
readme = { workspace = true }

20
crates/rmcp-macros/src/common.rs
View file

@ -55,17 +55,15 @@ pub fn extract_doc_line(
/// Returns the full Parameters<T> type if found
pub fn find_parameters_type_in_sig(sig: &Signature) -> Option<Box<Type>> {
sig.inputs.iter().find_map(|input| {
if let FnArg::Typed(pat_type) = input {
if let Type::Path(type_path) = &*pat_type.ty {
if type_path
.path
.segments
.last()
.is_some_and(|type_name| type_name.ident == "Parameters")
{
return Some(pat_type.ty.clone());
}
}
if let FnArg::Typed(pat_type) = input
&& let Type::Path(type_path) = &*pat_type.ty
&& type_path
.path
.segments
.last()
.is_some_and(|type_name| type_name.ident == "Parameters")
{
return Some(pat_type.ty.clone());
}
None
})

14
crates/rmcp-macros/src/prompt.rs
View file

@ -132,13 +132,13 @@ pub fn prompt(attr: TokenStream, input: TokenStream) -> syn::Result<TokenStream>
// 3. make body: { Box::pin(async move { #body }) }
let new_output = syn::parse2::<ReturnType>({
let mut lt = quote! { 'static };
if let Some(receiver) = fn_item.sig.receiver() {
if let Some((_, receiver_lt)) = receiver.reference.as_ref() {
if let Some(receiver_lt) = receiver_lt {
lt = quote! { #receiver_lt };
} else {
lt = quote! { '_ };
}
if let Some(receiver) = fn_item.sig.receiver()
&& let Some((_, receiver_lt)) = receiver.reference.as_ref()
{
if let Some(receiver_lt) = receiver_lt {
lt = quote! { #receiver_lt };
} else {
lt = quote! { '_ };
}
}
match &fn_item.sig.output {

31
crates/rmcp-macros/src/tool.rs
View file

@ -7,16 +7,13 @@ use crate::common::extract_doc_line;
/// Check if a type is Json<T> and extract the inner type T
fn extract_json_inner_type(ty: &syn::Type) -> Option<&syn::Type> {
if let syn::Type::Path(type_path) = ty {
if let Some(last_segment) = type_path.path.segments.last() {
if last_segment.ident == "Json" {
if let syn::PathArguments::AngleBracketed(args) = &last_segment.arguments {
if let Some(syn::GenericArgument::Type(inner_type)) = args.args.first() {
return Some(inner_type);
}
}
}
}
if let syn::Type::Path(type_path) = ty
&& let Some(last_segment) = type_path.path.segments.last()
&& last_segment.ident == "Json"
&& let syn::PathArguments::AngleBracketed(args) = &last_segment.arguments
&& let Some(syn::GenericArgument::Type(inner_type)) = args.args.first()
{
return Some(inner_type);
}
None
}
@ -286,13 +283,13 @@ pub fn tool(attr: TokenStream, input: TokenStream) -> syn::Result<TokenStream> {
let omit_send = cfg!(feature = "local") || attribute.local;
let new_output = syn::parse2::<ReturnType>({
let mut lt = quote! { 'static };
if let Some(receiver) = fn_item.sig.receiver() {
if let Some((_, receiver_lt)) = receiver.reference.as_ref() {
if let Some(receiver_lt) = receiver_lt {
lt = quote! { #receiver_lt };
} else {
lt = quote! { '_ };
}
if let Some(receiver) = fn_item.sig.receiver()
&& let Some((_, receiver_lt)) = receiver.reference.as_ref()
{
if let Some(receiver_lt) = receiver_lt {
lt = quote! { #receiver_lt };
} else {
lt = quote! { '_ };
}
}
match &fn_item.sig.output {

1
crates/rmcp/Cargo.toml
View file

@ -3,6 +3,7 @@ name = "rmcp"
license = { workspace = true }
version = { workspace = true }
edition = { workspace = true }
rust-version = { workspace = true }
repository = { workspace = true }
homepage = { workspace = true }
readme = { workspace = true }

8
crates/rmcp/src/handler/server/router/tool.rs
View file

@ -550,10 +550,10 @@ where
}
fn notify_if_visible(&self, name: &str) {
if self.map.contains_key(name) {
if let Some(notifier) = &self.notifier {
notifier();
}
if self.map.contains_key(name)
&& let Some(notifier) = &self.notifier
{
notifier();
}
}

18
crates/rmcp/src/model/elicitation_schema.rs
View file

@ -983,17 +983,15 @@ impl EnumSchemaBuilder<MultiSelect> {
return Err("One of the provided default values is not in enum values".to_string());
}
}
if let Some(min) = self.min_items {
if (default_values.len() as u64) < min {
return Err("Number of provided default values is less than min_items".to_string());
}
if let Some(min) = self.min_items
&& (default_values.len() as u64) < min
{
return Err("Number of provided default values is less than min_items".to_string());
}
if let Some(max) = self.max_items {
if (default_values.len() as u64) > max {
return Err(
"Number of provided default values is greater than max_items".to_string(),
);
}
if let Some(max) = self.max_items
&& (default_values.len() as u64) > max
{
return Err("Number of provided default values is greater than max_items".to_string());
}
self.default = default_values;
Ok(self)

18
crates/rmcp/src/service.rs
View file

@ -554,11 +554,10 @@ impl<R: ServiceRole> RequestHandle<R> {
None => None,
}
}, if reset_timeout_on_progress && idle_sleep.is_some() && self.progress_reset_rx.is_some() => {
if progress.is_some() {
if let Some((timeout, sleep)) = idle_sleep.as_mut() {
if progress.is_some()
&& let Some((timeout, sleep)) = idle_sleep.as_mut() {
sleep.as_mut().reset(tokio::time::Instant::now() + *timeout);
}
}
}
}
}
@ -1351,11 +1350,10 @@ where
tracing::trace!(?evt, "new event");
match evt {
Event::SendTaskResult(SendTaskResult::Request { id, result }) => {
if let Err(e) = result {
if let Some(responder) = local_responder_pool.remove(&id) {
if let Err(e) = result
&& let Some(responder) = local_responder_pool.remove(&id) {
let _ = responder.send(Err(ServiceError::TransportSend(e)));
}
}
}
Event::SendTaskResult(SendTaskResult::Notification {
responder,
@ -1368,16 +1366,14 @@ where
Ok(())
};
let _ = responder.send(response);
if let Some(param) = cancellation_param {
if let Some(request_id) = &param.request_id {
if let Some(responder) = local_responder_pool.remove(request_id) {
if let Some(param) = cancellation_param
&& let Some(request_id) = &param.request_id
&& let Some(responder) = local_responder_pool.remove(request_id) {
tracing::info!(id = %request_id, reason = param.reason, "cancelled");
let _response_result = responder.send(Err(ServiceError::Cancelled {
reason: param.reason.clone(),
}));
}
}
}
}
Event::ResponseSendTaskResult(result) => {
if let Err(error) = result {

12
crates/rmcp/src/transport/async_rw.rs
View file

@ -320,14 +320,12 @@ fn try_parse_with_compatibility<T: serde::de::DeserializeOwned>(
Ok(item) => Ok(Some(item)),
Err(e) => {
// Check if this is a notification that should be ignored for compatibility
if let Ok(json_value) = serde_json::from_str::<serde_json::Value>(line_str) {
if let Some(method) =
if let Ok(json_value) = serde_json::from_str::<serde_json::Value>(line_str)
&& let Some(method) =
json_value.get("method").and_then(serde_json::Value::as_str)
{
if should_ignore_notification(&json_value, method) {
return Ok(None);
}
}
&& should_ignore_notification(&json_value, method)
{
return Ok(None);
}
tracing::debug!(

140
crates/rmcp/src/transport/auth.rs
View file

@ -1248,56 +1248,56 @@ impl AuthorizationManager {
/// the client if credentials are found. Returns `false` when credentials
/// are absent or discarded after an authorization-server change.
pub async fn initialize_from_store(&mut self) -> Result<bool, AuthError> {
if let Some(stored) = self.credential_store.load().await? {
if stored.token_response.is_some() {
if self.metadata.is_none() {
let metadata = self.discover_metadata().await?;
self.metadata = Some(metadata);
}
if let (Some(stored_issuer), Some(current_issuer)) =
(stored.issuer.as_deref(), self.metadata_issuer().as_deref())
{
// A CIMD client ID is the client's metadata URL, so it is
// portable across authorization servers and exempt here.
if stored_issuer != current_issuer {
if is_https_url(&stored.client_id) {
// A CIMD client ID is the client's metadata URL, so it is
// portable across authorization servers — but the tokens
// were minted by the previous AS and must not be reused.
tracing::warn!(
stored_issuer,
current_issuer,
"authorization server issuer changed; discarding tokens but keeping portable CIMD client ID"
);
self.credential_store
.save(
StoredCredentials::new(
stored.client_id.clone(),
None,
vec![],
None,
)
.with_issuer(self.metadata_issuer()),
)
.await?;
self.configure_client_id(&stored.client_id)?;
return Ok(false);
}
if let Some(stored) = self.credential_store.load().await?
&& stored.token_response.is_some()
{
if self.metadata.is_none() {
let metadata = self.discover_metadata().await?;
self.metadata = Some(metadata);
}
if let (Some(stored_issuer), Some(current_issuer)) =
(stored.issuer.as_deref(), self.metadata_issuer().as_deref())
{
// A CIMD client ID is the client's metadata URL, so it is
// portable across authorization servers and exempt here.
if stored_issuer != current_issuer {
if is_https_url(&stored.client_id) {
// A CIMD client ID is the client's metadata URL, so it is
// portable across authorization servers — but the tokens
// were minted by the previous AS and must not be reused.
tracing::warn!(
stored_issuer,
current_issuer,
"authorization server issuer changed; clearing stored credentials bound to the previous issuer"
"authorization server issuer changed; discarding tokens but keeping portable CIMD client ID"
);
self.credential_store.clear().await?;
self.credential_store
.save(
StoredCredentials::new(
stored.client_id.clone(),
None,
vec![],
None,
)
.with_issuer(self.metadata_issuer()),
)
.await?;
self.configure_client_id(&stored.client_id)?;
return Ok(false);
}
}
self.configure_client_id(&stored.client_id)?;
return Ok(true);
tracing::warn!(
stored_issuer,
current_issuer,
"authorization server issuer changed; clearing stored credentials bound to the previous issuer"
);
self.credential_store.clear().await?;
return Ok(false);
}
}
self.configure_client_id(&stored.client_id)?;
return Ok(true);
}
Ok(false)
}
@ -1425,10 +1425,10 @@ impl AuthorizationManager {
// RFC 8414 RECOMMENDS response_types_supported in the metadata. This field is optional,
// but if present and does not include the flow we use ("code"), bail out early with a clear error.
if let Some(response_types_supported) = metadata.response_types_supported.as_ref() {
if !response_types_supported.contains(&response_type.to_string()) {
return Err(AuthError::InvalidScope(response_type.to_string()));
}
if let Some(response_types_supported) = metadata.response_types_supported.as_ref()
&& !response_types_supported.contains(&response_type.to_string())
{
return Err(AuthError::InvalidScope(response_type.to_string()));
}
// The client always sends an S256 challenge. A server that advertises
@ -1713,12 +1713,11 @@ impl AuthorizationManager {
}
// nothing requested or challenged yet: seed from AS metadata, then caller defaults
if let Some(metadata) = &self.metadata {
if let Some(scopes_supported) = &metadata.scopes_supported {
if !scopes_supported.is_empty() {
return scopes_supported.clone();
}
}
if let Some(metadata) = &self.metadata
&& let Some(scopes_supported) = &metadata.scopes_supported
&& !scopes_supported.is_empty()
{
return scopes_supported.clone();
}
default_scopes.iter().map(|s| s.to_string()).collect()
@ -1730,12 +1729,11 @@ impl AuthorizationManager {
if scopes.is_empty() || scopes.iter().any(|s| s == "offline_access") {
return;
}
if let Some(metadata) = &self.metadata {
if let Some(supported) = &metadata.scopes_supported {
if supported.iter().any(|s| s == "offline_access") {
scopes.push("offline_access".to_string());
}
}
if let Some(metadata) = &self.metadata
&& let Some(supported) = &metadata.scopes_supported
&& supported.iter().any(|s| s == "offline_access")
{
scopes.push("offline_access".to_string());
}
}
@ -2255,10 +2253,10 @@ impl AuthorizationManager {
self.validate_resource_metadata_resource(&resource_metadata)?;
// store scopes_supported from protected resource metadata for select_scopes()
if let Some(scopes) = resource_metadata.scopes_supported {
if !scopes.is_empty() {
*self.resource_scopes.write().await = scopes;
}
if let Some(scopes) = resource_metadata.scopes_supported
&& !scopes.is_empty()
{
*self.resource_scopes.write().await = scopes;
}
let mut candidates = Vec::new();
@ -2701,20 +2699,18 @@ impl AuthorizationManager {
if let ClientCredentialsConfig::PrivateKeyJwt {
signing_algorithm, ..
} = config
{
if let Some(algs) = metadata
&& let Some(algs) = metadata
.additional_fields
.get("token_endpoint_auth_signing_alg_values_supported")
.and_then(|v| v.as_array())
{
let alg_str = signing_algorithm.as_str();
if !algs.iter().any(|a| a.as_str() == Some(alg_str)) {
let supported: Vec<&str> = algs.iter().filter_map(|a| a.as_str()).collect();
return Err(AuthError::ClientCredentialsError(format!(
"Authorization server does not support signing algorithm '{}'. Supported: {:?}",
alg_str, supported
)));
}
{
let alg_str = signing_algorithm.as_str();
if !algs.iter().any(|a| a.as_str() == Some(alg_str)) {
let supported: Vec<&str> = algs.iter().filter_map(|a| a.as_str()).collect();
return Err(AuthError::ClientCredentialsError(format!(
"Authorization server does not support signing algorithm '{}'. Supported: {:?}",
alg_str, supported
)));
}
}

16
crates/rmcp/src/transport/common/client_side_sse.rs
View file

@ -178,10 +178,10 @@ pub struct FixedInterval {
impl SseRetryPolicy for FixedInterval {
fn retry(&self, current_times: usize) -> Option<Duration> {
if let Some(max_times) = self.max_times {
if current_times >= max_times {
return None;
}
if let Some(max_times) = self.max_times
&& current_times >= max_times
{
return None;
}
Some(self.duration)
}
@ -222,10 +222,10 @@ impl Default for ExponentialBackoff {
impl SseRetryPolicy for ExponentialBackoff {
fn retry(&self, current_times: usize) -> Option<Duration> {
if let Some(max_times) = self.max_times {
if current_times >= max_times {
return None;
}
if let Some(max_times) = self.max_times
&& current_times >= max_times
{
return None;
}
Some(self.base_duration * (2u32.pow(current_times as u32)))
}

80
crates/rmcp/src/transport/common/mcp_headers.rs
View file

@ -104,10 +104,10 @@ fn param_header_annotations(input_schema: &JsonObject) -> Vec<(String, String)>
let mut out = Vec::new();
if let Some(Value::Object(props)) = input_schema.get("properties") {
for (prop, schema) in props {
if let Some(Value::String(header)) = schema.get("x-mcp-header") {
if !header.is_empty() {
out.push((prop.clone(), header.clone()));
}
if let Some(Value::String(header)) = schema.get("x-mcp-header")
&& !header.is_empty()
{
out.push((prop.clone(), header.clone()));
}
}
}
@ -236,20 +236,18 @@ pub(crate) fn standard_request_headers(
push(HEADER_MCP_NAME, &encode_header_value(&name));
}
if method == "tools/call" {
if let (Some(schema), Some(arguments)) =
if method == "tools/call"
&& let (Some(schema), Some(arguments)) =
(tool_schema, params.and_then(|p| p.get("arguments")))
{
for (prop, header) in param_header_annotations(schema) {
let Some(arg) = arguments.get(&prop) else {
continue;
};
let Some(encoded) = primitive_to_string(arg).map(|s| encode_header_value(&s))
else {
continue;
};
push(&format!("{HEADER_MCP_PARAM_PREFIX}{header}"), &encoded);
}
{
for (prop, header) in param_header_annotations(schema) {
let Some(arg) = arguments.get(&prop) else {
continue;
};
let Some(encoded) = primitive_to_string(arg).map(|s| encode_header_value(&s)) else {
continue;
};
push(&format!("{HEADER_MCP_PARAM_PREFIX}{header}"), &encoded);
}
}
out
@ -296,34 +294,34 @@ pub(crate) fn validate_request_headers(
}
}
if method == "tools/call" {
if let Some(schema) = tool_schema {
let arguments = params.and_then(|p| p.get("arguments"));
for (prop, header) in param_header_annotations(schema) {
let full = format!("{HEADER_MCP_PARAM_PREFIX}{header}");
let header_value = header_str(headers, &full);
let arg = arguments.and_then(|a| a.get(&prop));
let body_value = arg.filter(|v| !v.is_null()).and_then(primitive_to_string);
if method == "tools/call"
&& let Some(schema) = tool_schema
{
let arguments = params.and_then(|p| p.get("arguments"));
for (prop, header) in param_header_annotations(schema) {
let full = format!("{HEADER_MCP_PARAM_PREFIX}{header}");
let header_value = header_str(headers, &full);
let arg = arguments.and_then(|a| a.get(&prop));
let body_value = arg.filter(|v| !v.is_null()).and_then(primitive_to_string);
match (header_value, body_value) {
(None, None) => {}
(Some(_), None) => {
match (header_value, body_value) {
(None, None) => {}
(Some(_), None) => {
return Err(format!(
"unexpected {full} header for absent or null `{prop}`"
));
}
(None, Some(_)) => {
return Err(format!("missing {full} header for `{prop}`"));
}
(Some(raw), Some(expected)) => {
let decoded = decode_header_value(raw)
.ok_or_else(|| format!("{full} header is not valid Base64"))?;
if decoded != expected {
return Err(format!(
"unexpected {full} header for absent or null `{prop}`"
"{full} header `{decoded}` does not match body value `{expected}`"
));
}
(None, Some(_)) => {
return Err(format!("missing {full} header for `{prop}`"));
}
(Some(raw), Some(expected)) => {
let decoded = decode_header_value(raw)
.ok_or_else(|| format!("{full} header is not valid Base64"))?;
if decoded != expected {
return Err(format!(
"{full} header `{decoded}` does not match body value `{expected}`"
));
}
}
}
}
}

54
crates/rmcp/src/transport/common/reqwest/streamable_http_client.rs
View file

@ -178,36 +178,36 @@ impl StreamableHttpClient for reqwest::Client {
request = request.header(HEADER_SESSION_ID, session_id.as_ref());
}
let response = request.json(&message).send().await?;
if response.status() == reqwest::StatusCode::UNAUTHORIZED {
if let Some(header) = response.headers().get(WWW_AUTHENTICATE) {
let header = header
.to_str()
.map_err(|_| {
StreamableHttpError::UnexpectedServerResponse(Cow::from(
"invalid www-authenticate header value",
))
})?
.to_string();
return Err(StreamableHttpError::AuthRequired(AuthRequiredError {
www_authenticate_header: header,
}));
}
}
if response.status() == reqwest::StatusCode::FORBIDDEN {
if let Some(header) = response.headers().get(WWW_AUTHENTICATE) {
let header_str = header.to_str().map_err(|_| {
if response.status() == reqwest::StatusCode::UNAUTHORIZED
&& let Some(header) = response.headers().get(WWW_AUTHENTICATE)
{
let header = header
.to_str()
.map_err(|_| {
StreamableHttpError::UnexpectedServerResponse(Cow::from(
"invalid www-authenticate header value",
))
})?;
let scope = extract_scope_from_header(header_str);
return Err(StreamableHttpError::InsufficientScope(
InsufficientScopeError {
www_authenticate_header: header_str.to_string(),
required_scope: scope,
},
));
}
})?
.to_string();
return Err(StreamableHttpError::AuthRequired(AuthRequiredError {
www_authenticate_header: header,
}));
}
if response.status() == reqwest::StatusCode::FORBIDDEN
&& let Some(header) = response.headers().get(WWW_AUTHENTICATE)
{
let header_str = header.to_str().map_err(|_| {
StreamableHttpError::UnexpectedServerResponse(Cow::from(
"invalid www-authenticate header value",
))
})?;
let scope = extract_scope_from_header(header_str);
return Err(StreamableHttpError::InsufficientScope(
InsufficientScopeError {
www_authenticate_header: header_str.to_string(),
required_scope: scope,
},
));
}
let status = response.status();
if matches!(

112
crates/rmcp/src/transport/common/unix_socket.rs
View file

@ -226,37 +226,37 @@ impl StreamableHttpClient for UnixSocketHttpClient {
let status = response.status();
if status == StatusCode::UNAUTHORIZED {
if let Some(header) = response.headers().get(WWW_AUTHENTICATE) {
let www_authenticate_header = header
.to_str()
.map_err(|_| {
StreamableHttpError::UnexpectedServerResponse(Cow::from(
"invalid www-authenticate header value",
))
})?
.to_string();
return Err(StreamableHttpError::AuthRequired(AuthRequiredError {
www_authenticate_header,
}));
}
}
if status == StatusCode::FORBIDDEN {
if let Some(header) = response.headers().get(WWW_AUTHENTICATE) {
let header_str = header.to_str().map_err(|_| {
if status == StatusCode::UNAUTHORIZED
&& let Some(header) = response.headers().get(WWW_AUTHENTICATE)
{
let www_authenticate_header = header
.to_str()
.map_err(|_| {
StreamableHttpError::UnexpectedServerResponse(Cow::from(
"invalid www-authenticate header value",
))
})?;
let scope = extract_scope_from_header(header_str);
return Err(StreamableHttpError::InsufficientScope(
InsufficientScopeError {
www_authenticate_header: header_str.to_string(),
required_scope: scope,
},
));
}
})?
.to_string();
return Err(StreamableHttpError::AuthRequired(AuthRequiredError {
www_authenticate_header,
}));
}
if status == StatusCode::FORBIDDEN
&& let Some(header) = response.headers().get(WWW_AUTHENTICATE)
{
let header_str = header.to_str().map_err(|_| {
StreamableHttpError::UnexpectedServerResponse(Cow::from(
"invalid www-authenticate header value",
))
})?;
let scope = extract_scope_from_header(header_str);
return Err(StreamableHttpError::InsufficientScope(
InsufficientScopeError {
www_authenticate_header: header_str.to_string(),
required_scope: scope,
},
));
}
if matches!(status, StatusCode::ACCEPTED | StatusCode::NO_CONTENT) {
@ -438,37 +438,37 @@ impl StreamableHttpClient for UnixSocketHttpClient {
return Err(StreamableHttpError::ServerDoesNotSupportSse);
}
if response.status() == StatusCode::UNAUTHORIZED {
if let Some(header) = response.headers().get(WWW_AUTHENTICATE) {
let www_authenticate_header = header
.to_str()
.map_err(|_| {
StreamableHttpError::UnexpectedServerResponse(Cow::from(
"invalid www-authenticate header value",
))
})?
.to_string();
return Err(StreamableHttpError::AuthRequired(AuthRequiredError {
www_authenticate_header,
}));
}
}
if response.status() == StatusCode::FORBIDDEN {
if let Some(header) = response.headers().get(WWW_AUTHENTICATE) {
let header_str = header.to_str().map_err(|_| {
if response.status() == StatusCode::UNAUTHORIZED
&& let Some(header) = response.headers().get(WWW_AUTHENTICATE)
{
let www_authenticate_header = header
.to_str()
.map_err(|_| {
StreamableHttpError::UnexpectedServerResponse(Cow::from(
"invalid www-authenticate header value",
))
})?;
let scope = extract_scope_from_header(header_str);
return Err(StreamableHttpError::InsufficientScope(
InsufficientScopeError {
www_authenticate_header: header_str.to_string(),
required_scope: scope,
},
));
}
})?
.to_string();
return Err(StreamableHttpError::AuthRequired(AuthRequiredError {
www_authenticate_header,
}));
}
if response.status() == StatusCode::FORBIDDEN
&& let Some(header) = response.headers().get(WWW_AUTHENTICATE)
{
let header_str = header.to_str().map_err(|_| {
StreamableHttpError::UnexpectedServerResponse(Cow::from(
"invalid www-authenticate header value",
))
})?;
let scope = extract_scope_from_header(header_str);
return Err(StreamableHttpError::InsufficientScope(
InsufficientScopeError {
www_authenticate_header: header_str.to_string(),
required_scope: scope,
},
));
}
if !response.status().is_success() {

50
crates/rmcp/src/transport/streamable_http_client.rs
View file

@ -42,20 +42,20 @@ fn build_request_headers(
use serde_json::Value;
let mut headers = base.clone();
if *version >= ProtocolVersion::STANDARD_HEADERS {
if let Ok(value) = serde_json::to_value(message) {
let schema = value
.get("method")
.and_then(Value::as_str)
.filter(|method| *method == "tools/call")
.and_then(|_| value.get("params"))
.and_then(|params| params.get("name"))
.and_then(Value::as_str)
.and_then(|name| tool_cache.get(name))
.map(Arc::as_ref);
for (name, val) in mcp_headers::standard_request_headers(&value, schema) {
headers.insert(name, val);
}
if *version >= ProtocolVersion::STANDARD_HEADERS
&& let Ok(value) = serde_json::to_value(message)
{
let schema = value
.get("method")
.and_then(Value::as_str)
.filter(|method| *method == "tools/call")
.and_then(|_| value.get("params"))
.and_then(|params| params.get("name"))
.and_then(Value::as_str)
.and_then(|name| tool_cache.get(name))
.map(Arc::as_ref);
for (name, val) in mcp_headers::standard_request_headers(&value, schema) {
headers.insert(name, val);
}
}
headers
@ -90,9 +90,10 @@ fn cache_tools_from_response(
if protocol_version < &ProtocolVersion::STANDARD_HEADERS {
return;
}
if let ServerJsonRpcMessage::Response(response) = message {
if let ServerResult::ListToolsResult(list) = &mut response.result {
list.tools.retain(|tool| {
if let ServerJsonRpcMessage::Response(response) = message
&& let ServerResult::ListToolsResult(list) = &mut response.result
{
list.tools.retain(|tool| {
let Err(reason) =
mcp_headers::validate_param_header_annotations(&tool.input_schema)
else {
@ -102,7 +103,6 @@ fn cache_tools_from_response(
tracing::warn!(tool = %tool.name, "rejecting invalid x-mcp-header annotations: {reason}");
false
});
}
}
}
@ -112,13 +112,13 @@ fn negotiate_version_headers(
) -> (ProtocolVersion, HashMap<HeaderName, HeaderValue>) {
let mut version = ProtocolVersion::default();
let mut headers = base;
if let ServerJsonRpcMessage::Response(response) = init_response {
if let ServerResult::InitializeResult(init_result) = &response.result {
version = init_result.protocol_version.clone();
// HeaderName::from_static requires lowercase
if let Ok(hv) = HeaderValue::from_str(init_result.protocol_version.as_str()) {
headers.insert(HeaderName::from_static("mcp-protocol-version"), hv);
}
if let ServerJsonRpcMessage::Response(response) = init_response
&& let ServerResult::InitializeResult(init_result) = &response.result
{
version = init_result.protocol_version.clone();
// HeaderName::from_static requires lowercase
if let Ok(hv) = HeaderValue::from_str(init_result.protocol_version.as_str()) {
headers.insert(HeaderName::from_static("mcp-protocol-version"), hv);
}
}
(version, headers)

18
crates/rmcp/src/transport/streamable_http_server/session/local.rs
View file

@ -479,11 +479,11 @@ impl LocalSessionWorker {
&mut self,
notification: &JsonRpcNotification<ClientNotification>,
) {
if let ClientNotification::CancelledNotification(n) = &notification.notification {
if let Some(request_id) = n.params.request_id.clone() {
let resource = ResourceKey::McpRequestId(request_id);
self.unregister_resource(&resource);
}
if let ClientNotification::CancelledNotification(n) = &notification.notification
&& let Some(request_id) = n.params.request_id.clone()
{
let resource = ResourceKey::McpRequestId(request_id);
self.unregister_resource(&resource);
}
}
fn evict_expired_channels(&mut self) {
@ -638,11 +638,9 @@ impl LocalSessionWorker {
OutboundChannel::RequestWise { id, close } => {
if let Some(request_wise) = self.tx_router.get_mut(&id) {
request_wise.tx.send(message).await?;
if close {
if let Some(channel) = self.tx_router.remove(&id) {
for resource in channel.resources {
self.resource_router.remove(&resource);
}
if close && let Some(channel) = self.tx_router.remove(&id) {
for resource in channel.resources {
self.resource_router.remove(&resource);
}
}
} else {

14
crates/rmcp/tests/test_mrtr_behavior.rs
View file

@ -238,13 +238,13 @@ impl ClientHandler for MrtrClient {
request: ElicitRequestParams,
_context: RequestContext<RoleClient>,
) -> Result<ElicitResult, ErrorData> {
if let ElicitRequestParams::FormElicitationParams { message, .. } = &request {
if message == "FAIL" {
return Err(ErrorData::internal_error(
"elicitation handler failed",
None,
));
}
if let ElicitRequestParams::FormElicitationParams { message, .. } = &request
&& message == "FAIL"
{
return Err(ErrorData::internal_error(
"elicitation handler failed",
None,
));
}
Ok(ElicitResult::new(ElicitationAction::Accept).with_content(json!({ "name": "Ferris" })))
}

Some files were not shown because too many files have changed in this diff Show more