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(); +}