diff --git a/Cargo.lock b/Cargo.lock index 228e132f..ad94e4ad 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -460,6 +460,17 @@ dependencies = [ "cpufeatures 0.2.17", ] +[[package]] +name = "chacha20" +version = "0.10.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6f8d983286843e49675a4b7a2d174efe136dc93a18d69130dd18198a6c167601" +dependencies = [ + "cfg-if", + "cpufeatures 0.3.0", + "rand_core 0.10.1", +] + [[package]] name = "chacha20poly1305" version = "0.10.1" @@ -467,7 +478,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "10cd79432192d1c0f4e1a0fef9527696cc039165d729fb41b3f4f4f354c2dc35" dependencies = [ "aead", - "chacha20", + "chacha20 0.9.1", "cipher 0.4.4", "poly1305", "zeroize", @@ -1193,6 +1204,7 @@ dependencies = [ "cfg-if", "libc", "r-efi 6.0.0", + "rand_core 0.10.1", "wasip2", "wasip3", ] @@ -2828,6 +2840,17 @@ dependencies = [ "rand_core 0.9.5", ] +[[package]] +name = "rand" +version = "0.10.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d2e8e8bcc7961af1fdac401278c6a831614941f6164ee3bf4ce61b7edb162207" +dependencies = [ + "chacha20 0.10.0", + "getrandom 0.4.2", + "rand_core 0.10.1", +] + [[package]] name = "rand_chacha" version = "0.3.1" @@ -2866,6 +2889,12 @@ dependencies = [ "getrandom 0.3.4", ] +[[package]] +name = "rand_core" +version = "0.10.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "63b8176103e19a2643978565ca18b50549f6101881c443590420e4dc998a3c69" + [[package]] name = "rand_xoshiro" version = "0.7.0" @@ -3014,18 +3043,27 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "0810a9f717d9828f475fe1f629f4c305c8464b7f496c3a854b58d29e65f4058e" dependencies = [ "async-trait", + "bytes", "chrono", "futures", + "http", + "http-body", + "http-body-util", "pastey", "pin-project-lite", + "rand 0.10.1", "rmcp-macros", "schemars", "serde", "serde_json", + "sse-stream", "thiserror 2.0.18", "tokio", + "tokio-stream", "tokio-util", + "tower-service", "tracing", + "uuid", ] [[package]] @@ -3628,6 +3666,19 @@ dependencies = [ "der", ] +[[package]] +name = "sse-stream" +version = "0.2.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f3962b63f038885f15bce2c6e02c0e7925c072f1ac86bb60fd44c5c6b762fb72" +dependencies = [ + "bytes", + "futures-util", + "http-body", + "http-body-util", + "pin-project-lite", +] + [[package]] name = "stable_deref_trait" version = "1.2.1" diff --git a/Cargo.toml b/Cargo.toml index 3fdc0642..44680642 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -36,6 +36,7 @@ rmcp = { version = "1.7", default-features = false, features = [ "server", "macros", "transport-io", + "transport-streamable-http-server", ] } rusqlite = { version = "0.37" } schemars = "1.0" diff --git a/hive-ag3nt/src/bin/hive.rs b/hive-ag3nt/src/bin/hive.rs index e31b26e7..2a67da85 100644 --- a/hive-ag3nt/src/bin/hive.rs +++ b/hive-ag3nt/src/bin/hive.rs @@ -40,10 +40,18 @@ enum Cmd { #[arg(long, default_value_t = 1000)] poll_ms: u64, }, - /// Run the MCP server on stdio. Spawned by `claude` via + /// Run the MCP server. Default is stdio — spawned by `claude` via /// `--mcp-config`; tools dispatch through `/run/hive/mcp.sock` back - /// into the hyperhive broker. - Mcp, + /// into the hyperhive broker. Pass `--http ` to instead run a + /// long-lived streamable-http listener (persistent daemon) that + /// claude reconnects to each turn, avoiding the per-turn stdio + /// re-registration race. + Mcp { + /// Serve over streamable-http on this address (e.g. + /// `127.0.0.1:8790`) instead of stdio. Bind loopback only. + #[arg(long)] + http: Option, + }, /// Inject a wake-up event into this harness's inbox so the next /// turn fires with the given body. Intended for extra MCP servers /// / helpers (matrix bridge, scraper, webhook listener, etc.) that @@ -70,7 +78,10 @@ async fn main() -> Result<()> { match cli.cmd { Cmd::Serve { poll_ms } => serve_main::(&cli.socket, poll_ms).await, - Cmd::Mcp => mcp::serve_agent_stdio(cli.socket).await, + Cmd::Mcp { http } => match http { + Some(addr) => mcp::serve_http(cli.socket, addr).await, + None => mcp::serve_agent_stdio(cli.socket).await, + }, Cmd::Wake { from, body } => wake::(&cli.socket, from, body).await, } } diff --git a/hive-ag3nt/src/mcp.rs b/hive-ag3nt/src/mcp.rs index 610e26c3..db21c244 100644 --- a/hive-ag3nt/src/mcp.rs +++ b/hive-ag3nt/src/mcp.rs @@ -1549,6 +1549,41 @@ pub async fn serve_agent_stdio(socket: PathBuf) -> Result<()> { serve_stdio(socket).await } +/// Run the MCP server over HTTP (rmcp streamable-http transport) on `addr`. +/// +/// Unlike [`serve_stdio`] — a fresh stdio child claude respawns every turn — +/// this is meant to run as a long-lived in-container daemon. claude reconnects +/// to the stable URL each turn instead of respawning and re-registering a stdio +/// subprocess, which removes the per-turn MCP registration race that can strand +/// an agent when the async `initialize`/`tools/list` loses to claude's first +/// tool call. `socket` is the hyperhive control socket every tool call dials +/// fresh (the handler holds only the path), so a host-side hive-c0re restart is +/// transparent — the next call just reconnects. +/// +/// Binds loopback only in practice; the default `allowed_hosts` +/// (`localhost`/`127.0.0.1`/`::1`) rejects Host headers from anywhere else. +/// +/// # Errors +/// +/// Returns an error if the listener cannot bind `addr` or the HTTP server +/// exits with a fatal error. +pub async fn serve_http(socket: PathBuf, addr: std::net::SocketAddr) -> Result<()> { + use rmcp::transport::streamable_http_server::{ + StreamableHttpServerConfig, StreamableHttpService, session::local::LocalSessionManager, + }; + let session_manager = std::sync::Arc::new(LocalSessionManager::default()); + let service = StreamableHttpService::new( + move || Ok(AgentServer::new(socket.clone())), + session_manager, + StreamableHttpServerConfig::default(), + ); + let app = axum::Router::new().nest_service("/mcp", service); + let listener = tokio::net::TcpListener::bind(addr).await?; + tracing::info!(%addr, "serving hyperhive MCP over streamable-http at /mcp"); + axum::serve(listener, app).await?; + Ok(()) +} + // ----------------------------------------------------------------------------- // Privileged tool arg types (lifecycle, approvals, scheduling, diagnostics) // -----------------------------------------------------------------------------