From a9e474c8fc7874cdb13b03ef8ffec0bead7c4fbd Mon Sep 17 00:00:00 2001 From: jif Date: Fri, 30 Oct 2026 13:25:51 +0000 Subject: [PATCH] Add an MCP adapter for message-board tools (#58341) ## What changed Add `codex-message-board-mcp` with a `MessageBoardMcpServer` that exposes selected message-board tools through `rmcp::ServerHandler`. Reuse the existing tool schemas, validation, permission checks, output budgets, and storage behavior. Bind the board connection and caller identity when constructing the adapter, leaving authentication and transport setup to the host. Return JSON text with matching structured content for object results, map validation and permission failures to MCP tool errors, and report unknown tools as protocol errors. Scope request deduplication to each MCP session so reused JSON-RPC IDs on new connections do not collide with earlier posts. Preserve the distinction between numeric and string request IDs. ## Testing Add integration tests for tool discovery, bounded results, caller identity validation, unknown tools, request deduplication within and across sessions, and read-only permission denials. Exercise discovery and calls against both local and in-memory boards. GitOrigin-RevId: 6c62c340b24198a00d0d3d0487cb2ebc5c397b06 --- codex-rs/Cargo.lock | 16 + codex-rs/Cargo.toml | 2 + codex-rs/message-board-mcp/BUILD.bazel | 6 + codex-rs/message-board-mcp/Cargo.toml | 24 ++ codex-rs/message-board-mcp/README.md | 41 +++ codex-rs/message-board-mcp/src/lib.rs | 173 +++++++++++ codex-rs/message-board/Cargo.toml | 4 +- codex-rs/message-board/tests/local_board.rs | 3 + .../tests/local_board/mcp_tests.rs | 290 ++++++++++++++++++ 9 files changed, 558 insertions(+), 1 deletion(-) create mode 100644 codex-rs/message-board-mcp/BUILD.bazel create mode 100644 codex-rs/message-board-mcp/Cargo.toml create mode 100644 codex-rs/message-board-mcp/README.md create mode 100644 codex-rs/message-board-mcp/src/lib.rs create mode 100644 codex-rs/message-board/tests/local_board/mcp_tests.rs diff --git a/codex-rs/Cargo.lock b/codex-rs/Cargo.lock index 7b8ae40ef4..deb61feea9 100644 --- a/codex-rs/Cargo.lock +++ b/codex-rs/Cargo.lock @@ -4164,12 +4164,14 @@ dependencies = [ "caseless", "chrono", "codex-attachment-store", + "codex-message-board-mcp", "codex-protocol", "codex-state", "codex-tools", "codex-utils-output-truncation", "futures", "pretty_assertions", + "rmcp", "serde", "serde_json", "sqlx", @@ -4179,6 +4181,20 @@ dependencies = [ "uuid", ] +[[package]] +name = "codex-message-board-mcp" +version = "0.0.0" +dependencies = [ + "codex-attachment-store", + "codex-message-board", + "codex-protocol", + "codex-tools", + "codex-utils-output-truncation", + "rmcp", + "serde_json", + "uuid", +] + [[package]] name = "codex-message-history" version = "0.0.0" diff --git a/codex-rs/Cargo.toml b/codex-rs/Cargo.toml index b13c02bb5a..8813801e20 100644 --- a/codex-rs/Cargo.toml +++ b/codex-rs/Cargo.toml @@ -92,6 +92,7 @@ members = [ "memories/write", "mermaid", "message-board", + "message-board-mcp", "model-provider-info", "mxc-sandbox", "models-manager", @@ -220,6 +221,7 @@ codex-connectors = { path = "connectors" } codex-agent-message-board-client = { path = "agent-message-board-client" } codex-agent-message-board-extension = { path = "ext/agent-message-board" } codex-message-board = { path = "message-board" } +codex-message-board-mcp = { path = "message-board-mcp" } codex-connectors-extension = { path = "ext/connectors" } codex-context-fragments = { path = "context-fragments" } codex-core = { path = "core" } diff --git a/codex-rs/message-board-mcp/BUILD.bazel b/codex-rs/message-board-mcp/BUILD.bazel new file mode 100644 index 0000000000..3938286b20 --- /dev/null +++ b/codex-rs/message-board-mcp/BUILD.bazel @@ -0,0 +1,6 @@ +load("//:defs.bzl", "codex_rust_crate") + +codex_rust_crate( + name = "message-board-mcp", + crate_name = "codex_message_board_mcp", +) diff --git a/codex-rs/message-board-mcp/Cargo.toml b/codex-rs/message-board-mcp/Cargo.toml new file mode 100644 index 0000000000..a4841b5d22 --- /dev/null +++ b/codex-rs/message-board-mcp/Cargo.toml @@ -0,0 +1,24 @@ +[package] +edition.workspace = true +license.workspace = true +name = "codex-message-board-mcp" +version.workspace = true + +[lib] +name = "codex_message_board_mcp" +path = "src/lib.rs" +test = false +doctest = false + +[lints] +workspace = true + +[dependencies] +codex-attachment-store = { workspace = true } +codex-message-board = { workspace = true } +codex-protocol = { workspace = true } +codex-tools = { workspace = true } +codex-utils-output-truncation = { workspace = true } +rmcp = { workspace = true, features = ["server"] } +serde_json = { workspace = true } +uuid = { workspace = true, features = ["v7"] } diff --git a/codex-rs/message-board-mcp/README.md b/codex-rs/message-board-mcp/README.md new file mode 100644 index 0000000000..ad7970be1b --- /dev/null +++ b/codex-rs/message-board-mcp/README.md @@ -0,0 +1,41 @@ +# Message-board MCP adapter + +`MessageBoardMcpServer` implements `rmcp::ServerHandler` over the existing +`codex-message-board` tool factory. It reuses the selected tools' descriptions, +input schemas, validation, permission checks, output budgets and storage behavior. +It has no dependency on `codex-core`. + +A host supplies an authorized board connection, caller ID and caller path, then +serves one adapter instance per MCP session using its chosen `rmcp` transport: + +```rust,ignore +use codex_message_board::MessageBoardTool; +use codex_message_board_mcp::MessageBoardMcpServer; +use rmcp::ServiceExt; + +let server = MessageBoardMcpServer::new( + &MessageBoardTool::DISCUSSION, + board, + caller, + caller_path, +); +let running = server.serve(transport).await?; +running.waiting().await?; +``` + +The host owns transport setup and authentication. Tool arguments and MCP metadata +cannot change the bound board or caller. Tool selection does not grant access; +for example, a read-only `LocalBoardConnection` still rejects writes. + +Posts retain the board's request deduplication within a session. Each new adapter +has a fresh request scope so reused JSON-RPC IDs on a new connection do not collide +with earlier posts. This does not provide deduplication across reconnects. + +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, +start turns, or change Core delivery. A host can adopt the adapter without +changing the native board integration. diff --git a/codex-rs/message-board-mcp/src/lib.rs b/codex-rs/message-board-mcp/src/lib.rs new file mode 100644 index 0000000000..77569b0c36 --- /dev/null +++ b/codex-rs/message-board-mcp/src/lib.rs @@ -0,0 +1,173 @@ +//! MCP transport adapter for the shared board's existing schemas and handlers. +//! The host binds board access and caller identity before serving a connection. +//! Storage, permissions and notification delivery remain owned by the board. + +use codex_attachment_store::InlineAttachmentStore; +use codex_message_board::AgentMessageBoard; +use codex_message_board::MessageBoardTool; +use codex_message_board::message_board_tools; +use codex_protocol::AgentPath; +use codex_protocol::ThreadId; +use codex_tools::ConversationHistory; +use codex_tools::FunctionCallError; +use codex_tools::NoopTurnItemEmitter; +use codex_tools::ToolCall; +use codex_tools::ToolCallSource; +use codex_tools::ToolExecutor; +use codex_tools::ToolPayload; +use codex_tools::ToolSpec; +use codex_utils_output_truncation::TruncationPolicy; +use rmcp::ErrorData; +use rmcp::ServerHandler; +use rmcp::model::CallToolRequestParams; +use rmcp::model::CallToolResponse; +use rmcp::model::CallToolResult; +use rmcp::model::ContentBlock; +use rmcp::model::Implementation; +use rmcp::model::ListToolsResult; +use rmcp::model::PaginatedRequestParams; +use rmcp::model::ServerCapabilities; +use rmcp::model::ServerConfig; +use rmcp::model::Tool; +use rmcp::service::RequestContext; +use rmcp::service::RoleServer; +use std::sync::Arc; +use uuid::Uuid; + +/// A caller-bound MCP service, usable with any `rmcp` server transport. +/// +/// Create one instance per MCP session. The host must authenticate the caller +/// and supply its authoritative path and board connection; request arguments or +/// metadata never select an identity. Selecting tools does not grant permissions. +/// The board's host continues to handle notifications; this service adds no +/// Core delivery, lifecycle registration, or MCP push notifications. +pub struct MessageBoardMcpServer { + tools: Vec ToolExecutor>>>, + invocation_scope: String, +} + +impl MessageBoardMcpServer { + pub fn new( + tools: &[MessageBoardTool], + board: Arc, + caller: ThreadId, + caller_path: AgentPath, + ) -> Self { + Self { + tools: message_board_tools( + tools, + board, + caller, + caller_path, + /*namespace*/ None, + /*namespace_description*/ "", + /*tool_messages*/ None, + ), + // JSON-RPC request IDs are connection-local, unlike board request IDs. + invocation_scope: Uuid::now_v7().to_string(), + } + } +} + +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")), + ) + } + + async fn list_tools( + &self, + 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)) + } + + async fn call_tool( + &self, + request: CallToolRequestParams, + context: RequestContext, + ) -> Result { + let executor = self + .tools + .iter() + .find(|executor| executor.tool_name().name == request.name) + .ok_or_else(|| { + ErrorData::invalid_params("Unknown message-board tool", /*data*/ None) + })?; + // Serialization preserves the distinction between numeric and string IDs. + let call_id = serde_json::to_string(&context.id).map_err(|_| { + ErrorData::internal_error("Invalid MCP request ID", /*data*/ None) + })?; + let payload = ToolPayload::Function { + arguments: serde_json::Value::Object(request.arguments.unwrap_or_default()).to_string(), + }; + let call = ToolCall { + turn_id: self.invocation_scope.clone(), + call_id, + tool_name: executor.tool_name(), + model: String::new(), + codex_turn_metadata: None, + // Let the shared handler impose its existing response budget. + truncation_policy: TruncationPolicy::Bytes(usize::MAX), + source: ToolCallSource::Direct, + conversation_history: ConversationHistory::default(), + turn_item_emitter: Arc::new(NoopTurnItemEmitter), + attachment_store: Arc::new(InlineAttachmentStore), + ephemeral: false, + environments: Vec::new(), + payload: payload.clone(), + }; + let result = match executor.handle(call).await { + Ok(output) => { + let value = output.code_mode_result(&payload); + let mut result = if value.is_object() { + CallToolResult::structured(value) + } else { + // Legacy MCP structuredContent must be an object (upsert returns null). + CallToolResult::success(vec![ContentBlock::text(value.to_string())]) + }; + result.is_error = Some(!output.success_for_logging()); + result + } + Err(FunctionCallError::RespondToModel(message)) => { + CallToolResult::error(vec![ContentBlock::text(message)]) + } + Err(FunctionCallError::Fatal(_)) => { + return Err(ErrorData::internal_error( + "Message-board tool failed", + /*data*/ None, + )); + } + }; + Ok(result.into()) + } +} diff --git a/codex-rs/message-board/Cargo.toml b/codex-rs/message-board/Cargo.toml index 52f991c07d..57a5c6fc57 100644 --- a/codex-rs/message-board/Cargo.toml +++ b/codex-rs/message-board/Cargo.toml @@ -29,7 +29,9 @@ uuid = { workspace = true, features = ["serde", "v7"] } [dev-dependencies] codex-attachment-store = { workspace = true } +codex-message-board-mcp = { workspace = true } codex-utils-output-truncation = { workspace = true } pretty_assertions = { workspace = true } +rmcp = { workspace = true, features = ["client", "server", "transport-async-rw"] } tempfile = { workspace = true } -tokio = { workspace = true, features = ["macros", "rt-multi-thread"] } +tokio = { workspace = true, features = ["io-util", "macros", "rt-multi-thread"] } diff --git a/codex-rs/message-board/tests/local_board.rs b/codex-rs/message-board/tests/local_board.rs index a676766200..9e11ebde85 100644 --- a/codex-rs/message-board/tests/local_board.rs +++ b/codex-rs/message-board/tests/local_board.rs @@ -2199,3 +2199,6 @@ async fn storage_namespaces_preserve_isolation_and_reopen() -> Result<()> { } Ok(()) } + +#[path = "local_board/mcp_tests.rs"] +mod mcp; diff --git a/codex-rs/message-board/tests/local_board/mcp_tests.rs b/codex-rs/message-board/tests/local_board/mcp_tests.rs new file mode 100644 index 0000000000..598519b5ed --- /dev/null +++ b/codex-rs/message-board/tests/local_board/mcp_tests.rs @@ -0,0 +1,290 @@ +//! Exercises the MCP adapter against the existing board and host fixtures. + +use super::*; +use codex_message_board_mcp::MessageBoardMcpServer; +use codex_tools::ToolSpec; +use pretty_assertions::assert_eq; +use rmcp::ServerHandler; +use rmcp::ServiceExt; +use rmcp::model::CallToolRequestParams; +use rmcp::model::CallToolResponse; +use rmcp::model::CallToolResult; +use rmcp::model::RequestId; +use rmcp::model::Tool; +use rmcp::service::RequestContext; +use serde_json::json; + +#[tokio::test] +async fn discovery_and_calls_reuse_board_schemas_identity_and_bounded_results() { + for backend in [Backend::Local, Backend::InMemory] { + let dir = tempfile::tempdir().unwrap(); + let sqlite = SqliteConfig::new_for_testing(dir.path().to_path_buf().try_into().unwrap()); + let caller = ThreadId::new(); + let path = AgentPath::root().join("worker").unwrap(); + let host = Arc::new(Host { + clock: AtomicI64::default(), + fail_clock: AtomicBool::new(false), + agent_path_calls: AtomicUsize::default(), + members: [(caller, path.clone())].into(), + active: AtomicBool::new(true), + fail_notifications: AtomicBool::new(false), + notifications: Mutex::default(), + }); + let board = backend.open(&sqlite, caller.into(), host).await; + let native = message_board_tools( + &MessageBoardTool::DISCUSSION, + board.clone(), + caller, + path.clone(), + /*namespace*/ None, + "", + /*tool_messages*/ None, + ); + let adapter = MessageBoardMcpServer::new( + &MessageBoardTool::DISCUSSION, + board.clone(), + caller, + path.clone(), + ); + let (server_io, client_io) = tokio::io::duplex(/*max_buf_size*/ 65536); + let (server, client) = tokio::join!(adapter.serve(server_io), ().serve(client_io)); + let server = server.unwrap(); + let mut client = client.unwrap(); + let expected = native + .iter() + .map(|executor| { + let ToolSpec::Function(spec) = executor.spec() else { + panic!("expected function") + }; + Tool::new( + spec.name, + spec.description, + Arc::new( + serde_json::from_value(serde_json::to_value(spec.parameters).unwrap()) + .unwrap(), + ), + ) + }) + .collect::>(); + assert_eq!(client.list_all_tools().await.unwrap(), expected); + + let text = "é\"".repeat(6_000); + let posted = client + .call_tool( + CallToolRequestParams::new("post").with_arguments( + json!({"new_channel_name":"work", "text":text}) + .as_object() + .unwrap() + .clone(), + ), + ) + .await + .unwrap(); + assert_eq!(posted.is_error, Some(false)); + let metadata: PostMetadata = + serde_json::from_value(posted.structured_content.unwrap()).unwrap(); + assert_eq!(metadata.author, path); + let arguments = json!({"message_id":metadata.message_id}); + let read = client + .call_tool( + CallToolRequestParams::new("read_post") + .with_arguments(arguments.as_object().unwrap().clone()), + ) + .await + .unwrap(); + let payload = board_tool_call("read_post", arguments); + let native_read = native + .iter() + .find(|executor| executor.tool_name().name == "read_post") + .unwrap() + .handle(payload.clone()) + .await + .unwrap(); + let mut expected = + CallToolResult::structured(native_read.code_mode_result(&payload.payload)); + // The negotiated legacy wire format omits the newer resultType discriminator. + expected.result_type = None; + assert_eq!(read, expected); + let value = read.structured_content.unwrap(); + assert!(value.to_string().len() <= 8_000); + let content: PostContent = serde_json::from_value(value).unwrap(); + assert!(content.next_offset_chars < content.n_chars); + assert_eq!( + content.text, + text.chars() + .take(content.next_offset_chars) + .collect::() + ); + + // Identity is bound by the host, never supplied in model arguments. + let invalid = client + .call_tool( + CallToolRequestParams::new("post").with_arguments( + json!({"new_channel_name":"spoofed", "text":"bad", "caller":ThreadId::new()}) + .as_object() + .unwrap() + .clone(), + ), + ) + .await + .unwrap(); + assert_eq!(invalid.is_error, Some(true)); + let channels = board + .list_channels( + caller, + ChannelQuery { + query: Some("spoofed".into()), + direction: SortDirection::NewestFirst, + page: PageRequest::default(), + }, + ) + .await + .unwrap(); + assert_eq!(channels.channels.results, Vec::new()); + assert!( + client + .call_tool(CallToolRequestParams::new("unknown")) + .await + .is_err() + ); + client.close().await.unwrap(); + server.waiting().await.unwrap(); + } +} + +#[tokio::test] +async fn request_identity_deduplicates_within_a_session_but_not_across_sessions() { + let caller = ThreadId::new(); + let host = Arc::new(Host { + clock: AtomicI64::default(), + fail_clock: AtomicBool::new(false), + agent_path_calls: AtomicUsize::default(), + members: [(caller, AgentPath::root())].into(), + active: AtomicBool::new(true), + fail_notifications: AtomicBool::new(false), + notifications: Mutex::default(), + }); + let board = Arc::new(InMemoryAgentMessageBoard::new(caller.into(), host)); + board + .create_channel( + caller, + CreateChannelRequest { + channel_name: "general".into(), + description: None, + subscription: SubscriptionChange::Unsubscribe, + }, + ) + .await + .unwrap(); + let mut posts = Vec::new(); + for _ in 0..2 { + let adapter = MessageBoardMcpServer::new( + &[MessageBoardTool::Post], + board.clone(), + caller, + AgentPath::root(), + ); + let (server_io, client_io) = tokio::io::duplex(/*max_buf_size*/ 65536); + let (server, client) = tokio::join!(adapter.serve(server_io), ().serve(client_io)); + let server = server.unwrap(); + let mut client = client.unwrap(); + // Use the handler's real session with explicit IDs to exercise retransmission + // and distinguish the JSON-RPC number 7 from the string "7". + for id in [RequestId::Number(7), RequestId::String("7".into())] { + let request = CallToolRequestParams::new("post").with_arguments( + json!({"channel_name":"general", "text":"same request"}) + .as_object() + .unwrap() + .clone(), + ); + let context = RequestContext::new(id, server.peer().clone()); + let first = server + .service() + .call_tool(request.clone(), context.clone()) + .await + .unwrap(); + let retry = server.service().call_tool(request, context).await.unwrap(); + let CallToolResponse::Complete(first) = first else { + panic!("expected completed call") + }; + let CallToolResponse::Complete(retry) = retry else { + panic!("expected completed retry") + }; + assert_eq!(first.is_error, Some(false)); + assert_eq!(first, retry); + posts.push(first); + } + client.close().await.unwrap(); + server.waiting().await.unwrap(); + } + for (index, post) in posts.iter().enumerate() { + assert!(posts[..index].iter().all(|previous| previous != post)); + } + let stored = board + .search_posts( + caller, + PostQuery { + channel_name: Some("general".into()), + query: Some("same request".into()), + after_message_id: None, + author: None, + max_chars_per_post: NonZeroU32::new(100).unwrap(), + page: PageRequest::default(), + }, + ) + .await + .unwrap(); + assert_eq!(stored.results.len(), 4); +} + +#[tokio::test] +async fn read_only_connection_returns_the_existing_structured_permission_denial() { + let dir = tempfile::tempdir().unwrap(); + let sqlite = SqliteConfig::new_for_testing(dir.path().to_path_buf().try_into().unwrap()); + let caller = ThreadId::new(); + let host = Arc::new(Host { + clock: AtomicI64::default(), + fail_clock: AtomicBool::new(false), + agent_path_calls: AtomicUsize::default(), + members: [(caller, AgentPath::root())].into(), + active: AtomicBool::new(true), + fail_notifications: AtomicBool::new(false), + notifications: Mutex::default(), + }); + let board = Arc::new( + LocalBoard::new(sqlite, caller.into()) + .read_only() + .open(host) + .await + .unwrap(), + ); + let adapter = MessageBoardMcpServer::new( + &MessageBoardTool::DISCUSSION, + board, + caller, + AgentPath::root(), + ); + let (server_io, client_io) = tokio::io::duplex(/*max_buf_size*/ 65536); + let (server, client) = tokio::join!(adapter.serve(server_io), ().serve(client_io)); + let server = server.unwrap(); + let mut client = client.unwrap(); + let result = client + .call_tool( + CallToolRequestParams::new("post").with_arguments( + json!({"new_channel_name":"blocked", "text":"bad"}) + .as_object() + .unwrap() + .clone(), + ), + ) + .await + .unwrap(); + let mut expected = CallToolResult::structured_error(json!({"error": { + "code":"permission_denied", "message":"This board connection is read-only.", + "action":"channel.post", "resource":format!("board:{}", SessionId::from(caller)), + }})); + expected.result_type = None; + assert_eq!(result, expected); + client.close().await.unwrap(); + server.waiting().await.unwrap(); +}