Handle SSE data fields without spaces (#60575)

Fixes #60270.

## Summary

- Accept MCP HTTP SSE `data:` fields when the optional space after the
colon is omitted.
- Add a regression test covering `data:{...}` streamable HTTP responses.

## Tests

- `cargo fmt --check`
- `cargo test -p context_server
test_sse_data_field_without_space_after_colon -- --nocapture`
- `cargo test -p context_server -- --nocapture`

Release Notes:

- Fixed MCP HTTP context servers timing out when SSE data fields omit
the optional space after the colon.

---------

Co-authored-by: dino <dinojoaocosta@gmail.com>
This commit is contained in:
迷渡anddino authored and GitHub committed 2026-07-29 10:46:31 +00:00
1 parent f85349be9c
commit 007ffc790e
1 file changed
+52 -2
+52 -2
View File
@@ -300,7 +300,7 @@ impl HttpTransport {
data_buffer.clear();
}
in_message = false;
} else if let Some(data) = line.strip_prefix("data: ") {
} else if let Some(data) = line.strip_prefix("data:") {
// Handle data lines
let data = data.trim();
if !data.is_empty() {
@@ -412,9 +412,13 @@ impl Drop for HttpTransport {
mod tests {
use super::*;
use async_trait::async_trait;
use futures::FutureExt as _;
use gpui::TestAppContext;
use parking_lot::Mutex as SyncMutex;
use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
use std::{
sync::atomic::{AtomicBool, AtomicUsize, Ordering},
time::Duration,
};
/// A mock token provider that returns a configurable token and tracks
/// refresh attempts.
@@ -492,6 +496,52 @@ mod tests {
.unwrap())
}
#[gpui::test]
async fn test_sse_data_field(cx: &mut TestAppContext) {
for data_prefix in ["data:", "data: "] {
let body = format!(
"id:1\nevent:message\n{data_prefix}{{\"jsonrpc\":\"2.0\",\"id\":1,\"result\":{{}}}}\n\n"
);
let client = make_fake_http_client(move |_req| {
let body = body.clone();
Box::pin(async move {
Ok(Response::builder()
.status(200)
.header("Content-Type", "text/event-stream;charset=UTF-8")
.body(AsyncBody::from(body.into_bytes()))
.unwrap())
})
});
let transport = HttpTransport::new(
client,
"http://mcp.example.com/mcp".to_string(),
HashMap::default(),
cx.background_executor.clone(),
);
transport
.send(r#"{"jsonrpc":"2.0","id":1,"method":"initialize"}"#.to_string())
.await
.expect("send should succeed");
let mut responses = transport.receive();
let next_response = responses.next().fuse();
let timeout = cx.background_executor.timer(Duration::from_secs(1)).fuse();
futures::pin_mut!(next_response, timeout);
let response = futures::select_biased! {
response = next_response => response.expect("expected SSE response"),
_ = timeout => panic!("timed out waiting for SSE response with {data_prefix:?}"),
};
assert_eq!(
response, r#"{"jsonrpc":"2.0","id":1,"result":{}}"#,
"unexpected SSE response for {data_prefix:?}",
);
}
}
#[gpui::test]
async fn test_bearer_token_attached_to_requests(cx: &mut TestAppContext) {
// Capture the Authorization header from the request.