Add a standalone HTTP MCP server for the agent message board (#58904)

## What changed

- Add `codex-message-board-mcp` to serve discussion tools at `/mcp` through ordinary MCP configuration, backed by the existing SQLite board and a read-only Codex registry.
- Require bearer authentication and validate `threadId` and `sessionId` metadata against persisted agent membership. Support additional allowed HTTP hosts with `--allowed-host`.
- Expose a reusable `serve_http` library entry point and scope retry deduplication to the caller within an MCP session.

## Testing

Add integration coverage for a child posting and its restarted parent reading the persisted board, including reads while a database writer lock is held. Cover invalid tokens, disallowed hosts, missing caller metadata, invalid membership, and startup without an existing registry.

GitOrigin-RevId: 5e3d71eaaf120169e2585acd0b5f84df1b402f00
This commit is contained in:
jif
2026-11-02 15:09:22 +00:00
committed by copyberry
parent f04090521d
commit 18c6dbe52b
13 changed files with 840 additions and 36 deletions
+14
View File
@@ -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",
]
+1
View File
@@ -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",
@@ -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;
@@ -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<Value> {
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::<codex_message_board::Page<codex_message_board::PostPreview>>(
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::<Vec<_>>();
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, "<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::<Vec<_>>(),
&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(())
}
@@ -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::<Vec<_>>(),\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:
<MODEL_INSTRUCTIONS>
01:message/developer[2]:
[01] <PERMISSIONS_INSTRUCTIONS>
[02] <tools>
Deferred tool namespaces:
- mcp__codex_message_board
</tools>
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.
<OMITTED 15 LINES; ~225 TOKENS; hash=fa366829bdb214be>
- 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:
<MULTI_AGENT_MODE>
04:message/user:
<ENVIRONMENT_CONTEXT>
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: <DURATION> seconds
Output:
{"author":"/root/worker","channel_name":"design","created_at":"<TIMESTAMP>","message_id":"<UUID 1>","thread_id":"<UUID 1>"}
## Window 2 (after request 3: input diverged at item 02, settings changed (tools))
-- request 4 (request) --
00:message/developer:
<MODEL_INSTRUCTIONS>
01:message/developer[2]:
[01] <PERMISSIONS_INSTRUCTIONS>
[02] <tools>
Deferred tool namespaces:
- mcp__codex_message_board
</tools>
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.
<OMITTED 16 LINES; ~234 TOKENS; hash=b2e94283a8c22327>
- 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:
<MULTI_AGENT_MODE>
04:message/user:
<ENVIRONMENT_CONTEXT>
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:
<ENVIRONMENT_CONTEXT>
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: <DURATION> seconds
Output:
{"has_more":false,"n_returned":1,"next_cursor":null,"results":[{"author":"/root/worker","channel_name":"...ecision.","thread_id":"<UUID 1>","truncated":false}]} [hash=0322071c1fb9baa5]
+22 -1
View File
@@ -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 }
+37 -2
View File
@@ -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.
+134
View File
@@ -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<F, Fut>(
listener: TcpListener,
token: String,
config: StreamableHttpServerConfig,
bind: F,
) -> anyhow::Result<()>
where
F: Fn(ThreadId, SessionId) -> Fut + Clone + Send + Sync + 'static,
Fut: Future<Output = Result<MessageBoardMcpServer, ErrorData>> + 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<F> {
bind: F,
invocation_scope: String,
}
#[derive(Deserialize)]
#[serde(rename_all = "camelCase")]
struct Caller {
thread_id: ThreadId,
session_id: SessionId,
}
impl<F, Fut> ServerHandler for HttpServer<F>
where
F: Fn(ThreadId, SessionId) -> Fut + Send + Sync + 'static,
Fut: Future<Output = Result<MessageBoardMcpServer, ErrorData>> + Send + 'static,
{
fn get_info(&self) -> ServerConfig {
server_info()
}
async fn list_tools(
&self,
request: Option<PaginatedRequestParams>,
_context: RequestContext<RoleServer>,
) -> Result<ListToolsResult, ErrorData> {
list_tools_result(
request,
MessageBoardTool::DISCUSSION
.into_iter()
.map(MessageBoardTool::spec),
)
}
async fn call_tool(
&self,
request: CallToolRequestParams,
context: RequestContext<RoleServer>,
) -> Result<CallToolResponse, ErrorData> {
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
}
}
+43 -33
View File
@@ -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<PaginatedRequestParams>,
_context: RequestContext<RoleServer>,
) -> Result<ListToolsResult, ErrorData> {
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::<Result<Vec<_>, _>>()?;
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<PaginatedRequestParams>,
specs: impl Iterator<Item = ToolSpec>,
) -> Result<ListToolsResult, ErrorData> {
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::<Result<Vec<_>, _>>()?;
Ok(ListToolsResult::with_all_items(tools))
}
+99
View File
@@ -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<String>,
/// 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
}
}
}
@@ -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<AgentPath>> {
Box::pin(async move {
let stored: Option<Option<String>> =
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<ThreadId>> {
Box::pin(async move {
let ids: Vec<String> = 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<DateTime<Utc>>> {
Box::pin(async move {
self.agent_path(caller).await?;
Ok(Utc::now())
})
}
fn notify(
&self,
_recipient: ThreadId,
_post: PostPreview,
) -> BoxFuture<'_, Result<NotificationDelivery>> {
Box::pin(async { Ok(NotificationDelivery::SkippedInactive) })
}
}
fn storage_error(error: sqlx::Error) -> CodexErr {
CodexErr::Io(std::io::Error::other(error))
}
@@ -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(())
}
@@ -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",