use std::net::SocketAddr;
use std::sync::Arc;
use axum::{
extract::State,
http::{HeaderMap, StatusCode},
response::{IntoResponse, Json},
routing::{get, post},
Router,
};
use serde_json::{json, Value};
use tokio::task::JoinHandle;
use car_mcp::error_codes::PARSE as E_PARSE;
use car_mcp::{Request as McpRequest, Server as McpServer};
const PROTOCOL_HEADER: &str = "mcp-protocol-version";
fn origin_allowed(origin: &str) -> bool {
if origin.is_empty() || origin == "null" {
return false;
}
let Some((scheme, rest)) = origin.split_once("://") else {
return false;
};
if !scheme.eq_ignore_ascii_case("http") && !scheme.eq_ignore_ascii_case("https") {
return false;
}
let host = match rest.strip_prefix('[') {
Some(after_bracket) => match after_bracket.split_once(']') {
Some((inner, tail)) if tail.is_empty() || tail.starts_with(':') => inner,
_ => return false,
},
None => match rest.split_once(':') {
Some((h, _port)) => h,
None => rest,
},
};
let host = host.to_ascii_lowercase();
if host == "localhost" || host == "::1" {
return true;
}
matches!(host.parse::<std::net::IpAddr>(), Ok(ip) if ip.is_loopback())
}
fn check_origin(headers: &HeaderMap) -> Result<(), axum::response::Response> {
let Some(raw) = headers.get(axum::http::header::ORIGIN) else {
return Ok(());
};
let allowed = raw.to_str().map(origin_allowed).unwrap_or(false);
if allowed {
return Ok(());
}
let shown = raw.to_str().unwrap_or("<non-utf8>");
tracing::warn!(origin = %shown, "MCP request rejected: origin not allowed");
Err((
StatusCode::FORBIDDEN,
Json(json!({ "error": "origin not allowed" })),
)
.into_response())
}
fn check_protocol_version(headers: &HeaderMap) -> Result<(), axum::response::Response> {
let Some(raw) = headers.get(PROTOCOL_HEADER) else {
return Ok(());
};
if raw
.to_str()
.map(|v| car_mcp::SUPPORTED_VERSIONS.contains(&v))
.unwrap_or(false)
{
return Ok(());
}
let shown = raw.to_str().unwrap_or("<non-utf8>");
tracing::warn!(version = %shown, "MCP request rejected: unsupported protocol version");
Err((
StatusCode::BAD_REQUEST,
Json(json!({
"error": "unsupported MCP-Protocol-Version",
"requested": shown,
"supported": car_mcp::SUPPORTED_VERSIONS,
})),
)
.into_response())
}
#[derive(Clone)]
struct McpState {
server: Arc<McpServer>,
}
pub async fn start_mcp(
server: Arc<McpServer>,
addr: SocketAddr,
) -> Result<(SocketAddr, JoinHandle<()>), String> {
let listener = tokio::net::TcpListener::bind(addr)
.await
.map_err(|e| format!("bind {addr}: {e}"))?;
let bound = listener
.local_addr()
.map_err(|e| format!("local_addr: {e}"))?;
let state = McpState { server };
let app: Router = Router::new()
.route("/mcp", post(handle_mcp_post).get(handle_mcp_get))
.route("/mcp/health", get(handle_health))
.with_state(state);
let task = tokio::spawn(async move {
if let Err(e) = axum::serve(listener, app).await {
tracing::warn!(error = %e, "mcp HTTP server exited");
}
});
Ok((bound, task))
}
async fn handle_health() -> impl IntoResponse {
Json(json!({
"status": "ok",
"protocol_version": car_mcp::PROTOCOL_VERSION,
"server_name": car_mcp::SERVER_NAME,
}))
}
async fn handle_mcp_get() -> impl IntoResponse {
(
StatusCode::METHOD_NOT_ALLOWED,
[(axum::http::header::ALLOW, "POST")],
Json(json!({
"error": "this server offers no SSE stream; POST JSON-RPC requests to /mcp",
})),
)
}
async fn handle_mcp_post(
State(state): State<McpState>,
headers: HeaderMap,
body: String,
) -> axum::response::Response {
if let Err(resp) = check_origin(&headers) {
return resp;
}
if let Err(resp) = check_protocol_version(&headers) {
return resp;
}
let req: McpRequest = match serde_json::from_str(&body) {
Ok(req) => req,
Err(e) => {
let resp = json!({
"jsonrpc": "2.0",
"id": Value::Null,
"error": {
"code": E_PARSE,
"message": format!("parse error: {e}"),
},
});
return (StatusCode::OK, Json(resp)).into_response();
}
};
match crate::mcp_daemon::handle_request(&state.server, req, &headers).await {
Some(resp) => match serde_json::to_value(&resp) {
Ok(v) => (StatusCode::OK, Json(v)).into_response(),
Err(e) => (
StatusCode::INTERNAL_SERVER_ERROR,
Json(json!({
"jsonrpc": "2.0",
"id": Value::Null,
"error": {
"code": -32603,
"message": format!("response serialization failed: {e}"),
},
})),
)
.into_response(),
},
None => StatusCode::ACCEPTED.into_response(),
}
}
#[cfg(test)]
mod tests {
use super::*;
use std::time::Duration;
async fn boot_test_server() -> (SocketAddr, JoinHandle<()>) {
let server = Arc::new(McpServer::new());
let addr: SocketAddr = "127.0.0.1:0".parse().unwrap();
let (bound, task) = start_mcp(server, addr).await.expect("start_mcp");
(bound, task)
}
async fn http_post(addr: SocketAddr, body: &str) -> (StatusCode, Value) {
http_post_with_origin(addr, body, None).await
}
async fn http_post_with_origin(
addr: SocketAddr,
body: &str,
origin: Option<&str>,
) -> (StatusCode, Value) {
let (status, text) = http_post_raw(addr, body, origin, None).await;
let value: Value = serde_json::from_str(&text).expect("json");
(status, value)
}
async fn http_post_with_protocol(
addr: SocketAddr,
body: &str,
version: Option<&str>,
) -> (StatusCode, Value) {
let (status, text) = http_post_raw(addr, body, None, version).await;
let value: Value = serde_json::from_str(&text).expect("json");
(status, value)
}
async fn http_post_raw(
addr: SocketAddr,
body: &str,
origin: Option<&str>,
version: Option<&str>,
) -> (StatusCode, String) {
let url = format!("http://{}/mcp", addr);
let client = reqwest::Client::new();
let mut req = client
.post(&url)
.header("Content-Type", "application/json")
.body(body.to_string());
if let Some(origin) = origin {
req = req.header("Origin", origin);
}
if let Some(version) = version {
req = req.header("MCP-Protocol-Version", version);
}
let resp = req.send().await.expect("post");
let status = resp.status();
let text = resp.text().await.expect("body");
(status, text)
}
#[tokio::test]
async fn health_endpoint_returns_ok() {
let (addr, _task) = boot_test_server().await;
tokio::time::sleep(Duration::from_millis(50)).await;
let url = format!("http://{}/mcp/health", addr);
let resp = reqwest::get(&url).await.expect("get");
assert_eq!(resp.status(), StatusCode::OK);
let body: Value = resp.json().await.expect("json");
assert_eq!(body["status"], "ok");
assert_eq!(body["protocol_version"], car_mcp::PROTOCOL_VERSION);
}
#[tokio::test]
async fn initialize_round_trips_over_http() {
let (addr, _task) = boot_test_server().await;
tokio::time::sleep(Duration::from_millis(50)).await;
let req = r#"{"jsonrpc":"2.0","id":1,"method":"initialize","params":{}}"#;
let (status, body) = http_post(addr, req).await;
assert_eq!(status, StatusCode::OK);
assert_eq!(body["jsonrpc"], "2.0");
assert_eq!(body["id"], 1);
assert_eq!(body["result"]["protocolVersion"], car_mcp::PROTOCOL_VERSION);
}
#[tokio::test]
async fn initialize_negotiates_over_http() {
let (addr, _task) = boot_test_server().await;
tokio::time::sleep(Duration::from_millis(50)).await;
let req = format!(
r#"{{"jsonrpc":"2.0","id":1,"method":"initialize","params":{{"protocolVersion":"{}","capabilities":{{}},"clientInfo":{{"name":"c","version":"0"}}}}}}"#,
car_mcp::PROTOCOL_VERSION
);
let (status, body) = http_post(addr, &req).await;
assert_eq!(status, StatusCode::OK);
assert_eq!(body["result"]["protocolVersion"], car_mcp::PROTOCOL_VERSION);
let req = r#"{"jsonrpc":"2.0","id":2,"method":"initialize","params":{"protocolVersion":"1999-01-01"}}"#;
let (status, body) = http_post(addr, req).await;
assert_eq!(status, StatusCode::OK);
assert!(body.get("error").is_none(), "{body}");
assert_eq!(body["result"]["protocolVersion"], car_mcp::PROTOCOL_VERSION);
}
#[tokio::test]
async fn notification_over_http_returns_202_and_no_body() {
let (addr, _task) = boot_test_server().await;
tokio::time::sleep(Duration::from_millis(50)).await;
let req = r#"{"jsonrpc":"2.0","method":"notifications/initialized"}"#;
let (status, body) = http_post_raw(addr, req, None, None).await;
assert_eq!(status, StatusCode::ACCEPTED);
assert!(body.is_empty(), "expected an empty body, got {body:?}");
}
#[tokio::test]
async fn tools_list_round_trips_over_http() {
let (addr, _task) = boot_test_server().await;
tokio::time::sleep(Duration::from_millis(50)).await;
let req = r#"{"jsonrpc":"2.0","id":2,"method":"tools/list"}"#;
let (status, body) = http_post(addr, req).await;
assert_eq!(status, StatusCode::OK);
let tools = body["result"]["tools"].as_array().expect("tools array");
assert_eq!(
tools.len(),
car_mcp::cached_tool_schemas().len(),
"HTTP tools/list must surface every defined tool, and no others"
);
let names: std::collections::HashSet<&str> =
tools.iter().filter_map(|t| t["name"].as_str()).collect();
assert_eq!(
names.len(),
tools.len(),
"tool names must be unique: {tools:?}"
);
}
struct ProbeTool;
#[async_trait::async_trait]
impl car_mcp::ToolHandler for ProbeTool {
async fn call(&self, _args: Value) -> Result<String, car_mcp::ToolError> {
Ok("served by the daemon".to_string())
}
}
#[tokio::test]
async fn a_registered_tool_is_reachable_over_http() {
let mut server = McpServer::new();
server
.register_tool(
json!({
"name": "daemon_probe",
"description": "a daemon-only tool",
"inputSchema": { "type": "object", "properties": {} },
"annotations": {
"readOnlyHint": true,
"destructiveHint": false,
"idempotentHint": true,
"openWorldHint": false,
},
}),
Arc::new(ProbeTool),
)
.expect("registers");
let addr: SocketAddr = "127.0.0.1:0".parse().unwrap();
let (addr, _task) = start_mcp(Arc::new(server), addr).await.expect("start_mcp");
tokio::time::sleep(Duration::from_millis(50)).await;
let (status, body) =
http_post(addr, r#"{"jsonrpc":"2.0","id":1,"method":"tools/list"}"#).await;
assert_eq!(status, StatusCode::OK);
let tools = body["result"]["tools"].as_array().expect("tools array");
assert_eq!(tools.len(), car_mcp::cached_tool_schemas().len() + 1);
assert!(tools
.iter()
.any(|t| t["name"].as_str() == Some("daemon_probe")));
let (status, body) = http_post(
addr,
r#"{"jsonrpc":"2.0","id":2,"method":"tools/call","params":{"name":"daemon_probe","arguments":{}}}"#,
)
.await;
assert_eq!(status, StatusCode::OK);
assert_eq!(body["result"]["isError"], false);
assert_eq!(body["result"]["content"][0]["text"], "served by the daemon");
}
#[tokio::test]
async fn malformed_json_returns_parse_error() {
let (addr, _task) = boot_test_server().await;
tokio::time::sleep(Duration::from_millis(50)).await;
let (status, body) = http_post(addr, "{not valid").await;
assert_eq!(status, StatusCode::OK);
assert_eq!(body["error"]["code"], -32700);
}
#[tokio::test]
async fn shared_memgine_lets_facts_persist_across_requests() {
let memgine = Arc::new(tokio::sync::Mutex::new(car_memgine::MemgineEngine::new(
None,
)));
let server = Arc::new(McpServer::with_memgine(memgine));
let addr: SocketAddr = "127.0.0.1:0".parse().unwrap();
let (addr, _task) = start_mcp(server, addr).await.expect("start");
tokio::time::sleep(Duration::from_millis(50)).await;
let add = r#"{"jsonrpc":"2.0","id":1,"method":"tools/call","params":{"name":"memory_add_fact","arguments":{"subject":"daemon","body":"shared engine works"}}}"#;
let (_, _) = http_post(addr, add).await;
let query = r#"{"jsonrpc":"2.0","id":2,"method":"tools/call","params":{"name":"memory_query","arguments":{"query":"daemon","k":5}}}"#;
let (_, body) = http_post(addr, query).await;
let text = body["result"]["content"][0]["text"].as_str().expect("text");
assert!(
text.contains("daemon"),
"expected query to find ingested fact: {text}"
);
}
#[tokio::test]
async fn get_mcp_returns_405_method_not_allowed() {
let (addr, _task) = boot_test_server().await;
tokio::time::sleep(Duration::from_millis(50)).await;
let url = format!("http://{}/mcp", addr);
let resp = reqwest::Client::new()
.get(&url)
.header("mcp-session-id", "hopeful-subscriber")
.send()
.await
.expect("get");
assert_eq!(resp.status(), StatusCode::METHOD_NOT_ALLOWED);
assert_eq!(
resp.headers()
.get(axum::http::header::ALLOW)
.and_then(|v| v.to_str().ok()),
Some("POST"),
"a 405 names what IS allowed"
);
let body: Value = resp.json().await.expect("json body");
assert!(
body["error"]
.as_str()
.is_some_and(|e| e.contains("no SSE stream")),
"the body should say why: {body}"
);
}
#[test]
fn origin_allowed_unit() {
for ok in [
"http://localhost",
"http://localhost:3000",
"https://localhost:8443",
"HTTP://LocalHost:3000",
"http://127.0.0.1",
"http://127.0.0.1:9102",
"https://127.0.0.53",
"http://[::1]",
"http://[::1]:3000",
] {
assert!(origin_allowed(ok), "should be allowed: {ok}");
}
for bad in [
"https://evil.example",
"http://sub.localhost.evil.example",
"http://localhost.evil.com",
"https://127.0.0.1.evil.com",
"null",
"",
"file://",
"file:///etc/passwd",
"ws://localhost:3000",
"localhost:3000",
"http://[::1",
"http://[::1]x",
"http://10.0.0.5",
] {
assert!(!origin_allowed(bad), "should be rejected: {bad}");
}
}
#[tokio::test]
async fn mcp_post_without_origin_is_allowed() {
let (addr, _task) = boot_test_server().await;
tokio::time::sleep(Duration::from_millis(50)).await;
let req = r#"{"jsonrpc":"2.0","id":1,"method":"tools/list"}"#;
let (status, body) = http_post_with_origin(addr, req, None).await;
assert_eq!(status, StatusCode::OK);
assert!(
body["result"]["tools"].as_array().is_some(),
"absent Origin must be allowed — no real MCP client sends one: {body}"
);
}
#[tokio::test]
async fn mcp_post_with_loopback_origin_is_allowed() {
let (addr, _task) = boot_test_server().await;
tokio::time::sleep(Duration::from_millis(50)).await;
let req = r#"{"jsonrpc":"2.0","id":1,"method":"tools/list"}"#;
for origin in [
"http://localhost:3000",
"http://127.0.0.1:9102",
"http://[::1]",
] {
let (status, body) = http_post_with_origin(addr, req, Some(origin)).await;
assert_eq!(status, StatusCode::OK, "origin {origin} must be allowed");
assert!(body["result"]["tools"].as_array().is_some(), "{body}");
}
}
#[tokio::test]
async fn mcp_post_with_foreign_origin_is_rejected() {
let memgine = Arc::new(tokio::sync::Mutex::new(car_memgine::MemgineEngine::new(
None,
)));
let server = Arc::new(McpServer::with_memgine(memgine));
let (addr, _task) = start_mcp(server, "127.0.0.1:0".parse().unwrap())
.await
.expect("start");
tokio::time::sleep(Duration::from_millis(50)).await;
let add = r#"{"jsonrpc":"2.0","id":1,"method":"tools/call","params":{"name":"memory_add_fact","arguments":{"subject":"rebind","body":"attacker planted this"}}}"#;
let (status, body) = http_post_with_origin(addr, add, Some("https://evil.example")).await;
assert_eq!(status, StatusCode::FORBIDDEN);
assert_eq!(body["error"], "origin not allowed");
let query = r#"{"jsonrpc":"2.0","id":2,"method":"tools/call","params":{"name":"memory_query","arguments":{"query":"rebind","k":5}}}"#;
let (_, body) = http_post(addr, query).await;
let text = body["result"]["content"][0]["text"]
.as_str()
.unwrap_or_default();
assert!(
!text.contains("attacker planted this"),
"rejected origin must not reach the tool: {text}"
);
}
#[tokio::test]
async fn mcp_post_with_null_origin_is_rejected() {
let (addr, _task) = boot_test_server().await;
tokio::time::sleep(Duration::from_millis(50)).await;
let req = r#"{"jsonrpc":"2.0","id":1,"method":"tools/list"}"#;
let (status, body) = http_post_with_origin(addr, req, Some("null")).await;
assert_eq!(
status,
StatusCode::FORBIDDEN,
"the opaque origin is not a safe origin"
);
assert_eq!(body["error"], "origin not allowed");
}
#[tokio::test]
async fn mcp_get_with_foreign_origin_gets_the_same_405() {
let (addr, _task) = boot_test_server().await;
tokio::time::sleep(Duration::from_millis(50)).await;
let url = format!("http://{}/mcp", addr);
let resp = reqwest::Client::new()
.get(&url)
.header("Origin", "https://evil.example")
.send()
.await
.expect("get");
assert_eq!(resp.status(), StatusCode::METHOD_NOT_ALLOWED);
}
#[tokio::test]
async fn health_stays_reachable_from_any_origin() {
let (addr, _task) = boot_test_server().await;
tokio::time::sleep(Duration::from_millis(50)).await;
let url = format!("http://{}/mcp/health", addr);
let resp = reqwest::Client::new()
.get(&url)
.header("Origin", "https://evil.example")
.send()
.await
.expect("get");
assert_eq!(resp.status(), StatusCode::OK);
}
#[tokio::test]
async fn mcp_post_without_protocol_version_is_allowed() {
let (addr, _task) = boot_test_server().await;
tokio::time::sleep(Duration::from_millis(50)).await;
let req = r#"{"jsonrpc":"2.0","id":1,"method":"tools/list"}"#;
let (status, body) = http_post_with_protocol(addr, req, None).await;
assert_eq!(status, StatusCode::OK);
assert!(
body["result"]["tools"].as_array().is_some(),
"an absent MCP-Protocol-Version must be allowed: {body}"
);
}
#[tokio::test]
async fn mcp_post_with_supported_protocol_version_is_allowed() {
let (addr, _task) = boot_test_server().await;
tokio::time::sleep(Duration::from_millis(50)).await;
let req = r#"{"jsonrpc":"2.0","id":1,"method":"tools/list"}"#;
let (status, body) =
http_post_with_protocol(addr, req, Some(car_mcp::PROTOCOL_VERSION)).await;
assert_eq!(status, StatusCode::OK);
assert!(body["result"]["tools"].as_array().is_some(), "{body}");
}
#[tokio::test]
async fn mcp_post_with_unsupported_protocol_version_is_rejected() {
let memgine = Arc::new(tokio::sync::Mutex::new(car_memgine::MemgineEngine::new(
None,
)));
let server = Arc::new(McpServer::with_memgine(memgine));
let (addr, _task) = start_mcp(server, "127.0.0.1:0".parse().unwrap())
.await
.expect("start");
tokio::time::sleep(Duration::from_millis(50)).await;
let add = r#"{"jsonrpc":"2.0","id":1,"method":"tools/call","params":{"name":"memory_add_fact","arguments":{"subject":"dialect","body":"wrong-revision write"}}}"#;
let (status, body) = http_post_with_protocol(addr, add, Some("1999-01-01")).await;
assert_eq!(status, StatusCode::BAD_REQUEST);
assert_eq!(body["error"], "unsupported MCP-Protocol-Version");
assert_eq!(body["requested"], "1999-01-01");
let supported = body["supported"].as_array().expect("supported list");
assert!(
supported
.iter()
.any(|v| v.as_str() == Some(car_mcp::PROTOCOL_VERSION)),
"the 400 must name what we do speak: {body}"
);
let query = r#"{"jsonrpc":"2.0","id":2,"method":"tools/call","params":{"name":"memory_query","arguments":{"query":"dialect","k":5}}}"#;
let (_, body) = http_post(addr, query).await;
let text = body["result"]["content"][0]["text"]
.as_str()
.unwrap_or_default();
assert!(
!text.contains("wrong-revision write"),
"a rejected protocol version must not reach the tool: {text}"
);
}
#[tokio::test]
async fn mcp_get_with_unsupported_protocol_version_gets_the_same_405() {
let (addr, _task) = boot_test_server().await;
tokio::time::sleep(Duration::from_millis(50)).await;
let url = format!("http://{}/mcp", addr);
let resp = reqwest::Client::new()
.get(&url)
.header("MCP-Protocol-Version", "1999-01-01")
.send()
.await
.expect("get");
assert_eq!(resp.status(), StatusCode::METHOD_NOT_ALLOWED);
}
#[tokio::test]
async fn health_ignores_protocol_version_header() {
let (addr, _task) = boot_test_server().await;
tokio::time::sleep(Duration::from_millis(50)).await;
let url = format!("http://{}/mcp/health", addr);
let resp = reqwest::Client::new()
.get(&url)
.header("MCP-Protocol-Version", "1999-01-01")
.send()
.await
.expect("get");
assert_eq!(resp.status(), StatusCode::OK);
let body: Value = resp.json().await.expect("json");
assert_eq!(body["status"], "ok");
assert_eq!(body["protocol_version"], car_mcp::PROTOCOL_VERSION);
}
}