diff --git a/codex-rs/Cargo.lock b/codex-rs/Cargo.lock index deb61feea9..1b56929610 100644 --- a/codex-rs/Cargo.lock +++ b/codex-rs/Cargo.lock @@ -4185,13 +4185,27 @@ dependencies = [ name = "codex-message-board-mcp" version = "0.0.0" dependencies = [ + "anyhow", + "axum", + "chrono", + "clap", "codex-attachment-store", "codex-message-board", "codex-protocol", + "codex-state", "codex-tools", + "codex-utils-absolute-path", + "codex-utils-cargo-bin", "codex-utils-output-truncation", + "codex-utils-process", + "futures", "rmcp", + "serde", "serde_json", + "sqlx", + "tempfile", + "tokio", + "tokio-util", "uuid", ] diff --git a/codex-rs/core/BUILD.bazel b/codex-rs/core/BUILD.bazel index 4c04c5d6b4..9b2f933b97 100644 --- a/codex-rs/core/BUILD.bazel +++ b/codex-rs/core/BUILD.bazel @@ -11,6 +11,7 @@ codex_rust_crate( "//codex-rs/bwrap:bwrap", "//codex-rs/code-mode-host:codex-code-mode-host", "//codex-rs/linux-sandbox:codex-linux-sandbox", + "//codex-rs/message-board-mcp:codex-message-board-mcp", "//codex-rs/rmcp-client:test_stdio_server", "//codex-rs/rmcp-client:test_streamable_http_server", "//codex-rs/cli:codex", diff --git a/codex-rs/core/tests/suite/scenarios_agent_message_board.rs b/codex-rs/core/tests/suite/scenarios_agent_message_board.rs index 8df6c97235..dd1fbe82f5 100644 --- a/codex-rs/core/tests/suite/scenarios_agent_message_board.rs +++ b/codex-rs/core/tests/suite/scenarios_agent_message_board.rs @@ -29,6 +29,9 @@ use serde_json::Value; use serde_json::json; use tokio::sync::oneshot; +#[path = "scenarios_agent_message_board_http.rs"] +mod http; + #[path = "scenarios_agent_message_board_remote.rs"] mod remote; diff --git a/codex-rs/core/tests/suite/scenarios_agent_message_board_http.rs b/codex-rs/core/tests/suite/scenarios_agent_message_board_http.rs new file mode 100644 index 0000000000..54c8c7fe8f --- /dev/null +++ b/codex-rs/core/tests/suite/scenarios_agent_message_board_http.rs @@ -0,0 +1,261 @@ +//! Two real agents share the detached board through ordinary HTTP MCP configuration. + +use super::*; +use pretty_assertions::assert_eq; +use std::sync::Arc; +use tokio::io::AsyncBufReadExt; + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn http_board_is_shared_with_children() -> anyhow::Result<()> { + let server = responses::start_mock_server().await; + let home = Arc::new(tempfile::tempdir()?); + let sqlite = codex_state::SqliteConfig::new_for_testing(home.path().to_path_buf().try_into()?); + let _state = codex_state::StateRuntime::init(sqlite.clone(), "openai".into()).await?; + let command = codex_utils_cargo_bin::cargo_bin("codex-message-board-mcp")?; + let mut process = + tokio::process::Command::from(codex_utils_process::background_command(command)) + .arg("--sqlite-home") + .arg(sqlite.home()) + .args(["--listen", "127.0.0.1:0", "--allowed-host", "board.example"]) + .env("CODEX_MESSAGE_BOARD_TOKEN", "http-board-test-token") + .stdout(std::process::Stdio::piped()) + .kill_on_drop(true) + .spawn()?; + let ready = tokio::io::BufReader::new(process.stdout.take().context("server stdout")?) + .lines() + .next_line() + .await? + .context("server readiness")?; + let ready: Value = serde_json::from_str(&ready)?; + let endpoint = json!({"url":ready["url"],"http_headers":{"Authorization":"Bearer http-board-test-token","Host":"board.example"},"required":true,"default_tools_approval_mode":"approve"}); + let configured_endpoint = endpoint.clone(); + let configured_sqlite = sqlite.clone(); + let configure_board = move |config: &mut codex_core::config::Config| { + configure(config); + config.sqlite = configured_sqlite.clone(); + config + .features + .disable(Feature::AgentMessageBoard) + .expect("disable native board"); + let mut servers = config.mcp_servers.get().clone(); + servers.insert( + "codex_message_board".into(), + serde_json::from_value(configured_endpoint.clone()).expect("HTTP config"), + ); + config.mcp_servers.set(servers).expect("configure endpoint"); + }; + let mut builder = test_codex() + .with_home(home) + .with_config(configure_board.clone()); + let root = builder.build_with_auto_env(&server).await?; + responses::mount_sse_sequence( + &server, + vec![ + tool( + "spawn-worker", + "spawn_agent", + json!({"task_name":"worker","message":"Say ready.","fork_turns":"none"}), + ), + done(), + done(), + ], + ) + .await; + root.submit_turn("Spawn a worker and finish your turn.") + .await?; + let root_id = root.session_configured.thread_id; + let child_id = root + .thread_manager + .list_thread_ids() + .await + .into_iter() + .find(|id| *id != root_id) + .expect("child runtime"); + let child = root.thread_manager.get_thread(child_id).await?; + wait_for_event(&child, |event| matches!(event, EventMsg::TurnComplete(_))).await; + + let board_turn = |id: &str, name: &str, arguments: Value| { + vec![ + sse(vec![ + responses::ev_tool_search_call("find-board", &json!({"query":name,"limit":1})), + ev_completed("find-board"), + ]), + sse(vec![ + ev_function_call_with_namespace( + id, + "mcp__codex_message_board", + name, + &arguments.to_string(), + ), + ev_completed(id), + ]), + done(), + ] + }; + let board_result = |request: &ResponsesRequest, id: &str| -> anyhow::Result { + let output = request + .function_call_output_text(id) + .with_context(|| format!("missing board result {id}"))?; + let json = output + .split_once("\nOutput:\n") + .map_or(output.as_str(), |(_, json)| json); + serde_json::from_str(json).with_context(|| format!("board result: {output}")) + }; + let posted = responses::mount_sse_sequence( + &server, + board_turn( + "worker-post", + "post", + json!({"new_channel_name":"design","text":"Worker's durable decision."}), + ), + ) + .await; + child + .start_or_steer_turn(TurnInputRequest::user_input(vec![UserInput::Text { + text: "Find the board tools and post your decision.".into(), + text_elements: vec![], + }])) + .await?; + wait_for_event(&child, |event| matches!(event, EventMsg::TurnComplete(_))).await; + let post = board_result( + posted.requests().last().context("child follow-up")?, + "worker-post", + )?; + assert_eq!(post["author"], "/root/worker"); + child.shutdown_and_wait().await?; + + // A restarted parent discovers and reads the same persisted board. + let resumed = test_codex() + .with_config(configure_board) + .restart_with_auto_env(&server, &root) + .await?; + // A read must not initialize the schema or acquire the database writer lock. + let pool = sqlite + .open_read_write_pool(&sqlite.home().join("agent_message_board_1.sqlite")) + .await?; + let writer = pool.begin_with("BEGIN IMMEDIATE").await?; + let read = responses::mount_sse_sequence( + &server, + board_turn( + "root-read", + "search_posts", + json!({"channel_name":"design"}), + ), + ) + .await; + resumed + .submit_turn("Find the board tools and read the worker's decision.") + .await?; + writer.rollback().await?; + let result = board_result( + read.requests().last().context("parent follow-up")?, + "root-read", + )?; + assert_eq!( + serde_json::from_value::>( + result + )?, + codex_message_board::Page { + results: vec![codex_message_board::PostPreview { + metadata: serde_json::from_value(post.clone())?, + text_preview: "Worker's durable decision.".into(), + n_chars: "Worker's durable decision.".chars().count(), + truncated: false, + }], + next_cursor: None, + } + ); + for request in [&posted.requests()[1], &read.requests()[1]] { + assert!( + request + .tool_search_output("find-board") + .to_string() + .contains("mcp__codex_message_board") + ); + assert!(request.tool_by_name("collaboration", "post").is_none()); + } + let timestamp = post["created_at"].as_str().context("post timestamp")?; + let mut bodies = posted + .requests() + .iter() + .chain(read.requests().iter()) + .map(ResponsesRequest::body_json) + .collect::>(); + for body in &mut bodies { + for item in body["input"].as_array_mut().context("request input")? { + if item["type"] == "function_call_output" + && let Some(output) = item["output"].as_str() + { + item["output"] = Value::String( + context_snapshot::normalize_json_lines(output) + .replace(timestamp, ""), + ); + } + } + } + insta::assert_snapshot!( + "agent_message_board_mcp_context", + context_snapshot::format_context_snapshot( + "A child posts through the HTTP MCP executable; its restarted parent discovers the same board and reads the decision.", + &bodies.iter().map(SnapshotEntry::body).collect::>(), + &ContextSnapshotOptions::default().rewrite_known_segments(), + ) + ); + // The endpoint runs on the app host, including when the executor is remote. + // Exercise authentication and membership through a host-local HTTP client. + use codex_rmcp_client::RmcpClient; + let url = endpoint["url"].as_str().context("endpoint URL")?; + for (token, host) in [ + ("wrong-token", "board.example"), + ("http-board-test-token", "unlisted.example"), + ("http-board-test-token", "127.0.0.1"), + ] { + let client = RmcpClient::new_streamable_http_client( + "board-auth-test", + url, + Some(token.to_owned()), + Some([("Host".into(), host.into())].into()), + /*env_http_headers*/ None, + codex_config::types::OAuthCredentialsStoreMode::File, + Default::default(), + codex_exec_server::Environment::default_for_tests().get_http_client(), + /*auth_provider*/ None, + ) + .await?; + let initialized = client + .initialize( + Default::default(), + /*timeout*/ None, + Box::new(|_, _| Box::pin(async { anyhow::bail!("unexpected elicitation") })), + ) + .await; + if token == "wrong-token" || host == "unlisted.example" { + assert!(initialized.is_err()); + continue; + } + initialized?; + for metadata in [ + None, + Some(json!({"threadId":root_id, "sessionId":child_id})), + ] { + let error = client + .call_tool( + "get_channels".into(), + /*arguments*/ None, + |_| Ok(metadata.clone()), + /*timeout*/ None, + ) + .await + .expect_err("invalid identity must fail"); + assert_eq!( + codex_rmcp_client::mcp_error(&error).map(|error| error.code), + Some(rmcp::model::ErrorCode::INVALID_PARAMS) + ); + } + client.shutdown().await; + } + resumed.codex.shutdown_and_wait().await?; + process.kill().await?; + process.wait().await?; + Ok(()) +} diff --git a/codex-rs/core/tests/suite/snapshots/all__suite__scenarios__agent_message_board__http__agent_message_board_mcp_context.snap b/codex-rs/core/tests/suite/snapshots/all__suite__scenarios__agent_message_board__http__agent_message_board_mcp_context.snap new file mode 100644 index 0000000000..0e7eea4d3a --- /dev/null +++ b/codex-rs/core/tests/suite/snapshots/all__suite__scenarios__agent_message_board__http__agent_message_board_mcp_context.snap @@ -0,0 +1,102 @@ +--- +source: core/tests/suite/scenarios_agent_message_board_http.rs +expression: "context_snapshot::format_context_snapshot(\"A child posts through the HTTP MCP executable; its restarted parent discovers the same board and reads the decision.\",\n&bodies.iter().map(SnapshotEntry::body).collect::>(),\n&ContextSnapshotOptions::default().rewrite_known_segments(),)" +--- +Scenario: A child posts through the HTTP MCP executable; its restarted parent discovers the same board and reads the decision. + +## Window 1 +-- request 1 (request) -- +00:message/developer: + +01:message/developer[2]: + [01] + [02] + Deferred tool namespaces: + - mcp__codex_message_board + +02:message/developer: + You are an agent in a team of agents collaborating to complete a task. + + You can spawn sub-agents to handle subtasks, and those sub-agents can spawn their own sub-agents. All ag...nd capable, and have access to the same set of tools. [hash=876a93c54d3f61d6] + + You can use `spawn_agent` to create a new agent, `followup_task` to give an existing agent a new task an... `send_message` to pass a message to a running agent. [hash=824c12425c7626dc] + Child agents can also spawn their own sub-agents. + + When you provide a response in the final channel, that content is immediately delivered back to your parent agent. + + - All agents use the same current working directory. + - As a result, edits made by one agent are immediately visible to all other agents. + + When calling `wait_agent`, prefer longer waits (minutes) to avoid busy polling. + + There are 4 available concurrency slots, meaning that up to 4 agents can be active at once, including you. + + Full-history forks (`fork_turns` omitted or `"all"`) inherit the parent model and reasoning effort and d...uctions; when doing so, set `fork_turns` to `"none"`. [hash=18ccf7e93ecabab9] +03:message/developer: + +04:message/user: + +05:agent_message +06:message/assistant: + Done. +07:message/user: + Find the board tools and post your decision. +-- request 2 (request) -- +08:tool_search_call +09:tool_search_output +-- request 3 (request) -- +10:function_call/mcp__codex_message_board.post:{"new_channel_name":"design","text":"Worker's durable decision."} +11:function_call_output:Wall time: seconds + Output: + {"author":"/root/worker","channel_name":"design","created_at":"","message_id":"","thread_id":""} + +## Window 2 (after request 3: input diverged at item 02, settings changed (tools)) +-- request 4 (request) -- +00:message/developer: + +01:message/developer[2]: + [01] + [02] + Deferred tool namespaces: + - mcp__codex_message_board + +02:message/developer: + You are `/root`, the primary agent in a team of agents collaborating to fulfill the user's goals. + + At the start of your turn, you are the active agent. + You can spawn sub-agents to handle subtasks, and those sub-agents can spawn their own sub-agents. + All agents in the team, including the agents that you can assign tasks to, are equally intelligent and capable, and have access to the same set of tools. + + You can use `spawn_agent` to create a new agent, `followup_task` to give an existing agent a new task an...message to a running agent without triggering a turn. [hash=7d40ff8b58977202] + Child agents can also spawn their own sub-agents. + + - All agents use the same current working directory. + - As a result, edits made by one agent are immediately visible to all other agents. + + When calling `wait_agent`, prefer longer waits (minutes) to avoid busy polling. + + There are 4 available concurrency slots, meaning that up to 4 agents can be active at once, including you. + + Full-history forks (`fork_turns` omitted or `"all"`) inherit the parent model and reasoning effort and d...uctions; when doing so, set `fork_turns` to `"none"`. [hash=18ccf7e93ecabab9] +03:message/developer: + +04:message/user: + +05:message/user: + Spawn a worker and finish your turn. +06:function_call/collaboration.spawn_agent:{"fork_turns":"none","message":"Say ready.","task_name":"worker"} +07:function_call_output:{"task_name":"/root/worker"} +08:message/assistant: + Done. +09:message/user: + +10:message/user: + Find the board tools and read the worker's decision. +-- request 5 (request) -- +11:tool_search_call +12:tool_search_output +-- request 6 (request) -- +13:function_call/mcp__codex_message_board.search_posts:{"channel_name":"design"} +14:function_call_output:Wall time: seconds + Output: + {"has_more":false,"n_returned":1,"next_cursor":null,"results":[{"author":"/root/worker","channel_name":"...ecision.","thread_id":"","truncated":false}]} [hash=0322071c1fb9baa5] diff --git a/codex-rs/message-board-mcp/Cargo.toml b/codex-rs/message-board-mcp/Cargo.toml index a4841b5d22..27febf6e2f 100644 --- a/codex-rs/message-board-mcp/Cargo.toml +++ b/codex-rs/message-board-mcp/Cargo.toml @@ -10,15 +10,36 @@ path = "src/lib.rs" test = false doctest = false +[[bin]] +name = "codex-message-board-mcp" +path = "src/main.rs" +test = false + [lints] workspace = true [dependencies] +anyhow = { workspace = true } +axum = { workspace = true, features = ["http1", "tokio"] } +chrono = { workspace = true } +clap = { workspace = true, features = ["derive"] } codex-attachment-store = { workspace = true } codex-message-board = { workspace = true } codex-protocol = { workspace = true } +codex-state = { workspace = true } codex-tools = { workspace = true } +codex-utils-absolute-path = { workspace = true } codex-utils-output-truncation = { workspace = true } -rmcp = { workspace = true, features = ["server"] } +futures = { workspace = true } +rmcp = { workspace = true, features = ["server", "transport-streamable-http-server"] } +serde = { workspace = true, features = ["derive"] } serde_json = { workspace = true } +sqlx = { workspace = true } +tokio = { workspace = true, features = ["macros", "net", "rt", "signal"] } +tokio-util = { workspace = true, features = ["rt"] } uuid = { workspace = true, features = ["v7"] } + +[dev-dependencies] +codex-utils-cargo-bin = { workspace = true } +codex-utils-process = { workspace = true } +tempfile = { workspace = true } diff --git a/codex-rs/message-board-mcp/README.md b/codex-rs/message-board-mcp/README.md index ad7970be1b..86d663f45c 100644 --- a/codex-rs/message-board-mcp/README.md +++ b/codex-rs/message-board-mcp/README.md @@ -35,7 +35,42 @@ Results include JSON text and, for object results, matching structured content. Board validation failures and permission denials use MCP tool errors; unknown tools use protocol errors. -Notifications still go through the board's existing `MessageBoardHost`. This -crate does not register agents, launch a server process, push MCP notifications, +Notifications still go through the board's existing `MessageBoardHost`. The adapter +does not register agents, push MCP notifications, start turns, or change Core delivery. A host can adopt the adapter without changing the native board integration. + +## HTTP server + +Build with `cargo build -p codex-message-board-mcp` from `codex-rs`. Set +`CODEX_MESSAGE_BOARD_TOKEN`, then run: + +```sh +target/debug/codex-message-board-mcp --sqlite-home /absolute/path/to/codex/sqlite-directory --listen 127.0.0.1:7777 +``` + +Connect with ordinary MCP config; keep the native `agent_message_board` feature disabled: + +```toml +[mcp_servers.codex_message_board] +url = "http://127.0.0.1:7777/mcp" +bearer_token_env_var = "CODEX_MESSAGE_BOARD_TOKEN" +required = true +``` + +The executable prints its URL and shuts down on Ctrl-C. It validates membership +against the existing read-only Codex registry and uses the existing SQLite board, +system time and no notification delivery. Core delivery is a separate change. + +The library's `serve_http(listener, token, config, bind)` exposes the shared +discussion tool contract at `/mcp`. Its async binding validates `(ThreadId, SessionId)` +and returns an existing `MessageBoardMcpServer`. SQLite wiring stays in the executable; +other implementations can reuse `AgentMessageBoard` and `MessageBoardHost`. + +Calls use Codex's `threadId` and `sessionId` metadata. The bearer authorizes a trusted +host to assert identity; use TLS outside loopback. Add `--allowed-host board.example.com` +for a remote hostname (repeat for multiple names); loopback hosts remain allowed. +Library callers configure this through `StreamableHttpServerConfig`. Normal MCP +approval policy applies. +Deduplication is scoped to MCP sessions, without guarantees across reconnects or for +stateless clients. Plugin packaging, defaults and in-memory launching are separate. diff --git a/codex-rs/message-board-mcp/src/http.rs b/codex-rs/message-board-mcp/src/http.rs new file mode 100644 index 0000000000..7c84d6ba21 --- /dev/null +++ b/codex-rs/message-board-mcp/src/http.rs @@ -0,0 +1,134 @@ +//! Streamable HTTP and trusted-host caller metadata for the shared MCP contract. +//! Backend selection and membership validation belong to the supplied adapter binding. + +use crate::MessageBoardMcpServer; +use crate::list_tools_result; +use crate::server_info; +use axum::Router; +use axum::extract::Request; +use axum::http::HeaderValue; +use axum::http::StatusCode; +use axum::http::header::AUTHORIZATION; +use axum::middleware; +use axum::middleware::Next; +use codex_message_board::MessageBoardTool; +use codex_protocol::SessionId; +use codex_protocol::ThreadId; +use rmcp::ErrorData; +use rmcp::ServerHandler; +use rmcp::model::CallToolRequestParams; +use rmcp::model::CallToolResponse; +use rmcp::model::ListToolsResult; +use rmcp::model::PaginatedRequestParams; +use rmcp::model::ServerConfig; +use rmcp::service::RequestContext; +use rmcp::service::RoleServer; +use rmcp::transport::StreamableHttpServerConfig; +use rmcp::transport::StreamableHttpService; +use rmcp::transport::streamable_http_server::session::local::LocalSessionManager; +use serde::Deserialize; +use std::future::Future; +use std::sync::Arc; +use tokio::net::TcpListener; + +/// Serves the discussion tools on `/mcp`. The binding must validate membership and +/// return the existing adapter with an authoritative board, caller and path. +/// The bearer authorizes a trusted host to assert `threadId` and `sessionId` metadata. +/// Use TLS at the deployment boundary outside loopback. Cancellation stops HTTP and MCP sessions. +pub async fn serve_http( + listener: TcpListener, + token: String, + config: StreamableHttpServerConfig, + bind: F, +) -> anyhow::Result<()> +where + F: Fn(ThreadId, SessionId) -> Fut + Clone + Send + Sync + 'static, + Fut: Future> + Send + 'static, +{ + anyhow::ensure!(!token.is_empty(), "MCP bearer token must not be empty"); + let mut authorization = HeaderValue::from_str(&format!("Bearer {token}"))?; + authorization.set_sensitive(true); + let cancellation = config.cancellation_token.clone(); + let app = Router::new() + .nest_service( + "/mcp", + StreamableHttpService::new( + move || { + Ok(HttpServer { + bind: bind.clone(), + invocation_scope: uuid::Uuid::now_v7().to_string(), + }) + }, + Arc::new(LocalSessionManager::default()), + config, + ), + ) + .layer(middleware::from_fn(move |request: Request, next: Next| { + let authorization = authorization.clone(); + async move { + if request.headers().get(AUTHORIZATION) != Some(&authorization) { + return Err(StatusCode::UNAUTHORIZED); + } + Ok(next.run(request).await) + } + })); + axum::serve(listener, app) + .with_graceful_shutdown(cancellation.cancelled_owned()) + .await?; + Ok(()) +} + +struct HttpServer { + bind: F, + invocation_scope: String, +} + +#[derive(Deserialize)] +#[serde(rename_all = "camelCase")] +struct Caller { + thread_id: ThreadId, + session_id: SessionId, +} + +impl ServerHandler for HttpServer +where + F: Fn(ThreadId, SessionId) -> Fut + Send + Sync + 'static, + Fut: Future> + Send + 'static, +{ + fn get_info(&self) -> ServerConfig { + server_info() + } + + async fn list_tools( + &self, + request: Option, + _context: RequestContext, + ) -> Result { + list_tools_result( + request, + MessageBoardTool::DISCUSSION + .into_iter() + .map(MessageBoardTool::spec), + ) + } + + async fn call_tool( + &self, + request: CallToolRequestParams, + context: RequestContext, + ) -> Result { + let caller: Caller = serde_json::from_value(serde_json::Value::Object( + context.meta.0.0.clone(), + )) + .map_err(|_| { + ErrorData::invalid_params( + "Calls require threadId and sessionId metadata", + /*data*/ None, + ) + })?; + let mut adapter = (self.bind)(caller.thread_id, caller.session_id).await?; + // Preserve retry identity across calls in this MCP session, scoped to the caller. + adapter.invocation_scope = format!("{}:{}", self.invocation_scope, caller.thread_id); + adapter.call_tool(request, context).await + } +} diff --git a/codex-rs/message-board-mcp/src/lib.rs b/codex-rs/message-board-mcp/src/lib.rs index 77569b0c36..7529b37591 100644 --- a/codex-rs/message-board-mcp/src/lib.rs +++ b/codex-rs/message-board-mcp/src/lib.rs @@ -1,7 +1,11 @@ -//! MCP transport adapter for the shared board's existing schemas and handlers. -//! The host binds board access and caller identity before serving a connection. +//! MCP adapters and HTTP transport for existing board tools. +//! Bound adapters fix caller identity; HTTP validates it against the configured host. //! Storage, permissions and notification delivery remain owned by the board. +mod http; + +pub use http::serve_http; + use codex_attachment_store::InlineAttachmentStore; use codex_message_board::AgentMessageBoard; use codex_message_board::MessageBoardTool; @@ -71,9 +75,7 @@ impl MessageBoardMcpServer { impl ServerHandler for MessageBoardMcpServer { fn get_info(&self) -> ServerConfig { - ServerConfig::new(ServerCapabilities::builder().enable_tools().build()).with_server_info( - Implementation::new("codex-message-board", env!("CARGO_PKG_VERSION")), - ) + server_info() } async fn list_tools( @@ -81,34 +83,7 @@ impl ServerHandler for MessageBoardMcpServer { request: Option, _context: RequestContext, ) -> Result { - if request.is_some_and(|request| request.cursor.is_some()) { - return Err(ErrorData::invalid_params( - "Message-board tools do not use pagination cursors", - /*data*/ None, - )); - } - let tools = self - .tools - .iter() - .map(|executor| { - let ToolSpec::Function(spec) = executor.spec() else { - return Err(ErrorData::internal_error( - "Message-board tool is not a function", - /*data*/ None, - )); - }; - let schema = serde_json::to_value(spec.parameters) - .and_then(serde_json::from_value) - .map_err(|_| { - ErrorData::internal_error( - "Invalid message-board schema", - /*data*/ None, - ) - })?; - Ok(Tool::new(spec.name, spec.description, Arc::new(schema))) - }) - .collect::, _>>()?; - Ok(ListToolsResult::with_all_items(tools)) + list_tools_result(request, self.tools.iter().map(|executor| executor.spec())) } async fn call_tool( @@ -171,3 +146,38 @@ impl ServerHandler for MessageBoardMcpServer { Ok(result.into()) } } + +fn server_info() -> ServerConfig { + ServerConfig::new(ServerCapabilities::builder().enable_tools().build()).with_server_info( + Implementation::new("codex-message-board", env!("CARGO_PKG_VERSION")), + ) +} + +fn list_tools_result( + request: Option, + specs: impl Iterator, +) -> Result { + if request.is_some_and(|request| request.cursor.is_some()) { + return Err(ErrorData::invalid_params( + "Message-board tools do not use pagination cursors", + /*data*/ None, + )); + } + let tools = specs + .map(|spec| { + let ToolSpec::Function(spec) = spec else { + return Err(ErrorData::internal_error( + "Message-board tool is not a function", + /*data*/ None, + )); + }; + let schema = serde_json::to_value(spec.parameters) + .and_then(serde_json::from_value) + .map_err(|_| { + ErrorData::internal_error("Invalid message-board schema", /*data*/ None) + })?; + Ok(Tool::new(spec.name, spec.description, Arc::new(schema))) + }) + .collect::, _>>()?; + Ok(ListToolsResult::with_all_items(tools)) +} diff --git a/codex-rs/message-board-mcp/src/main.rs b/codex-rs/message-board-mcp/src/main.rs new file mode 100644 index 0000000000..7ee3d52eb1 --- /dev/null +++ b/codex-rs/message-board-mcp/src/main.rs @@ -0,0 +1,99 @@ +//! Detached HTTP entry point; endpoint configuration and process lifetime belong to the deployer. + +mod persisted; + +use clap::Parser; +use codex_message_board::LocalAgentMessageBoard; +use codex_message_board::LocalBoardOptions; +use codex_message_board::MessageBoardHost; +use codex_message_board::MessageBoardTool; +use codex_message_board_mcp::MessageBoardMcpServer; +use codex_state::SqliteConfig; +use codex_utils_absolute_path::AbsolutePathBuf; +use rmcp::ErrorData; +use rmcp::transport::StreamableHttpServerConfig; +use std::sync::Arc; +use std::sync::Mutex; + +#[derive(Parser)] +struct Args { + /// Absolute directory containing Codex's state and message-board databases. + #[arg(long)] + sqlite_home: std::path::PathBuf, + #[arg(long, default_value = "127.0.0.1:7777")] + listen: std::net::SocketAddr, + /// Additional allowed HTTP Host names or host:port authorities. Loopback remains allowed. + #[arg(long = "allowed-host")] + allowed_hosts: Vec, + /// Name of the environment variable containing the trusted-host bearer token. + #[arg(long, default_value = "CODEX_MESSAGE_BOARD_TOKEN")] + bearer_token_env: String, +} + +#[tokio::main(flavor = "current_thread")] +async fn main() -> anyhow::Result<()> { + let args = Args::parse(); + let token = std::env::var(&args.bearer_token_env).map_err(|_| { + anyhow::anyhow!("MCP bearer token environment variable is missing or invalid") + })?; + let sqlite = SqliteConfig::from_sqlite_home_with_nfs_enforcement( + AbsolutePathBuf::try_from(args.sqlite_home)?, + /*enforce_delete_on_nfs*/ false, + ); + let registry = sqlite + .open_read_only_pool(&sqlite.state_db_path(), /*busy_timeout*/ None) + .await?; + let listener = tokio::net::TcpListener::bind(args.listen).await?; + println!( + "{}", + serde_json::json!({"url": format!("http://{}/mcp", listener.local_addr()?)}) + ); + let cancellation = tokio_util::sync::CancellationToken::new(); + let mut http = + StreamableHttpServerConfig::default().with_cancellation_token(cancellation.clone()); + http.allowed_hosts.extend(args.allowed_hosts); + // All sessions share one SQLite file. Retain a board between calls so the + // weak pool registry reuses its connections instead of initializing again. + let pool_owner = Arc::new(Mutex::new(None)); + let serving = + codex_message_board_mcp::serve_http(listener, token, http, move |caller, session_id| { + let pool_owner = Arc::clone(&pool_owner); + let sqlite = sqlite.clone(); + let host = Arc::new(persisted::PersistedHost { + registry: registry.clone(), + root: session_id.into(), + }); + async move { + let path = host.agent_path(caller).await.map_err(|error| { + ErrorData::invalid_params(error.to_string(), /*data*/ None) + })?; + let board = LocalAgentMessageBoard::open( + &sqlite, + &LocalBoardOptions::default(), + session_id, + host, + ) + .await + .map_err(|error| { + ErrorData::internal_error(error.to_string(), /*data*/ None) + })?; + *pool_owner + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) = Some(board.clone()); + Ok(MessageBoardMcpServer::new( + &MessageBoardTool::DISCUSSION, + Arc::new(board), + caller, + path, + )) + } + }); + tokio::pin!(serving); + tokio::select! { + result = &mut serving => result, + _ = tokio::signal::ctrl_c() => { + cancellation.cancel(); + serving.await + } + } +} diff --git a/codex-rs/message-board-mcp/src/persisted.rs b/codex-rs/message-board-mcp/src/persisted.rs new file mode 100644 index 0000000000..c1931f348b --- /dev/null +++ b/codex-rs/message-board-mcp/src/persisted.rs @@ -0,0 +1,99 @@ +//! Reads persisted membership without starting runtimes or modifying Codex's registry. +//! Standalone posts use system time; Core notification delivery is a separate integration. + +use chrono::DateTime; +use chrono::Utc; +use codex_message_board::MessageBoardHost; +use codex_message_board::NotificationDelivery; +use codex_message_board::PostPreview; +use codex_protocol::AgentPath; +use codex_protocol::ThreadId; +use codex_protocol::error::CodexErr; +use codex_protocol::error::Result; +use futures::future::BoxFuture; +use sqlx::SqlitePool; + +pub(super) struct PersistedHost { + pub registry: SqlitePool, + pub root: ThreadId, +} + +impl MessageBoardHost for PersistedHost { + fn agent_path(&self, caller: ThreadId) -> BoxFuture<'_, Result> { + Box::pin(async move { + let stored: Option> = + sqlx::query_scalar("SELECT agent_path FROM threads WHERE id = ?") + .bind(caller.to_string()) + .fetch_optional(&self.registry) + .await + .map_err(storage_error)?; + let path = match stored.ok_or(CodexErr::ThreadNotFound(caller))? { + Some(path) => AgentPath::from_string(path).map_err(CodexErr::InvalidRequest)?, + None if caller == self.root => AgentPath::root(), + None => { + return Err(CodexErr::InvalidRequest( + "caller has no persisted agent path".into(), + )); + } + }; + if self.resolve_agent(path.clone()).await? != caller { + return Err(CodexErr::InvalidRequest( + "caller does not belong to this board".into(), + )); + } + Ok(path) + }) + } + + fn resolve_agent(&self, path: AgentPath) -> BoxFuture<'_, Result> { + Box::pin(async move { + let ids: Vec = sqlx::query_scalar( + r#" +WITH RECURSIVE tree(id) AS ( + SELECT id FROM threads WHERE id = ? + UNION + SELECT child_thread_id FROM thread_spawn_edges JOIN tree ON parent_thread_id = tree.id +) +SELECT threads.id FROM tree JOIN threads ON threads.id = tree.id +WHERE (threads.id = ? AND ? = '/root' AND (agent_path IS NULL OR agent_path = '/root')) + OR (threads.id != ? AND agent_path = ?) +LIMIT 2 +"#, + ) + .bind(self.root.to_string()) + .bind(self.root.to_string()) + .bind(path.as_str()) + .bind(self.root.to_string()) + .bind(path.as_str()) + .fetch_all(&self.registry) + .await + .map_err(storage_error)?; + match ids.as_slice() { + [id] => ThreadId::from_string(id) + .map_err(|error| CodexErr::InvalidRequest(error.to_string())), + _ => Err(CodexErr::InvalidRequest( + "unknown or ambiguous agent path in this board".into(), + )), + } + }) + } + + fn current_time(&self, caller: ThreadId) -> BoxFuture<'_, Result>> { + Box::pin(async move { + self.agent_path(caller).await?; + Ok(Utc::now()) + }) + } + + fn notify( + &self, + _recipient: ThreadId, + _post: PostPreview, + ) -> BoxFuture<'_, Result> { + Box::pin(async { Ok(NotificationDelivery::SkippedInactive) }) + } +} + +fn storage_error(error: sqlx::Error) -> CodexErr { + CodexErr::Io(std::io::Error::other(error)) +} diff --git a/codex-rs/message-board-mcp/tests/startup.rs b/codex-rs/message-board-mcp/tests/startup.rs new file mode 100644 index 0000000000..b5b3c695c2 --- /dev/null +++ b/codex-rs/message-board-mcp/tests/startup.rs @@ -0,0 +1,17 @@ +//! The configured executable must not create a new Codex registry when pointed at the wrong directory. + +#[test] +fn requires_an_existing_registry_without_creating_databases() -> anyhow::Result<()> { + let directory = tempfile::tempdir()?; + let output = codex_utils_process::background_command(codex_utils_cargo_bin::cargo_bin( + "codex-message-board-mcp", + )?) + .env("CODEX_MESSAGE_BOARD_TOKEN", "startup-test-only-token") + .arg("--sqlite-home") + .arg(directory.path()) + .args(["--listen", "127.0.0.1:0"]) + .output()?; + assert!(!output.status.success()); + assert!(directory.path().read_dir()?.next().is_none()); + Ok(()) +} diff --git a/codex-rs/message-board/src/tools/selection.rs b/codex-rs/message-board/src/tools/selection.rs index cbce5dc7d9..e799b7563e 100644 --- a/codex-rs/message-board/src/tools/selection.rs +++ b/codex-rs/message-board/src/tools/selection.rs @@ -34,6 +34,14 @@ impl MessageBoardTool { Self::Post, ]; + /// Bundled schema for discovery before a caller has been bound to a board. + pub fn spec(self) -> codex_tools::ToolSpec { + super::spec::tool( + self, /*namespace*/ None, /*namespace_description*/ "", + /*description_override*/ None, + ) + } + pub(super) fn name(self) -> &'static str { match self { Self::CreateChannel => "create_channel",