diff --git a/codex-rs/Cargo.lock b/codex-rs/Cargo.lock index edd0b9a78b..31955db0c9 100644 --- a/codex-rs/Cargo.lock +++ b/codex-rs/Cargo.lock @@ -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" diff --git a/codex-rs/Cargo.toml b/codex-rs/Cargo.toml index 5cdadb1477..3c9156f088 100644 --- a/codex-rs/Cargo.toml +++ b/codex-rs/Cargo.toml @@ -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" } diff --git a/codex-rs/app-server/tests/common/test_app_server.rs b/codex-rs/app-server/tests/common/test_app_server.rs index f5538fac4b..048f2eeb8e 100644 --- a/codex-rs/app-server/tests/common/test_app_server.rs +++ b/codex-rs/app-server/tests/common/test_app_server.rs @@ -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; diff --git a/codex-rs/cli/Cargo.toml b/codex-rs/cli/Cargo.toml index f7f66b19d4..67e54b7dba 100644 --- a/codex-rs/cli/Cargo.toml +++ b/codex-rs/cli/Cargo.toml @@ -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 } diff --git a/codex-rs/cli/src/main.rs b/codex-rs/cli/src/main.rs index 3d1c846cd2..abcfb720ab 100644 --- a/codex-rs/cli/src/main.rs +++ b/codex-rs/cli/src/main.rs @@ -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"); diff --git a/codex-rs/cli/tests/features.rs b/codex-rs/cli/tests/features.rs index 3aad6fdc2b..83cf9ce325 100644 --- a/codex-rs/cli/tests/features.rs +++ b/codex-rs/cli/tests/features.rs @@ -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")); diff --git a/codex-rs/docs/codex_mcp_interface.md b/codex-rs/docs/codex_mcp_interface.md deleted file mode 100644 index e3eb814f36..0000000000 --- a/codex-rs/docs/codex_mcp_interface.md +++ /dev/null @@ -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/`. diff --git a/codex-rs/mcp-server/BUILD.bazel b/codex-rs/mcp-server/BUILD.bazel deleted file mode 100644 index 5bf39611b9..0000000000 --- a/codex-rs/mcp-server/BUILD.bazel +++ /dev/null @@ -1,6 +0,0 @@ -load("//:defs.bzl", "codex_rust_crate") - -codex_rust_crate( - name = "mcp-server", - crate_name = "codex_mcp_server", -) diff --git a/codex-rs/mcp-server/Cargo.toml b/codex-rs/mcp-server/Cargo.toml deleted file mode 100644 index 69fb1e8ce6..0000000000 --- a/codex-rs/mcp-server/Cargo.toml +++ /dev/null @@ -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 } diff --git a/codex-rs/mcp-server/src/active_turn_registry.rs b/codex-rs/mcp-server/src/active_turn_registry.rs deleted file mode 100644 index 74652810b4..0000000000 --- a/codex-rs/mcp-server/src/active_turn_registry.rs +++ /dev/null @@ -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, -} - -#[derive(Default)] -struct ActiveTurnState { - by_request: HashMap, - 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 { - 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; diff --git a/codex-rs/mcp-server/src/active_turn_registry_tests.rs b/codex-rs/mcp-server/src/active_turn_registry_tests.rs deleted file mode 100644 index 55c7ffe3df..0000000000 --- a/codex-rs/mcp-server/src/active_turn_registry_tests.rs +++ /dev/null @@ -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)); -} diff --git a/codex-rs/mcp-server/src/codex_tool_config.rs b/codex-rs/mcp-server/src/codex_tool_config.rs deleted file mode 100644 index b9c8aa8505..0000000000 --- a/codex-rs/mcp-server/src/codex_tool_config.rs +++ /dev/null @@ -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, - - /// 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, - - /// Approval policy for shell commands generated by the model: - /// `on-request`, `never`. - #[serde(default, skip_serializing_if = "Option::is_none")] - pub approval_policy: Option, - - /// Sandbox mode: `read-only`, `workspace-write`, or `danger-full-access`. - #[serde(default, skip_serializing_if = "Option::is_none")] - pub sandbox: Option, - - /// Individual config settings that will override what is in - /// CODEX_HOME/config.toml. - #[serde(default, skip_serializing_if = "Option::is_none")] - pub config: Option>, - - /// The set of instructions to use instead of the default ones. - #[serde(default, skip_serializing_if = "Option::is_none")] - pub base_instructions: Option, - - /// Developer instructions that should be injected as a developer role message. - #[serde(default, skip_serializing_if = "Option::is_none")] - pub developer_instructions: Option, - - /// Prompt used when compacting the conversation. - #[serde(default, skip_serializing_if = "Option::is_none")] - pub compact_prompt: Option, -} - -/// 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 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 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::(); - - 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 { - 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, - - /// 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, - - /// The *next user prompt* to continue the Codex conversation. - pub prompt: String, -} - -impl CodexToolCallReplyParam { - pub(crate) fn get_thread_id(&self) -> anyhow::Result { - 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::(); - - 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 { - #[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::(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); - } -} diff --git a/codex-rs/mcp-server/src/codex_tool_runner.rs b/codex-rs/mcp-server/src/codex_tool_runner.rs deleted file mode 100644 index fdc32c3a07..0000000000 --- a/codex-rs/mcp-server/src/codex_tool_runner.rs +++ /dev/null @@ -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, -) -> 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, - thread_manager: Arc, - active_turns: Arc, -) { - 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, - outgoing: Arc, - request_id: RequestId, - prompt: String, - active_turns: Arc, -) { - 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, - outgoing: Arc, - request_id: RequestId, - active_turns: Arc, -) { - 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", - })) - ); - } -} diff --git a/codex-rs/mcp-server/src/exec_approval.rs b/codex-rs/mcp-server/src/exec_approval.rs deleted file mode 100644 index 5296315363..0000000000 --- a/codex-rs/mcp-server/src/exec_approval.rs +++ /dev/null @@ -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, - pub codex_cwd: PathBuf, - pub codex_parsed_cmd: Vec, -} - -// 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, - cwd: PathBuf, - outgoing: Arc, - codex: Arc, - request_id: RequestId, - tool_call_id: String, - event_id: String, - call_id: String, - approval_id: String, - codex_parsed_cmd: Vec, - 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(¶ms) { - 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, - codex: Arc, -) { - 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::(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}"); - } -} diff --git a/codex-rs/mcp-server/src/extension_event_sink.rs b/codex-rs/mcp-server/src/extension_event_sink.rs deleted file mode 100644 index 7c802cd246..0000000000 --- a/codex-rs/mcp-server/src/extension_event_sink.rs +++ /dev/null @@ -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, - active_turns: Arc, -) -> Arc { - Arc::new(McpExtensionEventSink { - outgoing, - active_turns, - }) -} - -struct McpExtensionEventSink { - outgoing: Arc, - active_turns: Arc, -} - -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; diff --git a/codex-rs/mcp-server/src/extension_event_sink_tests.rs b/codex-rs/mcp-server/src/extension_event_sink_tests.rs deleted file mode 100644 index 806b573a6d..0000000000 --- a/codex-rs/mcp-server/src/extension_event_sink_tests.rs +++ /dev/null @@ -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, - Arc, - Arc, - mpsc::UnboundedReceiver, -) { - 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| ¶ms["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| ¶ms["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()); -} diff --git a/codex-rs/mcp-server/src/lib.rs b/codex-rs/mcp-server/src/lib.rs deleted file mode 100644 index b972521ddc..0000000000 --- a/codex-rs/mcp-server/src/lib.rs +++ /dev/null @@ -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; - -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::(CHANNEL_CAPACITY); - let (outgoing_tx, mut outgoing_rx) = mpsc::unbounded_channel::(); - 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::(&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(()) - } -} diff --git a/codex-rs/mcp-server/src/main.rs b/codex-rs/mcp-server/src/main.rs deleted file mode 100644 index 350647441a..0000000000 --- a/codex-rs/mcp-server/src/main.rs +++ /dev/null @@ -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(()) - }) -} diff --git a/codex-rs/mcp-server/src/message_processor.rs b/codex-rs/mcp-server/src/message_processor.rs deleted file mode 100644 index 3231f0c65d..0000000000 --- a/codex-rs/mcp-server/src/message_processor.rs +++ /dev/null @@ -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, - initialized: bool, - arg0_paths: Arg0DispatchPaths, - thread_manager: Arc, - active_turns: Arc, -} - -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, - environment_manager: Arc, - state_db: Option, - installation_id: String, - ) -> std::io::Result { - 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::::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) { - 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) { - 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, - ) { - 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) { - tracing::info!("resources/list -> params: {:?}", params); - } - - fn handle_list_resource_templates(&self, params: Option) { - 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) { - 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, - ) { - 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, - ) { - let arguments = arguments.map(serde_json::Value::Object); - let (initial_prompt, config): (String, Config) = match arguments { - Some(json_val) => match serde_json::from_value::(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, - ) { - 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::(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"); - } -} diff --git a/codex-rs/mcp-server/src/outgoing_message.rs b/codex-rs/mcp-server/src/outgoing_message.rs deleted file mode 100644 index 8e8a2c683c..0000000000 --- a/codex-rs/mcp-server/src/outgoing_message.rs +++ /dev/null @@ -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; - -/// Sends messages to the client and manages request callbacks. -pub(crate) struct OutgoingMessageSender { - next_request_id: AtomicI64, - sender: mpsc::UnboundedSender, - request_id_to_callback: Mutex>>, -} - -impl OutgoingMessageSender { - pub(crate) fn new(sender: mpsc::UnboundedSender) -> 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, - ) -> oneshot::Receiver { - 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(&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, - ) { - #[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 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, -} - -#[derive(Debug, Clone, PartialEq, Serialize)] -pub(crate) struct OutgoingNotification { - pub method: String, - #[serde(default, skip_serializing_if = "Option::is_none")] - pub params: Option, -} - -#[derive(Debug, Clone, PartialEq, Serialize)] -pub(crate) struct OutgoingNotificationParams { - #[serde(rename = "_meta", default, skip_serializing_if = "Option::is_none")] - pub meta: Option, - - #[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, - - /// 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, -} - -#[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::(); - 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::(); - 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::(); - 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::(); - 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(()) - } -} diff --git a/codex-rs/mcp-server/src/patch_approval.rs b/codex-rs/mcp-server/src/patch_approval.rs deleted file mode 100644 index 9cc1815908..0000000000 --- a/codex-rs/mcp-server/src/patch_approval.rs +++ /dev/null @@ -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, - #[serde(skip_serializing_if = "Option::is_none")] - pub codex_grant_root: Option, - pub codex_changes: HashMap, -} - -#[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, - grant_root: Option, - changes: HashMap, - outgoing: Arc, - codex: Arc, - 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(¶ms) { - 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, - codex: Arc, -) { - 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::(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}"); - } -} diff --git a/codex-rs/mcp-server/src/workload_identity_tests.rs b/codex-rs/mcp-server/src/workload_identity_tests.rs deleted file mode 100644 index 4e131c62b6..0000000000 --- a/codex-rs/mcp-server/src/workload_identity_tests.rs +++ /dev/null @@ -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"); -} diff --git a/codex-rs/mcp-server/tests/all.rs b/codex-rs/mcp-server/tests/all.rs deleted file mode 100644 index fdf98aa945..0000000000 --- a/codex-rs/mcp-server/tests/all.rs +++ /dev/null @@ -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; diff --git a/codex-rs/mcp-server/tests/common/BUILD.bazel b/codex-rs/mcp-server/tests/common/BUILD.bazel deleted file mode 100644 index d588b5b8dc..0000000000 --- a/codex-rs/mcp-server/tests/common/BUILD.bazel +++ /dev/null @@ -1,7 +0,0 @@ -load("//:defs.bzl", "codex_rust_crate") - -codex_rust_crate( - name = "common", - crate_name = "mcp_test_support", - crate_srcs = glob(["*.rs"]), -) diff --git a/codex-rs/mcp-server/tests/common/Cargo.toml b/codex-rs/mcp-server/tests/common/Cargo.toml deleted file mode 100644 index 515aa01b5a..0000000000 --- a/codex-rs/mcp-server/tests/common/Cargo.toml +++ /dev/null @@ -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 } diff --git a/codex-rs/mcp-server/tests/common/lib.rs b/codex-rs/mcp-server/tests/common/lib.rs deleted file mode 100644 index 503e753118..0000000000 --- a/codex-rs/mcp-server/tests/common/lib.rs +++ /dev/null @@ -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; diff --git a/codex-rs/mcp-server/tests/common/mcp_process.rs b/codex-rs/mcp-server/tests/common/mcp_process.rs deleted file mode 100644 index f1ad725d24..0000000000 --- a/codex-rs/mcp-server/tests/common/mcp_process.rs +++ /dev/null @@ -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, -} - -impl McpProcess { - pub async fn new(codex_home: &Path) -> anyhow::Result { - 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 { - 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 { - 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, - ) -> anyhow::Result { - 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, - ) -> 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> { - let mut line = String::new(); - self.stdout.read_line(&mut line).await?; - let message = serde_json::from_str::< - JsonRpcMessage, - >(&line)?; - eprintln!("read message from stdout: {message:?}"); - Ok(message) - } - - pub async fn read_stream_until_request_message( - &mut self, - ) -> anyhow::Result> { - 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> { - 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> { - 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> { - 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> { - 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, - } - } - } -} diff --git a/codex-rs/mcp-server/tests/common/mock_model_server.rs b/codex-rs/mcp-server/tests/common/mock_model_server.rs deleted file mode 100644 index 7734ae12cd..0000000000 --- a/codex-rs/mcp-server/tests/common/mock_model_server.rs +++ /dev/null @@ -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) -> 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, -} - -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") - } -} diff --git a/codex-rs/mcp-server/tests/common/responses.rs b/codex-rs/mcp-server/tests/common/responses.rs deleted file mode 100644 index 099919a838..0000000000 --- a/codex-rs/mcp-server/tests/common/responses.rs +++ /dev/null @@ -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, - workdir: Option<&Path>, - timeout_ms: Option, - call_id: &str, -) -> anyhow::Result { - 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 { - 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 { - 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), - ])) -} diff --git a/codex-rs/mcp-server/tests/suite/codex_tool.rs b/codex-rs/mcp-server/tests/suite/codex_tool.rs deleted file mode 100644 index d58bdd7a6e..0000000000 --- a/codex-rs/mcp-server/tests/suite/codex_tool.rs +++ /dev/null @@ -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::( - 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, - workdir: &Path, - codex_mcp_tool_call_id: String, - codex_event_id: String, - thread_id: codex_protocol::ThreadId, -) -> anyhow::Result { - 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::( - 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::()?; - 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 ") - .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, - grant_root: Option, - reason: Option, - codex_mcp_tool_call_id: String, - codex_event_id: String, - thread_id: codex_protocol::ThreadId, -) -> anyhow::Result { - 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) -> anyhow::Result { - 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] -"# - ), - ) -} diff --git a/codex-rs/mcp-server/tests/suite/mod.rs b/codex-rs/mcp-server/tests/suite/mod.rs deleted file mode 100644 index 6b50853b16..0000000000 --- a/codex-rs/mcp-server/tests/suite/mod.rs +++ /dev/null @@ -1 +0,0 @@ -mod codex_tool; diff --git a/justfile b/justfile index b8319d09a2..f85f1c0338 100644 --- a/justfile +++ b/justfile @@ -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 diff --git a/tools/argument-comment-lint/src/lib.rs b/tools/argument-comment-lint/src/lib.rs index 76b9e17732..d14d39b3a5 100644 --- a/tools/argument-comment-lint/src/lib.rs +++ b/tools/argument-comment-lint/src/lib.rs @@ -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]