mirror of
https://github.com/openai/codex.git
synced 2026-11-12 21:20:34 +00:00
Move single-use MCP helpers
Move manager-only startup update, origin, and startup error display helpers from mcp_connection.rs into manager.rs. Move client-only bearer token resolution and server-name validation helpers into client.rs. Co-authored-by: Codex <noreply@openai.com>
This commit is contained in:
@@ -25,8 +25,6 @@ use crate::mcp_connection::DEFAULT_TOOL_TIMEOUT;
|
||||
use crate::mcp_connection::MCP_SANDBOX_STATE_META_CAPABILITY;
|
||||
use crate::mcp_connection::McpRuntimeEnvironment;
|
||||
use crate::mcp_connection::emit_duration;
|
||||
use crate::mcp_connection::resolve_bearer_token;
|
||||
use crate::mcp_connection::validate_mcp_server_name;
|
||||
use crate::tools::ToolFilter;
|
||||
use crate::tools::ToolInfo;
|
||||
use crate::tools::filter_tools;
|
||||
@@ -298,6 +296,44 @@ pub(crate) fn elicitation_capability_for_server(
|
||||
})
|
||||
}
|
||||
|
||||
fn resolve_bearer_token(
|
||||
server_name: &str,
|
||||
bearer_token_env_var: Option<&str>,
|
||||
) -> Result<Option<String>> {
|
||||
let Some(env_var) = bearer_token_env_var else {
|
||||
return Ok(None);
|
||||
};
|
||||
|
||||
match env::var(env_var) {
|
||||
Ok(value) => {
|
||||
if value.is_empty() {
|
||||
Err(anyhow!(
|
||||
"Environment variable {env_var} for MCP server '{server_name}' is empty"
|
||||
))
|
||||
} else {
|
||||
Ok(Some(value))
|
||||
}
|
||||
}
|
||||
Err(env::VarError::NotPresent) => Err(anyhow!(
|
||||
"Environment variable {env_var} for MCP server '{server_name}' is not set"
|
||||
)),
|
||||
Err(env::VarError::NotUnicode(_)) => Err(anyhow!(
|
||||
"Environment variable {env_var} for MCP server '{server_name}' contains invalid Unicode"
|
||||
)),
|
||||
}
|
||||
}
|
||||
|
||||
fn validate_mcp_server_name(server_name: &str) -> Result<()> {
|
||||
let re = regex_lite::Regex::new(r"^[a-zA-Z0-9_-]+$")?;
|
||||
if !re.is_match(server_name) {
|
||||
return Err(anyhow!(
|
||||
"Invalid MCP server name '{server_name}': must match pattern {pattern}",
|
||||
pattern = re.as_str()
|
||||
));
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
async fn start_server_task(
|
||||
server_name: String,
|
||||
client: Arc<RmcpClient>,
|
||||
|
||||
@@ -17,12 +17,9 @@ use crate::client::list_tools_for_client_uncached;
|
||||
use crate::elicitation::ElicitationRequestManager;
|
||||
use crate::mcp::CODEX_APPS_MCP_SERVER_NAME;
|
||||
use crate::mcp::ToolPluginProvenance;
|
||||
use crate::mcp_connection::DEFAULT_STARTUP_TIMEOUT;
|
||||
use crate::mcp_connection::McpRuntimeEnvironment;
|
||||
use crate::mcp_connection::emit_duration;
|
||||
use crate::mcp_connection::emit_update;
|
||||
use crate::mcp_connection::mcp_init_error_display;
|
||||
use crate::mcp_connection::startup_outcome_error_message;
|
||||
use crate::mcp_connection::transport_origin;
|
||||
use crate::tools::ToolInfo;
|
||||
use crate::tools::filter_tools;
|
||||
use crate::tools::qualify_tools;
|
||||
@@ -59,6 +56,7 @@ use tokio::task::JoinSet;
|
||||
use tokio_util::sync::CancellationToken;
|
||||
use tracing::instrument;
|
||||
use tracing::warn;
|
||||
use url::Url;
|
||||
|
||||
/// A thin wrapper around a set of running [`RmcpClient`] instances.
|
||||
pub struct McpConnectionManager {
|
||||
@@ -67,6 +65,92 @@ pub struct McpConnectionManager {
|
||||
elicitation_requests: ElicitationRequestManager,
|
||||
}
|
||||
|
||||
async fn emit_update(
|
||||
submit_id: &str,
|
||||
tx_event: &Sender<Event>,
|
||||
update: McpStartupUpdateEvent,
|
||||
) -> Result<(), async_channel::SendError<Event>> {
|
||||
tx_event
|
||||
.send(Event {
|
||||
id: submit_id.to_string(),
|
||||
msg: EventMsg::McpStartupUpdate(update),
|
||||
})
|
||||
.await
|
||||
}
|
||||
|
||||
fn transport_origin(transport: &McpServerTransportConfig) -> Option<String> {
|
||||
match transport {
|
||||
McpServerTransportConfig::StreamableHttp { url, .. } => {
|
||||
let parsed = Url::parse(url).ok()?;
|
||||
Some(parsed.origin().ascii_serialization())
|
||||
}
|
||||
McpServerTransportConfig::Stdio { .. } => Some("stdio".to_string()),
|
||||
}
|
||||
}
|
||||
|
||||
fn mcp_init_error_display(
|
||||
server_name: &str,
|
||||
entry: Option<&McpAuthStatusEntry>,
|
||||
err: &StartupOutcomeError,
|
||||
) -> String {
|
||||
if let Some(McpServerTransportConfig::StreamableHttp {
|
||||
url,
|
||||
bearer_token_env_var,
|
||||
http_headers,
|
||||
..
|
||||
}) = &entry.map(|entry| &entry.config.transport)
|
||||
&& url == "https://api.githubcopilot.com/mcp/"
|
||||
&& bearer_token_env_var.is_none()
|
||||
&& http_headers.as_ref().map(HashMap::is_empty).unwrap_or(true)
|
||||
{
|
||||
format!(
|
||||
"GitHub MCP does not support OAuth. Log in by adding a personal access token (https://github.com/settings/personal-access-tokens) to your environment and config.toml:\n[mcp_servers.{server_name}]\nbearer_token_env_var = CODEX_GITHUB_PERSONAL_ACCESS_TOKEN"
|
||||
)
|
||||
} else if is_mcp_client_auth_required_error(err) {
|
||||
format!(
|
||||
"The {server_name} MCP server is not logged in. Run `codex mcp login {server_name}`."
|
||||
)
|
||||
} else if is_mcp_client_startup_timeout_error(err) {
|
||||
let startup_timeout_secs = match entry {
|
||||
Some(entry) => match entry.config.startup_timeout_sec {
|
||||
Some(timeout) => timeout,
|
||||
None => DEFAULT_STARTUP_TIMEOUT,
|
||||
},
|
||||
None => DEFAULT_STARTUP_TIMEOUT,
|
||||
}
|
||||
.as_secs();
|
||||
format!(
|
||||
"MCP client for `{server_name}` timed out after {startup_timeout_secs} seconds. Add or adjust `startup_timeout_sec` in your config.toml:\n[mcp_servers.{server_name}]\nstartup_timeout_sec = XX"
|
||||
)
|
||||
} else {
|
||||
format!("MCP client for `{server_name}` failed to start: {err:#}")
|
||||
}
|
||||
}
|
||||
|
||||
fn is_mcp_client_auth_required_error(error: &StartupOutcomeError) -> bool {
|
||||
match error {
|
||||
StartupOutcomeError::Failed { error } => error.contains("Auth required"),
|
||||
_ => false,
|
||||
}
|
||||
}
|
||||
|
||||
fn is_mcp_client_startup_timeout_error(error: &StartupOutcomeError) -> bool {
|
||||
match error {
|
||||
StartupOutcomeError::Failed { error } => {
|
||||
error.contains("request timed out")
|
||||
|| error.contains("timed out handshaking with MCP server")
|
||||
}
|
||||
_ => false,
|
||||
}
|
||||
}
|
||||
|
||||
fn startup_outcome_error_message(error: StartupOutcomeError) -> String {
|
||||
match error {
|
||||
StartupOutcomeError::Cancelled => "MCP startup cancelled".to_string(),
|
||||
StartupOutcomeError::Failed { error } => error,
|
||||
}
|
||||
}
|
||||
|
||||
impl McpConnectionManager {
|
||||
pub fn new_uninitialized(
|
||||
approval_policy: &Constrained<AskForApproval>,
|
||||
|
||||
@@ -2,29 +2,16 @@
|
||||
//!
|
||||
//! This module contains shared types and helpers used by [`McpConnectionManager`].
|
||||
|
||||
use std::collections::HashMap;
|
||||
use std::env;
|
||||
use std::path::PathBuf;
|
||||
use std::sync::Arc;
|
||||
use std::time::Duration;
|
||||
|
||||
use crate::McpAuthStatusEntry;
|
||||
use crate::client::StartupOutcomeError;
|
||||
use anyhow::Result;
|
||||
use anyhow::anyhow;
|
||||
use async_channel::Sender;
|
||||
use codex_exec_server::Environment;
|
||||
use codex_protocol::models::PermissionProfile;
|
||||
use codex_protocol::protocol::Event;
|
||||
use codex_protocol::protocol::EventMsg;
|
||||
use codex_protocol::protocol::McpStartupUpdateEvent;
|
||||
use codex_protocol::protocol::SandboxPolicy;
|
||||
|
||||
use serde::Deserialize;
|
||||
use serde::Serialize;
|
||||
use url::Url;
|
||||
|
||||
use codex_config::McpServerTransportConfig;
|
||||
|
||||
/// Default timeout for initializing MCP server & initially listing tools.
|
||||
pub(crate) const DEFAULT_STARTUP_TIMEOUT: Duration = Duration::from_secs(30);
|
||||
@@ -79,135 +66,8 @@ impl McpRuntimeEnvironment {
|
||||
}
|
||||
}
|
||||
|
||||
pub(crate) async fn emit_update(
|
||||
submit_id: &str,
|
||||
tx_event: &Sender<Event>,
|
||||
update: McpStartupUpdateEvent,
|
||||
) -> Result<(), async_channel::SendError<Event>> {
|
||||
tx_event
|
||||
.send(Event {
|
||||
id: submit_id.to_string(),
|
||||
msg: EventMsg::McpStartupUpdate(update),
|
||||
})
|
||||
.await
|
||||
}
|
||||
|
||||
pub(crate) fn resolve_bearer_token(
|
||||
server_name: &str,
|
||||
bearer_token_env_var: Option<&str>,
|
||||
) -> Result<Option<String>> {
|
||||
let Some(env_var) = bearer_token_env_var else {
|
||||
return Ok(None);
|
||||
};
|
||||
|
||||
match env::var(env_var) {
|
||||
Ok(value) => {
|
||||
if value.is_empty() {
|
||||
Err(anyhow!(
|
||||
"Environment variable {env_var} for MCP server '{server_name}' is empty"
|
||||
))
|
||||
} else {
|
||||
Ok(Some(value))
|
||||
}
|
||||
}
|
||||
Err(env::VarError::NotPresent) => Err(anyhow!(
|
||||
"Environment variable {env_var} for MCP server '{server_name}' is not set"
|
||||
)),
|
||||
Err(env::VarError::NotUnicode(_)) => Err(anyhow!(
|
||||
"Environment variable {env_var} for MCP server '{server_name}' contains invalid Unicode"
|
||||
)),
|
||||
}
|
||||
}
|
||||
|
||||
pub(crate) fn emit_duration(metric: &str, duration: Duration, tags: &[(&str, &str)]) {
|
||||
if let Some(metrics) = codex_otel::global() {
|
||||
let _ = metrics.record_duration(metric, duration, tags);
|
||||
}
|
||||
}
|
||||
|
||||
pub(crate) fn transport_origin(transport: &McpServerTransportConfig) -> Option<String> {
|
||||
match transport {
|
||||
McpServerTransportConfig::StreamableHttp { url, .. } => {
|
||||
let parsed = Url::parse(url).ok()?;
|
||||
Some(parsed.origin().ascii_serialization())
|
||||
}
|
||||
McpServerTransportConfig::Stdio { .. } => Some("stdio".to_string()),
|
||||
}
|
||||
}
|
||||
|
||||
pub(crate) fn validate_mcp_server_name(server_name: &str) -> Result<()> {
|
||||
let re = regex_lite::Regex::new(r"^[a-zA-Z0-9_-]+$")?;
|
||||
if !re.is_match(server_name) {
|
||||
return Err(anyhow!(
|
||||
"Invalid MCP server name '{server_name}': must match pattern {pattern}",
|
||||
pattern = re.as_str()
|
||||
));
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
pub(crate) fn mcp_init_error_display(
|
||||
server_name: &str,
|
||||
entry: Option<&McpAuthStatusEntry>,
|
||||
err: &StartupOutcomeError,
|
||||
) -> String {
|
||||
if let Some(McpServerTransportConfig::StreamableHttp {
|
||||
url,
|
||||
bearer_token_env_var,
|
||||
http_headers,
|
||||
..
|
||||
}) = &entry.map(|entry| &entry.config.transport)
|
||||
&& url == "https://api.githubcopilot.com/mcp/"
|
||||
&& bearer_token_env_var.is_none()
|
||||
&& http_headers.as_ref().map(HashMap::is_empty).unwrap_or(true)
|
||||
{
|
||||
format!(
|
||||
"GitHub MCP does not support OAuth. Log in by adding a personal access token (https://github.com/settings/personal-access-tokens) to your environment and config.toml:\n[mcp_servers.{server_name}]\nbearer_token_env_var = CODEX_GITHUB_PERSONAL_ACCESS_TOKEN"
|
||||
)
|
||||
} else if is_mcp_client_auth_required_error(err) {
|
||||
format!(
|
||||
"The {server_name} MCP server is not logged in. Run `codex mcp login {server_name}`."
|
||||
)
|
||||
} else if is_mcp_client_startup_timeout_error(err) {
|
||||
let startup_timeout_secs = match entry {
|
||||
Some(entry) => match entry.config.startup_timeout_sec {
|
||||
Some(timeout) => timeout,
|
||||
None => DEFAULT_STARTUP_TIMEOUT,
|
||||
},
|
||||
None => DEFAULT_STARTUP_TIMEOUT,
|
||||
}
|
||||
.as_secs();
|
||||
format!(
|
||||
"MCP client for `{server_name}` timed out after {startup_timeout_secs} seconds. Add or adjust `startup_timeout_sec` in your config.toml:\n[mcp_servers.{server_name}]\nstartup_timeout_sec = XX"
|
||||
)
|
||||
} else {
|
||||
format!("MCP client for `{server_name}` failed to start: {err:#}")
|
||||
}
|
||||
}
|
||||
|
||||
fn is_mcp_client_auth_required_error(error: &StartupOutcomeError) -> bool {
|
||||
match error {
|
||||
StartupOutcomeError::Failed { error } => error.contains("Auth required"),
|
||||
_ => false,
|
||||
}
|
||||
}
|
||||
|
||||
fn is_mcp_client_startup_timeout_error(error: &StartupOutcomeError) -> bool {
|
||||
match error {
|
||||
StartupOutcomeError::Failed { error } => {
|
||||
error.contains("request timed out")
|
||||
|| error.contains("timed out handshaking with MCP server")
|
||||
}
|
||||
_ => false,
|
||||
}
|
||||
}
|
||||
|
||||
pub(crate) fn startup_outcome_error_message(error: StartupOutcomeError) -> String {
|
||||
match error {
|
||||
StartupOutcomeError::Cancelled => "MCP startup cancelled".to_string(),
|
||||
StartupOutcomeError::Failed { error } => error,
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod mcp_init_error_display_tests {}
|
||||
|
||||
@@ -11,7 +11,6 @@ use crate::client::elicitation_capability_for_server;
|
||||
use crate::declared_openai_file_input_param_names;
|
||||
use crate::elicitation::ElicitationRequestManager;
|
||||
use crate::elicitation::elicitation_is_rejected_by_policy;
|
||||
use crate::mcp_connection::transport_origin;
|
||||
use crate::tools::ToolFilter;
|
||||
use crate::tools::ToolInfo;
|
||||
use crate::tools::filter_tools;
|
||||
|
||||
Reference in New Issue
Block a user