oxml-mcp 0.0.9

// SPDX-License-Identifier: MIT OR Apache-2.0
// Copyright (c) 2026 oxml. All rights reserved.

//! The two HTTP transports, through the real binary.
//!
//! Each test starts `oxml-mcp --transport ... --port 0`, reads the
//! port the server announces, and talks HTTP/1.1 to it over a plain
//! socket. The client here is deliberately small and literal -- one
//! request, one response, chunked bodies decoded by hand -- because
//! what is being checked is the wire format a client will see, and a
//! client library would hide it.

use std::fmt::Write as _;
use std::io::{BufRead, BufReader, Read, Write};
use std::net::TcpStream;
use std::process::{Child, Command, Stdio};
use std::time::Duration;

use serde_json::{Value, json};

/// A running server, killed when dropped.
struct Server {
    child: Child,
    addr: String,
}

impl Server {
    fn start(transport: &str) -> Self {
        let mut child = Command::new(env!("CARGO_BIN_EXE_oxml-mcp"))
            .args(["--transport", transport, "--port", "0"])
            .stdin(Stdio::null())
            .stdout(Stdio::null())
            .stderr(Stdio::piped())
            .spawn()
            .expect("server starts");
        // The port is announced on stderr, which is the one line a
        // test needs and the reason `--port 0` exists.
        let stderr = child.stderr.take().expect("stderr");
        let mut line = String::new();
        let _ = BufReader::new(stderr).read_line(&mut line).expect("read");
        let addr = line
            .trim()
            .strip_prefix("listening on http://")
            .and_then(|rest| rest.split('/').next())
            .unwrap_or_else(|| panic!("no address in {line:?}"))
            .to_owned();
        Self { child, addr }
    }
}

impl Drop for Server {
    fn drop(&mut self) {
        // Ask first, with the interrupt the transport handles, and only
        // then kill. A server killed outright never writes its coverage
        // profile, which is how the HTTP paths read as untested while
        // every test drove them.
        #[cfg(unix)]
        {
            let _ = Command::new("kill")
                .args(["-INT", &self.child.id().to_string()])
                .status();
            for _ in 0..50 {
                if matches!(self.child.try_wait(), Ok(Some(_))) {
                    return;
                }
                std::thread::sleep(Duration::from_millis(100));
            }
        }
        let _ = self.child.kill();
        let _ = self.child.wait();
    }
}

/// One HTTP/1.1 exchange, with the body read incrementally so an
/// event stream can be consumed as it arrives.
struct Http {
    status: u16,
    headers: Vec<(String, String)>,
    reader: BufReader<TcpStream>,
    chunked: bool,
    remaining: Option<usize>,
    /// Decoded body bytes not yet handed out.
    pending: Vec<u8>,
}

impl Http {
    fn send(
        addr: &str,
        method: &str,
        path: &str,
        headers: &[(&str, &str)],
        body: &str,
    ) -> Self {
        let mut stream = TcpStream::connect(addr).expect("connect");
        stream
            .set_read_timeout(Some(Duration::from_secs(10)))
            .expect("timeout");
        let mut request = format!(
            "{method} {path} HTTP/1.1\r\nHost: {addr}\r\nConnection: close\r\n"
        );
        for (name, value) in headers {
            let _ = write!(request, "{name}: {value}\r\n");
        }
        let _ = write!(request, "Content-Length: {}\r\n\r\n{body}", body.len());
        stream.write_all(request.as_bytes()).expect("write");

        let mut reader = BufReader::new(stream);
        let mut line = String::new();
        let _ = reader.read_line(&mut line).expect("status line");
        let status: u16 = line
            .split_whitespace()
            .nth(1)
            .and_then(|s| s.parse().ok())
            .unwrap_or_else(|| panic!("bad status line {line:?}"));
        let mut headers = Vec::new();
        loop {
            line.clear();
            let _ = reader.read_line(&mut line).expect("header");
            let trimmed = line.trim_end();
            if trimmed.is_empty() {
                break;
            }
            let (name, value) = trimmed.split_once(':').expect("header");
            headers.push((name.to_ascii_lowercase(), value.trim().to_owned()));
        }
        let header = |name: &str| {
            headers
                .iter()
                .find(|(n, _)| n == name)
                .map(|(_, v)| v.clone())
        };
        let chunked =
            header("transfer-encoding").is_some_and(|v| v.contains("chunked"));
        let remaining = header("content-length").and_then(|v| v.parse().ok());
        Self {
            status,
            headers,
            reader,
            chunked,
            remaining,
            pending: Vec::new(),
        }
    }

    fn header(&self, name: &str) -> Option<&str> {
        self.headers
            .iter()
            .find(|(n, _)| n == name)
            .map(|(_, v)| v.as_str())
    }

    /// Pull the next piece of body into `pending`. False at the end.
    fn fill(&mut self) -> bool {
        if self.chunked {
            let mut size = String::new();
            let _ = self.reader.read_line(&mut size).expect("chunk size");
            let size =
                usize::from_str_radix(size.trim(), 16).expect("hex size");
            if size == 0 {
                return false;
            }
            let mut chunk = vec![0; size + 2];
            self.reader.read_exact(&mut chunk).expect("chunk");
            chunk.truncate(size);
            self.pending.extend_from_slice(&chunk);
            true
        } else {
            let want = self.remaining.unwrap_or(usize::MAX).min(4096);
            if want == 0 {
                return false;
            }
            let mut buf = vec![0; want];
            let n = self.reader.read(&mut buf).expect("read");
            if n == 0 {
                return false;
            }
            self.pending.extend_from_slice(&buf[..n]);
            if let Some(r) = self.remaining.as_mut() {
                *r -= n;
            }
            true
        }
    }

    /// The whole body, for a response that ends.
    fn body(mut self) -> String {
        while self.fill() {}
        String::from_utf8(self.pending).expect("utf-8")
    }

    /// The next `event` with a `data` field, as (event name, data).
    ///
    /// Keep-alive comments carry no data and are skipped.
    fn next_event(&mut self) -> (String, String) {
        loop {
            let text = String::from_utf8_lossy(&self.pending).into_owned();
            if let Some(end) = text.find("\n\n") {
                let block = text[..end].to_owned();
                let _ = self.pending.drain(..end + 2);
                let mut event = "message".to_owned();
                let mut data = Vec::new();
                for line in block.lines() {
                    if let Some(v) = line.strip_prefix("event:") {
                        v.trim().clone_into(&mut event);
                    } else if let Some(v) = line.strip_prefix("data:") {
                        data.push(v.trim().to_owned());
                    }
                }
                // A priming event carries an id and no data; a
                // keep-alive is a comment. Neither is a message.
                if data.iter().any(|d| !d.is_empty()) {
                    return (event, data.join("\n"));
                }
                continue;
            }
            assert!(self.fill(), "stream ended before an event arrived");
        }
    }

    /// The next JSON-RPC message on the stream.
    fn next_message(&mut self) -> Value {
        let (event, data) = self.next_event();
        assert_eq!(event, "message", "unexpected event {event}: {data}");
        serde_json::from_str(&data).unwrap_or_else(|e| panic!("{e}: {data}"))
    }
}

const ACCEPT_BOTH: (&str, &str) =
    ("Accept", "application/json, text/event-stream");
const JSON: (&str, &str) = ("Content-Type", "application/json");

fn initialize(version: &str) -> String {
    json!({"jsonrpc": "2.0", "id": 1, "method": "initialize", "params": {
        "protocolVersion": version, "capabilities": {},
        "clientInfo": {"name": "test", "version": "0"}}})
    .to_string()
}

fn stateless(id: u32, method: &str, params: Value) -> String {
    let mut params = params;
    // Both keys are required by the revision; a request naming only
    // its version is refused.
    params["_meta"] = json!({
        "io.modelcontextprotocol/protocolVersion": "2026-07-28",
        "io.modelcontextprotocol/clientCapabilities": {}
    });
    json!({"jsonrpc": "2.0", "id": id, "method": method, "params": params})
        .to_string()
}

// --- Streamable HTTP, 2025-11-25: initialize and a session ------------

#[test]
fn streamable_http_holds_a_session_with_a_handshake() {
    let server = Server::start("streamable-http");
    let mut r = Http::send(
        &server.addr,
        "POST",
        "/mcp",
        &[ACCEPT_BOTH, JSON],
        &initialize("2025-11-25"),
    );
    assert_eq!(r.status, 200);
    assert!(
        r.header("content-type")
            .is_some_and(|c| c.starts_with("text/event-stream")),
        "{:?}",
        r.headers
    );
    let session = r.header("mcp-session-id").expect("session id").to_owned();
    let init = r.next_message();
    assert_eq!(init["id"], 1);
    assert_eq!(init["result"]["protocolVersion"], "2025-11-25");
    assert_eq!(init["result"]["serverInfo"]["name"], "oxml-mcp");
    drop(r);

    let session_header = ("Mcp-Session-Id", session.as_str());
    let version_header = ("MCP-Protocol-Version", "2025-11-25");
    let r = Http::send(
        &server.addr,
        "POST",
        "/mcp",
        &[ACCEPT_BOTH, JSON, session_header, version_header],
        r#"{"jsonrpc":"2.0","method":"notifications/initialized"}"#,
    );
    assert_eq!(r.status, 202);

    let mut r = Http::send(
        &server.addr,
        "POST",
        "/mcp",
        &[ACCEPT_BOTH, JSON, session_header, version_header],
        r#"{"jsonrpc":"2.0","id":2,"method":"tools/list"}"#,
    );
    assert_eq!(r.status, 200);
    let list = r.next_message();
    let tools = list["result"]["tools"].as_array().expect("tools");
    assert_eq!(tools.len(), 4, "{list}");
    assert!(
        tools.iter().all(|t| t["outputSchema"].is_object()),
        "{list}"
    );
    drop(r);

    // GET opens the server-to-client stream for the session.
    let r = Http::send(
        &server.addr,
        "GET",
        "/mcp",
        &[
            ("Accept", "text/event-stream"),
            session_header,
            version_header,
        ],
        "",
    );
    assert_eq!(r.status, 200);
    assert!(
        r.header("content-type")
            .is_some_and(|c| c.starts_with("text/event-stream")),
        "{:?}",
        r.headers
    );
    drop(r);

    // A session id the server never issued is not found.
    let r = Http::send(
        &server.addr,
        "POST",
        "/mcp",
        &[
            ACCEPT_BOTH,
            JSON,
            ("Mcp-Session-Id", "nope"),
            version_header,
        ],
        r#"{"jsonrpc":"2.0","id":3,"method":"ping"}"#,
    );
    assert_eq!(r.status, 404);

    // DELETE ends the session.
    let r = Http::send(
        &server.addr,
        "DELETE",
        "/mcp",
        &[session_header, version_header],
        "",
    );
    assert!(r.status < 300, "{}", r.status);
}

// --- Streamable HTTP, 2026-07-28: stateless, per-request _meta ----------

#[test]
fn streamable_http_serves_stateless_requests_without_a_session() {
    let server = Server::start("streamable-http");
    let version = ("MCP-Protocol-Version", "2026-07-28");
    // SEP-2243: this revision mirrors the method (and, for a tool call,
    // the tool name) in headers so a gateway can route without reading
    // the body. The SDK requires them.

    let mut r = Http::send(
        &server.addr,
        "POST",
        "/mcp",
        &[
            ACCEPT_BOTH,
            JSON,
            version,
            ("Mcp-Method", "server/discover"),
        ],
        &stateless(1, "server/discover", json!({})),
    );
    assert_eq!(r.status, 200);
    assert!(
        r.header("mcp-session-id").is_none(),
        "no session in this revision"
    );
    let discover = r.next_message();
    let versions = discover["result"]["supportedVersions"]
        .as_array()
        .expect("supportedVersions");
    assert!(versions.contains(&json!("2026-07-28")), "{discover}");
    assert!(versions.contains(&json!("2025-11-25")), "{discover}");
    assert_eq!(
        discover["result"]["_meta"]["io.modelcontextprotocol/serverInfo"]["name"],
        "oxml-mcp"
    );
    drop(r);

    let mut r = Http::send(
        &server.addr,
        "POST",
        "/mcp",
        &[ACCEPT_BOTH, JSON, version, ("Mcp-Method", "tools/list")],
        &stateless(2, "tools/list", json!({})),
    );
    assert_eq!(r.status, 200);
    let list = r.next_message();
    assert_eq!(list["result"]["tools"].as_array().map(Vec::len), Some(4));
    drop(r);

    let call_headers = [
        ACCEPT_BOTH,
        JSON,
        version,
        ("Mcp-Method", "tools/call"),
        ("Mcp-Name", "xml_check"),
    ];
    let mut r = Http::send(
        &server.addr,
        "POST",
        "/mcp",
        &call_headers,
        &stateless(
            3,
            "tools/call",
            json!({"name": "xml_check", "arguments": {"xml": "<a/>"}}),
        ),
    );
    assert_eq!(r.status, 200);
    let call = r.next_message();
    assert_eq!(call["result"]["isError"], false, "{call}");
    assert_eq!(call["result"]["structuredContent"]["well_formed"], true);
    drop(r);

    // A tool failure is a result the model can read.
    let mut r = Http::send(
        &server.addr,
        "POST",
        "/mcp",
        &call_headers,
        &stateless(
            4,
            "tools/call",
            json!({"name": "xml_check", "arguments": {"xml": "<a>"}}),
        ),
    );
    let call = r.next_message();
    assert_eq!(call["result"]["isError"], true, "{call}");
    drop(r);
}

#[test]
fn streamable_http_refuses_what_the_stateless_revision_forbids() {
    let server = Server::start("streamable-http");
    let version = ("MCP-Protocol-Version", "2026-07-28");

    // So is a tool the server does not have: in this revision the
    // SDK's `-32602` would be an HTTP 400, which no model reads.
    let mut r = Http::send(
        &server.addr,
        "POST",
        "/mcp",
        &[
            ACCEPT_BOTH,
            JSON,
            version,
            ("Mcp-Method", "tools/call"),
            ("Mcp-Name", "no_such_tool"),
        ],
        &stateless(
            5,
            "tools/call",
            json!({"name": "no_such_tool", "arguments": {}}),
        ),
    );
    assert_eq!(r.status, 200);
    let call = r.next_message();
    assert_eq!(call["result"]["isError"], true, "{call}");
    assert!(
        call["result"]["content"][0]["text"]
            .as_str()
            .is_some_and(|t| t.contains("Unknown tool: no_such_tool")),
        "{call}"
    );
    drop(r);

    // A request naming its revision without the rest of the required
    // metadata is refused before any handler runs.
    let r = Http::send(
        &server.addr,
        "POST",
        "/mcp",
        &[ACCEPT_BOTH, JSON, version, ("Mcp-Method", "tools/list")],
        r#"{"jsonrpc":"2.0","id":5,"method":"tools/list","params":{"_meta":{"io.modelcontextprotocol/protocolVersion":"2026-07-28"}}}"#,
    );
    assert_eq!(r.status, 400);
    assert!(r.body().contains("clientCapabilities"));

    // Without a session, GET has nothing to stream: 405.
    let r = Http::send(
        &server.addr,
        "GET",
        "/mcp",
        &[("Accept", "text/event-stream"), version],
        "",
    );
    assert_eq!(r.status, 405);

    // A body that is not JSON is refused at the HTTP layer. The SDK
    // says 415, an unsupported media type, which is how it reads a
    // body that claims to be JSON and is not.
    let r = Http::send(
        &server.addr,
        "POST",
        "/mcp",
        &[ACCEPT_BOTH, JSON, version],
        "{not json",
    );
    assert_eq!(r.status, 415);
}

#[test]
fn streamable_http_mirrors_routing_headers() {
    // SEP-2243: a request may carry its method and tool name as
    // headers for routing. A header that disagrees with the body is a
    // lie one of the two is telling, and is refused.
    let server = Server::start("streamable-http");
    let version = ("MCP-Protocol-Version", "2026-07-28");
    let mut r = Http::send(
        &server.addr,
        "POST",
        "/mcp",
        &[
            ACCEPT_BOTH,
            JSON,
            version,
            ("Mcp-Method", "tools/call"),
            ("Mcp-Name", "xml_check"),
        ],
        &stateless(
            1,
            "tools/call",
            json!({"name": "xml_check", "arguments": {"xml": "<a/>"}}),
        ),
    );
    assert_eq!(r.status, 200);
    assert_eq!(r.next_message()["result"]["isError"], false);
    drop(r);

    let r = Http::send(
        &server.addr,
        "POST",
        "/mcp",
        &[ACCEPT_BOTH, JSON, version, ("Mcp-Method", "tools/list")],
        &stateless(2, "server/discover", json!({})),
    );
    assert_eq!(r.status, 400, "a mismatched Mcp-Method must be refused");
    assert!(r.body().contains("-32020"));
}

// --- The 2024-11-05 HTTP+SSE transport -----------------------------------

#[test]
fn sse_opens_a_stream_and_answers_posts_on_it() {
    let server = Server::start("sse");
    let mut stream = Http::send(&server.addr, "GET", "/sse", &[], "");
    assert_eq!(stream.status, 200);
    assert!(
        stream
            .header("content-type")
            .is_some_and(|c| c.starts_with("text/event-stream")),
        "{:?}",
        stream.headers
    );
    let (event, endpoint) = stream.next_event();
    assert_eq!(event, "endpoint");
    assert!(endpoint.starts_with("/messages/?sessionId="), "{endpoint}");

    let r = Http::send(
        &server.addr,
        "POST",
        &endpoint,
        &[JSON],
        &initialize("2024-11-05"),
    );
    assert_eq!(r.status, 202);
    let init = stream.next_message();
    assert_eq!(init["id"], 1);
    assert_eq!(init["result"]["protocolVersion"], "2024-11-05");
    assert_eq!(init["result"]["serverInfo"]["name"], "oxml-mcp");

    let r = Http::send(
        &server.addr,
        "POST",
        &endpoint,
        &[JSON],
        r#"{"jsonrpc":"2.0","method":"notifications/initialized"}"#,
    );
    assert_eq!(r.status, 202);

    let r = Http::send(
        &server.addr,
        "POST",
        &endpoint,
        &[JSON],
        r#"{"jsonrpc":"2.0","id":2,"method":"tools/list"}"#,
    );
    assert_eq!(r.status, 202);
    let list = stream.next_message();
    assert_eq!(list["id"], 2);
    assert_eq!(list["result"]["tools"].as_array().map(Vec::len), Some(4));

    let r = Http::send(
        &server.addr,
        "POST",
        &endpoint,
        &[JSON],
        r#"{"jsonrpc":"2.0","id":3,"method":"tools/call","params":{"name":"xml_query","arguments":{"xml":"<a><b>Dune</b></a>","xpath":"//b"}}}"#,
    );
    assert_eq!(r.status, 202);
    let call = stream.next_message();
    assert_eq!(call["result"]["content"][0]["text"], "Dune", "{call}");

    // The endpoint without its trailing slash works too.
    let bare = endpoint.replacen("/messages/?", "/messages?", 1);
    let r = Http::send(
        &server.addr,
        "POST",
        &bare,
        &[JSON],
        r#"{"jsonrpc":"2.0","id":4,"method":"ping"}"#,
    );
    assert_eq!(r.status, 202);
    assert_eq!(stream.next_message()["id"], 4);
}

#[test]
fn sse_refuses_what_it_cannot_route() {
    let server = Server::start("sse");
    let r = Http::send(
        &server.addr,
        "POST",
        "/messages/?sessionId=unknown",
        &[JSON],
        r#"{"jsonrpc":"2.0","id":1,"method":"ping"}"#,
    );
    assert_eq!(r.status, 404);

    let mut stream = Http::send(&server.addr, "GET", "/sse", &[], "");
    let (_, endpoint) = stream.next_event();
    let r = Http::send(&server.addr, "POST", &endpoint, &[JSON], "{not json");
    assert_eq!(r.status, 400);
    assert!(r.body().contains("invalid JSON-RPC"));

    // Hanging up ends the session: the endpoint stops existing.
    drop(stream);
    let mut gone = 0;
    for _ in 0..50 {
        let r = Http::send(
            &server.addr,
            "POST",
            &endpoint,
            &[JSON],
            r#"{"jsonrpc":"2.0","id":2,"method":"ping"}"#,
        );
        gone = r.status;
        if gone == 404 {
            break;
        }
        std::thread::sleep(Duration::from_millis(100));
    }
    assert_eq!(gone, 404, "the session outlived its stream");
}

#[test]
fn a_port_in_use_is_reported_not_swallowed() {
    let holder = std::net::TcpListener::bind("127.0.0.1:0").expect("bind");
    let port = holder.local_addr().expect("addr").port().to_string();
    let out = Command::new(env!("CARGO_BIN_EXE_oxml-mcp"))
        .args(["--transport", "streamable-http", "--port", &port])
        .output()
        .expect("runs");
    assert_eq!(out.status.code(), Some(1));
    let text = String::from_utf8_lossy(&out.stderr);
    assert!(text.contains("cannot listen on"), "{text}");
    assert!(text.contains(&port), "{text}");
}