feat: mcp sdk conformance (#687)
* adds conformance server and client * adds results from initial run of https://github.com/modelcontextprotocol/conformance/tree/main/.claude/skills/mcp-sdk-tier-audit skill * various small changes applied during the testing loop Co-authored-by: Dale Seo <5466341+DaleSeo@users.noreply.github.com>
This commit is contained in:
parent
b967c132ae
commit
a7e4ae3203
4 changed files with 2392 additions and 10 deletions
|
|
@ -153,8 +153,7 @@ impl ProtocolVersion {
|
|||
pub const V_2025_06_18: Self = Self(Cow::Borrowed("2025-06-18"));
|
||||
pub const V_2025_03_26: Self = Self(Cow::Borrowed("2025-03-26"));
|
||||
pub const V_2024_11_05: Self = Self(Cow::Borrowed("2024-11-05"));
|
||||
// Keep LATEST at 2025-03-26 until full 2025-06-18 compliance and automated testing are in place.
|
||||
pub const LATEST: Self = Self::V_2025_03_26;
|
||||
pub const LATEST: Self = Self::V_2025_06_18;
|
||||
|
||||
/// All protocol versions known to this SDK.
|
||||
pub const KNOWN_VERSIONS: &[Self] =
|
||||
|
|
|
|||
|
|
@ -255,8 +255,23 @@ where
|
|||
}
|
||||
}
|
||||
None => {
|
||||
tracing::debug!("sse stream terminated");
|
||||
return Poll::Ready(None);
|
||||
// Per SEP-1699, a graceful stream close is
|
||||
// reconnectable. If the server sent a `retry` field
|
||||
// we MUST wait that long before reconnecting.
|
||||
let interval = this
|
||||
.server_retry_interval
|
||||
.take()
|
||||
.or_else(|| this.retry_policy.retry(0));
|
||||
if let Some(interval) = interval {
|
||||
tracing::debug!(?interval, "sse stream ended gracefully, reconnecting");
|
||||
SseAutoReconnectStreamState::WaitingNextRetry {
|
||||
sleep: tokio::time::sleep(interval),
|
||||
retry_times: 0,
|
||||
}
|
||||
} else {
|
||||
tracing::debug!("sse stream terminated, no reconnect policy");
|
||||
return Poll::Ready(None);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -197,8 +197,18 @@ impl StreamableHttpClient for reqwest::Client {
|
|||
Ok(StreamableHttpPostResponse::Sse(event_stream, session_id))
|
||||
}
|
||||
Some(ct) if ct.as_bytes().starts_with(JSON_MIME_TYPE.as_bytes()) => {
|
||||
let message: ServerJsonRpcMessage = response.json().await?;
|
||||
Ok(StreamableHttpPostResponse::Json(message, session_id))
|
||||
// Try to parse as a valid JSON-RPC message. If the body is
|
||||
// malformed (e.g. a 200 response to a notification that lacks
|
||||
// an `id` field), treat it as accepted rather than failing.
|
||||
match response.json::<ServerJsonRpcMessage>().await {
|
||||
Ok(message) => Ok(StreamableHttpPostResponse::Json(message, session_id)),
|
||||
Err(e) => {
|
||||
tracing::warn!(
|
||||
"could not parse JSON response as ServerJsonRpcMessage, treating as accepted: {e}"
|
||||
);
|
||||
Ok(StreamableHttpPostResponse::Accepted)
|
||||
}
|
||||
}
|
||||
}
|
||||
_ => {
|
||||
// unexpected content type
|
||||
|
|
|
|||
|
|
@ -154,14 +154,16 @@ impl StreamableHttpPostResponse {
|
|||
}
|
||||
}
|
||||
|
||||
pub fn expect_accepted<E>(self) -> Result<(), StreamableHttpError<E>>
|
||||
pub fn expect_accepted_or_json<E>(self) -> Result<(), StreamableHttpError<E>>
|
||||
where
|
||||
E: std::error::Error + Send + Sync + 'static,
|
||||
{
|
||||
match self {
|
||||
Self::Accepted => Ok(()),
|
||||
// Tolerate servers that return 200 with JSON for notifications
|
||||
Self::Json(..) => Ok(()),
|
||||
got => Err(StreamableHttpError::UnexpectedServerResponse(
|
||||
format!("expect accepted, got {got:?}").into(),
|
||||
format!("expect accepted or json, got {got:?}").into(),
|
||||
)),
|
||||
}
|
||||
}
|
||||
|
|
@ -410,7 +412,7 @@ impl<C: StreamableHttpClient> Worker for StreamableHttpClientWorker<C> {
|
|||
.map_err(WorkerQuitReason::fatal_context(
|
||||
"send initialized notification",
|
||||
))?
|
||||
.expect_accepted::<C::Error>()
|
||||
.expect_accepted_or_json::<C::Error>()
|
||||
.map_err(WorkerQuitReason::fatal_context(
|
||||
"process initialized notification response",
|
||||
))?;
|
||||
|
|
|
|||
Loading…
Reference in a new issue