valknar/llmx
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

[MCP] Introduce an experimental official rust sdk based mcp client (#4252)

Browse Source 
The [official Rust
SDK](57fc428c57)
has come a long way since we first started our mcp client implementation
5 months ago and, today, it is much more complete than our own
stdio-only implementation.

This PR introduces a new config flag `experimental_use_rmcp_client`
which will use a new mcp client powered by the sdk instead of our own.

To keep this PR simple, I've only implemented the same stdio MCP
functionality that we had but will expand on it with future PRs.

---------

Co-authored-by: pakrym-oai <pakrym@openai.com>
This commit is contained in:
Gabriel Peal
2025-09-26 10:13:37 -07:00
committed by GitHub
parent ea095e30c1
commit e555a36c6a
17 changed files with 1136 additions and 62 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
codex-rs/core/Cargo.toml
View File

@@ -22,6 +22,7 @@ chrono = { workspace = true, features = ["serde"] }
codex-apply-patch = { workspace = true }
codex-file-search = { workspace = true }
codex-mcp-client = { workspace = true }
codex-rmcp-client = { workspace = true }
codex-protocol = { workspace = true }
dirs = { workspace = true }
env-flags = { workspace = true }
@@ -82,6 +83,7 @@ openssl-sys = { workspace = true, features = ["vendored"] }
[dev-dependencies]
assert_cmd = { workspace = true }
core_test_support = { workspace = true }
escargot = { workspace = true }
maplit = { workspace = true }
predicates = { workspace = true }
pretty_assertions = { workspace = true }

5
codex-rs/core/src/codex.rs
View File

@@ -377,7 +377,10 @@ impl Session {
// - load history metadata
let rollout_fut = RolloutRecorder::new(&config, rollout_params);
let mcp_fut = McpConnectionManager::new(config.mcp_servers.clone());
let mcp_fut = McpConnectionManager::new(
config.mcp_servers.clone(),
config.use_experimental_use_rmcp_client,
);
let default_shell_fut = shell::default_user_shell();
let history_meta_fut = crate::message_history::history_metadata(&config);

10
codex-rs/core/src/config.rs
View File

@@ -184,6 +184,10 @@ pub struct Config {
/// If set to `true`, used only the experimental unified exec tool.
pub use_experimental_unified_exec_tool: bool,
/// If set to `true`, use the experimental official Rust MCP client.
/// https://github.com/modelcontextprotocol/rust-sdk
pub use_experimental_use_rmcp_client: bool,
/// Include the `view_image` tool that lets the agent attach a local image path to context.
pub include_view_image_tool: bool,
@@ -693,6 +697,7 @@ pub struct ConfigToml {
pub experimental_use_exec_command_tool: Option<bool>,
pub experimental_use_unified_exec_tool: Option<bool>,
pub experimental_use_rmcp_client: Option<bool>,
pub projects: Option<HashMap<String, ProjectConfig>>,
@@ -1043,6 +1048,7 @@ impl Config {
use_experimental_unified_exec_tool: cfg
.experimental_use_unified_exec_tool
.unwrap_or(false),
use_experimental_use_rmcp_client: cfg.experimental_use_rmcp_client.unwrap_or(false),
include_view_image_tool,
active_profile: active_profile_name,
disable_paste_burst: cfg.disable_paste_burst.unwrap_or(false),
@@ -1651,6 +1657,7 @@ model_verbosity = "high"
tools_web_search_request: false,
use_experimental_streamable_shell_tool: false,
use_experimental_unified_exec_tool: false,
use_experimental_use_rmcp_client: false,
include_view_image_tool: true,
active_profile: Some("o3".to_string()),
disable_paste_burst: false,
@@ -1709,6 +1716,7 @@ model_verbosity = "high"
tools_web_search_request: false,
use_experimental_streamable_shell_tool: false,
use_experimental_unified_exec_tool: false,
use_experimental_use_rmcp_client: false,
include_view_image_tool: true,
active_profile: Some("gpt3".to_string()),
disable_paste_burst: false,
@@ -1782,6 +1790,7 @@ model_verbosity = "high"
tools_web_search_request: false,
use_experimental_streamable_shell_tool: false,
use_experimental_unified_exec_tool: false,
use_experimental_use_rmcp_client: false,
include_view_image_tool: true,
active_profile: Some("zdr".to_string()),
disable_paste_burst: false,
@@ -1841,6 +1850,7 @@ model_verbosity = "high"
tools_web_search_request: false,
use_experimental_streamable_shell_tool: false,
use_experimental_unified_exec_tool: false,
use_experimental_use_rmcp_client: false,
include_view_image_tool: true,
active_profile: Some("gpt5".to_string()),
disable_paste_burst: false,

138
codex-rs/core/src/mcp_connection_manager.rs
View File

@@ -16,6 +16,7 @@ use anyhow::Context;
use anyhow::Result;
use anyhow::anyhow;
use codex_mcp_client::McpClient;
use codex_rmcp_client::RmcpClient;
use mcp_types::ClientCapabilities;
use mcp_types::Implementation;
use mcp_types::Tool;
@@ -86,11 +87,64 @@ struct ToolInfo {
}
struct ManagedClient {
client: Arc<McpClient>,
client: McpClientAdapter,
startup_timeout: Duration,
tool_timeout: Option<Duration>,
}
#[derive(Clone)]
enum McpClientAdapter {
Legacy(Arc<McpClient>),
Rmcp(Arc<RmcpClient>),
}
impl McpClientAdapter {
async fn new_stdio_client(
use_rmcp_client: bool,
program: OsString,
args: Vec<OsString>,
env: Option<HashMap<String, String>>,
params: mcp_types::InitializeRequestParams,
startup_timeout: Duration,
) -> Result<Self> {
tracing::error!(
"new_stdio_client use_rmcp_client: {use_rmcp_client} program: {program:?} args: {args:?} env: {env:?} params: {params:?} startup_timeout: {startup_timeout:?}"
);
if use_rmcp_client {
let client = Arc::new(RmcpClient::new_stdio_client(program, args, env).await?);
client.initialize(params, Some(startup_timeout)).await?;
Ok(McpClientAdapter::Rmcp(client))
} else {
let client = Arc::new(McpClient::new_stdio_client(program, args, env).await?);
client.initialize(params, Some(startup_timeout)).await?;
Ok(McpClientAdapter::Legacy(client))
}
}
async fn list_tools(
&self,
params: Option<mcp_types::ListToolsRequestParams>,
timeout: Option<Duration>,
) -> Result<mcp_types::ListToolsResult> {
match self {
McpClientAdapter::Legacy(client) => client.list_tools(params, timeout).await,
McpClientAdapter::Rmcp(client) => client.list_tools(params, timeout).await,
}
}
async fn call_tool(
&self,
name: String,
arguments: Option<serde_json::Value>,
timeout: Option<Duration>,
) -> Result<mcp_types::CallToolResult> {
match self {
McpClientAdapter::Legacy(client) => client.call_tool(name, arguments, timeout).await,
McpClientAdapter::Rmcp(client) => client.call_tool(name, arguments, timeout).await,
}
}
}
/// A thin wrapper around a set of running [`McpClient`] instances.
#[derive(Default)]
pub(crate) struct McpConnectionManager {
@@ -115,12 +169,15 @@ impl McpConnectionManager {
/// user should be informed about these errors.
pub async fn new(
mcp_servers: HashMap<String, McpServerConfig>,
use_rmcp_client: bool,
) -> Result<(Self, ClientStartErrors)> {
// Early exit if no servers are configured.
if mcp_servers.is_empty() {
return Ok((Self::default(), ClientStartErrors::default()));
}
tracing::error!("new mcp_servers: {mcp_servers:?} use_rmcp_client: {use_rmcp_client}");
// Launch all configured servers concurrently.
let mut join_set = JoinSet::new();
let mut errors = ClientStartErrors::new();
@@ -137,57 +194,48 @@ impl McpConnectionManager {
}
let startup_timeout = cfg.startup_timeout_sec.unwrap_or(DEFAULT_STARTUP_TIMEOUT);
let tool_timeout = cfg.tool_timeout_sec.unwrap_or(DEFAULT_TOOL_TIMEOUT);
let use_rmcp_client_flag = use_rmcp_client;
join_set.spawn(async move {
let McpServerConfig {
command, args, env, ..
} = cfg;
let client_res = McpClient::new_stdio_client(
command.into(),
args.into_iter().map(OsString::from).collect(),
let command_os: OsString = command.into();
let args_os: Vec<OsString> = args.into_iter().map(Into::into).collect();
let params = mcp_types::InitializeRequestParams {
capabilities: ClientCapabilities {
experimental: None,
roots: None,
sampling: None,
// https://modelcontextprotocol.io/specification/2025-06-18/client/elicitation#capabilities
// indicates this should be an empty object.
elicitation: Some(json!({})),
},
client_info: Implementation {
name: "codex-mcp-client".to_owned(),
version: env!("CARGO_PKG_VERSION").to_owned(),
title: Some("Codex".into()),
// This field is used by Codex when it is an MCP
// server: it should not be used when Codex is
// an MCP client.
user_agent: None,
},
protocol_version: mcp_types::MCP_SCHEMA_VERSION.to_owned(),
};
let client = McpClientAdapter::new_stdio_client(
use_rmcp_client_flag,
command_os,
args_os,
env,
params,
startup_timeout,
)
.await;
match client_res {
Ok(client) => {
// Initialize the client.
let params = mcp_types::InitializeRequestParams {
capabilities: ClientCapabilities {
experimental: None,
roots: None,
sampling: None,
// https://modelcontextprotocol.io/specification/2025-06-18/client/elicitation#capabilities
// indicates this should be an empty object.
elicitation: Some(json!({})),
},
client_info: Implementation {
name: "codex-mcp-client".to_owned(),
version: env!("CARGO_PKG_VERSION").to_owned(),
title: Some("Codex".into()),
// This field is used by Codex when it is an MCP
// server: it should not be used when Codex is
// an MCP client.
user_agent: None,
},
protocol_version: mcp_types::MCP_SCHEMA_VERSION.to_owned(),
};
let initialize_notification_params = None;
let init_result = client
.initialize(
params,
initialize_notification_params,
Some(startup_timeout),
)
.await;
(
(server_name, tool_timeout),
init_result.map(|_| (client, startup_timeout)),
)
}
Err(e) => ((server_name, tool_timeout), Err(e.into())),
}
.await
.map(|c| (c, startup_timeout));
((server_name, tool_timeout), client)
});
}
@@ -207,7 +255,7 @@ impl McpConnectionManager {
clients.insert(
server_name,
ManagedClient {
client: Arc::new(client),
client,
startup_timeout,
tool_timeout: Some(tool_timeout),
},

1
codex-rs/core/tests/suite/mod.rs
View File

@@ -14,6 +14,7 @@ mod live_cli;
mod model_overrides;
mod prompt_caching;
mod review;
mod rmcp_client;
mod rollout_list_find;
mod seatbelt;
mod stream_error_allows_next_turn;

162
codex-rs/core/tests/suite/rmcp_client.rs Normal file
View File

@@ -0,0 +1,162 @@
use std::collections::HashMap;
use std::time::Duration;
use codex_core::config_types::McpServerConfig;
use codex_core::protocol::AskForApproval;
use codex_core::protocol::EventMsg;
use codex_core::protocol::InputItem;
use codex_core::protocol::Op;
use codex_core::protocol::SandboxPolicy;
use codex_protocol::config_types::ReasoningSummary;
use core_test_support::responses;
use core_test_support::responses::mount_sse_once;
use core_test_support::skip_if_no_network;
use core_test_support::test_codex::test_codex;
use core_test_support::wait_for_event;
use core_test_support::wait_for_event_with_timeout;
use escargot::CargoBuild;
use serde_json::Value;
use wiremock::matchers::any;
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn rmcp_tool_call_round_trip() -> anyhow::Result<()> {
skip_if_no_network!(Ok(()));
let server = responses::start_mock_server().await;
let call_id = "call-123";
let server_name = "rmcp";
let tool_name = format!("{server_name}__echo");
mount_sse_once(
&server,
any(),
responses::sse(vec![
serde_json::json!({
"type": "response.created",
"response": {"id": "resp-1"}
}),
responses::ev_function_call(call_id, &tool_name, "{\"message\":\"ping\"}"),
responses::ev_completed("resp-1"),
]),
)
.await;
mount_sse_once(
&server,
any(),
responses::sse(vec![
responses::ev_assistant_message("msg-1", "rmcp echo tool completed successfully."),
responses::ev_completed("resp-2"),
]),
)
.await;
let expected_env_value = "propagated-env";
let rmcp_test_server_bin = CargoBuild::new()
.package("codex-rmcp-client")
.bin("rmcp_test_server")
.run()?
.path()
.to_string_lossy()
.into_owned();
let fixture = test_codex()
.with_config(move |config| {
config.use_experimental_use_rmcp_client = true;
config.mcp_servers.insert(
server_name.to_string(),
McpServerConfig {
command: rmcp_test_server_bin.clone(),
args: Vec::new(),
env: Some(HashMap::from([(
"MCP_TEST_VALUE".to_string(),
expected_env_value.to_string(),
)])),
startup_timeout_sec: Some(Duration::from_secs(10)),
tool_timeout_sec: None,
},
);
})
.build(&server)
.await?;
let session_model = fixture.session_configured.model.clone();
fixture
.codex
.submit(Op::UserTurn {
items: vec![InputItem::Text {
text: "call the rmcp echo tool".into(),
}],
final_output_json_schema: None,
cwd: fixture.cwd.path().to_path_buf(),
approval_policy: AskForApproval::Never,
sandbox_policy: SandboxPolicy::DangerFullAccess,
model: session_model,
effort: None,
summary: ReasoningSummary::Auto,
})
.await?;
eprintln!("waiting for mcp tool call begin event");
let begin_event = wait_for_event_with_timeout(
&fixture.codex,
|ev| {
eprintln!("ev: {ev:?}");
matches!(ev, EventMsg::McpToolCallBegin(_))
},
Duration::from_secs(10),
)
.await;
eprintln!("mcp tool call begin event: {begin_event:?}");
let EventMsg::McpToolCallBegin(begin) = begin_event else {
unreachable!("event guard guarantees McpToolCallBegin");
};
assert_eq!(begin.invocation.server, server_name);
assert_eq!(begin.invocation.tool, "echo");
let end_event = wait_for_event(&fixture.codex, |ev| {
matches!(ev, EventMsg::McpToolCallEnd(_))
})
.await;
eprintln!("end_event: {end_event:?}");
let EventMsg::McpToolCallEnd(end) = end_event else {
unreachable!("event guard guarantees McpToolCallEnd");
};
let result = end
.result
.as_ref()
.expect("rmcp echo tool should return success");
assert_eq!(result.is_error, Some(false));
assert!(
result.content.is_empty(),
"content should default to an empty array"
);
let structured = result
.structured_content
.as_ref()
.expect("structured content");
let Value::Object(map) = structured else {
panic!("structured content should be an object: {structured:?}");
};
let echo_value = map
.get("echo")
.and_then(Value::as_str)
.expect("echo payload present");
assert_eq!(echo_value, "ping");
let env_value = map
.get("env")
.and_then(Value::as_str)
.expect("env snapshot inserted");
assert_eq!(env_value, expected_env_value);
let task_complete_event =
wait_for_event(&fixture.codex, |ev| matches!(ev, EventMsg::TaskComplete(_))).await;
eprintln!("task_complete_event: {task_complete_event:?}");
server.verify().await;
Ok(())
}