fix: accept 200 with empty body in response to notifications in addition to 202 (#849)
This commit is contained in:
parent
3529c3675f
commit
cc66e3091e
3 changed files with 79 additions and 0 deletions
|
|
@ -178,11 +178,25 @@ impl StreamableHttpClient for reqwest::Client {
|
||||||
.headers()
|
.headers()
|
||||||
.get(reqwest::header::CONTENT_TYPE)
|
.get(reqwest::header::CONTENT_TYPE)
|
||||||
.map(|ct| String::from_utf8_lossy(ct.as_bytes()).to_string());
|
.map(|ct| String::from_utf8_lossy(ct.as_bytes()).to_string());
|
||||||
|
let content_length = response.content_length();
|
||||||
let session_id = response
|
let session_id = response
|
||||||
.headers()
|
.headers()
|
||||||
.get(HEADER_SESSION_ID)
|
.get(HEADER_SESSION_ID)
|
||||||
.and_then(|v| v.to_str().ok())
|
.and_then(|v| v.to_str().ok())
|
||||||
.map(|s| s.to_string());
|
.map(|s| s.to_string());
|
||||||
|
// Spec requires 202 Accepted for these, but some servers return an empty 200.
|
||||||
|
// Treat empty success responses as equivalent to Accepted.
|
||||||
|
if status.is_success()
|
||||||
|
&& content_length == Some(0)
|
||||||
|
&& matches!(
|
||||||
|
message,
|
||||||
|
ClientJsonRpcMessage::Notification(_)
|
||||||
|
| ClientJsonRpcMessage::Response(_)
|
||||||
|
| ClientJsonRpcMessage::Error(_)
|
||||||
|
)
|
||||||
|
{
|
||||||
|
return Ok(StreamableHttpPostResponse::Accepted);
|
||||||
|
}
|
||||||
// Non-success responses may carry valid JSON-RPC error payloads that
|
// Non-success responses may carry valid JSON-RPC error payloads that
|
||||||
// should be surfaced as McpError rather than lost in TransportSend.
|
// should be surfaced as McpError rather than lost in TransportSend.
|
||||||
if !status.is_success() {
|
if !status.is_success() {
|
||||||
|
|
|
||||||
|
|
@ -257,12 +257,29 @@ impl StreamableHttpClient for UnixSocketHttpClient {
|
||||||
}
|
}
|
||||||
|
|
||||||
let content_type = response.headers().get(http::header::CONTENT_TYPE).cloned();
|
let content_type = response.headers().get(http::header::CONTENT_TYPE).cloned();
|
||||||
|
let content_length = response
|
||||||
|
.headers()
|
||||||
|
.get(http::header::CONTENT_LENGTH)
|
||||||
|
.and_then(|v| v.to_str().ok())
|
||||||
|
.and_then(|v| v.parse::<u64>().ok());
|
||||||
let session_id = response
|
let session_id = response
|
||||||
.headers()
|
.headers()
|
||||||
.get(HEADER_SESSION_ID)
|
.get(HEADER_SESSION_ID)
|
||||||
.and_then(|v| v.to_str().ok())
|
.and_then(|v| v.to_str().ok())
|
||||||
.map(|s| s.to_string());
|
.map(|s| s.to_string());
|
||||||
|
|
||||||
|
if status.is_success()
|
||||||
|
&& content_length == Some(0)
|
||||||
|
&& matches!(
|
||||||
|
message,
|
||||||
|
ClientJsonRpcMessage::Notification(_)
|
||||||
|
| ClientJsonRpcMessage::Response(_)
|
||||||
|
| ClientJsonRpcMessage::Error(_)
|
||||||
|
)
|
||||||
|
{
|
||||||
|
return Ok(StreamableHttpPostResponse::Accepted);
|
||||||
|
}
|
||||||
|
|
||||||
match content_type {
|
match content_type {
|
||||||
Some(ref ct) if ct.as_bytes().starts_with(EVENT_STREAM_MIME_TYPE.as_bytes()) => {
|
Some(ref ct) if ct.as_bytes().starts_with(EVENT_STREAM_MIME_TYPE.as_bytes()) => {
|
||||||
let sse_stream = SseStream::new(response.into_body()).boxed();
|
let sse_stream = SseStream::new(response.into_body()).boxed();
|
||||||
|
|
|
||||||
|
|
@ -0,0 +1,48 @@
|
||||||
|
#![cfg(all(
|
||||||
|
feature = "transport-streamable-http-client",
|
||||||
|
feature = "transport-streamable-http-client-reqwest",
|
||||||
|
not(feature = "local")
|
||||||
|
))]
|
||||||
|
|
||||||
|
use std::{collections::HashMap, sync::Arc};
|
||||||
|
|
||||||
|
use rmcp::{
|
||||||
|
model::{ClientJsonRpcMessage, ClientNotification, InitializedNotification},
|
||||||
|
transport::streamable_http_client::{StreamableHttpClient, StreamableHttpPostResponse},
|
||||||
|
};
|
||||||
|
|
||||||
|
async fn spawn_empty_ok_server() -> String {
|
||||||
|
use axum::{Router, http::StatusCode, routing::post};
|
||||||
|
|
||||||
|
let router = Router::new().route("/mcp", post(|| async { StatusCode::OK }));
|
||||||
|
|
||||||
|
let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
|
||||||
|
let addr = listener.local_addr().unwrap();
|
||||||
|
tokio::spawn(async move {
|
||||||
|
axum::serve(listener, router).await.unwrap();
|
||||||
|
});
|
||||||
|
|
||||||
|
format!("http://{addr}/mcp")
|
||||||
|
}
|
||||||
|
|
||||||
|
#[tokio::test]
|
||||||
|
async fn empty_success_response_to_notification_is_accepted() {
|
||||||
|
let url = spawn_empty_ok_server().await;
|
||||||
|
let client = reqwest::Client::new();
|
||||||
|
let result = client
|
||||||
|
.post_message(
|
||||||
|
Arc::from(url.as_str()),
|
||||||
|
ClientJsonRpcMessage::notification(ClientNotification::InitializedNotification(
|
||||||
|
InitializedNotification::default(),
|
||||||
|
)),
|
||||||
|
None,
|
||||||
|
None,
|
||||||
|
HashMap::new(),
|
||||||
|
)
|
||||||
|
.await;
|
||||||
|
|
||||||
|
match result {
|
||||||
|
Ok(StreamableHttpPostResponse::Accepted) => {}
|
||||||
|
other => panic!("expected Accepted, got: {other:?}"),
|
||||||
|
}
|
||||||
|
}
|
||||||
Loading…
Reference in a new issue