Remove the deprecated codex mcp-server command (#42993)

## What changed

Remove the `codex mcp-server` subcommand and the standalone `codex-mcp-server` crate, including its tests, interface documentation, build dependencies, and run recipe.

## Testing

Update the strict configuration test to use `codex exec hello`, preserving coverage for rejecting unknown configuration overrides.

GitOrigin-RevId: 8688b1f41862d7c2073ea2812a257c0087c55d9c
This commit is contained in:
jif
2026-09-05 12:17:32 +00:00
committed by copyberry
parent 2bd71f96d4
commit 531f3836a1
33 changed files with 3 additions and 4528 deletions
-59
View File
@@ -2533,7 +2533,6 @@ dependencies = [
"codex-install-context",
"codex-login",
"codex-mcp",
"codex-mcp-server",
"codex-memories-write",
"codex-model-provider",
"codex-models-manager",
@@ -3813,45 +3812,6 @@ dependencies = [
"tracing",
]
[[package]]
name = "codex-mcp-server"
version = "0.0.0"
dependencies = [
"anyhow",
"app_test_support",
"codex-arg0",
"codex-config",
"codex-core",
"codex-exec-server",
"codex-extension-api",
"codex-features",
"codex-git-attribution",
"codex-home",
"codex-image-generation-extension",
"codex-login",
"codex-otel",
"codex-protocol",
"codex-shell-command",
"codex-skills-extension",
"codex-utils-absolute-path",
"codex-utils-cli",
"codex-utils-json-to-toml",
"core_test_support",
"mcp_test_support",
"os_info",
"pretty_assertions",
"rmcp",
"schemars 0.8.22",
"serde",
"serde_json",
"shlex",
"tempfile",
"tokio",
"tracing",
"tracing-subscriber",
"wiremock",
]
[[package]]
name = "codex-memories-extension"
version = "0.0.0"
@@ -9675,25 +9635,6 @@ dependencies = [
"syn 2.0.117",
]
[[package]]
name = "mcp_test_support"
version = "0.0.0"
dependencies = [
"anyhow",
"codex-login",
"codex-mcp-server",
"codex-terminal-detection",
"codex-utils-cargo-bin",
"core_test_support",
"os_info",
"pretty_assertions",
"rmcp",
"serde_json",
"shlex",
"tokio",
"wiremock",
]
[[package]]
name = "md-5"
version = "0.10.6"
-3
View File
@@ -79,7 +79,6 @@ members = [
"lmstudio",
"login",
"codex-mcp",
"mcp-server",
"memories/read",
"memories/write",
"model-provider-info",
@@ -238,7 +237,6 @@ codex-memories-read = { path = "memories/read" }
codex-memories-write = { path = "memories/write" }
codex-mcp = { path = "codex-mcp" }
codex-mcp-extension = { path = "ext/mcp" }
codex-mcp-server = { path = "mcp-server" }
codex-model-provider-info = { path = "model-provider-info" }
codex-models-manager = { path = "models-manager" }
codex-mxc-sandbox = { path = "mxc-sandbox" }
@@ -301,7 +299,6 @@ codex-v8-poc = { path = "v8-poc" }
codex-workload-identity = { path = "workload-identity" }
codex-windows-sandbox = { path = "windows-sandbox-rs" }
core_test_support = { path = "core/tests/common" }
mcp_test_support = { path = "mcp-server/tests/common" }
# External
learning_mode_windows = { git = "https://github.com/microsoft/mxc", rev = "6cd3d58f05d3447e67109cfb75e042803b843ca4" }
@@ -278,7 +278,7 @@ impl TestAppServer {
.is_err_and(|error| error.kind() == std::io::ErrorKind::ExecutableFileBusy)
|| retries == 2
{
break process.context("codex-mcp-server proc should start")?;
break process.context("codex app-server proc should start")?;
}
retries += 1;
tokio::time::sleep(Duration::from_millis(10)).await;
-1
View File
@@ -56,7 +56,6 @@ codex-install-context = { workspace = true }
codex-login = { workspace = true }
codex-memories-write = { workspace = true }
codex-mcp = { workspace = true }
codex-mcp-server = { workspace = true }
codex-model-provider = { workspace = true }
codex-models-manager = { workspace = true }
codex-plugin = { workspace = true }
-36
View File
@@ -167,9 +167,6 @@ enum Subcommand {
/// Manage Codex plugins.
Plugin(PluginCli),
/// Start Codex as an MCP server (stdio).
McpServer(McpServerCommand),
/// [experimental] Run the app server or related tooling.
AppServer(AppServerCommand),
@@ -322,13 +319,6 @@ struct ReviewCommand {
args: ReviewArgs,
}
#[derive(Debug, Parser)]
struct McpServerCommand {
/// Error out when config.toml contains fields that are not recognized by this version of Codex.
#[arg(long = "strict-config", default_value_t = false)]
strict_config: bool,
}
#[derive(Debug, Parser)]
struct DebugTraceReduceCommand {
/// Trace bundle directory containing manifest.json and trace.jsonl.
@@ -1270,22 +1260,6 @@ async fn cli_main(
);
codex_exec::run_main(exec_cli, arg0_paths.clone()).await?;
}
Some(Subcommand::McpServer(McpServerCommand { strict_config })) => {
eprintln!(
"warning: `codex mcp-server` is deprecated and will be removed in a future release."
);
reject_remote_mode_for_subcommand(
root_remote.as_deref(),
root_remote_auth_token_env.as_deref(),
"mcp-server",
)?;
codex_mcp_server::run_main(
arg0_paths.clone(),
root_config_overrides,
strict_config || root_strict_config,
)
.await?;
}
Some(Subcommand::Mcp(mut mcp_cli)) => {
reject_remote_mode_for_subcommand(
root_remote.as_deref(),
@@ -2563,7 +2537,6 @@ fn unsupported_subcommand_name_for_strict_config(
| Some(Subcommand::Agents(_))
| Some(Subcommand::Exec(_))
| Some(Subcommand::Review(_))
| Some(Subcommand::McpServer(_))
| Some(Subcommand::ExecServer(_))
| Some(Subcommand::Resume(_))
| Some(Subcommand::Queue(_))
@@ -4411,15 +4384,6 @@ mod tests {
let cli = MultitoolCli::try_parse_from(["codex", "--strict-config"]).expect("parse");
assert!(cli.interactive.strict_config);
let cli = MultitoolCli::try_parse_from(["codex", "mcp-server", "--strict-config"])
.expect("parse");
assert_matches!(
cli.subcommand,
Some(Subcommand::McpServer(McpServerCommand {
strict_config: true,
}))
);
let cli =
MultitoolCli::try_parse_from(["codex", "review", "--strict-config", "--uncommitted"])
.expect("parse");
+1 -1
View File
@@ -30,7 +30,7 @@ fn strict_config_rejects_unknown_config_override() -> Result<()> {
let codex_home = TempDir::new()?;
let mut cmd = codex_command(codex_home.path())?;
cmd.args(["--strict-config", "-c", "foo=bar", "mcp-server"])
cmd.args(["--strict-config", "-c", "foo=bar", "exec", "hello"])
.assert()
.failure()
.stderr(contains("unknown configuration field"));
-144
View File
@@ -1,144 +0,0 @@
# Codex MCP Server Interface [experimental]
This document describes Codex's experimental MCP server interface: a JSON-RPC API that runs over the Model Context Protocol (MCP) transport to control a local Codex engine.
- Status: experimental and subject to change without notice
- Server binary: `codex mcp-server` (or `codex-mcp-server`)
- Transport: standard MCP over stdio (JSON-RPC 2.0, line-delimited)
## Overview
Codex exposes MCP-compatible methods to manage threads, turns, accounts, config, and approvals. The types live in `app-server-protocol/src/protocol/{common,v1,v2}.rs` and are consumed by the app server implementation in `app-server/`.
At a glance:
- Primary v2 RPCs
- `thread/start`, `thread/resume`, `thread/fork`, `thread/read`, `thread/list`
- `turn/start`, `turn/steer`, `turn/interrupt`
- `account/read`, `account/login/start`, `account/login/cancel`, `account/logout`, `account/rateLimits/read`
- `config/read`, `config/value/write`, `config/batchWrite`
- `model/list`, `app/list`, `collaborationMode/list`
- Remaining v1 compatibility RPCs
- `getConversationSummary`
- `getAuthStatus`
- `gitDiffToRemote`
- `fuzzyFileSearch`, `fuzzyFileSearch/sessionStart`, `fuzzyFileSearch/sessionUpdate`, `fuzzyFileSearch/sessionStop`
- Notifications
- v2 typed notifications such as `thread/started`, `turn/completed`, `account/login/completed`
- `codex/event/*` stream notifications for live agent events
- `fuzzyFileSearch/sessionUpdated`, `fuzzyFileSearch/sessionCompleted`
- Approvals (server -> client requests)
- `applyPatchApproval`, `execCommandApproval`
See code for full type definitions and exact shapes: `app-server-protocol/src/protocol/{common,v1,v2}.rs`.
## Starting the server
Run Codex as an MCP server and connect an MCP client:
```bash
codex mcp-server | your_mcp_client
```
For a simple inspection UI, you can also try:
```bash
npx @modelcontextprotocol/inspector codex mcp-server
```
Use the separate `codex mcp` subcommand to manage configured MCP server launchers in `config.toml`.
## Threads and turns
Use the v2 thread and turn APIs for all new integrations. `thread/start` creates a thread, `turn/start` submits user input, `turn/interrupt` stops an in-flight turn, and `thread/list` / `thread/read` expose persisted history.
`getConversationSummary` remains as a compatibility helper for clients that still need a summary lookup by `conversationId` or `rolloutPath`. Lookups by `conversationId` are preferred; lookups by `rolloutPath` won't work with non-local thread stores.
For complete request and response shapes, see the app-server README and the protocol definitions in `app-server-protocol/src/protocol/v2.rs`.
## Models
Fetch the catalog of models available in the current Codex build with `model/list`. The request accepts optional pagination inputs:
- `limit` - number of models to return (defaults to a server-selected value)
- `cursor` - opaque string from the previous response's `nextCursor`
Each response yields:
- `data` - ordered list of models. A model includes:
- `id`, `model`, `displayName`, `description`
- `supportedReasoningEfforts` - array of objects with:
- `reasoningEffort` - a string value advertised by the model; common values are `none|minimal|low|medium|high|xhigh`
- `description` - human-friendly label for the effort
- `defaultReasoningEffort` - suggested effort for the UI
- `inputModalities` - accepted input types for the model
- `supportsPersonality` - whether the model supports personality-specific instructions
- `isDefault` - whether the model is recommended for most users
- `upgrade` - optional recommended upgrade model id
- `upgradeInfo` - optional upgrade metadata object with:
- `model` - recommended upgrade model id
- `upgradeCopy` - optional display copy for the upgrade recommendation
- `modelLink` - optional link for the upgrade recommendation
- `migrationMarkdown` - optional markdown shown when presenting the upgrade
- `nextCursor` - pass into the next request to continue paging (optional)
## Collaboration modes (experimental)
Fetch the built-in collaboration mode presets with `collaborationMode/list`. This endpoint does not accept pagination and returns the full list in one response:
- `data` - ordered list of collaboration mode masks (partial settings to apply on top of the base mode)
- For tri-state fields like `reasoning_effort` and `developer_instructions`, omit the field to keep the current value, set it to `null` to clear it, or set a concrete value to update it.
- Built-in presets do not set `model`. The Plan preset sets `reasoning_effort` to medium; clients keep or override model separately.
When sending `turn/start` with `collaborationMode`, `settings.developer_instructions: null` means "use built-in instructions for the selected mode".
## Event stream
While a conversation runs, the server sends notifications:
- `codex/event` with the serialized Codex event payload. The shape matches `core/src/protocol.rs`'s `Event` and `EventMsg` types. Some notifications include a `_meta.requestId` to correlate with the originating request.
- `fuzzyFileSearch/sessionUpdated` and `fuzzyFileSearch/sessionCompleted` for the legacy fuzzy search flow.
Clients should render events and, when present, surface approval requests (see next section).
## Tool responses
The `codex` and `codex-reply` tools return standard MCP `CallToolResult` payloads. For compatibility with MCP clients that prefer `structuredContent`, Codex mirrors the content blocks inside `structuredContent` alongside the `threadId`.
Example:
```json
{
"content": [{ "type": "text", "text": "Hello from Codex" }],
"structuredContent": {
"threadId": "019bbed6-1e9e-7f31-984c-a05b65045719",
"content": "Hello from Codex"
}
}
```
## Approvals (server -> client)
When Codex needs approval to apply changes or run commands, the server issues JSON-RPC requests to the client:
- `applyPatchApproval { conversationId, callId, fileChanges, reason?, grantRoot? }`
- `execCommandApproval { conversationId, callId, approvalId?, command, cwd, reason? }`
The client must reply with `{ decision: "allow" | "deny" }` for each request.
## Auth helpers
For the complete request/response shapes and flow examples, see the [Auth endpoints (v2) section in the app-server README](../app-server/README.md#auth-endpoints-v2).
## Legacy compatibility methods
The server still accepts a narrow v1 compatibility surface for existing app clients:
- `getConversationSummary`
- `getAuthStatus`
- `gitDiffToRemote`
- `fuzzyFileSearch`, `fuzzyFileSearch/sessionStart`, `fuzzyFileSearch/sessionUpdate`, `fuzzyFileSearch/sessionStop`
## Compatibility and stability
This interface is experimental. Method names, fields, and event shapes may evolve. For the authoritative schema, consult `app-server-protocol/src/protocol/{common,v1,v2}.rs` and the corresponding server wiring in `app-server/`.
-6
View File
@@ -1,6 +0,0 @@
load("//:defs.bzl", "codex_rust_crate")
codex_rust_crate(
name = "mcp-server",
crate_name = "codex_mcp_server",
)
-60
View File
@@ -1,60 +0,0 @@
[package]
name = "codex-mcp-server"
version.workspace = true
edition.workspace = true
license.workspace = true
[[bin]]
name = "codex-mcp-server"
path = "src/main.rs"
[lib]
name = "codex_mcp_server"
path = "src/lib.rs"
doctest = false
[lints]
workspace = true
[dependencies]
anyhow = { workspace = true }
codex-arg0 = { workspace = true }
codex-config = { workspace = true }
codex-core = { workspace = true }
codex-home = { workspace = true }
codex-image-generation-extension = { workspace = true }
codex-exec-server = { workspace = true }
codex-extension-api = { workspace = true }
codex-features = { workspace = true }
codex-git-attribution = { workspace = true }
codex-login = { workspace = true }
codex-otel = { workspace = true }
codex-protocol = { workspace = true }
codex-skills-extension = { workspace = true }
codex-utils-cli = { workspace = true }
codex-utils-json-to-toml = { workspace = true }
rmcp = { workspace = true }
schemars = { workspace = true }
serde = { workspace = true, features = ["derive"] }
serde_json = { workspace = true }
shlex = { workspace = true }
tokio = { workspace = true, features = [
"io-std",
"macros",
"process",
"rt-multi-thread",
"signal",
] }
tracing = { workspace = true, features = ["log"] }
tracing-subscriber = { workspace = true, features = ["env-filter", "fmt"] }
[dev-dependencies]
app_test_support = { workspace = true }
codex-utils-absolute-path = { workspace = true }
codex-shell-command = { workspace = true }
core_test_support = { workspace = true }
mcp_test_support = { workspace = true }
os_info = { workspace = true }
pretty_assertions = { workspace = true }
tempfile = { workspace = true }
wiremock = { workspace = true }
@@ -1,97 +0,0 @@
use std::collections::HashMap;
use std::sync::Mutex;
use std::sync::MutexGuard;
use std::sync::PoisonError;
use codex_protocol::ThreadId;
use rmcp::model::RequestId;
/// Tracks active MCP requests by both request ID and Codex turn.
///
/// Warning and response enqueue callbacks run while holding the same lock. This
/// makes route removal and response enqueue atomic with respect to warning
/// enqueue, so a warning is either queued before its tool response or dropped.
#[derive(Default)]
pub(crate) struct ActiveTurnRegistry {
state: Mutex<ActiveTurnState>,
}
#[derive(Default)]
struct ActiveTurnState {
by_request: HashMap<RequestId, ActiveTurn>,
by_thread_and_turn: HashMap<(ThreadId, String), RequestId>,
}
struct ActiveTurn {
thread_id: ThreadId,
turn_id: String,
}
impl ActiveTurnRegistry {
pub(crate) fn register(&self, request_id: RequestId, thread_id: ThreadId, turn_id: String) {
let active_turn = ActiveTurn {
thread_id,
turn_id: turn_id.clone(),
};
let mut state = self.lock();
if let Some(previous_turn) = state.by_request.insert(request_id.clone(), active_turn) {
state
.by_thread_and_turn
.remove(&(previous_turn.thread_id, previous_turn.turn_id));
}
state
.by_thread_and_turn
.insert((thread_id, turn_id), request_id);
}
pub(crate) fn thread_id(&self, request_id: &RequestId) -> Option<ThreadId> {
self.lock()
.by_request
.get(request_id)
.map(|active_turn| active_turn.thread_id)
}
pub(crate) fn with_request_id(
&self,
thread_id: ThreadId,
turn_id: &str,
f: impl FnOnce(&RequestId),
) -> bool {
let state = self.lock();
let Some(request_id) = state
.by_thread_and_turn
.get(&(thread_id, turn_id.to_string()))
else {
return false;
};
f(request_id);
true
}
pub(crate) fn unregister(&self, request_id: &RequestId) {
let mut state = self.lock();
Self::remove(&mut state, request_id);
}
pub(crate) fn finish(&self, request_id: &RequestId, finish: impl FnOnce()) {
let mut state = self.lock();
Self::remove(&mut state, request_id);
finish();
}
fn remove(state: &mut ActiveTurnState, request_id: &RequestId) {
if let Some(active_turn) = state.by_request.remove(request_id) {
state
.by_thread_and_turn
.remove(&(active_turn.thread_id, active_turn.turn_id));
}
}
fn lock(&self) -> MutexGuard<'_, ActiveTurnState> {
self.state.lock().unwrap_or_else(PoisonError::into_inner)
}
}
#[cfg(test)]
#[path = "active_turn_registry_tests.rs"]
mod tests;
@@ -1,70 +0,0 @@
use codex_protocol::ThreadId;
use pretty_assertions::assert_eq;
use rmcp::model::RequestId;
use super::ActiveTurnRegistry;
#[test]
fn colliding_request_id_strings_use_independent_turn_routes() {
let active_turns = ActiveTurnRegistry::default();
let thread_id = ThreadId::new();
let numeric_request_id = RequestId::Number(7);
let string_request_id = RequestId::String("7".into());
let numeric_turn_id = "numeric-turn".to_string();
let string_turn_id = "string-turn".to_string();
active_turns.register(
numeric_request_id.clone(),
thread_id,
numeric_turn_id.clone(),
);
active_turns.register(string_request_id.clone(), thread_id, string_turn_id.clone());
let mut routed_request_ids = Vec::new();
assert!(
active_turns.with_request_id(thread_id, &numeric_turn_id, |request_id| {
routed_request_ids.push(request_id.clone());
})
);
assert!(
active_turns.with_request_id(thread_id, &string_turn_id, |request_id| {
routed_request_ids.push(request_id.clone());
})
);
assert_eq!(
routed_request_ids,
vec![numeric_request_id.clone(), string_request_id.clone()]
);
active_turns.unregister(&numeric_request_id);
assert!(!active_turns.with_request_id(thread_id, &numeric_turn_id, |_| {}));
let mut remaining_request_id = None;
assert!(
active_turns.with_request_id(thread_id, &string_turn_id, |request_id| {
remaining_request_id = Some(request_id.clone());
})
);
assert_eq!(remaining_request_id, Some(string_request_id));
}
#[test]
fn reused_request_id_replaces_the_previous_turn_route() {
let active_turns = ActiveTurnRegistry::default();
let thread_id = ThreadId::new();
let request_id = RequestId::Number(7);
let first_turn_id = "first-turn".to_string();
let second_turn_id = "second-turn".to_string();
active_turns.register(request_id.clone(), thread_id, first_turn_id.clone());
active_turns.register(request_id.clone(), thread_id, second_turn_id.clone());
assert!(!active_turns.with_request_id(thread_id, &first_turn_id, |_| {}));
let mut routed_request_id = None;
assert!(
active_turns.with_request_id(thread_id, &second_turn_id, |request_id| {
routed_request_id = Some(request_id.clone());
})
);
assert_eq!(routed_request_id, Some(request_id));
}
@@ -1,432 +0,0 @@
//! Configuration object accepted by the `codex` MCP tool-call.
use codex_arg0::Arg0DispatchPaths;
use codex_core::config::Config;
use codex_core::config::ConfigBuilder;
use codex_core::config::ConfigOverrides;
use codex_protocol::ThreadId;
use codex_protocol::config_types::SandboxMode;
use codex_protocol::protocol::AskForApproval;
use codex_utils_json_to_toml::json_to_toml;
use rmcp::model::JsonObject;
use rmcp::model::Tool;
use schemars::JsonSchema;
use schemars::r#gen::SchemaSettings;
use serde::Deserialize;
use serde::Serialize;
use std::collections::HashMap;
use std::path::PathBuf;
use std::sync::Arc;
/// Client-supplied configuration for a `codex` tool-call.
#[derive(Debug, Clone, Serialize, Deserialize, JsonSchema, Default)]
#[serde(rename_all = "kebab-case", deny_unknown_fields)]
#[schemars(deny_unknown_fields)]
pub struct CodexToolCallParam {
/// The *initial user prompt* to start the Codex conversation.
pub prompt: String,
/// Optional override for the model name (e.g. 'gpt-5.2', 'gpt-5.2-codex').
#[serde(default, skip_serializing_if = "Option::is_none")]
pub model: Option<String>,
/// Working directory for the session. If relative, it is resolved against
/// the server process's current working directory.
#[serde(default, skip_serializing_if = "Option::is_none")]
pub cwd: Option<String>,
/// Approval policy for shell commands generated by the model:
/// `on-request`, `never`.
#[serde(default, skip_serializing_if = "Option::is_none")]
pub approval_policy: Option<CodexToolCallApprovalPolicy>,
/// Sandbox mode: `read-only`, `workspace-write`, or `danger-full-access`.
#[serde(default, skip_serializing_if = "Option::is_none")]
pub sandbox: Option<CodexToolCallSandboxMode>,
/// Individual config settings that will override what is in
/// CODEX_HOME/config.toml.
#[serde(default, skip_serializing_if = "Option::is_none")]
pub config: Option<HashMap<String, serde_json::Value>>,
/// The set of instructions to use instead of the default ones.
#[serde(default, skip_serializing_if = "Option::is_none")]
pub base_instructions: Option<String>,
/// Developer instructions that should be injected as a developer role message.
#[serde(default, skip_serializing_if = "Option::is_none")]
pub developer_instructions: Option<String>,
/// Prompt used when compacting the conversation.
#[serde(default, skip_serializing_if = "Option::is_none")]
pub compact_prompt: Option<String>,
}
/// Custom enum mirroring [`AskForApproval`], but has an extra dependency on
/// [`JsonSchema`].
#[derive(Debug, Clone, Serialize, Deserialize, JsonSchema, PartialEq)]
#[serde(rename_all = "kebab-case")]
pub enum CodexToolCallApprovalPolicy {
OnRequest,
Never,
}
impl From<CodexToolCallApprovalPolicy> for AskForApproval {
fn from(value: CodexToolCallApprovalPolicy) -> Self {
match value {
CodexToolCallApprovalPolicy::OnRequest => AskForApproval::OnRequest,
CodexToolCallApprovalPolicy::Never => AskForApproval::Never,
}
}
}
/// Custom enum mirroring [`SandboxMode`] from config_types.rs, but with
/// `JsonSchema` support.
#[derive(Debug, Clone, Serialize, Deserialize, JsonSchema, PartialEq)]
#[serde(rename_all = "kebab-case")]
pub enum CodexToolCallSandboxMode {
ReadOnly,
WorkspaceWrite,
DangerFullAccess,
}
impl From<CodexToolCallSandboxMode> for SandboxMode {
fn from(value: CodexToolCallSandboxMode) -> Self {
match value {
CodexToolCallSandboxMode::ReadOnly => SandboxMode::ReadOnly,
CodexToolCallSandboxMode::WorkspaceWrite => SandboxMode::WorkspaceWrite,
CodexToolCallSandboxMode::DangerFullAccess => SandboxMode::DangerFullAccess,
}
}
}
/// Builds a `Tool` definition (JSON schema etc.) for the Codex tool-call.
pub(crate) fn create_tool_for_codex_tool_call_param() -> Tool {
let schema = SchemaSettings::draft2019_09()
.with(|s| {
s.inline_subschemas = true;
s.option_add_null_type = false;
})
.into_generator()
.into_root_schema_for::<CodexToolCallParam>();
let input_schema = create_tool_input_schema(schema, "Codex tool schema should serialize");
Tool::new(
"codex",
"Run a Codex session. Accepts configuration parameters matching the Codex Config struct.",
input_schema,
)
.with_title("Codex")
.with_raw_output_schema(codex_tool_output_schema())
}
fn codex_tool_output_schema() -> Arc<JsonObject> {
let schema = serde_json::json!({
"type": "object",
"properties": {
"threadId": { "type": "string" },
"content": { "type": "string" }
},
"required": ["threadId", "content"],
});
match schema {
serde_json::Value::Object(map) => Arc::new(map),
_ => unreachable!("json literal must be an object"),
}
}
impl CodexToolCallParam {
/// Returns the initial user prompt to start the Codex conversation and the
/// effective Config object generated from the supplied parameters.
pub async fn into_config(
self,
arg0_paths: Arg0DispatchPaths,
) -> std::io::Result<(String, Config)> {
let Self {
prompt,
model,
cwd,
approval_policy,
sandbox,
config: cli_overrides,
base_instructions,
developer_instructions,
compact_prompt,
} = self;
// Build the `ConfigOverrides` recognized by codex-core.
let overrides = ConfigOverrides {
model,
cwd: cwd.map(PathBuf::from),
approval_policy: approval_policy.map(Into::into),
sandbox_mode: sandbox.map(Into::into),
codex_self_exe: arg0_paths.codex_self_exe.clone(),
codex_linux_sandbox_exe: arg0_paths.codex_linux_sandbox_exe.clone(),
main_execve_wrapper_exe: arg0_paths.main_execve_wrapper_exe.clone(),
base_instructions,
developer_instructions,
compact_prompt,
..Default::default()
};
let cli_overrides = cli_overrides
.unwrap_or_default()
.into_iter()
.map(|(k, v)| (k, json_to_toml(v)))
.collect();
let cfg = ConfigBuilder::default()
.cli_overrides(cli_overrides)
.harness_overrides(overrides)
.build()
.await?;
Ok((prompt, cfg))
}
}
#[derive(Debug, Clone, Serialize, Deserialize, JsonSchema)]
#[serde(rename_all = "camelCase")]
pub struct CodexToolCallReplyParam {
/// DEPRECATED: use threadId instead.
#[serde(default, skip_serializing_if = "Option::is_none")]
conversation_id: Option<String>,
/// The thread id for this Codex session.
/// This field is required, but we keep it optional here for backward
/// compatibility for clients that still use conversationId.
#[serde(default, skip_serializing_if = "Option::is_none")]
thread_id: Option<String>,
/// The *next user prompt* to continue the Codex conversation.
pub prompt: String,
}
impl CodexToolCallReplyParam {
pub(crate) fn get_thread_id(&self) -> anyhow::Result<ThreadId> {
if let Some(thread_id) = &self.thread_id {
let thread_id = ThreadId::from_string(thread_id)?;
Ok(thread_id)
} else if let Some(conversation_id) = &self.conversation_id {
let thread_id = ThreadId::from_string(conversation_id)?;
Ok(thread_id)
} else {
Err(anyhow::anyhow!(
"either threadId or conversationId must be provided"
))
}
}
}
/// Builds a `Tool` definition for the `codex-reply` tool-call.
pub(crate) fn create_tool_for_codex_tool_call_reply_param() -> Tool {
let schema = SchemaSettings::draft2019_09()
.with(|s| {
s.inline_subschemas = true;
s.option_add_null_type = false;
})
.into_generator()
.into_root_schema_for::<CodexToolCallReplyParam>();
let input_schema = create_tool_input_schema(schema, "Codex reply tool schema should serialize");
Tool::new(
"codex-reply",
"Continue a Codex conversation by providing the thread id and prompt.",
input_schema,
)
.with_title("Codex Reply")
.with_raw_output_schema(codex_tool_output_schema())
}
fn create_tool_input_schema(
schema: schemars::schema::RootSchema,
panic_message: &str,
) -> Arc<JsonObject> {
#[expect(clippy::expect_used)]
let schema_value = serde_json::to_value(&schema).expect(panic_message);
let mut schema_object = match schema_value {
serde_json::Value::Object(object) => object,
_ => panic!("tool schema should serialize to a JSON object"),
};
// Prefer keeping the "core" JSON Schema keys while still preserving `$defs`
// in case any `$ref` leaks into the generated schema (even though we try
// to inline subschemas).
let mut input_schema = JsonObject::new();
for key in [
"additionalProperties",
"properties",
"required",
"type",
"$defs",
"definitions",
] {
if let Some(value) = schema_object.remove(key) {
input_schema.insert(key.to_string(), value);
}
}
Arc::new(input_schema)
}
#[cfg(test)]
mod tests {
use super::*;
use pretty_assertions::assert_eq;
/// We include a test to verify the exact JSON schema as "executable
/// documentation" for the schema. When can track changes to this test as a
/// way to audit changes to the generated schema.
///
/// Seeing the fully expanded schema makes it easier to casually verify that
/// the generated JSON for enum types such as "approval-policy" is compact.
/// Ideally, modelcontextprotocol/inspector would provide a simpler UI for
/// enum fields versus open string fields to take advantage of this.
///
/// As of 2025-05-04, there is an open PR for this:
/// https://github.com/modelcontextprotocol/inspector/pull/196
#[test]
fn verify_codex_tool_json_schema() {
let tool = create_tool_for_codex_tool_call_param();
let tool_json = serde_json::to_value(&tool).expect("tool serializes");
let expected_tool_json = serde_json::json!({
"description": "Run a Codex session. Accepts configuration parameters matching the Codex Config struct.",
"inputSchema": {
"additionalProperties": false,
"properties": {
"approval-policy": {
"description": "Approval policy for shell commands generated by the model: `on-request`, `never`.",
"enum": [
"on-request",
"never"
],
"type": "string"
},
"base-instructions": {
"description": "The set of instructions to use instead of the default ones.",
"type": "string"
},
"compact-prompt": {
"description": "Prompt used when compacting the conversation.",
"type": "string"
},
"config": {
"additionalProperties": true,
"description": "Individual config settings that will override what is in CODEX_HOME/config.toml.",
"type": "object"
},
"cwd": {
"description": "Working directory for the session. If relative, it is resolved against the server process's current working directory.",
"type": "string"
},
"developer-instructions": {
"description": "Developer instructions that should be injected as a developer role message.",
"type": "string"
},
"model": {
"description": "Optional override for the model name (e.g. 'gpt-5.2', 'gpt-5.2-codex').",
"type": "string"
},
"prompt": {
"description": "The *initial user prompt* to start the Codex conversation.",
"type": "string"
},
"sandbox": {
"description": "Sandbox mode: `read-only`, `workspace-write`, or `danger-full-access`.",
"enum": [
"read-only",
"workspace-write",
"danger-full-access"
],
"type": "string"
}
},
"required": [
"prompt"
],
"type": "object"
},
"name": "codex",
"outputSchema": {
"properties": {
"content": {
"type": "string"
},
"threadId": {
"type": "string"
}
},
"required": [
"threadId",
"content"
],
"type": "object"
},
"title": "Codex"
});
assert_eq!(expected_tool_json, tool_json);
}
#[test]
fn codex_tool_call_param_rejects_removed_profile_field() {
let err = serde_json::from_value::<CodexToolCallParam>(serde_json::json!({
"prompt": "hello",
"profile": "work"
}))
.expect_err("removed profile field should fail");
assert!(
err.to_string().contains("unknown field `profile`"),
"unexpected error: {err}"
);
}
#[test]
fn verify_codex_tool_reply_json_schema() {
let tool = create_tool_for_codex_tool_call_reply_param();
let tool_json = serde_json::to_value(&tool).expect("tool serializes");
let expected_tool_json = serde_json::json!({
"description": "Continue a Codex conversation by providing the thread id and prompt.",
"inputSchema": {
"properties": {
"conversationId": {
"description": "DEPRECATED: use threadId instead.",
"type": "string"
},
"prompt": {
"description": "The *next user prompt* to continue the Codex conversation.",
"type": "string"
},
"threadId": {
"description": "The thread id for this Codex session. This field is required, but we keep it optional here for backward compatibility for clients that still use conversationId.",
"type": "string"
}
},
"required": [
"prompt",
],
"type": "object",
},
"name": "codex-reply",
"outputSchema": {
"properties": {
"content": {
"type": "string"
},
"threadId": {
"type": "string"
}
},
"required": [
"threadId",
"content"
],
"type": "object"
},
"title": "Codex Reply",
});
assert_eq!(expected_tool_json, tool_json);
}
}
@@ -1,423 +0,0 @@
//! Asynchronous worker that executes a **Codex** tool-call inside a spawned
//! Tokio task. Separated from `message_processor.rs` to keep that file small
//! and to make future feature-growth easier to manage.
use std::sync::Arc;
use crate::active_turn_registry::ActiveTurnRegistry;
use crate::exec_approval::handle_exec_approval_request;
use crate::outgoing_message::OutgoingMessageSender;
use crate::outgoing_message::OutgoingNotificationMeta;
use crate::patch_approval::handle_patch_approval_request;
use codex_core::CodexThread;
use codex_core::NewThread;
use codex_core::StartIfIdleSubmission;
use codex_core::StartThreadOptions;
use codex_core::ThreadManager;
use codex_core::TurnInputRequest;
use codex_core::TurnInputSubmission;
use codex_core::config::Config as CodexConfig;
use codex_protocol::ThreadId;
use codex_protocol::protocol::AgentMessageEvent;
use codex_protocol::protocol::ApplyPatchApprovalRequestEvent;
use codex_protocol::protocol::Event;
use codex_protocol::protocol::EventMsg;
use codex_protocol::protocol::ExecApprovalRequestEvent;
use codex_protocol::protocol::TurnCompleteEvent;
use codex_protocol::user_input::UserInput;
use rmcp::model::CallToolResult;
use rmcp::model::ContentBlock;
use rmcp::model::RequestId;
use serde_json::json;
/// To adhere to MCP `tools/call` response format, include the Codex
/// `threadId` in the `structured_content` field of the response.
/// Some MCP clients ignore `content` when `structuredContent` is present, so
/// mirror the text there as well.
pub(crate) fn create_call_tool_result_with_thread_id(
thread_id: ThreadId,
text: String,
is_error: Option<bool>,
) -> CallToolResult {
let content_text = text;
let content = vec![ContentBlock::text(content_text.clone())];
let structured_content = json!({
"threadId": thread_id,
"content": content_text,
});
let mut result = CallToolResult::success(content);
result.is_error = is_error;
result.structured_content = Some(structured_content);
result
}
fn prompt_request(prompt: String) -> TurnInputRequest {
TurnInputRequest::user_input(vec![UserInput::Text {
text: prompt,
// MCP tool prompts are plain text with no UI element ranges.
text_elements: Vec::new(),
}])
}
/// Run a complete Codex session and stream events back to the client.
///
/// On completion (success or error) the function sends the appropriate
/// `tools/call` response so the LLM can continue the conversation.
pub async fn run_codex_tool_session(
id: RequestId,
initial_prompt: String,
config: CodexConfig,
outgoing: Arc<OutgoingMessageSender>,
thread_manager: Arc<ThreadManager>,
active_turns: Arc<ActiveTurnRegistry>,
) {
let NewThread {
thread_id,
thread,
session_configured,
} = match thread_manager
.start_thread(StartThreadOptions::new(config.clone()))
.await
{
Ok(res) => res,
Err(e) => {
let result = CallToolResult::error(vec![ContentBlock::text(format!(
"Failed to start Codex session: {e}"
))]);
outgoing.send_response(id.clone(), result);
return;
}
};
let session_configured_event = Event {
// Use a fake id value for now.
id: "".to_string(),
msg: EventMsg::SessionConfigured(session_configured.clone()),
};
outgoing.send_event_as_notification(
&session_configured_event,
Some(OutgoingNotificationMeta {
request_id: Some(id.clone()),
thread_id: Some(thread_id),
}),
);
let turn_id = match thread
.start_turn_if_idle(prompt_request(initial_prompt))
.await
{
Ok(StartIfIdleSubmission::Started { turn_id }) => turn_id,
Ok(StartIfIdleSubmission::NotSubmitted { reason }) => {
tracing::error!("Failed to submit initial prompt: {reason:?}");
let result = create_call_tool_result_with_thread_id(
thread_id,
format!("Failed to submit initial prompt: {reason:?}"),
Some(true),
);
outgoing.send_response(id.clone(), result);
return;
}
Err(e) => {
tracing::error!("Failed to submit initial prompt: {e}");
let result = create_call_tool_result_with_thread_id(
thread_id,
format!("Failed to submit initial prompt: {e}"),
Some(true),
);
outgoing.send_response(id.clone(), result);
return;
}
};
active_turns.register(id.clone(), thread_id, turn_id);
run_codex_tool_session_inner(thread_id, thread, outgoing, id, active_turns).await;
}
pub async fn run_codex_tool_session_reply(
thread_id: ThreadId,
thread: Arc<CodexThread>,
outgoing: Arc<OutgoingMessageSender>,
request_id: RequestId,
prompt: String,
active_turns: Arc<ActiveTurnRegistry>,
) {
let turn_id = match thread.start_or_steer_turn(prompt_request(prompt)).await {
Ok(TurnInputSubmission::Started { turn_id } | TurnInputSubmission::Steered { turn_id }) => {
turn_id
}
Ok(TurnInputSubmission::NotSubmitted { reason }) => {
tracing::error!("Failed to submit user input: {reason:?}");
let result = create_call_tool_result_with_thread_id(
thread_id,
format!("Failed to submit user input: {reason:?}"),
Some(true),
);
outgoing.send_response(request_id.clone(), result);
return;
}
Err(e) => {
tracing::error!("Failed to submit user input: {e}");
let result = create_call_tool_result_with_thread_id(
thread_id,
format!("Failed to submit user input: {e}"),
Some(true),
);
outgoing.send_response(request_id.clone(), result);
return;
}
};
active_turns.register(request_id.clone(), thread_id, turn_id);
run_codex_tool_session_inner(thread_id, thread, outgoing, request_id, active_turns).await;
}
async fn run_codex_tool_session_inner(
thread_id: ThreadId,
thread: Arc<CodexThread>,
outgoing: Arc<OutgoingMessageSender>,
request_id: RequestId,
active_turns: Arc<ActiveTurnRegistry>,
) {
let request_id_str = request_id.to_string();
// Stream events until the task needs to pause for user interaction or
// completes.
loop {
match thread.next_event().await {
Ok(event) => {
outgoing.send_event_as_notification(
&event,
Some(OutgoingNotificationMeta {
request_id: Some(request_id.clone()),
thread_id: Some(thread_id),
}),
);
match event.msg {
EventMsg::ExecApprovalRequest(ev) => {
let approval_id = ev.effective_approval_id();
let ExecApprovalRequestEvent {
kind: _,
turn_id: _,
environment_id: _,
started_at_ms: _,
command,
cwd,
call_id,
plugin_id: _,
script_path: _,
approval_id: _,
reason: _,
proposed_execpolicy_amendment: _,
proposed_network_policy_amendments: _,
parsed_cmd,
network_approval_context: _,
additional_permissions: _,
available_decisions: _,
} = ev;
handle_exec_approval_request(
command,
std::path::PathBuf::from(cwd.into_string()),
outgoing.clone(),
thread.clone(),
request_id.clone(),
request_id_str.clone(),
event.id.clone(),
call_id,
approval_id,
parsed_cmd,
thread_id,
)
.await;
continue;
}
EventMsg::PlanDelta(_) => {
continue;
}
EventMsg::Error(err_event) => {
// Always respond in tools/call's expected shape, and include conversationId so the client can resume.
let result = create_call_tool_result_with_thread_id(
thread_id,
err_event.message,
Some(true),
);
active_turns.finish(&request_id, || {
outgoing.send_response(request_id.clone(), result);
});
break;
}
EventMsg::Warning(_)
| EventMsg::AuthRecoveryStarted(_)
| EventMsg::AuthRecoveryCompleted(_)
| EventMsg::GuardianWarning(_)
| EventMsg::ModelVerification(_)
| EventMsg::SafetyBuffering(_)
| EventMsg::TurnModerationMetadata(_) => {
continue;
}
EventMsg::GuardianAssessment(_) => {
continue;
}
EventMsg::ElicitationRequest(_) => {
// TODO: forward elicitation requests to the client?
continue;
}
EventMsg::ApplyPatchApprovalRequest(ApplyPatchApprovalRequestEvent {
call_id,
turn_id: _,
started_at_ms: _,
reason,
grant_root,
changes,
}) => {
handle_patch_approval_request(
call_id,
reason,
grant_root,
changes,
outgoing.clone(),
thread.clone(),
request_id.clone(),
request_id_str.clone(),
event.id.clone(),
thread_id,
)
.await;
continue;
}
EventMsg::TurnComplete(TurnCompleteEvent {
last_agent_message, ..
}) => {
let text = match last_agent_message {
Some(msg) => msg,
None => "".to_string(),
};
let result = create_call_tool_result_with_thread_id(
thread_id, text, /*is_error*/ None,
);
active_turns.finish(&request_id, || {
outgoing.send_response(request_id.clone(), result);
});
break;
}
EventMsg::SessionConfigured(_) => {
tracing::error!("unexpected SessionConfigured event");
}
EventMsg::ThreadGoalUpdated(_) | EventMsg::ThreadQueueChanged(_) => {
// Ignore thread-scoped metadata updates in MCP tool runner.
}
EventMsg::McpStartupUpdate(_) | EventMsg::McpStartupComplete(_) => {
// Ignored in MCP tool runner.
}
EventMsg::AgentMessage(AgentMessageEvent { .. }) => {
// TODO: think how we want to support this in the MCP
}
EventMsg::AgentReasoningRawContent(_)
| EventMsg::TurnStarted(_)
| EventMsg::ThreadSettingsApplied(_)
| EventMsg::EnvironmentConnected(_)
| EventMsg::EnvironmentDisconnected(_)
| EventMsg::TokenCount(_)
| EventMsg::AgentReasoning(_)
| EventMsg::AgentReasoningSectionBreak(_)
| EventMsg::McpToolCallBegin(_)
| EventMsg::McpToolCallEnd(_)
| EventMsg::RealtimeConversationListVoicesResponse(_)
| EventMsg::ExecCommandBegin(_)
| EventMsg::TerminalInteraction(_)
| EventMsg::ExecCommandOutputDelta(_)
| EventMsg::ExecCommandEnd(_)
| EventMsg::StreamError(_)
| EventMsg::PatchApplyBegin(_)
| EventMsg::PatchApplyUpdated(_)
| EventMsg::PatchApplyEnd(_)
| EventMsg::TurnDiff(_)
| EventMsg::WebSearchBegin(_)
| EventMsg::WebSearchEnd(_)
| EventMsg::PlanUpdate(_)
| EventMsg::TurnAborted(_)
| EventMsg::UserMessage(_)
| EventMsg::ShutdownComplete
| EventMsg::ImageGenerationBegin(_)
| EventMsg::ImageGenerationEnd(_)
| EventMsg::ViewImageToolCall(_)
| EventMsg::RawResponseItem(_)
| EventMsg::RawResponseCompleted(_)
| EventMsg::EnteredReviewMode(_)
| EventMsg::ItemStarted(_)
| EventMsg::ItemCompleted(_)
| EventMsg::HookStarted(_)
| EventMsg::HookCompleted(_)
| EventMsg::AgentMessageContentDelta(_)
| EventMsg::ReasoningContentDelta(_)
| EventMsg::ReasoningRawContentDelta(_)
| EventMsg::ExitedReviewMode(_)
| EventMsg::RequestUserInput(_)
| EventMsg::RequestPermissions(_)
| EventMsg::DynamicToolCallRequest(_)
| EventMsg::DynamicToolCallResponse(_)
| EventMsg::ContextCompacted(_)
| EventMsg::ModelReroute(_)
| EventMsg::ThreadRolledBack(_)
| EventMsg::CollabAgentSpawnBegin(_)
| EventMsg::CollabAgentSpawnEnd(_)
| EventMsg::CollabAgentInteractionBegin(_)
| EventMsg::CollabAgentInteractionEnd(_)
| EventMsg::CollabWaitingBegin(_)
| EventMsg::CollabWaitingEnd(_)
| EventMsg::CollabCloseBegin(_)
| EventMsg::CollabCloseEnd(_)
| EventMsg::CollabResumeBegin(_)
| EventMsg::CollabResumeEnd(_)
| EventMsg::SubAgentActivity(_)
| EventMsg::RealtimeConversationStarted(_)
| EventMsg::RealtimeConversationSdp(_)
| EventMsg::RealtimeConversationRealtime(_)
| EventMsg::RealtimeConversationClosed(_)
| EventMsg::DeprecationNotice(_) => {
// For now, we do not do anything extra for these
// events. Note that
// send(codex_event_to_notification(&event)) above has
// already dispatched these events as notifications,
// though we may want to do give different treatment to
// individual events in the future.
}
}
}
Err(e) => {
let result = create_call_tool_result_with_thread_id(
thread_id,
format!("Codex runtime error: {e}"),
Some(true),
);
active_turns.finish(&request_id, || {
outgoing.send_response(request_id.clone(), result);
});
break;
}
}
}
}
#[cfg(test)]
mod tests {
use super::*;
use pretty_assertions::assert_eq;
#[test]
fn call_tool_result_includes_thread_id_in_structured_content() {
let thread_id = ThreadId::new();
let result = create_call_tool_result_with_thread_id(
thread_id,
"done".to_string(),
/*is_error*/ None,
);
assert_eq!(
result.structured_content,
Some(json!({
"threadId": thread_id,
"content": "done",
}))
);
}
}
-145
View File
@@ -1,145 +0,0 @@
use std::path::PathBuf;
use std::sync::Arc;
use codex_core::CodexThread;
use codex_protocol::ThreadId;
use codex_protocol::parse_command::ParsedCommand;
use codex_protocol::protocol::Op;
use codex_protocol::protocol::ReviewDecision;
use rmcp::model::ErrorData;
use rmcp::model::RequestId;
use serde::Deserialize;
use serde::Serialize;
use serde_json::Value;
use serde_json::json;
use tracing::error;
/// Conforms to the MCP elicitation request params shape, so it can be used as
/// the `params` field of an `elicitation/create` request.
#[derive(Debug, Deserialize, Serialize)]
pub struct ExecApprovalElicitRequestParams {
// These fields are required so that `params`
// conforms to ElicitRequestParams.
pub message: String,
#[serde(rename = "requestedSchema")]
pub requested_schema: Value,
// These are additional fields the client can use to
// correlate the request with the codex tool call.
#[serde(rename = "threadId")]
pub thread_id: ThreadId,
pub codex_elicitation: String,
pub codex_mcp_tool_call_id: String,
pub codex_event_id: String,
pub codex_call_id: String,
pub codex_command: Vec<String>,
pub codex_cwd: PathBuf,
pub codex_parsed_cmd: Vec<ParsedCommand>,
}
// TODO(mbolin): ExecApprovalResponse does not conform to ElicitResult. See:
// - https://github.com/modelcontextprotocol/modelcontextprotocol/blob/f962dc1780fa5eed7fb7c8a0232f1fc83ef220cd/schema/2025-06-18/schema.json#L617-L636
// - https://modelcontextprotocol.io/specification/draft/client/elicitation#protocol-messages
// It should have "action" and "content" fields.
#[derive(Debug, Serialize, Deserialize)]
pub struct ExecApprovalResponse {
pub decision: ReviewDecision,
}
#[allow(clippy::too_many_arguments)]
pub(crate) async fn handle_exec_approval_request(
command: Vec<String>,
cwd: PathBuf,
outgoing: Arc<crate::outgoing_message::OutgoingMessageSender>,
codex: Arc<CodexThread>,
request_id: RequestId,
tool_call_id: String,
event_id: String,
call_id: String,
approval_id: String,
codex_parsed_cmd: Vec<ParsedCommand>,
thread_id: ThreadId,
) {
let escaped_command =
shlex::try_join(command.iter().map(String::as_str)).unwrap_or_else(|_| command.join(" "));
let message = format!(
"Allow Codex to run `{escaped_command}` in `{cwd}`?",
cwd = cwd.to_string_lossy()
);
let params = ExecApprovalElicitRequestParams {
message,
requested_schema: json!({"type":"object","properties":{}}),
thread_id,
codex_elicitation: "exec-approval".to_string(),
codex_mcp_tool_call_id: tool_call_id.clone(),
codex_event_id: event_id.clone(),
codex_call_id: call_id,
codex_command: command,
codex_cwd: cwd,
codex_parsed_cmd,
};
let params_json = match serde_json::to_value(&params) {
Ok(value) => value,
Err(err) => {
let message = format!("Failed to serialize ExecApprovalElicitRequestParams: {err}");
error!("{message}");
outgoing.send_error(request_id.clone(), ErrorData::invalid_params(message, None));
return;
}
};
let on_response = outgoing
.send_request("elicitation/create", Some(params_json))
.await;
// Listen for the response on a separate task so we don't block the main agent loop.
{
let codex = codex.clone();
let approval_id = approval_id.clone();
let event_id = event_id.clone();
tokio::spawn(async move {
on_exec_approval_response(approval_id, event_id, on_response, codex).await;
});
}
}
async fn on_exec_approval_response(
approval_id: String,
event_id: String,
receiver: tokio::sync::oneshot::Receiver<serde_json::Value>,
codex: Arc<CodexThread>,
) {
let response = receiver.await;
let value = match response {
Ok(value) => value,
Err(err) => {
error!("request failed: {err:?}");
return;
}
};
// Try to deserialize `value` and then make the appropriate call to `codex`.
let response = serde_json::from_value::<ExecApprovalResponse>(value).unwrap_or_else(|err| {
error!("failed to deserialize ExecApprovalResponse: {err}");
// If we cannot deserialize the response, we deny the request to be
// conservative.
ExecApprovalResponse {
decision: ReviewDecision::denied("approval request failed"),
}
});
if let Err(err) = codex
.submit(Op::ExecApproval {
id: approval_id,
turn_id: Some(event_id),
decision: response.decision,
})
.await
{
error!("failed to submit ExecApproval: {err}");
}
}
@@ -1,89 +0,0 @@
use std::sync::Arc;
use codex_extension_api::ExtensionEventSink;
use codex_extension_api::ExtensionWarning;
use codex_protocol::ThreadId;
use codex_protocol::protocol::Event;
use codex_protocol::protocol::EventMsg;
use codex_protocol::protocol::WarningEvent;
use crate::active_turn_registry::ActiveTurnRegistry;
use crate::outgoing_message::OutgoingMessageSender;
use crate::outgoing_message::OutgoingNotificationMeta;
const MAX_EXTENSION_WARNING_BYTES: usize = 256;
pub(crate) fn extension_event_sink(
outgoing: Arc<OutgoingMessageSender>,
active_turns: Arc<ActiveTurnRegistry>,
) -> Arc<dyn ExtensionEventSink> {
Arc::new(McpExtensionEventSink {
outgoing,
active_turns,
})
}
struct McpExtensionEventSink {
outgoing: Arc<OutgoingMessageSender>,
active_turns: Arc<ActiveTurnRegistry>,
}
impl ExtensionEventSink for McpExtensionEventSink {
fn emit(&self, event: Event) {
tracing::debug!(
event_id = %event.id,
msg = ?event.msg,
"dropping unsupported extension event"
);
}
fn emit_warning(&self, warning: ExtensionWarning) {
let ExtensionWarning {
thread_id,
turn_id,
message,
} = warning;
let Ok(thread_id) = ThreadId::from_string(&thread_id) else {
tracing::warn!(%thread_id, "dropping extension warning with invalid thread id");
return;
};
let Some(turn_id) = turn_id else {
tracing::debug!(%thread_id, "dropping extension warning without a turn id");
return;
};
let mut message = message;
if message.len() > MAX_EXTENSION_WARNING_BYTES {
let mut truncate_at = MAX_EXTENSION_WARNING_BYTES;
while !message.is_char_boundary(truncate_at) {
truncate_at -= 1;
}
message.truncate(truncate_at);
}
let found_active_turn =
self.active_turns
.with_request_id(thread_id, &turn_id, |request_id| {
let event = Event {
id: turn_id.clone(),
msg: EventMsg::Warning(WarningEvent { message }),
};
self.outgoing.send_event_as_notification(
&event,
Some(OutgoingNotificationMeta {
request_id: Some(request_id.clone()),
thread_id: Some(thread_id),
}),
);
});
if !found_active_turn {
tracing::debug!(
%thread_id,
%turn_id,
"dropping extension warning without a matching active turn"
);
}
}
}
#[cfg(test)]
#[path = "extension_event_sink_tests.rs"]
mod tests;
@@ -1,190 +0,0 @@
use std::sync::Arc;
use codex_extension_api::ExtensionEventSink;
use codex_extension_api::ExtensionWarning;
use codex_protocol::ThreadId;
use codex_protocol::protocol::Event;
use codex_protocol::protocol::EventMsg;
use codex_protocol::protocol::WarningEvent;
use pretty_assertions::assert_eq;
use rmcp::model::RequestId;
use serde_json::json;
use tokio::sync::mpsc;
use super::extension_event_sink;
use crate::active_turn_registry::ActiveTurnRegistry;
use crate::outgoing_message::OutgoingMessage;
use crate::outgoing_message::OutgoingMessageSender;
use crate::outgoing_message::OutgoingNotification;
fn test_sink() -> (
Arc<dyn ExtensionEventSink>,
Arc<OutgoingMessageSender>,
Arc<ActiveTurnRegistry>,
mpsc::UnboundedReceiver<OutgoingMessage>,
) {
let (outgoing_tx, outgoing_rx) = mpsc::unbounded_channel();
let outgoing = Arc::new(OutgoingMessageSender::new(outgoing_tx));
let active_turns = Arc::new(ActiveTurnRegistry::default());
(
extension_event_sink(Arc::clone(&outgoing), Arc::clone(&active_turns)),
outgoing,
active_turns,
outgoing_rx,
)
}
#[tokio::test(flavor = "current_thread")]
async fn warning_is_enqueued_before_following_response() {
let (sink, outgoing, active_turns, mut outgoing_rx) = test_sink();
let request_id = RequestId::Number(7);
let thread_id = ThreadId::new();
let turn_id = "turn-7".to_string();
active_turns.register(request_id.clone(), thread_id, turn_id.clone());
sink.emit_warning(ExtensionWarning {
thread_id: thread_id.to_string(),
turn_id: Some(turn_id),
message: "warning".to_string(),
});
active_turns.finish(&request_id, || {
outgoing.send_response(request_id.clone(), json!({}));
});
let Some(OutgoingMessage::Notification(notification)) = outgoing_rx.recv().await else {
panic!("warning notification should be enqueued before the response");
};
assert_eq!(notification.method, "codex/event");
assert_eq!(
notification.params.as_ref().map(|params| &params["msg"]),
Some(&json!({
"message": "warning",
"type": "warning",
}))
);
}
#[tokio::test(flavor = "current_thread")]
async fn colliding_request_id_strings_route_warnings_independently() {
let (sink, _outgoing, active_turns, mut outgoing_rx) = test_sink();
let thread_id = ThreadId::new();
let numeric_turn_id = "019c0000-0000-7000-8000-000000000001".to_string();
let string_turn_id = "019c0000-0000-7000-8000-000000000002".to_string();
active_turns.register(RequestId::Number(7), thread_id, numeric_turn_id.clone());
active_turns.register(
RequestId::String("7".into()),
thread_id,
string_turn_id.clone(),
);
sink.emit_warning(ExtensionWarning {
thread_id: thread_id.to_string(),
turn_id: Some(numeric_turn_id.clone()),
message: "numeric warning".to_string(),
});
sink.emit_warning(ExtensionWarning {
thread_id: thread_id.to_string(),
turn_id: Some(string_turn_id.clone()),
message: "string warning".to_string(),
});
let Some(OutgoingMessage::Notification(numeric_warning)) = outgoing_rx.recv().await else {
panic!("expected warning for numeric request id");
};
assert_eq!(
numeric_warning,
OutgoingNotification {
method: "codex/event".to_string(),
params: Some(json!({
"_meta": {
"requestId": 7,
"threadId": thread_id,
},
"id": numeric_turn_id,
"msg": {
"message": "numeric warning",
"type": "warning",
},
})),
}
);
let Some(OutgoingMessage::Notification(string_warning)) = outgoing_rx.recv().await else {
panic!("expected warning for string request id");
};
assert_eq!(
string_warning,
OutgoingNotification {
method: "codex/event".to_string(),
params: Some(json!({
"_meta": {
"requestId": "7",
"threadId": thread_id,
},
"id": string_turn_id,
"msg": {
"message": "string warning",
"type": "warning",
},
})),
}
);
}
#[tokio::test(flavor = "current_thread")]
async fn warning_requires_a_matching_active_turn() {
let (sink, _outgoing, active_turns, mut outgoing_rx) = test_sink();
let request_id = RequestId::Number(7);
let thread_id = ThreadId::new();
active_turns.register(request_id, thread_id, "turn-7".to_string());
sink.emit_warning(ExtensionWarning {
thread_id: thread_id.to_string(),
turn_id: Some("different-turn".to_string()),
message: "warning".to_string(),
});
tokio::task::yield_now().await;
assert!(outgoing_rx.try_recv().is_err());
}
#[tokio::test(flavor = "current_thread")]
async fn warning_is_truncated_to_256_utf8_bytes() {
let (sink, _outgoing, active_turns, mut outgoing_rx) = test_sink();
let request_id = RequestId::Number(7);
let thread_id = ThreadId::new();
let turn_id = "turn-7".to_string();
active_turns.register(request_id.clone(), thread_id, turn_id.clone());
sink.emit_warning(ExtensionWarning {
thread_id: thread_id.to_string(),
turn_id: Some(turn_id),
message: format!("{}é", "a".repeat(255)),
});
let Some(OutgoingMessage::Notification(notification)) = outgoing_rx.recv().await else {
panic!("expected warning notification");
};
assert_eq!(
notification.params.as_ref().map(|params| &params["msg"]),
Some(&json!({
"message": "a".repeat(255),
"type": "warning",
}))
);
}
#[tokio::test(flavor = "current_thread")]
async fn generic_extension_events_are_dropped() {
let (sink, _outgoing, _running_requests, mut outgoing_rx) = test_sink();
sink.emit(Event {
id: "turn".to_string(),
msg: EventMsg::Warning(WarningEvent {
message: "warning".to_string(),
}),
});
tokio::task::yield_now().await;
assert!(outgoing_rx.try_recv().is_err());
}
-275
View File
@@ -1,275 +0,0 @@
//! Prototype MCP server.
#![recursion_limit = "256"]
#![deny(clippy::print_stdout, clippy::print_stderr)]
use std::io::ErrorKind;
use std::io::Result as IoResult;
use std::sync::Arc;
use codex_arg0::Arg0DispatchPaths;
use codex_core::config::ConfigBuilder;
use codex_core::resolve_installation_id;
use codex_exec_server::EnvironmentManager;
use codex_exec_server::ExecServerRuntimePaths;
use codex_login::default_client::set_default_client_residency_requirement;
use codex_utils_cli::CliConfigOverrides;
use rmcp::model::ClientNotification;
use rmcp::model::ClientRequest;
use rmcp::model::JsonRpcMessage;
use serde_json::Value;
use tokio::io::AsyncBufReadExt;
use tokio::io::AsyncWriteExt;
use tokio::io::BufReader;
use tokio::io::{self};
use tokio::sync::mpsc;
use tracing::debug;
use tracing::error;
use tracing::info;
use tracing_subscriber::EnvFilter;
use tracing_subscriber::prelude::*;
mod active_turn_registry;
mod codex_tool_config;
mod codex_tool_runner;
mod exec_approval;
mod extension_event_sink;
pub(crate) mod message_processor;
mod outgoing_message;
mod patch_approval;
use crate::message_processor::MessageProcessor;
use crate::outgoing_message::OutgoingJsonRpcMessage;
use crate::outgoing_message::OutgoingMessage;
use crate::outgoing_message::OutgoingMessageSender;
pub use crate::codex_tool_config::CodexToolCallParam;
pub use crate::codex_tool_config::CodexToolCallReplyParam;
pub use crate::exec_approval::ExecApprovalElicitRequestParams;
pub use crate::exec_approval::ExecApprovalResponse;
pub use crate::patch_approval::PatchApprovalElicitRequestParams;
pub use crate::patch_approval::PatchApprovalResponse;
/// Size of the bounded channels used to communicate between tasks. The value
/// is a balance between throughput and memory usage – 128 messages should be
/// plenty for an interactive CLI.
const CHANNEL_CAPACITY: usize = 128;
const DEFAULT_ANALYTICS_ENABLED: bool = true;
const OTEL_SERVICE_NAME: &str = "codex_mcp_server";
type IncomingMessage = JsonRpcMessage<ClientRequest, Value, ClientNotification>;
pub async fn run_main(
arg0_paths: Arg0DispatchPaths,
cli_config_overrides: CliConfigOverrides,
strict_config: bool,
) -> IoResult<()> {
reject_workload_identity(codex_login::is_workload_identity_selected())?;
// Parse CLI overrides once and derive the base Config eagerly so later
// components do not need to work with raw TOML values.
let cli_kv_overrides = cli_config_overrides.parse_overrides().map_err(|e| {
std::io::Error::new(
ErrorKind::InvalidInput,
format!("error parsing -c overrides: {e}"),
)
})?;
let config = ConfigBuilder::default()
.cli_overrides(cli_kv_overrides)
.strict_config(strict_config)
.build()
.await
.map_err(|e| {
std::io::Error::new(ErrorKind::InvalidData, format!("error loading config: {e}"))
})?;
config.auth_config().validate()?;
set_default_client_residency_requirement(config.enforce_residency.value());
let otel = codex_core::otel_init::build_provider(
&config,
env!("CARGO_PKG_VERSION"),
Some(OTEL_SERVICE_NAME),
DEFAULT_ANALYTICS_ENABLED,
)
.map_err(|e| {
std::io::Error::new(
ErrorKind::InvalidData,
format!("error loading otel config: {e}"),
)
})?;
codex_core::otel_init::record_process_start(otel.as_ref(), OTEL_SERVICE_NAME);
codex_core::otel_init::install_sqlite_telemetry(otel.as_ref(), OTEL_SERVICE_NAME);
let state_db = codex_core::init_state_db(&config).await;
let environment_manager = Arc::new(
EnvironmentManager::from_codex_home(
config.codex_home.clone(),
Some(ExecServerRuntimePaths::from_optional_paths(
arg0_paths.codex_self_exe.clone(),
arg0_paths.codex_linux_sandbox_exe.clone(),
)?),
config.http_client_factory(),
)
.await
.map_err(std::io::Error::other)?,
);
let fmt_layer = tracing_subscriber::fmt::layer()
.with_writer(std::io::stderr)
.with_filter(EnvFilter::from_default_env());
let otel_logger_layer = otel.as_ref().and_then(|provider| provider.logger_layer());
let otel_tracing_layer = otel.as_ref().and_then(|provider| provider.tracing_layer());
let _ = tracing_subscriber::registry()
.with(fmt_layer)
.with(otel_logger_layer)
.with(otel_tracing_layer)
.try_init();
// Set up channels.
let (incoming_tx, mut incoming_rx) = mpsc::channel::<IncomingMessage>(CHANNEL_CAPACITY);
let (outgoing_tx, mut outgoing_rx) = mpsc::unbounded_channel::<OutgoingMessage>();
let installation_id = resolve_installation_id(&config.codex_home).await?;
// Task: read from stdin, push to `incoming_tx`.
let stdin_reader_handle = tokio::spawn({
async move {
let stdin = io::stdin();
let reader = BufReader::new(stdin);
let mut lines = reader.lines();
while let Some(line) = lines.next_line().await.unwrap_or_default() {
match serde_json::from_str::<IncomingMessage>(&line) {
Ok(msg) => {
if incoming_tx.send(msg).await.is_err() {
// Receiver gone – nothing left to do.
break;
}
}
Err(e) => error!("Failed to deserialize JSON-RPC message: {e}"),
}
}
debug!("stdin reader finished (EOF)");
}
});
// Task: process incoming messages.
let processor_handle = tokio::spawn({
let outgoing_message_sender = OutgoingMessageSender::new(outgoing_tx);
let mut processor = MessageProcessor::new(
outgoing_message_sender,
arg0_paths,
Arc::new(config),
environment_manager,
state_db,
installation_id,
)
.await?;
async move {
while let Some(msg) = incoming_rx.recv().await {
match msg {
JsonRpcMessage::Request(r) => processor.process_request(r).await,
JsonRpcMessage::Response(r) => processor.process_response(r).await,
JsonRpcMessage::Notification(n) => processor.process_notification(n).await,
JsonRpcMessage::Error(e) => processor.process_error(e),
}
}
info!("processor task exited (channel closed)");
}
});
// Task: write outgoing messages to stdout.
let stdout_writer_handle = tokio::spawn(async move {
let mut stdout = io::stdout();
while let Some(outgoing_message) = outgoing_rx.recv().await {
let msg: OutgoingJsonRpcMessage = outgoing_message.into();
match serde_json::to_string(&msg) {
Ok(json) => {
if let Err(e) = stdout.write_all(json.as_bytes()).await {
error!("Failed to write to stdout: {e}");
break;
}
if let Err(e) = stdout.write_all(b"\n").await {
error!("Failed to write newline to stdout: {e}");
break;
}
}
Err(e) => error!("Failed to serialize JSON-RPC message: {e}"),
}
}
info!("stdout writer exited (channel closed)");
});
// Wait for all tasks to finish. The typical exit path is the stdin reader
// hitting EOF which, once it drops `incoming_tx`, propagates shutdown to
// the processor and then to the stdout task.
let _ = tokio::join!(stdin_reader_handle, processor_handle, stdout_writer_handle);
Ok(())
}
fn reject_workload_identity(workload_identity_selected: bool) -> IoResult<()> {
if workload_identity_selected {
return Err(std::io::Error::new(
ErrorKind::Unsupported,
"workload identity is not supported by `codex mcp-server`",
));
}
Ok(())
}
#[cfg(test)]
#[path = "workload_identity_tests.rs"]
mod workload_identity_tests;
#[cfg(test)]
mod tests {
use super::*;
use codex_config::types::OtelExporterKind;
use codex_core::config::ConfigBuilder;
use pretty_assertions::assert_eq;
use std::collections::HashMap;
use tempfile::TempDir;
#[test]
fn mcp_server_defaults_analytics_to_enabled() {
assert_eq!(DEFAULT_ANALYTICS_ENABLED, true);
}
#[tokio::test]
async fn mcp_server_builds_otel_provider_with_logs_traces_and_metrics() -> anyhow::Result<()> {
let codex_home = TempDir::new()?;
let mut config = ConfigBuilder::default()
.codex_home(codex_home.path().to_path_buf())
.build()
.await?;
let exporter = OtelExporterKind::OtlpGrpc {
endpoint: "http://localhost:4317".to_string(),
headers: HashMap::new(),
tls: None,
};
config.otel.exporter = exporter.clone();
config.otel.trace_exporter = exporter.clone();
config.otel.metrics_exporter = exporter;
config.analytics_enabled = None;
let provider = codex_core::otel_init::build_provider(
&config,
"0.0.0-test",
Some(OTEL_SERVICE_NAME),
DEFAULT_ANALYTICS_ENABLED,
)
.map_err(|err| anyhow::anyhow!(err.to_string()))?
.expect("otel provider");
assert!(provider.logger.is_some(), "expected log exporter");
assert!(
provider.tracer_provider.is_some(),
"expected trace exporter"
);
assert!(provider.metrics().is_some(), "expected metrics exporter");
provider.shutdown();
Ok(())
}
}
-18
View File
@@ -1,18 +0,0 @@
#![recursion_limit = "256"]
use codex_arg0::Arg0DispatchPaths;
use codex_arg0::arg0_dispatch_or_else;
use codex_mcp_server::run_main;
use codex_utils_cli::CliConfigOverrides;
fn main() -> anyhow::Result<()> {
arg0_dispatch_or_else(|arg0_paths: Arg0DispatchPaths| async move {
run_main(
arg0_paths,
CliConfigOverrides::default(),
/*strict_config*/ false,
)
.await?;
Ok(())
})
}
@@ -1,585 +0,0 @@
use std::sync::Arc;
use codex_arg0::Arg0DispatchPaths;
use codex_core::StateDbHandle;
use codex_core::ThreadManager;
use codex_core::config::Config;
use codex_exec_server::EnvironmentManager;
use codex_extension_api::ExtensionRegistryBuilder;
use codex_home::CodexHomeUserInstructionsProvider;
use codex_login::AuthManager;
use codex_login::default_client::USER_AGENT_SUFFIX;
use codex_login::default_client::get_codex_user_agent;
use codex_protocol::protocol::SessionSource;
use rmcp::model::CallToolRequestParams;
use rmcp::model::CallToolResult;
use rmcp::model::ClientNotification;
use rmcp::model::ClientRequest;
use rmcp::model::ErrorCode;
use rmcp::model::ErrorData;
use rmcp::model::Implementation;
use rmcp::model::InitializeResult;
use rmcp::model::JsonRpcError;
use rmcp::model::JsonRpcNotification;
use rmcp::model::JsonRpcRequest;
use rmcp::model::JsonRpcResponse;
use rmcp::model::RequestId;
use rmcp::model::ServerCapabilities;
use serde_json::json;
use tokio::task;
use crate::active_turn_registry::ActiveTurnRegistry;
use crate::codex_tool_config::CodexToolCallParam;
use crate::codex_tool_config::CodexToolCallReplyParam;
use crate::codex_tool_config::create_tool_for_codex_tool_call_param;
use crate::codex_tool_config::create_tool_for_codex_tool_call_reply_param;
use crate::extension_event_sink::extension_event_sink;
use crate::outgoing_message::OutgoingMessageSender;
pub(crate) struct MessageProcessor {
outgoing: Arc<OutgoingMessageSender>,
initialized: bool,
arg0_paths: Arg0DispatchPaths,
thread_manager: Arc<ThreadManager>,
active_turns: Arc<ActiveTurnRegistry>,
}
impl MessageProcessor {
/// Create a new `MessageProcessor`, retaining a handle to the outgoing
/// `Sender` so handlers can enqueue messages to be written to stdout.
pub(crate) async fn new(
outgoing: OutgoingMessageSender,
arg0_paths: Arg0DispatchPaths,
config: Arc<Config>,
environment_manager: Arc<EnvironmentManager>,
state_db: Option<StateDbHandle>,
installation_id: String,
) -> std::io::Result<Self> {
let outgoing = Arc::new(outgoing);
let auth_manager = AuthManager::shared_from_config(
config.as_ref(),
/*enable_codex_api_key_env*/ false,
)
.await
.map_err(std::io::Error::other)?;
let user_instructions_provider = Arc::new(CodexHomeUserInstructionsProvider::new(
config.codex_home.clone(),
));
let active_turns = Arc::new(ActiveTurnRegistry::default());
let mut extensions = ExtensionRegistryBuilder::<Config>::with_event_sink(
extension_event_sink(Arc::clone(&outgoing), Arc::clone(&active_turns)),
);
codex_git_attribution::install(
&mut extensions,
auth_manager.clone(),
config.chatgpt_base_url.clone(),
config.http_client_factory(),
);
codex_image_generation_extension::install(
&mut extensions,
auth_manager.clone(),
|config: &Config| Some(config.codex_home.clone()),
);
let skill_providers = codex_skills_extension::SkillProviders::new()
.with_host_provider(Arc::new(codex_skills_extension::HostSkillProvider::new()));
codex_skills_extension::install_with_providers_and_metrics(
&mut extensions,
skill_providers,
codex_otel::global(),
|config: &Config| codex_skills_extension::SkillsExtensionConfig {
include_instructions: config.include_skill_instructions,
max_context_tokens: config.skill_max_context_tokens,
bundled_skills_enabled: config.bundled_skills_enabled(),
orchestrator_skills_enabled: config.orchestrator_skills_enabled,
shadow_selection_enabled: config
.features
.enabled(codex_features::Feature::SkillSearch),
},
);
let thread_manager = Arc::new(ThreadManager::new(
config.as_ref(),
Arc::clone(&auth_manager),
codex_core::build_models_manager(config.as_ref(), auth_manager),
codex_core::CodexAppsToolsCache::default(),
SessionSource::Mcp,
environment_manager,
Arc::new(extensions.build()),
user_instructions_provider,
/*analytics_events_client*/ None,
codex_core::passthrough_image_store(),
codex_core::thread_store_from_config(config.as_ref(), state_db.clone()),
codex_core::local_agent_graph_store_from_state_db(state_db.as_ref()),
installation_id,
/*attestation_provider*/ None,
/*external_time_provider*/ None,
));
Ok(Self {
outgoing,
initialized: false,
arg0_paths,
thread_manager,
active_turns,
})
}
pub(crate) async fn process_request(&mut self, request: JsonRpcRequest<ClientRequest>) {
let request_id = request.id.clone();
let client_request = request.request;
match client_request {
ClientRequest::InitializeRequest(params) => {
self.handle_initialize(request_id, params.params).await;
}
ClientRequest::PingRequest(_params) => {
self.handle_ping(request_id).await;
}
ClientRequest::ListResourcesRequest(params) => {
self.handle_list_resources(params.params);
}
ClientRequest::ListResourceTemplatesRequest(params) => {
self.handle_list_resource_templates(params.params);
}
ClientRequest::ReadResourceRequest(params) => {
self.handle_read_resource(params.params);
}
ClientRequest::SubscribeRequest(params) => {
self.handle_subscribe(params.params);
}
ClientRequest::UnsubscribeRequest(params) => {
self.handle_unsubscribe(params.params);
}
ClientRequest::ListPromptsRequest(params) => {
self.handle_list_prompts(params.params);
}
ClientRequest::GetPromptRequest(params) => {
self.handle_get_prompt(params.params);
}
ClientRequest::ListToolsRequest(params) => {
self.handle_list_tools(request_id, params.params).await;
}
ClientRequest::CallToolRequest(params) => {
self.handle_call_tool(request_id, params.params).await;
}
ClientRequest::SetLevelRequest(params) => {
self.handle_set_level(params.params);
}
ClientRequest::CompleteRequest(params) => {
self.handle_complete(params.params);
}
ClientRequest::GetTaskRequest(_) => {
self.handle_unsupported_request(request_id, "tasks/get")
.await;
}
ClientRequest::CancelTaskRequest(_) => {
self.handle_unsupported_request(request_id, "tasks/cancel")
.await;
}
ClientRequest::CustomRequest(custom) => {
let method = custom.method;
self.outgoing.send_error(
request_id,
ErrorData::new(
ErrorCode::METHOD_NOT_FOUND,
format!("method not found: {method}"),
Some(json!({ "method": method })),
),
);
}
request => {
self.handle_unsupported_request(request_id, request.method())
.await;
}
}
}
pub(crate) async fn process_response(&mut self, response: JsonRpcResponse<serde_json::Value>) {
tracing::info!("<- response: {:?}", response);
let JsonRpcResponse { id, result, .. } = response;
self.outgoing.notify_client_response(id, result).await
}
pub(crate) async fn process_notification(
&mut self,
notification: JsonRpcNotification<ClientNotification>,
) {
match notification.notification {
ClientNotification::CancelledNotification(params) => {
self.handle_cancelled_notification(params.params).await;
}
ClientNotification::ProgressNotification(params) => {
self.handle_progress_notification(params.params);
}
ClientNotification::RootsListChangedNotification(_params) => {
self.handle_roots_list_changed();
}
ClientNotification::InitializedNotification(_) => {
self.handle_initialized_notification();
}
ClientNotification::CustomNotification(_) => {
tracing::warn!("ignoring custom client notification");
}
_ => {
tracing::warn!("ignoring unsupported client notification");
}
}
}
pub(crate) fn process_error(&mut self, err: JsonRpcError) {
tracing::error!("<- error: {:?}", err);
}
async fn handle_initialize(
&mut self,
id: RequestId,
params: rmcp::model::InitializeRequestParams,
) {
tracing::info!("initialize -> params: {:?}", params);
if self.initialized {
self.outgoing.send_error(
id,
ErrorData::invalid_request("initialize called more than once", None),
);
return;
}
let client_info = params.client_info;
let name = client_info.name;
let version = client_info.version;
let user_agent_suffix = format!("{name}; {version}");
if let Ok(mut suffix) = USER_AGENT_SUFFIX.lock() {
*suffix = Some(user_agent_suffix);
}
let server_info =
Implementation::new("codex-mcp-server", env!("CARGO_PKG_VERSION")).with_title("Codex");
// Preserve Codex's existing non-spec `serverInfo.user_agent` field.
let mut server_info_value = match serde_json::to_value(&server_info) {
Ok(value) => value,
Err(err) => {
self.outgoing.send_error(
id,
ErrorData::internal_error(
format!("failed to serialize server info: {err}"),
None,
),
);
return;
}
};
if let serde_json::Value::Object(ref mut obj) = server_info_value {
obj.insert("user_agent".to_string(), json!(get_codex_user_agent()));
}
let capabilities = ServerCapabilities::builder()
.enable_tools()
.enable_tool_list_changed()
.build();
let result = InitializeResult::new(capabilities)
.with_protocol_version(params.protocol_version)
.with_server_info(server_info);
let mut result_value = match serde_json::to_value(result) {
Ok(value) => value,
Err(err) => {
self.outgoing.send_error(
id,
ErrorData::internal_error(
format!("failed to serialize initialize response: {err}"),
None,
),
);
return;
}
};
if let serde_json::Value::Object(ref mut obj) = result_value {
obj.insert("serverInfo".to_string(), server_info_value);
}
self.initialized = true;
self.outgoing.send_response(id, result_value);
}
async fn handle_ping(&self, id: RequestId) {
tracing::info!("ping");
self.outgoing.send_response(id, json!({}));
}
fn handle_list_resources(&self, params: Option<rmcp::model::PaginatedRequestParams>) {
tracing::info!("resources/list -> params: {:?}", params);
}
fn handle_list_resource_templates(&self, params: Option<rmcp::model::PaginatedRequestParams>) {
tracing::info!("resources/templates/list -> params: {:?}", params);
}
fn handle_read_resource(&self, params: rmcp::model::ReadResourceRequestParams) {
tracing::info!("resources/read -> params: {:?}", params);
}
fn handle_subscribe(&self, params: rmcp::model::SubscribeRequestParams) {
tracing::info!("resources/subscribe -> params: {:?}", params);
}
fn handle_unsubscribe(&self, params: rmcp::model::UnsubscribeRequestParams) {
tracing::info!("resources/unsubscribe -> params: {:?}", params);
}
fn handle_list_prompts(&self, params: Option<rmcp::model::PaginatedRequestParams>) {
tracing::info!("prompts/list -> params: {:?}", params);
}
fn handle_get_prompt(&self, params: rmcp::model::GetPromptRequestParams) {
tracing::info!("prompts/get -> params: {:?}", params);
}
async fn handle_list_tools(
&self,
id: RequestId,
params: Option<rmcp::model::PaginatedRequestParams>,
) {
tracing::trace!("tools/list -> {params:?}");
let result = rmcp::model::ListToolsResult::with_all_items(vec![
create_tool_for_codex_tool_call_param(),
create_tool_for_codex_tool_call_reply_param(),
]);
self.outgoing.send_response(id, result);
}
async fn handle_call_tool(&self, id: RequestId, params: CallToolRequestParams) {
tracing::info!("tools/call -> params: {:?}", params);
let CallToolRequestParams {
name, arguments, ..
} = params;
match name.as_ref() {
"codex" => self.handle_tool_call_codex(id, arguments).await,
"codex-reply" => {
self.handle_tool_call_codex_session_reply(id, arguments)
.await
}
_ => {
let result = CallToolResult::error(vec![rmcp::model::ContentBlock::text(format!(
"Unknown tool '{name}'"
))]);
self.outgoing.send_response(id, result);
}
}
}
async fn handle_tool_call_codex(
&self,
id: RequestId,
arguments: Option<rmcp::model::JsonObject>,
) {
let arguments = arguments.map(serde_json::Value::Object);
let (initial_prompt, config): (String, Config) = match arguments {
Some(json_val) => match serde_json::from_value::<CodexToolCallParam>(json_val) {
Ok(tool_cfg) => match tool_cfg.into_config(self.arg0_paths.clone()).await {
Ok(cfg) => cfg,
Err(e) => {
let result = CallToolResult::error(vec![rmcp::model::ContentBlock::text(
format!("Failed to load Codex configuration from overrides: {e}"),
)]);
self.outgoing.send_response(id, result);
return;
}
},
Err(e) => {
let result = CallToolResult::error(vec![rmcp::model::ContentBlock::text(
format!("Failed to parse configuration for Codex tool: {e}"),
)]);
self.outgoing.send_response(id, result);
return;
}
},
None => {
let result = CallToolResult::error(vec![rmcp::model::ContentBlock::text(
"Missing arguments for codex tool-call; the `prompt` field is required.",
)]);
self.outgoing.send_response(id, result);
return;
}
};
// Clone outgoing and server to move into async task.
let outgoing = self.outgoing.clone();
let thread_manager = self.thread_manager.clone();
let active_turns = Arc::clone(&self.active_turns);
// Spawn an async task to handle the Codex session so that we do not
// block the synchronous message-processing loop.
task::spawn(async move {
// Run the Codex session and stream events back to the client.
crate::codex_tool_runner::run_codex_tool_session(
id,
initial_prompt,
config,
outgoing,
thread_manager,
active_turns,
)
.await;
});
}
async fn handle_tool_call_codex_session_reply(
&self,
request_id: RequestId,
arguments: Option<rmcp::model::JsonObject>,
) {
let arguments = arguments.map(serde_json::Value::Object);
tracing::info!("tools/call -> params: {:?}", arguments);
// parse arguments
let codex_tool_call_reply_param: CodexToolCallReplyParam = match arguments {
Some(json_val) => match serde_json::from_value::<CodexToolCallReplyParam>(json_val) {
Ok(params) => params,
Err(e) => {
tracing::error!("Failed to parse Codex tool call reply parameters: {e}");
let result = CallToolResult::error(vec![rmcp::model::ContentBlock::text(
format!("Failed to parse configuration for Codex tool: {e}"),
)]);
self.outgoing.send_response(request_id, result);
return;
}
},
None => {
tracing::error!(
"Missing arguments for codex-reply tool-call; the `thread_id` and `prompt` fields are required."
);
let result = CallToolResult::error(vec![rmcp::model::ContentBlock::text(
"Missing arguments for codex-reply tool-call; the `thread_id` and `prompt` fields are required.",
)]);
self.outgoing.send_response(request_id, result);
return;
}
};
let thread_id = match codex_tool_call_reply_param.get_thread_id() {
Ok(id) => id,
Err(e) => {
tracing::error!("Failed to parse thread_id: {e}");
let result = CallToolResult::error(vec![rmcp::model::ContentBlock::text(format!(
"Failed to parse thread_id: {e}"
))]);
self.outgoing.send_response(request_id, result);
return;
}
};
// Clone outgoing to move into async task.
let outgoing = self.outgoing.clone();
let active_turns = Arc::clone(&self.active_turns);
let codex = match self.thread_manager.get_thread(thread_id).await {
Ok(c) => c,
Err(_) => {
tracing::warn!("Session not found for thread_id: {thread_id}");
let result = crate::codex_tool_runner::create_call_tool_result_with_thread_id(
thread_id,
format!("Session not found for thread_id: {thread_id}"),
Some(true),
);
outgoing.send_response(request_id, result);
return;
}
};
// Spawn the long-running reply handler.
let prompt = codex_tool_call_reply_param.prompt.clone();
tokio::spawn({
let outgoing = outgoing.clone();
let active_turns = Arc::clone(&active_turns);
async move {
crate::codex_tool_runner::run_codex_tool_session_reply(
thread_id,
codex,
outgoing,
request_id,
prompt,
active_turns,
)
.await;
}
});
}
#[allow(deprecated)]
fn handle_set_level(&self, params: rmcp::model::SetLevelRequestParams) {
tracing::info!("logging/setLevel -> params: {:?}", params);
}
fn handle_complete(&self, params: rmcp::model::CompleteRequestParams) {
tracing::info!("completion/complete -> params: {:?}", params);
}
async fn handle_unsupported_request(&self, id: RequestId, method: &str) {
self.outgoing.send_error(
id,
ErrorData::new(
ErrorCode::METHOD_NOT_FOUND,
format!("method not found: {method}"),
Some(json!({ "method": method })),
),
);
}
// ---------------------------------------------------------------------
// Notification handlers
// ---------------------------------------------------------------------
async fn handle_cancelled_notification(&self, params: rmcp::model::CancelledNotificationParam) {
let Some(request_id) = params.request_id else {
tracing::warn!("ignoring cancellation without a request id");
return;
};
// Create a stable string form early for logging.
let request_id_string = request_id.to_string();
// Resolve the thread for the active MCP request.
let thread_id = match self.active_turns.thread_id(&request_id) {
Some(id) => id,
None => {
tracing::warn!("Session not found for request_id: {request_id_string}");
return;
}
};
tracing::info!("thread_id: {thread_id}");
// Obtain the Codex thread from the server.
let codex_arc = match self.thread_manager.get_thread(thread_id).await {
Ok(c) => c,
Err(_) => {
tracing::warn!("Session not found for thread_id: {thread_id}");
return;
}
};
// Submit interrupt to Codex.
if let Err(e) = codex_arc
.submit(codex_protocol::protocol::Op::Interrupt)
.await
{
tracing::error!("Failed to submit interrupt to Codex: {e}");
return;
}
// Stop routing extension events to the cancelled turn.
self.active_turns.unregister(&request_id);
}
fn handle_progress_notification(&self, params: rmcp::model::ProgressNotificationParam) {
tracing::info!("notifications/progress -> params: {:?}", params);
}
fn handle_roots_list_changed(&self) {
tracing::info!("notifications/roots/list_changed");
}
fn handle_initialized_notification(&self) {
tracing::info!("notifications/initialized");
}
}
-492
View File
@@ -1,492 +0,0 @@
use std::collections::HashMap;
use std::sync::atomic::AtomicI64;
use std::sync::atomic::Ordering;
use codex_protocol::ThreadId;
use codex_protocol::protocol::Event;
use rmcp::model::CustomNotification;
use rmcp::model::CustomRequest;
use rmcp::model::ErrorData;
use rmcp::model::JsonRpcError;
use rmcp::model::JsonRpcMessage;
use rmcp::model::JsonRpcNotification;
use rmcp::model::JsonRpcRequest;
use rmcp::model::JsonRpcResponse;
use rmcp::model::JsonRpcVersion2_0;
use rmcp::model::RequestId;
use serde::Serialize;
use serde_json::Value;
use tokio::sync::Mutex;
use tokio::sync::mpsc;
use tokio::sync::oneshot;
use tracing::warn;
pub(crate) type OutgoingJsonRpcMessage = JsonRpcMessage<CustomRequest, Value, CustomNotification>;
/// Sends messages to the client and manages request callbacks.
pub(crate) struct OutgoingMessageSender {
next_request_id: AtomicI64,
sender: mpsc::UnboundedSender<OutgoingMessage>,
request_id_to_callback: Mutex<HashMap<RequestId, oneshot::Sender<Value>>>,
}
impl OutgoingMessageSender {
pub(crate) fn new(sender: mpsc::UnboundedSender<OutgoingMessage>) -> Self {
Self {
next_request_id: AtomicI64::new(0),
sender,
request_id_to_callback: Mutex::new(HashMap::new()),
}
}
pub(crate) async fn send_request(
&self,
method: &str,
params: Option<serde_json::Value>,
) -> oneshot::Receiver<Value> {
let id = RequestId::Number(self.next_request_id.fetch_add(1, Ordering::Relaxed));
let outgoing_message_id = id.clone();
let (tx_approve, rx_approve) = oneshot::channel();
{
let mut request_id_to_callback = self.request_id_to_callback.lock().await;
request_id_to_callback.insert(id, tx_approve);
}
let outgoing_message = OutgoingMessage::Request(OutgoingRequest {
id: outgoing_message_id,
method: method.to_string(),
params,
});
let _ = self.sender.send(outgoing_message);
rx_approve
}
pub(crate) async fn notify_client_response(&self, id: RequestId, result: Value) {
let entry = {
let mut request_id_to_callback = self.request_id_to_callback.lock().await;
request_id_to_callback.remove_entry(&id)
};
match entry {
Some((id, sender)) => {
if let Err(err) = sender.send(result) {
warn!("could not notify callback for {id:?} due to: {err:?}");
}
}
None => {
warn!("could not find callback for {id:?}");
}
}
}
pub(crate) fn send_response<T: Serialize>(&self, id: RequestId, response: T) {
let mut result = match serde_json::to_value(response) {
Ok(result) => result,
Err(err) => {
self.send_error(
id,
ErrorData::internal_error(format!("failed to serialize response: {err}"), None),
);
return;
}
};
// rmcp result constructors include the modern discriminator by default.
// This legacy server serializes responses directly, bypassing rmcp's
// protocol-aware response handling, so preserve the historical wire shape:
// https://github.com/modelcontextprotocol/rust-sdk/issues/1036
if let Value::Object(object) = &mut result
&& object.get("resultType").and_then(Value::as_str) == Some("complete")
{
object.remove("resultType");
}
let outgoing_message = OutgoingMessage::Response(OutgoingResponse { id, result });
let _ = self.sender.send(outgoing_message);
}
/// This is used with the MCP server, but not the more general JSON-RPC app
/// server. Prefer [`OutgoingMessageSender::send_server_notification`] where
/// possible.
pub(crate) fn send_event_as_notification(
&self,
event: &Event,
meta: Option<OutgoingNotificationMeta>,
) {
#[expect(clippy::expect_used)]
let event_json = serde_json::to_value(event).expect("Event must serialize");
let params = if let Ok(params) = serde_json::to_value(OutgoingNotificationParams {
meta,
event: event_json.clone(),
}) {
params
} else {
warn!("Failed to serialize event as OutgoingNotificationParams");
event_json
};
self.send_notification(OutgoingNotification {
method: "codex/event".to_string(),
params: Some(params),
});
}
pub(crate) fn send_notification(&self, notification: OutgoingNotification) {
let outgoing_message = OutgoingMessage::Notification(notification);
let _ = self.sender.send(outgoing_message);
}
pub(crate) fn send_error(&self, id: RequestId, error: ErrorData) {
let outgoing_message = OutgoingMessage::Error(OutgoingError { id, error });
let _ = self.sender.send(outgoing_message);
}
}
/// Outgoing message from the server to the client.
pub(crate) enum OutgoingMessage {
Request(OutgoingRequest),
Notification(OutgoingNotification),
Response(OutgoingResponse),
Error(OutgoingError),
}
impl From<OutgoingMessage> for OutgoingJsonRpcMessage {
fn from(val: OutgoingMessage) -> Self {
use OutgoingMessage::*;
match val {
Request(OutgoingRequest { id, method, params }) => {
JsonRpcMessage::Request(JsonRpcRequest {
jsonrpc: JsonRpcVersion2_0,
id,
request: CustomRequest::new(method, params),
})
}
Notification(OutgoingNotification { method, params }) => {
JsonRpcMessage::Notification(JsonRpcNotification {
jsonrpc: JsonRpcVersion2_0,
notification: CustomNotification::new(method, params),
})
}
Response(OutgoingResponse { id, result }) => {
JsonRpcMessage::Response(JsonRpcResponse {
jsonrpc: JsonRpcVersion2_0,
id,
result,
})
}
Error(OutgoingError { id, error }) => JsonRpcMessage::Error(JsonRpcError {
jsonrpc: JsonRpcVersion2_0,
id: Some(id),
error,
}),
}
}
}
#[derive(Debug, Clone, PartialEq, Serialize)]
pub(crate) struct OutgoingRequest {
pub id: RequestId,
pub method: String,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub params: Option<serde_json::Value>,
}
#[derive(Debug, Clone, PartialEq, Serialize)]
pub(crate) struct OutgoingNotification {
pub method: String,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub params: Option<serde_json::Value>,
}
#[derive(Debug, Clone, PartialEq, Serialize)]
pub(crate) struct OutgoingNotificationParams {
#[serde(rename = "_meta", default, skip_serializing_if = "Option::is_none")]
pub meta: Option<OutgoingNotificationMeta>,
#[serde(flatten)]
pub event: serde_json::Value,
}
// Additional mcp-specific data to be added to a [`codex_protocol::protocol::Event`] as notification.params._meta
// MCP Spec: https://modelcontextprotocol.io/specification/2025-06-18/basic#meta
// Typescript Schema: https://github.com/modelcontextprotocol/modelcontextprotocol/blob/0695a497eb50a804fc0e88c18a93a21a675d6b3e/schema/2025-06-18/schema.ts
#[derive(Debug, Clone, PartialEq, Serialize)]
#[serde(rename_all = "camelCase")]
pub(crate) struct OutgoingNotificationMeta {
pub request_id: Option<RequestId>,
/// Because multiple threads may be multiplexed over a single MCP connection,
/// include the `threadId` in the notification meta.
#[serde(default, skip_serializing_if = "Option::is_none")]
pub thread_id: Option<ThreadId>,
}
#[derive(Debug, Clone, PartialEq, Serialize)]
pub(crate) struct OutgoingResponse {
pub id: RequestId,
pub result: Value,
}
#[derive(Debug, Clone, PartialEq, Serialize)]
pub(crate) struct OutgoingError {
pub error: ErrorData,
pub id: RequestId,
}
#[cfg(test)]
mod tests {
use anyhow::Result;
use codex_protocol::ThreadId;
use codex_protocol::models::PermissionProfile;
use codex_protocol::openai_models::ReasoningEffort;
use codex_protocol::protocol::AskForApproval;
use codex_protocol::protocol::EventMsg;
use codex_protocol::protocol::SessionConfiguredEvent;
use codex_utils_absolute_path::test_support::PathBufExt;
use codex_utils_absolute_path::test_support::test_path_buf;
use pretty_assertions::assert_eq;
use serde_json::json;
use tempfile::NamedTempFile;
use super::*;
#[test]
fn outgoing_request_serializes_as_jsonrpc_request() {
let msg: OutgoingJsonRpcMessage = OutgoingMessage::Request(OutgoingRequest {
id: RequestId::Number(1),
method: "elicitation/create".to_string(),
params: Some(json!({ "k": "v" })),
})
.into();
let value = serde_json::to_value(msg).expect("message should serialize");
let obj = value.as_object().expect("json object");
assert_eq!(obj.get("jsonrpc"), Some(&json!("2.0")));
assert_eq!(obj.get("id"), Some(&json!(1)));
assert_eq!(obj.get("method"), Some(&json!("elicitation/create")));
assert_eq!(obj.get("params"), Some(&json!({ "k": "v" })));
assert!(
obj.get("request").is_none(),
"rmcp request must flatten to JSON-RPC method/params"
);
}
#[test]
fn outgoing_notification_serializes_as_jsonrpc_notification() {
let msg: OutgoingJsonRpcMessage = OutgoingMessage::Notification(OutgoingNotification {
method: "notifications/initialized".to_string(),
params: None,
})
.into();
let value = serde_json::to_value(msg).expect("message should serialize");
let obj = value.as_object().expect("json object");
assert_eq!(obj.get("jsonrpc"), Some(&json!("2.0")));
assert_eq!(obj.get("method"), Some(&json!("notifications/initialized")));
assert_eq!(obj.get("params"), Some(&serde_json::Value::Null));
assert!(
obj.get("notification").is_none(),
"rmcp notification must flatten to JSON-RPC method/params"
);
}
#[tokio::test]
async fn outgoing_tool_response_preserves_legacy_wire_format() {
let (outgoing_tx, mut outgoing_rx) = mpsc::unbounded_channel::<OutgoingMessage>();
let outgoing_message_sender = OutgoingMessageSender::new(outgoing_tx);
outgoing_message_sender.send_response(
RequestId::Number(1),
rmcp::model::CallToolResult::success(Vec::new()),
);
let Some(OutgoingMessage::Response(response)) = outgoing_rx.recv().await else {
panic!("expected a tool-call response");
};
assert_eq!(response.id, RequestId::Number(1));
assert_eq!(response.result, json!({ "content": [], "isError": false }));
}
#[tokio::test]
async fn test_send_event_as_notification() -> Result<()> {
let (outgoing_tx, mut outgoing_rx) = mpsc::unbounded_channel::<OutgoingMessage>();
let outgoing_message_sender = OutgoingMessageSender::new(outgoing_tx);
let thread_id = ThreadId::new();
let rollout_file = NamedTempFile::new()?;
let event = Event {
id: "1".to_string(),
msg: EventMsg::SessionConfigured(SessionConfiguredEvent {
session_id: codex_protocol::SessionId::new(),
thread_id,
forked_from_id: None,
parent_thread_id: None,
thread_source: None,
thread_name: None,
model: "gpt-4o".to_string(),
model_provider_id: "test-provider".to_string(),
service_tier: None,
approval_policy: AskForApproval::Never,
approvals_reviewer: codex_protocol::config_types::ApprovalsReviewer::User,
permission_profile: PermissionProfile::read_only(),
active_permission_profile: None,
cwd: test_path_buf("/home/user/project").abs(),
reasoning_effort: Some(ReasoningEffort::default()),
initial_messages: None,
network_proxy: None,
rollout_path: Some(rollout_file.path().to_path_buf()),
}),
};
outgoing_message_sender.send_event_as_notification(&event, /*meta*/ None);
let result = outgoing_rx.recv().await.unwrap();
let OutgoingMessage::Notification(OutgoingNotification { method, params }) = result else {
panic!("expected Notification for first message");
};
assert_eq!(method, "codex/event");
let Ok(expected_params) = serde_json::to_value(&event) else {
panic!("Event must serialize");
};
assert_eq!(params, Some(expected_params));
Ok(())
}
#[tokio::test]
async fn test_send_event_as_notification_with_meta() -> Result<()> {
let (outgoing_tx, mut outgoing_rx) = mpsc::unbounded_channel::<OutgoingMessage>();
let outgoing_message_sender = OutgoingMessageSender::new(outgoing_tx);
let thread_id = ThreadId::new();
let rollout_file = NamedTempFile::new()?;
let session_configured_event = SessionConfiguredEvent {
session_id: codex_protocol::SessionId::new(),
thread_id,
forked_from_id: None,
parent_thread_id: None,
thread_source: None,
thread_name: None,
model: "gpt-4o".to_string(),
model_provider_id: "test-provider".to_string(),
service_tier: None,
approval_policy: AskForApproval::Never,
approvals_reviewer: codex_protocol::config_types::ApprovalsReviewer::User,
permission_profile: PermissionProfile::read_only(),
active_permission_profile: None,
cwd: test_path_buf("/home/user/project").abs(),
reasoning_effort: Some(ReasoningEffort::default()),
initial_messages: None,
network_proxy: None,
rollout_path: Some(rollout_file.path().to_path_buf()),
};
let event = Event {
id: "1".to_string(),
msg: EventMsg::SessionConfigured(session_configured_event.clone()),
};
let meta = OutgoingNotificationMeta {
request_id: Some(RequestId::String("123".into())),
thread_id: None,
};
outgoing_message_sender.send_event_as_notification(&event, Some(meta));
let result = outgoing_rx.recv().await.unwrap();
let OutgoingMessage::Notification(OutgoingNotification { method, params }) = result else {
panic!("expected Notification for first message");
};
assert_eq!(method, "codex/event");
let expected_params = json!({
"_meta": {
"requestId": "123",
},
"id": "1",
"msg": {
"type": "session_configured",
"session_id": session_configured_event.session_id,
"thread_id": session_configured_event.thread_id,
"model": "gpt-4o",
"model_provider_id": "test-provider",
"approval_policy": "never",
"approvals_reviewer": "user",
"permission_profile": session_configured_event.permission_profile,
"cwd": test_path_buf("/home/user/project"),
"reasoning_effort": session_configured_event.reasoning_effort,
"rollout_path": rollout_file.path().to_path_buf(),
}
});
assert_eq!(params.unwrap(), expected_params);
Ok(())
}
#[tokio::test]
async fn test_send_event_as_notification_with_meta_and_thread_id() -> Result<()> {
let (outgoing_tx, mut outgoing_rx) = mpsc::unbounded_channel::<OutgoingMessage>();
let outgoing_message_sender = OutgoingMessageSender::new(outgoing_tx);
let thread_id = ThreadId::new();
let rollout_file = NamedTempFile::new()?;
let session_configured_event = SessionConfiguredEvent {
session_id: codex_protocol::SessionId::new(),
thread_id,
forked_from_id: None,
parent_thread_id: None,
thread_source: None,
thread_name: None,
model: "gpt-4o".to_string(),
model_provider_id: "test-provider".to_string(),
service_tier: None,
approval_policy: AskForApproval::Never,
approvals_reviewer: codex_protocol::config_types::ApprovalsReviewer::User,
permission_profile: PermissionProfile::read_only(),
active_permission_profile: None,
cwd: test_path_buf("/home/user/project").abs(),
reasoning_effort: Some(ReasoningEffort::default()),
initial_messages: None,
network_proxy: None,
rollout_path: Some(rollout_file.path().to_path_buf()),
};
let event = Event {
id: "1".to_string(),
msg: EventMsg::SessionConfigured(session_configured_event.clone()),
};
let meta = OutgoingNotificationMeta {
request_id: Some(RequestId::String("123".into())),
thread_id: Some(thread_id),
};
outgoing_message_sender.send_event_as_notification(&event, Some(meta));
let result = outgoing_rx.recv().await.unwrap();
let OutgoingMessage::Notification(OutgoingNotification { method, params }) = result else {
panic!("expected Notification for first message");
};
assert_eq!(method, "codex/event");
let expected_params = json!({
"_meta": {
"requestId": "123",
"threadId": thread_id.to_string(),
},
"id": "1",
"msg": {
"type": "session_configured",
"session_id": session_configured_event.session_id,
"thread_id": session_configured_event.thread_id,
"model": "gpt-4o",
"model_provider_id": "test-provider",
"approval_policy": "never",
"approvals_reviewer": "user",
"permission_profile": session_configured_event.permission_profile,
"cwd": test_path_buf("/home/user/project"),
"reasoning_effort": session_configured_event.reasoning_effort,
"rollout_path": rollout_file.path().to_path_buf(),
}
});
assert_eq!(params.unwrap(), expected_params);
Ok(())
}
}
-140
View File
@@ -1,140 +0,0 @@
use std::collections::HashMap;
use std::path::PathBuf;
use std::sync::Arc;
use codex_core::CodexThread;
use codex_protocol::ThreadId;
use codex_protocol::protocol::FileChange;
use codex_protocol::protocol::Op;
use codex_protocol::protocol::ReviewDecision;
use rmcp::model::ErrorData;
use rmcp::model::RequestId;
use serde::Deserialize;
use serde::Serialize;
use serde_json::Value;
use serde_json::json;
use tracing::error;
use crate::outgoing_message::OutgoingMessageSender;
#[derive(Debug, Deserialize, Serialize)]
pub struct PatchApprovalElicitRequestParams {
pub message: String,
#[serde(rename = "requestedSchema")]
pub requested_schema: Value,
#[serde(rename = "threadId")]
pub thread_id: ThreadId,
pub codex_elicitation: String,
pub codex_mcp_tool_call_id: String,
pub codex_event_id: String,
pub codex_call_id: String,
#[serde(skip_serializing_if = "Option::is_none")]
pub codex_reason: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
pub codex_grant_root: Option<PathBuf>,
pub codex_changes: HashMap<PathBuf, FileChange>,
}
#[derive(Debug, Deserialize, Serialize)]
pub struct PatchApprovalResponse {
pub decision: ReviewDecision,
}
#[allow(clippy::too_many_arguments)]
pub(crate) async fn handle_patch_approval_request(
call_id: String,
reason: Option<String>,
grant_root: Option<PathBuf>,
changes: HashMap<PathBuf, FileChange>,
outgoing: Arc<OutgoingMessageSender>,
codex: Arc<CodexThread>,
request_id: RequestId,
tool_call_id: String,
event_id: String,
thread_id: ThreadId,
) {
let approval_id = call_id.clone();
let mut message_lines = Vec::new();
if let Some(r) = &reason {
message_lines.push(r.clone());
}
message_lines.push("Allow Codex to apply proposed code changes?".to_string());
let params = PatchApprovalElicitRequestParams {
message: message_lines.join("\n"),
requested_schema: json!({"type":"object","properties":{}}),
thread_id,
codex_elicitation: "patch-approval".to_string(),
codex_mcp_tool_call_id: tool_call_id.clone(),
codex_event_id: event_id.clone(),
codex_call_id: call_id,
codex_reason: reason,
codex_grant_root: grant_root,
codex_changes: changes,
};
let params_json = match serde_json::to_value(&params) {
Ok(value) => value,
Err(err) => {
let message = format!("Failed to serialize PatchApprovalElicitRequestParams: {err}");
error!("{message}");
outgoing.send_error(request_id.clone(), ErrorData::invalid_params(message, None));
return;
}
};
let on_response = outgoing
.send_request("elicitation/create", Some(params_json))
.await;
// Listen for the response on a separate task so we don't block the main agent loop.
{
let codex = codex.clone();
let approval_id = approval_id.clone();
tokio::spawn(async move {
on_patch_approval_response(approval_id, on_response, codex).await;
});
}
}
pub(crate) async fn on_patch_approval_response(
approval_id: String,
receiver: tokio::sync::oneshot::Receiver<serde_json::Value>,
codex: Arc<CodexThread>,
) {
let response = receiver.await;
let value = match response {
Ok(value) => value,
Err(err) => {
error!("request failed: {err:?}");
if let Err(submit_err) = codex
.submit(Op::PatchApproval {
id: approval_id.clone(),
decision: ReviewDecision::denied("approval request failed"),
})
.await
{
error!("failed to submit denied PatchApproval after request failure: {submit_err}");
}
return;
}
};
let response = serde_json::from_value::<PatchApprovalResponse>(value).unwrap_or_else(|err| {
error!("failed to deserialize PatchApprovalResponse: {err}");
PatchApprovalResponse {
decision: ReviewDecision::denied("approval request failed"),
}
});
if let Err(err) = codex
.submit(Op::PatchApproval {
id: approval_id,
decision: response.decision,
})
.await
{
error!("failed to submit PatchApproval: {err}");
}
}
@@ -1,15 +0,0 @@
use super::reject_workload_identity;
use pretty_assertions::assert_eq;
#[test]
fn workload_identity_markers_are_rejected() {
let error = reject_workload_identity(/*workload_identity_selected*/ true)
.expect_err("mcp-server does not support workload identity");
assert_eq!(
error.to_string(),
"workload identity is not supported by `codex mcp-server`"
);
reject_workload_identity(/*workload_identity_selected*/ false)
.expect("mcp-server remains available without workload identity");
}
-5
View File
@@ -1,5 +0,0 @@
#![allow(clippy::expect_used)]
// Single integration test binary that aggregates all test modules.
// The submodules live in `tests/suite/`.
mod suite;
@@ -1,7 +0,0 @@
load("//:defs.bzl", "codex_rust_crate")
codex_rust_crate(
name = "common",
crate_name = "mcp_test_support",
crate_srcs = glob(["*.rs"]),
)
@@ -1,33 +0,0 @@
[package]
name = "mcp_test_support"
version.workspace = true
edition.workspace = true
license.workspace = true
[lib]
path = "lib.rs"
test = false
doctest = false
[lints]
workspace = true
[dependencies]
anyhow = { workspace = true }
codex-login = { workspace = true }
codex-mcp-server = { workspace = true }
codex-terminal-detection = { workspace = true }
codex-utils-cargo-bin = { workspace = true }
rmcp = { workspace = true }
os_info = { workspace = true }
pretty_assertions = { workspace = true }
serde_json = { workspace = true }
tokio = { workspace = true, features = [
"io-std",
"macros",
"process",
"rt-multi-thread",
] }
wiremock = { workspace = true }
core_test_support = { path = "../../../core/tests/common" }
shlex = { workspace = true }
-14
View File
@@ -1,14 +0,0 @@
#![allow(clippy::expect_used)]
mod mcp_process;
mod mock_model_server;
mod responses;
pub use core_test_support::format_with_current_shell;
pub use core_test_support::format_with_current_shell_display_non_login;
pub use core_test_support::format_with_current_shell_non_login;
pub use mcp_process::McpProcess;
pub use mock_model_server::create_mock_responses_server;
pub use responses::create_apply_patch_sse_response;
pub use responses::create_command_execution_sse_response;
pub use responses::create_final_assistant_message_sse_response;
@@ -1,398 +0,0 @@
use std::path::Path;
use std::process::Stdio;
use std::sync::atomic::AtomicI64;
use std::sync::atomic::Ordering;
use tokio::io::AsyncBufReadExt;
use tokio::io::AsyncWriteExt;
use tokio::io::BufReader;
use tokio::process::Child;
use tokio::process::ChildStdin;
use tokio::process::ChildStdout;
use anyhow::Context;
use codex_mcp_server::CodexToolCallParam;
use codex_terminal_detection::user_agent;
use pretty_assertions::assert_eq;
use rmcp::model::CallToolRequestParams;
use rmcp::model::ClientCapabilities;
use rmcp::model::CustomNotification;
use rmcp::model::CustomRequest;
use rmcp::model::ElicitationCapability;
use rmcp::model::FormElicitationCapability;
use rmcp::model::Implementation;
use rmcp::model::InitializeRequestParams;
use rmcp::model::JsonRpcMessage;
use rmcp::model::JsonRpcNotification;
use rmcp::model::JsonRpcRequest;
use rmcp::model::JsonRpcResponse;
use rmcp::model::JsonRpcVersion2_0;
use rmcp::model::ProtocolVersion;
use rmcp::model::RequestId;
use serde_json::json;
use tokio::process::Command;
pub struct McpProcess {
next_request_id: AtomicI64,
/// Retain this child process until the client is dropped. The Tokio runtime
/// will make a "best effort" to reap the process after it exits, but it is
/// not a guarantee. See the `kill_on_drop` documentation for details.
#[allow(dead_code)]
process: Child,
stdin: ChildStdin,
stdout: BufReader<ChildStdout>,
}
impl McpProcess {
pub async fn new(codex_home: &Path) -> anyhow::Result<Self> {
Self::new_with_env(codex_home, &[]).await
}
/// Creates a new MCP process, allowing tests to override or remove
/// specific environment variables for the child process only.
///
/// Pass a tuple of (key, Some(value)) to set/override, or (key, None) to
/// remove a variable from the child's environment.
pub async fn new_with_env(
codex_home: &Path,
env_overrides: &[(&str, Option<&str>)],
) -> anyhow::Result<Self> {
let program = codex_utils_cargo_bin::cargo_bin("codex-mcp-server")
.context("should find binary for codex-mcp-server")?;
let mut cmd = Command::new(program);
cmd.stdin(Stdio::piped());
cmd.stdout(Stdio::piped());
cmd.stderr(Stdio::piped());
cmd.env("CODEX_HOME", codex_home);
cmd.env("RUST_LOG", "debug");
for (k, v) in env_overrides {
match v {
Some(val) => {
cmd.env(k, val);
}
None => {
cmd.env_remove(k);
}
}
}
let mut process = cmd
.kill_on_drop(true)
.spawn()
.context("codex-mcp-server proc should start")?;
let stdin = process
.stdin
.take()
.ok_or_else(|| anyhow::format_err!("mcp should have stdin fd"))?;
let stdout = process
.stdout
.take()
.ok_or_else(|| anyhow::format_err!("mcp should have stdout fd"))?;
let stdout = BufReader::new(stdout);
// Forward child's stderr to our stderr so failures are visible even
// when stdout/stderr are captured by the test harness.
if let Some(stderr) = process.stderr.take() {
let mut stderr_reader = BufReader::new(stderr).lines();
tokio::spawn(async move {
while let Ok(Some(line)) = stderr_reader.next_line().await {
eprintln!("[mcp stderr] {line}");
}
});
}
Ok(Self {
next_request_id: AtomicI64::new(0),
process,
stdin,
stdout,
})
}
/// Performs the initialization handshake with the MCP server.
pub async fn initialize(&mut self) -> anyhow::Result<()> {
let request_id = self.next_request_id.fetch_add(1, Ordering::Relaxed);
let mut capabilities = ClientCapabilities::default();
capabilities.elicitation =
Some(ElicitationCapability::new().with_form(FormElicitationCapability::new()));
let params = InitializeRequestParams::new(
capabilities,
Implementation::new("elicitation test", "0.0.0").with_title("Elicitation Test"),
)
.with_protocol_version(ProtocolVersion::V_2025_03_26);
let params_value = serde_json::to_value(params)?;
self.send_jsonrpc_message(JsonRpcMessage::Request(JsonRpcRequest {
jsonrpc: JsonRpcVersion2_0,
id: RequestId::Number(request_id),
request: CustomRequest::new("initialize", Some(params_value)),
}))
.await?;
let initialized = self.read_jsonrpc_message().await?;
let os_info = os_info::get();
let build_version = env!("CARGO_PKG_VERSION");
let originator = codex_login::default_client::originator().value;
let user_agent = format!(
"{originator}/{build_version} ({} {}; {}) {} (elicitation test; 0.0.0)",
os_info.os_type(),
os_info.version(),
os_info.architecture().unwrap_or("unknown"),
user_agent()
);
let JsonRpcMessage::Response(JsonRpcResponse {
jsonrpc,
id,
result,
}) = initialized
else {
anyhow::bail!("expected initialize response message, got: {initialized:?}")
};
assert_eq!(jsonrpc, JsonRpcVersion2_0);
assert_eq!(id, RequestId::Number(request_id));
assert_eq!(
result,
json!({
"capabilities": {
"tools": {
"listChanged": true
},
},
"serverInfo": {
"name": "codex-mcp-server",
"title": "Codex",
"version": "0.0.0",
"user_agent": user_agent
},
"protocolVersion": ProtocolVersion::V_2025_03_26
})
);
// Send notifications/initialized to ack the response.
self.send_jsonrpc_message(JsonRpcMessage::Notification(JsonRpcNotification {
jsonrpc: JsonRpcVersion2_0,
notification: CustomNotification::new("notifications/initialized", None),
}))
.await?;
Ok(())
}
/// Returns the id used to make the request so it can be used when
/// correlating notifications.
pub async fn send_codex_tool_call(
&mut self,
params: CodexToolCallParam,
) -> anyhow::Result<i64> {
let codex_tool_call_params = CallToolRequestParams::new("codex").with_arguments(
match serde_json::to_value(params)? {
serde_json::Value::Object(map) => map,
_ => unreachable!("params serialize to object"),
},
);
self.send_request(
"tools/call",
Some(serde_json::to_value(codex_tool_call_params)?),
)
.await
}
async fn send_request(
&mut self,
method: &str,
params: Option<serde_json::Value>,
) -> anyhow::Result<i64> {
let request_id = self.next_request_id.fetch_add(1, Ordering::Relaxed);
let message = JsonRpcMessage::Request(JsonRpcRequest {
jsonrpc: JsonRpcVersion2_0,
id: RequestId::Number(request_id),
request: CustomRequest::new(method, params),
});
self.send_jsonrpc_message(message).await?;
Ok(request_id)
}
pub async fn send_response(
&mut self,
id: RequestId,
result: serde_json::Value,
) -> anyhow::Result<()> {
self.send_jsonrpc_message(JsonRpcMessage::Response(JsonRpcResponse {
jsonrpc: JsonRpcVersion2_0,
id,
result,
}))
.await
}
async fn send_jsonrpc_message(
&mut self,
message: JsonRpcMessage<CustomRequest, serde_json::Value, CustomNotification>,
) -> anyhow::Result<()> {
eprintln!("writing message to stdin: {message:?}");
let payload = serde_json::to_string(&message)?;
self.stdin.write_all(payload.as_bytes()).await?;
self.stdin.write_all(b"\n").await?;
self.stdin.flush().await?;
Ok(())
}
async fn read_jsonrpc_message(
&mut self,
) -> anyhow::Result<JsonRpcMessage<CustomRequest, serde_json::Value, CustomNotification>> {
let mut line = String::new();
self.stdout.read_line(&mut line).await?;
let message = serde_json::from_str::<
JsonRpcMessage<CustomRequest, serde_json::Value, CustomNotification>,
>(&line)?;
eprintln!("read message from stdout: {message:?}");
Ok(message)
}
pub async fn read_stream_until_request_message(
&mut self,
) -> anyhow::Result<JsonRpcRequest<CustomRequest>> {
eprintln!("in read_stream_until_request_message()");
loop {
let message = self.read_jsonrpc_message().await?;
match message {
JsonRpcMessage::Notification(_) => {
eprintln!("notification: {message:?}");
}
JsonRpcMessage::Request(jsonrpc_request) => {
return Ok(jsonrpc_request);
}
JsonRpcMessage::Error(_) => {
anyhow::bail!("unexpected JSONRPCMessage::Error: {message:?}");
}
JsonRpcMessage::Response(_) => {
anyhow::bail!("unexpected JSONRPCMessage::Response: {message:?}");
}
}
}
}
pub async fn read_stream_until_response_message(
&mut self,
request_id: RequestId,
) -> anyhow::Result<JsonRpcResponse<serde_json::Value>> {
eprintln!("in read_stream_until_response_message({request_id:?})");
loop {
let message = self.read_jsonrpc_message().await?;
match message {
JsonRpcMessage::Notification(_) => {
eprintln!("notification: {message:?}");
}
JsonRpcMessage::Request(_) => {
anyhow::bail!("unexpected JSONRPCMessage::Request: {message:?}");
}
JsonRpcMessage::Error(_) => {
anyhow::bail!("unexpected JSONRPCMessage::Error: {message:?}");
}
JsonRpcMessage::Response(jsonrpc_response) => {
if jsonrpc_response.id == request_id {
return Ok(jsonrpc_response);
}
}
}
}
}
/// Reads notifications until a legacy TurnComplete event is observed:
/// Method "codex/event" with params.msg.type == "task_complete".
pub async fn read_stream_until_legacy_task_complete_notification(
&mut self,
) -> anyhow::Result<JsonRpcNotification<CustomNotification>> {
eprintln!("in read_stream_until_legacy_task_complete_notification()");
self.read_stream_until_codex_event("task_complete").await
}
pub async fn read_stream_until_codex_event(
&mut self,
event_type: &str,
) -> anyhow::Result<JsonRpcNotification<CustomNotification>> {
self.read_stream_until_codex_event_matching(event_type, |_| true)
.await
}
pub async fn read_stream_until_codex_event_matching(
&mut self,
event_type: &str,
predicate: impl Fn(&serde_json::Value) -> bool,
) -> anyhow::Result<JsonRpcNotification<CustomNotification>> {
eprintln!("in read_stream_until_codex_event({event_type})");
loop {
let message = self.read_jsonrpc_message().await?;
match message {
JsonRpcMessage::Notification(notification) => {
let is_match = notification.notification.method == "codex/event"
&& notification
.notification
.params
.as_ref()
.is_some_and(|params| {
params
.get("msg")
.and_then(|m| m.get("type"))
.and_then(|t| t.as_str())
== Some(event_type)
&& predicate(params)
});
if is_match {
return Ok(notification);
} else {
eprintln!("ignoring notification: {notification:?}");
}
}
JsonRpcMessage::Request(_) => {
anyhow::bail!("unexpected JSONRPCMessage::Request: {message:?}");
}
JsonRpcMessage::Error(_) => {
anyhow::bail!("unexpected JSONRPCMessage::Error: {message:?}");
}
JsonRpcMessage::Response(_) => {
anyhow::bail!("unexpected JSONRPCMessage::Response: {message:?}");
}
}
}
}
}
impl Drop for McpProcess {
fn drop(&mut self) {
// These tests spawn a `codex-mcp-server` child process.
//
// We keep that child alive for the test and rely on Tokio's `kill_on_drop(true)` when this
// helper is dropped. Tokio documents kill-on-drop as best-effort: dropping requests
// termination, but it does not guarantee the child has fully exited and been reaped before
// teardown continues.
//
// That makes cleanup timing nondeterministic. Leak detection can occasionally observe the
// child still alive at teardown and report `LEAK`, which makes the test flaky.
//
// Drop can't be async, so we do a bounded synchronous cleanup:
//
// 1. Request termination with `start_kill()`.
// 2. Poll `try_wait()` until the OS reports the child exited, with a short timeout.
let _ = self.process.start_kill();
let start = std::time::Instant::now();
let timeout = std::time::Duration::from_secs(5);
while start.elapsed() < timeout {
match self.process.try_wait() {
Ok(Some(_)) => return,
Ok(None) => std::thread::sleep(std::time::Duration::from_millis(10)),
Err(_) => return,
}
}
}
}
@@ -1,48 +0,0 @@
use std::sync::atomic::AtomicUsize;
use std::sync::atomic::Ordering;
use wiremock::Mock;
use wiremock::MockServer;
use wiremock::Respond;
use wiremock::ResponseTemplate;
use wiremock::matchers::method;
use wiremock::matchers::path;
/// Create a mock server that will provide the responses, in order, for
/// requests to the `/v1/responses` endpoint.
pub async fn create_mock_responses_server(responses: Vec<String>) -> MockServer {
let server = MockServer::start().await;
let num_calls = responses.len();
let seq_responder = SeqResponder {
num_calls: AtomicUsize::new(0),
responses,
};
Mock::given(method("POST"))
.and(path("/v1/responses"))
.respond_with(seq_responder)
.expect(num_calls as u64)
.mount(&server)
.await;
server
}
struct SeqResponder {
num_calls: AtomicUsize,
responses: Vec<String>,
}
impl Respond for SeqResponder {
fn respond(&self, _: &wiremock::Request) -> ResponseTemplate {
let call_num = self.num_calls.fetch_add(1, Ordering::SeqCst);
let response = self
.responses
.get(call_num)
.expect("mock model response should exist");
ResponseTemplate::new(200)
.insert_header("content-type", "text/event-stream")
.set_body_raw(response.clone(), "text/event-stream")
}
}
@@ -1,52 +0,0 @@
use std::path::Path;
use core_test_support::responses;
use serde_json::json;
pub fn create_command_execution_sse_response(
command: Vec<String>,
workdir: Option<&Path>,
timeout_ms: Option<u64>,
call_id: &str,
) -> anyhow::Result<String> {
let command_str = shlex::try_join(command.iter().map(String::as_str))?;
let mut tool_call_arguments = json!({
"cmd": command_str,
"workdir": workdir.map(|w| w.to_string_lossy()),
"sandbox_permissions": "require_escalated",
"justification": "Test approval request.",
});
if let Some(timeout_ms) = timeout_ms {
tool_call_arguments["yield_time_ms"] = json!(timeout_ms);
}
let arguments = serde_json::to_string(&tool_call_arguments)?;
let response_id = format!("resp-{call_id}");
Ok(responses::sse(vec![
responses::ev_response_created(&response_id),
responses::ev_function_call(call_id, "exec_command", &arguments),
responses::ev_completed(&response_id),
]))
}
pub fn create_final_assistant_message_sse_response(message: &str) -> anyhow::Result<String> {
let response_id = "resp-final";
Ok(responses::sse(vec![
responses::ev_response_created(response_id),
responses::ev_assistant_message("msg-final", message),
responses::ev_completed(response_id),
]))
}
pub fn create_apply_patch_sse_response(
patch_content: &str,
call_id: &str,
) -> anyhow::Result<String> {
let command = format!("apply_patch <<'EOF'\n{patch_content}\nEOF");
let arguments = serde_json::to_string(&json!({ "cmd": command }))?;
let response_id = format!("resp-{call_id}");
Ok(responses::sse(vec![
responses::ev_response_created(&response_id),
responses::ev_function_call(call_id, "exec_command", &arguments),
responses::ev_completed(&response_id),
]))
}
@@ -1,679 +0,0 @@
use std::collections::HashMap;
use std::env;
use std::path::Path;
use std::path::PathBuf;
use app_test_support::ChatGptAuthFixture;
use app_test_support::write_chatgpt_auth;
use codex_config::types::AuthCredentialsStoreMode;
use codex_core::spawn::CODEX_SANDBOX_NETWORK_DISABLED_ENV_VAR;
use codex_mcp_server::CodexToolCallParam;
use codex_mcp_server::ExecApprovalElicitRequestParams;
use codex_mcp_server::ExecApprovalResponse;
use codex_mcp_server::PatchApprovalElicitRequestParams;
use codex_mcp_server::PatchApprovalResponse;
use codex_protocol::protocol::FileChange;
use codex_protocol::protocol::ReviewDecision;
use codex_shell_command::parse_command;
use pretty_assertions::assert_eq;
use rmcp::model::JsonRpcResponse;
use rmcp::model::JsonRpcVersion2_0;
use rmcp::model::RequestId;
use serde_json::json;
use tempfile::TempDir;
use tokio::time::timeout;
use wiremock::Mock;
use wiremock::MockServer;
use wiremock::ResponseTemplate;
use wiremock::matchers::method;
use wiremock::matchers::path;
use core_test_support::skip_if_no_network;
use mcp_test_support::McpProcess;
use mcp_test_support::create_apply_patch_sse_response;
use mcp_test_support::create_command_execution_sse_response;
use mcp_test_support::create_final_assistant_message_sse_response;
use mcp_test_support::create_mock_responses_server;
use mcp_test_support::format_with_current_shell;
// Windows CI can spend tens of seconds in session startup before the first
// mock model request is sent.
const DEFAULT_READ_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(60);
/// Test that an explicitly escalated exec command triggers MCP elicitation and
/// that approving the request runs the command.
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn test_exec_command_approval_triggers_elicitation() {
if env::var(CODEX_SANDBOX_NETWORK_DISABLED_ENV_VAR).is_ok() {
println!(
"Skipping test because it cannot execute when network is disabled in a Codex sandbox."
);
return;
}
// Apparently `#[tokio::test]` must return `()`, so we create a helper
// function that returns `Result` so we can use `?` in favor of `unwrap`.
exec_command_approval_triggers_elicitation()
.await
.expect("exec command approval should trigger elicitation");
}
async fn exec_command_approval_triggers_elicitation() -> anyhow::Result<()> {
// Use a command that creates a file so we can observe its side effect.
let workdir_for_shell_function_call = TempDir::new()?;
let created_filename = "created_by_shell_tool.txt";
let created_file = workdir_for_shell_function_call
.path()
.join(created_filename);
let (shell_command, timeout_ms) = if cfg!(windows) {
(
vec![
"New-Item".to_string(),
"-ItemType".to_string(),
"File".to_string(),
"-Path".to_string(),
created_filename.to_string(),
"-Force".to_string(),
],
// `powershell.exe` startup can be slow on loaded Windows CI workers
10_000,
)
} else {
(
vec!["touch".to_string(), created_filename.to_string()],
5_000,
)
};
let expected_shell_command =
format_with_current_shell(&shlex::try_join(shell_command.iter().map(String::as_str))?);
let McpHandle {
process: mut mcp_process,
server: _server,
dir: _dir,
} = create_mcp_process(vec![
create_command_execution_sse_response(
shell_command.clone(),
Some(workdir_for_shell_function_call.path()),
Some(timeout_ms),
"call1234",
)?,
create_final_assistant_message_sse_response("File created!")?,
])
.await?;
// Send a "codex" tool request, which should hit the responses endpoint.
// In turn, it should reply with a tool call, which the MCP should forward
// as an elicitation.
let codex_request_id = mcp_process
.send_codex_tool_call(CodexToolCallParam {
prompt: "run `git init`".to_string(),
// Exercise MCP elicitation even when the surrounding environment enables auto-review.
config: Some(HashMap::from([
("approvals_reviewer".to_string(), json!("user")),
("features.guardian_approval".to_string(), json!(false)),
])),
..Default::default()
})
.await?;
let elicitation_request = timeout(
DEFAULT_READ_TIMEOUT,
mcp_process.read_stream_until_request_message(),
)
.await??;
assert_eq!(elicitation_request.jsonrpc, JsonRpcVersion2_0);
assert_eq!(elicitation_request.request.method, "elicitation/create");
let elicitation_request_id = elicitation_request.id.clone();
let params = serde_json::from_value::<ExecApprovalElicitRequestParams>(
elicitation_request
.request
.params
.clone()
.ok_or_else(|| anyhow::anyhow!("elicitation_request.params must be set"))?,
)?;
assert_eq!(
elicitation_request.request.params,
Some(create_expected_elicitation_request_params(
expected_shell_command,
workdir_for_shell_function_call.path(),
codex_request_id.to_string(),
params.codex_event_id.clone(),
params.thread_id,
)?)
);
// Accept the `git init` request by responding to the elicitation.
mcp_process
.send_response(
elicitation_request_id,
serde_json::to_value(ExecApprovalResponse {
decision: ReviewDecision::Approved,
})?,
)
.await?;
// Verify task_complete notification arrives before the tool call completes.
let _task_complete = timeout(
DEFAULT_READ_TIMEOUT,
mcp_process.read_stream_until_legacy_task_complete_notification(),
)
.await
.expect("task_complete_notification timeout")
.expect("task_complete_notification resp");
// Verify the original `codex` tool call completes and that the file was created.
let codex_response = timeout(
DEFAULT_READ_TIMEOUT,
mcp_process.read_stream_until_response_message(RequestId::Number(codex_request_id)),
)
.await??;
assert_eq!(
JsonRpcResponse {
jsonrpc: JsonRpcVersion2_0,
id: RequestId::Number(codex_request_id),
result: json!({
"content": [
{
"text": "File created!",
"type": "text"
}
],
"structuredContent": {
"threadId": params.thread_id,
"content": "File created!"
}
}),
},
codex_response
);
assert!(created_file.is_file(), "created file should exist");
Ok(())
}
fn create_expected_elicitation_request_params(
command: Vec<String>,
workdir: &Path,
codex_mcp_tool_call_id: String,
codex_event_id: String,
thread_id: codex_protocol::ThreadId,
) -> anyhow::Result<serde_json::Value> {
let expected_message = format!(
"Allow Codex to run `{}` in `{}`?",
shlex::try_join(command.iter().map(std::convert::AsRef::as_ref))?,
workdir.to_string_lossy()
);
let codex_parsed_cmd = parse_command::parse_command(&command);
let params_json = serde_json::to_value(ExecApprovalElicitRequestParams {
message: expected_message,
requested_schema: json!({"type":"object","properties":{}}),
thread_id,
codex_elicitation: "exec-approval".to_string(),
codex_mcp_tool_call_id,
codex_event_id,
codex_command: command,
codex_cwd: workdir.to_path_buf(),
codex_call_id: "call1234".to_string(),
codex_parsed_cmd,
})?;
Ok(params_json)
}
/// Test that patch approval triggers an elicitation request to the MCP and that
/// sending a denial leaves the patch unapplied, as expected.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn test_patch_approval_triggers_elicitation() {
if env::var(CODEX_SANDBOX_NETWORK_DISABLED_ENV_VAR).is_ok() {
println!(
"Skipping test because it cannot execute when network is disabled in a Codex sandbox."
);
return;
}
patch_approval_triggers_elicitation()
.await
.expect("patch approval should trigger elicitation");
}
async fn patch_approval_triggers_elicitation() -> anyhow::Result<()> {
if cfg!(windows) {
// powershell apply_patch shell calls are not parsed into apply patch approvals
return Ok(());
}
let cwd = TempDir::new()?;
let test_file = Path::new("/").join(
cwd.path()
.file_name()
.ok_or_else(|| anyhow::anyhow!("temporary directory must have a file name"))?,
);
let patch_content = format!(
"*** Begin Patch\n*** Add File: {}\n+new content\n*** End Patch",
test_file.as_path().to_string_lossy()
);
let McpHandle {
process: mut mcp_process,
server: _server,
dir: _dir,
} = create_mcp_process(vec![
create_apply_patch_sse_response(&patch_content, "call1234")?,
create_final_assistant_message_sse_response("Patch was not applied.")?,
])
.await?;
// Send a "codex" tool request that will trigger the apply_patch command
let codex_request_id = mcp_process
.send_codex_tool_call(CodexToolCallParam {
cwd: Some(cwd.path().to_string_lossy().to_string()),
prompt: "please modify the test file".to_string(),
// Use the user reviewer so this test exercises MCP elicitation even when auto-review
// is enabled by the surrounding environment.
config: Some(HashMap::from([
("approvals_reviewer".to_string(), json!("user")),
("features.guardian_approval".to_string(), json!(false)),
])),
..Default::default()
})
.await?;
let elicitation_request = timeout(
DEFAULT_READ_TIMEOUT,
mcp_process.read_stream_until_request_message(),
)
.await??;
assert_eq!(elicitation_request.jsonrpc, JsonRpcVersion2_0);
assert_eq!(elicitation_request.request.method, "elicitation/create");
let elicitation_request_id = elicitation_request.id.clone();
let params = serde_json::from_value::<PatchApprovalElicitRequestParams>(
elicitation_request
.request
.params
.clone()
.ok_or_else(|| anyhow::anyhow!("elicitation_request.params must be set"))?,
)?;
let mut expected_changes = HashMap::new();
expected_changes.insert(
test_file.as_path().to_path_buf(),
FileChange::Add {
content: "new content\n".to_string(),
},
);
assert_eq!(
elicitation_request.request.params,
Some(create_expected_patch_approval_elicitation_request_params(
expected_changes,
/*grant_root*/ None, // No grant_root expected
/*reason*/ None,
codex_request_id.to_string(),
params.codex_event_id.clone(),
params.thread_id,
)?)
);
// Deny the patch approval request so this test does not require the Linux sandbox helper.
mcp_process
.send_response(
elicitation_request_id,
serde_json::to_value(PatchApprovalResponse {
decision: ReviewDecision::denied("rejected by test"),
})?,
)
.await?;
// Verify the original `codex` tool call completes
let codex_response = timeout(
DEFAULT_READ_TIMEOUT,
mcp_process.read_stream_until_response_message(RequestId::Number(codex_request_id)),
)
.await??;
assert_eq!(
JsonRpcResponse {
jsonrpc: JsonRpcVersion2_0,
id: RequestId::Number(codex_request_id),
result: json!({
"content": [
{
"text": "Patch was not applied.",
"type": "text"
}
],
"structuredContent": {
"threadId": params.thread_id,
"content": "Patch was not applied."
}
}),
},
codex_response
);
assert!(!test_file.exists());
Ok(())
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn test_codex_tool_passes_base_instructions() {
skip_if_no_network!();
// Apparently `#[tokio::test]` must return `()`, so we create a helper
// function that returns `Result` so we can use `?` in favor of `unwrap`.
codex_tool_passes_base_instructions()
.await
.expect("codex tool should pass base instructions");
}
async fn codex_tool_passes_base_instructions() -> anyhow::Result<()> {
#![expect(clippy::unwrap_used)]
let server =
create_mock_responses_server(vec![create_final_assistant_message_sse_response("Enjoy!")?])
.await;
let caller_server = MockServer::start().await;
// Run `codex mcp` with a specific config.toml.
let codex_home = TempDir::new()?;
create_config_toml(codex_home.path(), &server.uri())?;
let skill_dir = codex_home.path().join("skills").join("demo");
std::fs::create_dir_all(&skill_dir)?;
std::fs::write(
skill_dir.join("SKILL.md"),
"---\nname: demo\ndescription: Demo skill.\n---\n# Demo\n\nUse this skill.\n",
)?;
write_chatgpt_auth(
codex_home.path(),
ChatGptAuthFixture::new("chatgpt-token").account_id("workspace-123"),
AuthCredentialsStoreMode::File,
)?;
Mock::given(method("GET"))
.and(path("/backend-api/wham/settings/user"))
.respond_with(ResponseTemplate::new(200).set_body_json(json!({
"commit_attribution_enabled": true,
})))
.expect(1)
.mount(&server)
.await;
let mut mcp_process = McpProcess::new_with_env(
codex_home.path(),
&[("OPENAI_API_KEY", None), ("CODEX_ACCESS_TOKEN", None)],
)
.await?;
timeout(DEFAULT_READ_TIMEOUT, mcp_process.initialize()).await??;
// Send a "codex" tool request, which should hit the responses endpoint.
let codex_request_id = mcp_process
.send_codex_tool_call(CodexToolCallParam {
prompt: "How are you?".to_string(),
config: Some(HashMap::from([(
"chatgpt_base_url".to_string(),
json!(format!("{}/backend-api", caller_server.uri())),
)])),
base_instructions: Some("You are a helpful assistant.".to_string()),
developer_instructions: Some("Foreshadow upcoming tool calls.".to_string()),
..Default::default()
})
.await?;
let codex_response = timeout(
DEFAULT_READ_TIMEOUT,
mcp_process.read_stream_until_response_message(RequestId::Number(codex_request_id)),
)
.await??;
assert_eq!(codex_response.jsonrpc, JsonRpcVersion2_0);
assert_eq!(codex_response.id, RequestId::Number(codex_request_id));
assert_eq!(
codex_response.result,
json!({
"content": [
{
"text": "Enjoy!",
"type": "text"
}
],
"structuredContent": {
"threadId": codex_response
.result
.get("structuredContent")
.and_then(|v| v.get("threadId"))
.and_then(serde_json::Value::as_str)
.expect("codex tool response should include structuredContent.threadId"),
"content": "Enjoy!"
}
})
);
let requests = server.received_requests().await.unwrap();
let request = requests
.iter()
.find(|request| request.url.path() == "/v1/responses")
.expect("mock model request should be recorded")
.body_json::<serde_json::Value>()?;
let instructions = request["instructions"]
.as_str()
.expect("responses request should include instructions");
assert!(instructions.starts_with("You are a helpful assistant."));
let developer_messages: Vec<&serde_json::Value> = request["input"]
.as_array()
.expect("responses request should include input items")
.iter()
.filter(|msg| msg.get("role").and_then(|role| role.as_str()) == Some("developer"))
.collect();
let developer_contents: Vec<&str> = developer_messages
.iter()
.filter_map(|msg| msg.get("content").and_then(serde_json::Value::as_array))
.flat_map(|content| content.iter())
.filter(|span| span.get("type").and_then(serde_json::Value::as_str) == Some("input_text"))
.filter_map(|span| span.get("text").and_then(serde_json::Value::as_str))
.collect();
let developer_text = developer_contents.join("\n");
assert_eq!(
developer_text
.matches("Co-authored-by: Codex <noreply@openai.com>")
.count(),
1
);
assert_eq!(
developer_text
.matches("Generated with [Codex](https://openai.com/codex/).")
.count(),
1
);
assert_eq!(
developer_text.matches("- demo: Demo skill.").count(),
1,
"host skill catalog should be included exactly once"
);
assert!(
developer_contents
.iter()
.any(|content| content.contains("`sandbox_mode`")),
"expected permissions developer message, got {developer_contents:?}"
);
assert!(
developer_contents.contains(&"Foreshadow upcoming tool calls."),
"expected developer instructions in developer messages, got {developer_contents:?}"
);
let caller_requests = caller_server.received_requests().await.unwrap();
assert!(
caller_requests
.iter()
.all(|request| request.url.path() != "/backend-api/wham/settings/user"),
"attribution settings must use the process-level base URL"
);
Ok(())
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn test_codex_tool_forwards_skills_extension_warnings() {
skip_if_no_network!();
codex_tool_forwards_skills_extension_warnings()
.await
.expect("codex tool should forward skills extension warnings");
}
async fn codex_tool_forwards_skills_extension_warnings() -> anyhow::Result<()> {
let server =
create_mock_responses_server(vec![create_final_assistant_message_sse_response("Enjoy!")?])
.await;
let codex_home = TempDir::new()?;
create_config_toml(codex_home.path(), &server.uri())?;
let skills_dir = codex_home.path().join("skills");
for index in 0..200 {
let name = format!("skill-{index:03}");
let skill_dir = skills_dir.join(&name);
std::fs::create_dir_all(&skill_dir)?;
let description = format!("Skill {index}: {}", "x".repeat(200));
std::fs::write(
skill_dir.join("SKILL.md"),
format!(
"---\nname: {name}\ndescription: {description}\n---\n# {name}\n\nUse this skill.\n"
),
)?;
}
let mut mcp_process = McpProcess::new(codex_home.path()).await?;
timeout(DEFAULT_READ_TIMEOUT, mcp_process.initialize()).await??;
let codex_request_id = mcp_process
.send_codex_tool_call(CodexToolCallParam {
prompt: "How are you?".to_string(),
..Default::default()
})
.await?;
let warning = timeout(
DEFAULT_READ_TIMEOUT,
mcp_process.read_stream_until_codex_event_matching("warning", |params| {
params["msg"]["message"]
.as_str()
.is_some_and(|message| message.contains("skills context budget"))
}),
)
.await??;
let warning_json = serde_json::to_value(&warning)?;
let params = warning
.notification
.params
.ok_or_else(|| anyhow::anyhow!("warning notification should include params"))?;
assert_eq!(
warning_json["params"]["_meta"]["requestId"],
codex_request_id
);
assert!(
warning_json["params"]["id"]
.as_str()
.is_some_and(|turn_id| !turn_id.is_empty())
);
assert!(
warning_json["params"]["_meta"]["threadId"]
.as_str()
.is_some_and(|thread_id| !thread_id.is_empty())
);
assert_eq!(params["msg"]["type"], "warning");
assert!(
params["msg"]["message"]
.as_str()
.is_some_and(|message| message.contains("skills context budget"))
);
timeout(
DEFAULT_READ_TIMEOUT,
mcp_process.read_stream_until_response_message(RequestId::Number(codex_request_id)),
)
.await??;
Ok(())
}
fn create_expected_patch_approval_elicitation_request_params(
changes: HashMap<PathBuf, FileChange>,
grant_root: Option<PathBuf>,
reason: Option<String>,
codex_mcp_tool_call_id: String,
codex_event_id: String,
thread_id: codex_protocol::ThreadId,
) -> anyhow::Result<serde_json::Value> {
let mut message_lines = Vec::new();
if let Some(r) = &reason {
message_lines.push(r.clone());
}
message_lines.push("Allow Codex to apply proposed code changes?".to_string());
let params_json = serde_json::to_value(PatchApprovalElicitRequestParams {
message: message_lines.join("\n"),
requested_schema: json!({"type":"object","properties":{}}),
thread_id,
codex_elicitation: "patch-approval".to_string(),
codex_mcp_tool_call_id,
codex_event_id,
codex_reason: reason,
codex_grant_root: grant_root,
codex_changes: changes,
codex_call_id: "call1234".to_string(),
})?;
Ok(params_json)
}
/// This handle is used to ensure that the MockServer and TempDir are not dropped while
/// the McpProcess is still running.
pub struct McpHandle {
pub process: McpProcess,
/// Retain the server for the lifetime of the McpProcess.
#[allow(dead_code)]
server: MockServer,
/// Retain the temporary directory for the lifetime of the McpProcess.
#[allow(dead_code)]
dir: TempDir,
}
async fn create_mcp_process(responses: Vec<String>) -> anyhow::Result<McpHandle> {
let server = create_mock_responses_server(responses).await;
let codex_home = TempDir::new()?;
create_config_toml(codex_home.path(), &server.uri())?;
let mut mcp_process = McpProcess::new(codex_home.path()).await?;
timeout(DEFAULT_READ_TIMEOUT, mcp_process.initialize()).await??;
Ok(McpHandle {
process: mcp_process,
server,
dir: codex_home,
})
}
/// Create a Codex config that uses the mock server as the model provider.
/// The command explicitly requests escalation so that we exercise the
/// elicitation code path.
fn create_config_toml(codex_home: &Path, server_uri: &str) -> std::io::Result<()> {
let config_toml = codex_home.join("config.toml");
std::fs::write(
config_toml,
format!(
r#"
model = "mock-model"
approval_policy = "on-request"
sandbox_policy = "workspace-write"
model_provider = "mock_provider"
chatgpt_base_url = "{server_uri}/backend-api"
cli_auth_credentials_store = "file"
[model_providers.mock_provider]
name = "Mock provider for test"
base_url = "{server_uri}/v1"
wire_api = "responses"
request_max_retries = 0
stream_max_retries = 0
[features]
"#
),
)
}
-1
View File
@@ -1 +0,0 @@
mod codex_tool;
-4
View File
@@ -169,10 +169,6 @@ bazel-argument-comment-lint:
build-for-release:
bazel build //codex-rs/cli:release_binaries
# Run the MCP server
mcp-server-run *args:
cargo run -p codex-mcp-server -- {args}
# Regenerate the json schema for config.toml from the current config types.
write-config-schema:
cargo run -p codex-config-schema --bin codex-write-config-schema
+1 -5
View File
@@ -278,11 +278,7 @@ fn is_meaningful_parameter_name(name: &str) -> bool {
}
fn is_workspace_crate_name(name: &str) -> bool {
name.starts_with("codex_")
|| matches!(
name,
"app_test_support" | "core_test_support" | "mcp_test_support"
)
name.starts_with("codex_") || matches!(name, "app_test_support" | "core_test_support")
}
#[test]