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

fix: preserve negotiated progress responses (#1005)

Browse source
This commit is contained in:
Dale Seo 2026-07-20 16:20:12 -04:00 • committed by GitHub
commit 5b053416b8
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
2 changed files with 156 additions and 15 deletions
Download patch file Download diff file Expand all files Collapse all files

8
crates/rmcp/src/transport/streamable_http_server/tower.rs
View file

@ -947,7 +947,13 @@ where
))
})?;
if self.config.json_response || jsonrpc_http_status(&first) != http::StatusCode::OK {
let terminal = matches!(
&first,
ServerJsonRpcMessage::Response(_) | ServerJsonRpcMessage::Error(_)
);
if terminal
&& (self.config.json_response || jsonrpc_http_status(&first) != http::StatusCode::OK)
{
// This message is the whole reply, so `receiver` is dropped here and
// anything the handler emits afterwards is undeliverable. Cancel it so
// a still-running handler stops instead of running on unobserved: its

163
crates/rmcp/tests/test_streamable_http_json_response.rs
View file

@ -17,6 +17,41 @@ use common::calculator::Calculator;
const INIT_BODY: &str = r#"{"jsonrpc":"2.0","id":1,"method":"initialize","params":{"protocolVersion":"2025-03-26","capabilities":{},"clientInfo":{"name":"test","version":"1.0"}}}"#;
const CALL_WITH_PROGRESS_BODY: &str = r#"{"jsonrpc":"2.0","id":2,"method":"tools/call","params":{"name":"progress","arguments":{},"_meta":{"progressToken":"progress-test-1"}}}"#;
const NEGOTIATED_CALL_BODY: &str = r#"{
"jsonrpc": "2.0",
"id": 2,
"method": "tools/call",
"params": {
"name": "terminal",
"arguments": {},
"_meta": {
"io.modelcontextprotocol/protocolVersion": "2026-07-28",
"io.modelcontextprotocol/clientInfo": {
"name": "test",
"version": "1.0"
},
"io.modelcontextprotocol/clientCapabilities": {}
}
}
}"#;
const NEGOTIATED_CALL_WITH_PROGRESS_BODY: &str = r#"{
"jsonrpc": "2.0",
"id": 2,
"method": "tools/call",
"params": {
"name": "progress",
"arguments": {},
"_meta": {
"progressToken": "progress-test-1",
"io.modelcontextprotocol/protocolVersion": "2026-07-28",
"io.modelcontextprotocol/clientInfo": {
"name": "test",
"version": "1.0"
},
"io.modelcontextprotocol/clientCapabilities": {}
}
}
}"#;
#[derive(Clone)]
struct ProgressServer;
@ -28,22 +63,24 @@ impl ServerHandler for ProgressServer {
async fn call_tool(
&self,
_request: CallToolRequestParams,
request: CallToolRequestParams,
context: RequestContext<rmcp::RoleServer>,
) -> Result<CallToolResponse, ErrorData> {
let progress_token = context
.meta
.get_progress_token()
.expect("request includes progressToken");
context
.peer
.notify_progress(
ProgressNotificationParam::new(progress_token, 50.0)
.with_total(100.0)
.with_message("working"),
)
.await
.expect("progress notification is delivered");
if request.name == "progress" {
let progress_token = context
.meta
.get_progress_token()
.expect("request includes progressToken");
context
.peer
.notify_progress(
ProgressNotificationParam::new(progress_token, 50.0)
.with_total(100.0)
.with_message("working"),
)
.await
.expect("progress notification is delivered");
}
Ok(CallToolResult::success(vec![ContentBlock::text("done")]).into())
}
}
@ -140,6 +177,49 @@ async fn stateless_json_response_returns_application_json() -> anyhow::Result<()
Ok(())
}
#[tokio::test]
async fn stateless_negotiated_terminal_response_returns_application_json() -> anyhow::Result<()> {
let ct = CancellationToken::new();
let (client, url, ct) = spawn_progress_server(
StreamableHttpServerConfig::default()
.with_stateful_mode(false)
.with_json_response(true)
.with_sse_keep_alive(None)
.with_cancellation_token(ct.child_token()),
)
.await;
let response = client
.post(&url)
.header("Content-Type", "application/json")
.header("Accept", "application/json, text/event-stream")
.header("MCP-Protocol-Version", "2026-07-28")
.header("Mcp-Method", "tools/call")
.header("Mcp-Name", "terminal")
.body(NEGOTIATED_CALL_BODY)
.send()
.await?;
assert_eq!(response.status(), 200);
let content_type = response
.headers()
.get("content-type")
.and_then(|value| value.to_str().ok())
.unwrap_or("");
assert!(
content_type.contains("application/json"),
"Expected application/json, got: {content_type}"
);
let body: serde_json::Value = response.json().await?;
assert_eq!(body["id"], 2);
assert!(body["result"].is_object(), "Expected result object");
ct.cancel();
Ok(())
}
#[tokio::test]
async fn stateless_json_response_falls_back_to_sse_for_progress() -> anyhow::Result<()> {
let ct = CancellationToken::new();
@ -192,6 +272,61 @@ async fn stateless_json_response_falls_back_to_sse_for_progress() -> anyhow::Res
Ok(())
}
#[tokio::test]
async fn stateless_negotiated_json_response_falls_back_to_sse_for_progress() -> anyhow::Result<()> {
let ct = CancellationToken::new();
let (client, url, ct) = spawn_progress_server(
StreamableHttpServerConfig::default()
.with_stateful_mode(false)
.with_json_response(true)
.with_sse_keep_alive(None)
.with_cancellation_token(ct.child_token()),
)
.await;
let response = client
.post(&url)
.header("Content-Type", "application/json")
.header("Accept", "application/json, text/event-stream")
.header("MCP-Protocol-Version", "2026-07-28")
.header("Mcp-Method", "tools/call")
.header("Mcp-Name", "progress")
.body(NEGOTIATED_CALL_WITH_PROGRESS_BODY)
.send()
.await?;
assert_eq!(response.status(), 200);
let content_type = response
.headers()
.get("content-type")
.and_then(|value| value.to_str().ok())
.unwrap_or("");
assert!(
content_type.contains("text/event-stream"),
"Expected SSE fallback, got: {content_type}"
);
let body = response.text().await?;
let messages: Vec<serde_json::Value> = body
.lines()
.filter_map(|line| line.strip_prefix("data:"))
.map(str::trim)
.filter(|data| !data.is_empty())
.map(serde_json::from_str)
.collect::<Result<_, _>>()?;
assert_eq!(messages.len(), 2, "Expected progress and result: {body}");
assert_eq!(messages[0]["method"], "notifications/progress");
assert_eq!(messages[1]["id"], 2);
assert!(
messages[1]["result"].is_object(),
"Expected result object: {body}"
);
ct.cancel();
Ok(())
}
#[tokio::test]
async fn stateless_sse_mode_default_unchanged() -> anyhow::Result<()> {
let ct = CancellationToken::new();