mirror of
https://github.com/hanzoai/dev.git
synced 2026-10-25 03:32:32 +00:00
fix(core/mcp): make rmcp client default
This commit is contained in:
Generated
-1
@@ -1171,7 +1171,6 @@ dependencies = [
|
||||
"code-browser",
|
||||
"code-file-search",
|
||||
"code-git-tooling",
|
||||
"code-mcp-client",
|
||||
"code-mcp-types",
|
||||
"code-otel",
|
||||
"code-protocol",
|
||||
|
||||
@@ -13,8 +13,6 @@ use code_core::config::write_global_mcp_servers;
|
||||
use code_core::config_types::McpServerConfig;
|
||||
use code_core::config_types::McpServerTransportConfig;
|
||||
|
||||
/// [experimental] Launch Codex as an MCP server or manage configured MCP servers.
|
||||
///
|
||||
/// Subcommands:
|
||||
/// - `serve` — run the MCP server on stdio
|
||||
/// - `list` — list configured servers (with `--json`)
|
||||
@@ -32,16 +30,12 @@ pub struct McpCli {
|
||||
|
||||
#[derive(Debug, clap::Subcommand)]
|
||||
pub enum McpSubcommand {
|
||||
/// [experimental] List configured MCP servers.
|
||||
List(ListArgs),
|
||||
|
||||
/// [experimental] Show details for a configured MCP server.
|
||||
Get(GetArgs),
|
||||
|
||||
/// [experimental] Add a global MCP server entry.
|
||||
Add(AddArgs),
|
||||
|
||||
/// [experimental] Remove a global MCP server entry.
|
||||
Remove(RemoveArgs),
|
||||
}
|
||||
|
||||
|
||||
@@ -22,7 +22,6 @@ chrono = { workspace = true, features = ["serde"] }
|
||||
chardetng = { workspace = true }
|
||||
code-apply-patch = { workspace = true }
|
||||
code-file-search = { workspace = true }
|
||||
code-mcp-client = { workspace = true }
|
||||
code-protocol = { workspace = true }
|
||||
code-rmcp-client = { workspace = true }
|
||||
code-app-server-protocol = { workspace = true }
|
||||
|
||||
@@ -1345,9 +1345,6 @@ fn codex_beta_features_header_value(config: &Config) -> Option<HeaderValue> {
|
||||
if config.tools_web_search_request {
|
||||
enabled.push("web_search_request");
|
||||
}
|
||||
if config.use_experimental_use_rmcp_client {
|
||||
enabled.push("rmcp_client");
|
||||
}
|
||||
|
||||
let value = enabled.join(",");
|
||||
if value.is_empty() {
|
||||
|
||||
@@ -4875,7 +4875,6 @@ async fn submission_loop(
|
||||
|
||||
let (mcp_connection_manager, failed_clients) = match McpConnectionManager::new(
|
||||
config.mcp_servers.clone(),
|
||||
config.use_experimental_use_rmcp_client,
|
||||
excluded_tools,
|
||||
)
|
||||
.await
|
||||
|
||||
@@ -407,8 +407,6 @@ pub struct Config {
|
||||
pub tools_web_search_allowed_domains: Option<Vec<String>>,
|
||||
/// Experimental: enable streamable shell tool selection (off by default).
|
||||
pub use_experimental_streamable_shell_tool: bool,
|
||||
/// Experimental: opt into the RMCP client implementation for MCP servers.
|
||||
pub use_experimental_use_rmcp_client: bool,
|
||||
/// Enable the `view_image` tool that lets the agent attach local images.
|
||||
pub include_view_image_tool: bool,
|
||||
|
||||
@@ -2155,9 +2153,6 @@ pub struct ConfigToml {
|
||||
#[serde(default)]
|
||||
pub confirm_guard: Option<ConfirmGuardConfig>,
|
||||
|
||||
#[serde(default)]
|
||||
pub experimental_use_rmcp_client: Option<bool>,
|
||||
|
||||
/// Disable server-side response storage (sends the full conversation
|
||||
/// context with every request). Currently necessary for OpenAI customers
|
||||
/// who have opted into Zero Data Retention (ZDR).
|
||||
@@ -3131,9 +3126,6 @@ impl Config {
|
||||
use_experimental_streamable_shell_tool: cfg
|
||||
.experimental_use_exec_command_tool
|
||||
.unwrap_or(false),
|
||||
use_experimental_use_rmcp_client: cfg
|
||||
.experimental_use_rmcp_client
|
||||
.unwrap_or(false),
|
||||
include_view_image_tool: include_view_image_tool_flag,
|
||||
skills_enabled,
|
||||
env_ctx_v2: env_ctx_v2_flag,
|
||||
|
||||
@@ -1,6 +1,6 @@
|
||||
//! Connection manager for Model Context Protocol (MCP) servers.
|
||||
//!
|
||||
//! The [`McpConnectionManager`] owns one [`code_mcp_client::McpClient`] per
|
||||
//! The [`McpConnectionManager`] owns one [`code_rmcp_client::RmcpClient`] per
|
||||
//! configured server (keyed by the *server name*). It offers convenience
|
||||
//! helpers to query the available tools across *all* servers and returns them
|
||||
//! in a single aggregated map using the fully-qualified tool name
|
||||
@@ -15,7 +15,6 @@ use std::time::Duration;
|
||||
use anyhow::Context;
|
||||
use anyhow::Result;
|
||||
use anyhow::anyhow;
|
||||
use code_mcp_client::McpClient;
|
||||
use code_rmcp_client::RmcpClient;
|
||||
use mcp_types::ClientCapabilities;
|
||||
use mcp_types::Implementation;
|
||||
@@ -26,7 +25,6 @@ use sha1::Digest;
|
||||
use sha1::Sha1;
|
||||
use tokio::sync::RwLock;
|
||||
use tokio::task::JoinSet;
|
||||
use tokio::task::yield_now;
|
||||
use tracing::info;
|
||||
use tracing::warn;
|
||||
|
||||
@@ -94,13 +92,11 @@ struct ManagedClient {
|
||||
|
||||
#[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>>,
|
||||
@@ -108,17 +104,11 @@ impl McpClientAdapter {
|
||||
startup_timeout: Duration,
|
||||
) -> Result<Self> {
|
||||
tracing::debug!(
|
||||
"new_stdio_client use_rmcp_client: {use_rmcp_client} program: {program:?} args: {args:?} env: {env:?} params: {params:?} startup_timeout: {startup_timeout:?}"
|
||||
"new_stdio_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))
|
||||
}
|
||||
let client = Arc::new(RmcpClient::new_stdio_client(program, args, env).await?);
|
||||
client.initialize(params, Some(startup_timeout)).await?;
|
||||
Ok(McpClientAdapter::Rmcp(client))
|
||||
}
|
||||
|
||||
async fn new_streamable_http_client(
|
||||
@@ -138,7 +128,6 @@ impl McpClientAdapter {
|
||||
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,
|
||||
}
|
||||
}
|
||||
@@ -150,19 +139,12 @@ impl McpClientAdapter {
|
||||
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,
|
||||
}
|
||||
}
|
||||
|
||||
async fn into_shutdown(self) {
|
||||
match self {
|
||||
McpClientAdapter::Legacy(client) => {
|
||||
drop(client);
|
||||
// Yield so Tokio can propagate the kill-on-drop signal before
|
||||
// the next session spins up a replacement server.
|
||||
yield_now().await;
|
||||
}
|
||||
McpClientAdapter::Rmcp(client) => {
|
||||
client.shutdown().await;
|
||||
}
|
||||
@@ -170,7 +152,7 @@ impl McpClientAdapter {
|
||||
}
|
||||
}
|
||||
|
||||
/// A thin wrapper around a set of running [`McpClient`] instances.
|
||||
/// A thin wrapper around a set of running [`RmcpClient`] instances.
|
||||
#[derive(Default)]
|
||||
pub struct McpConnectionManager {
|
||||
/// Server-name -> client instance.
|
||||
@@ -184,7 +166,7 @@ pub struct McpConnectionManager {
|
||||
}
|
||||
|
||||
impl McpConnectionManager {
|
||||
/// Spawn a [`McpClient`] for each configured server.
|
||||
/// Spawn a [`RmcpClient`] for each configured server.
|
||||
///
|
||||
/// * `mcp_servers` – Map loaded from the user configuration where *keys*
|
||||
/// are human-readable server identifiers and *values* are the spawn
|
||||
@@ -194,7 +176,6 @@ impl McpConnectionManager {
|
||||
/// user should be informed about these errors.
|
||||
pub async fn new(
|
||||
mcp_servers: HashMap<String, McpServerConfig>,
|
||||
use_rmcp_client: bool,
|
||||
excluded_tools: HashSet<(String, String)>,
|
||||
) -> Result<(Self, ClientStartErrors)> {
|
||||
// Early exit if no servers are configured.
|
||||
@@ -216,22 +197,9 @@ impl McpConnectionManager {
|
||||
continue;
|
||||
}
|
||||
|
||||
if matches!(
|
||||
cfg.transport,
|
||||
McpServerTransportConfig::StreamableHttp { .. }
|
||||
) && !use_rmcp_client
|
||||
{
|
||||
info!(
|
||||
"skipping MCP server `{}` configured with url because rmcp client is disabled",
|
||||
server_name
|
||||
);
|
||||
continue;
|
||||
}
|
||||
|
||||
let startup_timeout = cfg.startup_timeout_sec.unwrap_or(DEFAULT_STARTUP_TIMEOUT);
|
||||
let tool_timeout = cfg.tool_timeout_sec;
|
||||
|
||||
let use_rmcp_client_flag = use_rmcp_client;
|
||||
join_set.spawn(async move {
|
||||
let McpServerConfig { transport, .. } = cfg;
|
||||
let server_name_for_error = server_name.clone();
|
||||
@@ -263,7 +231,6 @@ impl McpConnectionManager {
|
||||
let command_os: OsString = command.into();
|
||||
let args_os: Vec<OsString> = args.into_iter().map(Into::into).collect();
|
||||
McpClientAdapter::new_stdio_client(
|
||||
use_rmcp_client_flag,
|
||||
command_os,
|
||||
args_os,
|
||||
env,
|
||||
@@ -569,7 +536,7 @@ mod tests {
|
||||
},
|
||||
);
|
||||
|
||||
let (_manager, errors) = McpConnectionManager::new(servers, false, HashSet::new())
|
||||
let (_manager, errors) = McpConnectionManager::new(servers, HashSet::new())
|
||||
.await
|
||||
.expect("manager creation should succeed even when servers fail");
|
||||
|
||||
|
||||
@@ -61,7 +61,8 @@ async fn main() -> Result<()> {
|
||||
servers.insert("slow_ok".to_string(), slow_ok);
|
||||
servers.insert("slow_fail".to_string(), slow_fail);
|
||||
|
||||
let (mgr, errs) = McpConnectionManager::new(servers, false, std::collections::HashSet::new()).await?;
|
||||
let (mgr, errs) =
|
||||
McpConnectionManager::new(servers, std::collections::HashSet::new()).await?;
|
||||
println!("Errors: {}", errs.len());
|
||||
for (name, e) in &errs {
|
||||
println!(" {}: {}", name, e);
|
||||
|
||||
@@ -121,13 +121,8 @@ async fn mcp_stdio_server_exits_before_next_session() {
|
||||
.expect("load config");
|
||||
|
||||
let servers_map = config.mcp_servers.clone();
|
||||
let use_rmcp = config.use_experimental_use_rmcp_client;
|
||||
|
||||
let (manager1, _) = McpConnectionManager::new(
|
||||
servers_map.clone(),
|
||||
use_rmcp,
|
||||
HashSet::new(),
|
||||
)
|
||||
let (manager1, _) = McpConnectionManager::new(servers_map.clone(), HashSet::new())
|
||||
.await
|
||||
.expect("start first MCP manager");
|
||||
|
||||
@@ -144,7 +139,7 @@ async fn mcp_stdio_server_exits_before_next_session() {
|
||||
std::fs::read_to_string(&log_path).unwrap_or_default()
|
||||
);
|
||||
|
||||
let (_manager2, _) = McpConnectionManager::new(servers_map, use_rmcp, HashSet::new())
|
||||
let (_manager2, _) = McpConnectionManager::new(servers_map, HashSet::new())
|
||||
.await
|
||||
.expect("start second MCP manager");
|
||||
}
|
||||
|
||||
@@ -415,8 +415,6 @@ env = { "API_KEY" = "value" }
|
||||
#### Streamable HTTP
|
||||
|
||||
```toml
|
||||
# Streamable HTTP requires the experimental rmcp client
|
||||
experimental_use_rmcp_client = true
|
||||
[mcp_servers.figma]
|
||||
url = "http://127.0.0.1:3845/mcp"
|
||||
# Optional bearer token to be passed into an `Authorization: Bearer <token>` header
|
||||
@@ -424,8 +422,6 @@ url = "http://127.0.0.1:3845/mcp"
|
||||
bearer_token = "<token>"
|
||||
```
|
||||
|
||||
Refer to the MCP CLI commands for oauth login
|
||||
|
||||
### Other configuration options
|
||||
|
||||
```toml
|
||||
|
||||
Reference in New Issue
Block a user