mirror of
https://github.com/openai/codex.git
synced 2026-11-11 21:10:39 +00:00
feat(mcp): refresh one server in place
Co-authored-by: Codex noreply@openai.com
This commit is contained in:
co-authored by
Codex noreply@openai.com
parent
f2f80ef442
commit
75b188502d
@@ -10,6 +10,7 @@ use std::collections::HashMap;
|
||||
use std::path::Path;
|
||||
use std::path::PathBuf;
|
||||
use std::sync::Arc;
|
||||
use std::sync::atomic::AtomicBool;
|
||||
use std::sync::atomic::Ordering;
|
||||
use std::time::Duration;
|
||||
use std::time::Instant;
|
||||
@@ -107,6 +108,53 @@ pub fn tool_is_model_visible(tool: &ToolInfo) -> bool {
|
||||
}
|
||||
|
||||
/// A thin wrapper around a set of running [`RmcpClient`] instances.
|
||||
#[derive(Clone)]
|
||||
struct McpClientRuntime {
|
||||
store_mode: OAuthCredentialsStoreMode,
|
||||
keyring_backend_kind: AuthKeyringBackendKind,
|
||||
tx_event: Sender<Event>,
|
||||
runtime_context: McpRuntimeContext,
|
||||
codex_home: PathBuf,
|
||||
codex_apps_tools_cache_key: CodexAppsToolsCacheKey,
|
||||
client_elicitation_capability: ElicitationCapability,
|
||||
supports_openai_form_elicitation: bool,
|
||||
chatgpt_auth_provider: Option<SharedAuthProvider>,
|
||||
}
|
||||
|
||||
impl McpClientRuntime {
|
||||
fn create_client(
|
||||
&self,
|
||||
server_name: String,
|
||||
server: EffectiveMcpServer,
|
||||
cancel_token: CancellationToken,
|
||||
elicitation_requests: ElicitationRequestManager,
|
||||
tool_plugin_provenance: Arc<ToolPluginProvenance>,
|
||||
) -> AsyncManagedClient {
|
||||
let tools_cache_context = if server_name == CODEX_APPS_MCP_SERVER_NAME {
|
||||
codex_apps_tools_cache_context(&self.codex_home, &self.codex_apps_tools_cache_key)
|
||||
} else {
|
||||
regular_mcp_tools_cache_context()
|
||||
};
|
||||
let runtime_auth_provider =
|
||||
chatgpt_auth_provider_for_server(&server, self.chatgpt_auth_provider.clone());
|
||||
AsyncManagedClient::new(
|
||||
server_name,
|
||||
server,
|
||||
self.store_mode,
|
||||
self.keyring_backend_kind,
|
||||
cancel_token,
|
||||
self.tx_event.clone(),
|
||||
elicitation_requests,
|
||||
tools_cache_context,
|
||||
tool_plugin_provenance,
|
||||
self.runtime_context.clone(),
|
||||
runtime_auth_provider,
|
||||
self.client_elicitation_capability.clone(),
|
||||
self.supports_openai_form_elicitation,
|
||||
)
|
||||
}
|
||||
}
|
||||
|
||||
pub struct McpConnectionManager {
|
||||
clients: HashMap<String, AsyncManagedClient>,
|
||||
server_metadata: HashMap<String, McpServerMetadata>,
|
||||
@@ -115,6 +163,9 @@ pub struct McpConnectionManager {
|
||||
prefix_mcp_tool_names: bool,
|
||||
elicitation_requests: ElicitationRequestManager,
|
||||
startup_cancellation_token: CancellationToken,
|
||||
client_runtime: Option<Arc<McpClientRuntime>>,
|
||||
cancel_on_drop: AtomicBool,
|
||||
pending_replacement_cancellation_token: Option<CancellationToken>,
|
||||
}
|
||||
|
||||
impl McpConnectionManager {
|
||||
@@ -158,6 +209,17 @@ impl McpConnectionManager {
|
||||
let chatgpt_auth_provider = auth
|
||||
.filter(|auth| auth.uses_codex_backend())
|
||||
.map(codex_model_provider::auth_provider_from_auth);
|
||||
let client_runtime = Arc::new(McpClientRuntime {
|
||||
store_mode,
|
||||
keyring_backend_kind,
|
||||
tx_event: tx_event.clone(),
|
||||
runtime_context,
|
||||
codex_home,
|
||||
codex_apps_tools_cache_key,
|
||||
client_elicitation_capability,
|
||||
supports_openai_form_elicitation,
|
||||
chatgpt_auth_provider,
|
||||
});
|
||||
let mcp_servers = mcp_servers.clone();
|
||||
for (server_name, server) in mcp_servers
|
||||
.into_iter()
|
||||
@@ -174,62 +236,25 @@ impl McpConnectionManager {
|
||||
},
|
||||
)
|
||||
.await;
|
||||
let codex_apps_tools_cache_context = if server_name == CODEX_APPS_MCP_SERVER_NAME {
|
||||
codex_apps_tools_cache_context(&codex_home, &codex_apps_tools_cache_key)
|
||||
} else {
|
||||
regular_mcp_tools_cache_context()
|
||||
};
|
||||
let runtime_auth_provider =
|
||||
chatgpt_auth_provider_for_server(&server, chatgpt_auth_provider.clone());
|
||||
let async_managed_client = AsyncManagedClient::new(
|
||||
let async_managed_client = client_runtime.create_client(
|
||||
server_name.clone(),
|
||||
server,
|
||||
store_mode,
|
||||
keyring_backend_kind,
|
||||
cancel_token.clone(),
|
||||
tx_event.clone(),
|
||||
elicitation_requests.clone(),
|
||||
codex_apps_tools_cache_context,
|
||||
Arc::clone(&tool_plugin_provenance),
|
||||
runtime_context.clone(),
|
||||
runtime_auth_provider,
|
||||
client_elicitation_capability.clone(),
|
||||
supports_openai_form_elicitation,
|
||||
);
|
||||
clients.insert(server_name.clone(), async_managed_client.clone());
|
||||
let tx_event = tx_event.clone();
|
||||
let submit_id = startup_submit_id.clone();
|
||||
let auth_entry = auth_entries.get(&server_name).cloned();
|
||||
join_set.spawn(async move {
|
||||
let mut outcome = async_managed_client.client().await;
|
||||
if cancel_token.is_cancelled() {
|
||||
outcome = Err(StartupOutcomeError::Cancelled);
|
||||
}
|
||||
let status = match &outcome {
|
||||
Ok(_) => McpStartupStatus::Ready,
|
||||
Err(StartupOutcomeError::Cancelled) => McpStartupStatus::Cancelled,
|
||||
Err(error) => {
|
||||
let error_str = mcp_init_error_display(
|
||||
server_name.as_str(),
|
||||
auth_entry.as_ref(),
|
||||
error,
|
||||
);
|
||||
McpStartupStatus::Failed { error: error_str }
|
||||
}
|
||||
};
|
||||
|
||||
let _ = emit_update(
|
||||
submit_id.as_str(),
|
||||
&tx_event,
|
||||
McpStartupUpdateEvent {
|
||||
server: server_name.clone(),
|
||||
status,
|
||||
},
|
||||
)
|
||||
.await;
|
||||
|
||||
(server_name, outcome)
|
||||
});
|
||||
join_set.spawn(observe_startup(
|
||||
server_name,
|
||||
async_managed_client,
|
||||
cancel_token,
|
||||
auth_entry,
|
||||
submit_id,
|
||||
tx_event,
|
||||
));
|
||||
}
|
||||
let manager = Self {
|
||||
clients,
|
||||
@@ -239,21 +264,15 @@ impl McpConnectionManager {
|
||||
prefix_mcp_tool_names,
|
||||
elicitation_requests: elicitation_requests.clone(),
|
||||
startup_cancellation_token: startup_cancellation_token.clone(),
|
||||
client_runtime: Some(client_runtime),
|
||||
cancel_on_drop: AtomicBool::new(true),
|
||||
pending_replacement_cancellation_token: None,
|
||||
};
|
||||
tokio::spawn(async move {
|
||||
let outcomes = join_set.join_all().await;
|
||||
let mut summary = McpStartupCompleteEvent::default();
|
||||
for (server_name, outcome) in outcomes {
|
||||
match outcome {
|
||||
Ok(_) => summary.ready.push(server_name),
|
||||
Err(StartupOutcomeError::Cancelled) => summary.cancelled.push(server_name),
|
||||
Err(StartupOutcomeError::Failed { error }) => {
|
||||
summary.failed.push(McpStartupFailure {
|
||||
server: server_name,
|
||||
error,
|
||||
})
|
||||
}
|
||||
}
|
||||
record_startup_outcome(&mut summary, server_name, outcome);
|
||||
}
|
||||
let _ = tx_event
|
||||
.send(Event {
|
||||
@@ -328,6 +347,9 @@ impl McpConnectionManager {
|
||||
/*reviewer*/ None,
|
||||
),
|
||||
startup_cancellation_token: CancellationToken::new(),
|
||||
client_runtime: None,
|
||||
cancel_on_drop: AtomicBool::new(true),
|
||||
pending_replacement_cancellation_token: None,
|
||||
}
|
||||
}
|
||||
|
||||
@@ -335,6 +357,115 @@ impl McpConnectionManager {
|
||||
!self.clients.is_empty()
|
||||
}
|
||||
|
||||
/// Builds a new manager that replaces only `server_name`.
|
||||
///
|
||||
/// Unchanged clients and manager-scoped elicitation state are shared with
|
||||
/// the replacement. The caller must swap the returned manager into place
|
||||
/// before allowing this manager to drop, then retire the old named client
|
||||
/// with [`Self::shutdown_server_after_readers_release`].
|
||||
pub async fn prepare_server_replacement(
|
||||
&self,
|
||||
server_name: String,
|
||||
server: EffectiveMcpServer,
|
||||
auth_entry: Option<McpAuthStatusEntry>,
|
||||
submit_id: String,
|
||||
) -> Result<Self> {
|
||||
let runtime = self
|
||||
.client_runtime
|
||||
.as_ref()
|
||||
.context("MCP client runtime is unavailable")?;
|
||||
let mut clients = self.clients.clone();
|
||||
clients.remove(&server_name);
|
||||
let mut server_metadata = self.server_metadata.clone();
|
||||
server_metadata.remove(&server_name);
|
||||
let mut required_servers = self.required_servers.clone();
|
||||
required_servers.retain(|required| required != &server_name);
|
||||
let mut replacement_cancellation_token = None;
|
||||
|
||||
if server.enabled() {
|
||||
if server.required() {
|
||||
required_servers.push(server_name.clone());
|
||||
required_servers.sort();
|
||||
}
|
||||
server_metadata.insert(server_name.clone(), McpServerMetadata::from(&server));
|
||||
let cancel_token = self.startup_cancellation_token.child_token();
|
||||
replacement_cancellation_token = Some(cancel_token.clone());
|
||||
let _ = emit_update(
|
||||
submit_id.as_str(),
|
||||
&runtime.tx_event,
|
||||
McpStartupUpdateEvent {
|
||||
server: server_name.clone(),
|
||||
status: McpStartupStatus::Starting,
|
||||
},
|
||||
)
|
||||
.await;
|
||||
let replacement_client = runtime.create_client(
|
||||
server_name.clone(),
|
||||
server,
|
||||
cancel_token.clone(),
|
||||
self.elicitation_requests.clone(),
|
||||
Arc::clone(&self.tool_plugin_provenance),
|
||||
);
|
||||
clients.insert(server_name.clone(), replacement_client.clone());
|
||||
|
||||
let tx_event = runtime.tx_event.clone();
|
||||
tokio::spawn(async move {
|
||||
let (server_name, outcome) = observe_startup(
|
||||
server_name,
|
||||
replacement_client,
|
||||
cancel_token,
|
||||
auth_entry,
|
||||
submit_id.clone(),
|
||||
tx_event.clone(),
|
||||
)
|
||||
.await;
|
||||
let mut summary = McpStartupCompleteEvent::default();
|
||||
record_startup_outcome(&mut summary, server_name, outcome);
|
||||
let _ = tx_event
|
||||
.send(Event {
|
||||
id: submit_id,
|
||||
msg: EventMsg::McpStartupComplete(summary),
|
||||
})
|
||||
.await;
|
||||
});
|
||||
}
|
||||
|
||||
Ok(Self {
|
||||
clients,
|
||||
server_metadata,
|
||||
required_servers,
|
||||
tool_plugin_provenance: Arc::clone(&self.tool_plugin_provenance),
|
||||
prefix_mcp_tool_names: self.prefix_mcp_tool_names,
|
||||
elicitation_requests: self.elicitation_requests.clone(),
|
||||
startup_cancellation_token: self.startup_cancellation_token.clone(),
|
||||
client_runtime: Some(Arc::clone(runtime)),
|
||||
cancel_on_drop: AtomicBool::new(false),
|
||||
pending_replacement_cancellation_token: replacement_cancellation_token,
|
||||
})
|
||||
}
|
||||
|
||||
/// Transfers ownership of the shared startup lifecycle after a successful swap.
|
||||
pub fn take_cancellation_ownership_from(&self, superseded: &Self) {
|
||||
self.cancel_on_drop.store(true, Ordering::Release);
|
||||
superseded.cancel_on_drop.store(false, Ordering::Release);
|
||||
}
|
||||
|
||||
/// Stops the replaced client after all readers release this manager snapshot.
|
||||
pub async fn shutdown_server_after_readers_release(self: Arc<Self>, server_name: String) {
|
||||
let client = self.clients.get(&server_name).cloned();
|
||||
let weak_manager = Arc::downgrade(&self);
|
||||
drop(self);
|
||||
|
||||
// A request keeps its ArcSwap snapshot for the complete MCP call. Wait
|
||||
// for those snapshots before terminating the old server transport.
|
||||
while weak_manager.strong_count() > 0 {
|
||||
tokio::time::sleep(Duration::from_millis(10)).await;
|
||||
}
|
||||
if let Some(client) = client {
|
||||
client.shutdown().await;
|
||||
}
|
||||
}
|
||||
|
||||
pub(crate) fn contains_server(&self, server_name: &str) -> bool {
|
||||
self.clients.contains_key(server_name)
|
||||
}
|
||||
@@ -844,7 +975,12 @@ impl McpConnectionManager {
|
||||
|
||||
impl Drop for McpConnectionManager {
|
||||
fn drop(&mut self) {
|
||||
self.startup_cancellation_token.cancel();
|
||||
if self.cancel_on_drop.load(Ordering::Acquire) {
|
||||
self.startup_cancellation_token.cancel();
|
||||
} else if let Some(cancel_token) = &self.pending_replacement_cancellation_token {
|
||||
// A prepared replacement that was never committed owns only its new client.
|
||||
cancel_token.cancel();
|
||||
}
|
||||
self.clients.clear();
|
||||
}
|
||||
}
|
||||
@@ -880,6 +1016,52 @@ fn chatgpt_auth_provider_for_server(
|
||||
chatgpt_auth_provider
|
||||
}
|
||||
|
||||
async fn observe_startup(
|
||||
server_name: String,
|
||||
client: AsyncManagedClient,
|
||||
cancel_token: CancellationToken,
|
||||
auth_entry: Option<McpAuthStatusEntry>,
|
||||
submit_id: String,
|
||||
tx_event: Sender<Event>,
|
||||
) -> (String, Result<ManagedClient, StartupOutcomeError>) {
|
||||
let mut outcome = client.client().await;
|
||||
if cancel_token.is_cancelled() {
|
||||
outcome = Err(StartupOutcomeError::Cancelled);
|
||||
}
|
||||
let status = match &outcome {
|
||||
Ok(_) => McpStartupStatus::Ready,
|
||||
Err(StartupOutcomeError::Cancelled) => McpStartupStatus::Cancelled,
|
||||
Err(error) => McpStartupStatus::Failed {
|
||||
error: mcp_init_error_display(&server_name, auth_entry.as_ref(), error),
|
||||
},
|
||||
};
|
||||
let _ = emit_update(
|
||||
&submit_id,
|
||||
&tx_event,
|
||||
McpStartupUpdateEvent {
|
||||
server: server_name.clone(),
|
||||
status,
|
||||
},
|
||||
)
|
||||
.await;
|
||||
(server_name, outcome)
|
||||
}
|
||||
|
||||
fn record_startup_outcome(
|
||||
summary: &mut McpStartupCompleteEvent,
|
||||
server_name: String,
|
||||
outcome: Result<ManagedClient, StartupOutcomeError>,
|
||||
) {
|
||||
match outcome {
|
||||
Ok(_) => summary.ready.push(server_name),
|
||||
Err(StartupOutcomeError::Cancelled) => summary.cancelled.push(server_name),
|
||||
Err(StartupOutcomeError::Failed { error }) => summary.failed.push(McpStartupFailure {
|
||||
server: server_name,
|
||||
error,
|
||||
}),
|
||||
}
|
||||
}
|
||||
|
||||
async fn emit_update(
|
||||
submit_id: &str,
|
||||
tx_event: &Sender<Event>,
|
||||
|
||||
@@ -101,6 +101,33 @@ fn create_test_server_info(title: &str) -> McpServerInfo {
|
||||
}
|
||||
}
|
||||
|
||||
fn local_stdio_server(command: &str) -> EffectiveMcpServer {
|
||||
EffectiveMcpServer::configured(McpServerConfig {
|
||||
auth: Default::default(),
|
||||
transport: McpServerTransportConfig::Stdio {
|
||||
command: command.to_string(),
|
||||
args: Vec::new(),
|
||||
env: None,
|
||||
env_vars: Vec::new(),
|
||||
cwd: None,
|
||||
},
|
||||
environment_id: codex_config::DEFAULT_MCP_SERVER_ENVIRONMENT_ID.to_string(),
|
||||
enabled: true,
|
||||
required: false,
|
||||
supports_parallel_tool_calls: false,
|
||||
disabled_reason: None,
|
||||
startup_timeout_sec: None,
|
||||
tool_timeout_sec: None,
|
||||
default_tools_approval_mode: None,
|
||||
enabled_tools: None,
|
||||
disabled_tools: None,
|
||||
scopes: None,
|
||||
oauth: None,
|
||||
oauth_resource: None,
|
||||
tools: HashMap::new(),
|
||||
})
|
||||
}
|
||||
|
||||
fn model_tool_names(tools: &[ToolInfo]) -> HashSet<ToolName> {
|
||||
tools
|
||||
.iter()
|
||||
@@ -1269,33 +1296,7 @@ async fn no_local_runtime_fails_local_stdio_but_keeps_local_http_server() {
|
||||
drop(rx_event);
|
||||
let codex_home = tempdir().expect("tempdir");
|
||||
let mcp_servers = HashMap::from([
|
||||
(
|
||||
"stdio".to_string(),
|
||||
EffectiveMcpServer::configured(McpServerConfig {
|
||||
auth: Default::default(),
|
||||
transport: McpServerTransportConfig::Stdio {
|
||||
command: "echo".to_string(),
|
||||
args: Vec::new(),
|
||||
env: None,
|
||||
env_vars: Vec::new(),
|
||||
cwd: None,
|
||||
},
|
||||
environment_id: codex_config::DEFAULT_MCP_SERVER_ENVIRONMENT_ID.to_string(),
|
||||
enabled: true,
|
||||
required: false,
|
||||
supports_parallel_tool_calls: false,
|
||||
disabled_reason: None,
|
||||
startup_timeout_sec: None,
|
||||
tool_timeout_sec: None,
|
||||
default_tools_approval_mode: None,
|
||||
enabled_tools: None,
|
||||
disabled_tools: None,
|
||||
scopes: None,
|
||||
oauth: None,
|
||||
oauth_resource: None,
|
||||
tools: HashMap::new(),
|
||||
}),
|
||||
),
|
||||
("stdio".to_string(), local_stdio_server("echo")),
|
||||
(
|
||||
"http".to_string(),
|
||||
EffectiveMcpServer::configured(McpServerConfig {
|
||||
@@ -1378,6 +1379,106 @@ async fn no_local_runtime_fails_local_stdio_but_keeps_local_http_server() {
|
||||
cancel_token.cancel();
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn server_replacement_preserves_and_does_not_cancel_unrelated_clients() {
|
||||
let approval_policy = Constrained::allow_any(AskForApproval::OnRequest);
|
||||
let (tx_event, rx_event) = async_channel::unbounded();
|
||||
drop(rx_event);
|
||||
let codex_home = tempdir().expect("tempdir");
|
||||
let cancel_token = CancellationToken::new();
|
||||
let manager = Arc::new(
|
||||
McpConnectionManager::new(
|
||||
&HashMap::from([
|
||||
("alpha".to_string(), local_stdio_server("alpha")),
|
||||
("beta".to_string(), local_stdio_server("beta")),
|
||||
]),
|
||||
OAuthCredentialsStoreMode::default(),
|
||||
AuthKeyringBackendKind::default(),
|
||||
HashMap::new(),
|
||||
&approval_policy,
|
||||
"initial".to_string(),
|
||||
tx_event,
|
||||
cancel_token.clone(),
|
||||
PermissionProfile::default(),
|
||||
McpRuntimeContext::new(
|
||||
Arc::new(EnvironmentManager::without_environments()),
|
||||
PathBuf::from("/tmp"),
|
||||
),
|
||||
codex_home.path().to_path_buf(),
|
||||
CodexAppsToolsCacheKey {
|
||||
account_id: None,
|
||||
chatgpt_user_id: None,
|
||||
is_workspace_account: false,
|
||||
},
|
||||
/*prefix_mcp_tool_names*/ true,
|
||||
ElicitationCapability::default(),
|
||||
/*supports_openai_form_elicitation*/ false,
|
||||
ToolPluginProvenance::default(),
|
||||
/*auth*/ None,
|
||||
/*elicitation_reviewer*/ None,
|
||||
)
|
||||
.await,
|
||||
);
|
||||
let alpha_startup = Arc::clone(&manager.clients["alpha"].startup_complete);
|
||||
let old_beta_startup = Arc::clone(&manager.clients["beta"].startup_complete);
|
||||
let alpha_cancel = manager.clients["alpha"].cancel_token.clone();
|
||||
let old_beta_cancel = manager.clients["beta"].cancel_token.clone();
|
||||
|
||||
let discarded_replacement = manager
|
||||
.prepare_server_replacement(
|
||||
"beta".to_string(),
|
||||
local_stdio_server("discarded"),
|
||||
None,
|
||||
"discarded-refresh".to_string(),
|
||||
)
|
||||
.await
|
||||
.expect("replacement manager should build");
|
||||
let discarded_cancel = discarded_replacement.clients["beta"].cancel_token.clone();
|
||||
assert!(manager.cancel_on_drop.load(Ordering::Acquire));
|
||||
assert!(!discarded_replacement.cancel_on_drop.load(Ordering::Acquire));
|
||||
drop(discarded_replacement);
|
||||
assert!(discarded_cancel.is_cancelled());
|
||||
assert!(!old_beta_cancel.is_cancelled());
|
||||
assert!(!cancel_token.is_cancelled());
|
||||
|
||||
let replacement = Arc::new(
|
||||
manager
|
||||
.prepare_server_replacement(
|
||||
"beta".to_string(),
|
||||
local_stdio_server("replacement"),
|
||||
None,
|
||||
"refresh".to_string(),
|
||||
)
|
||||
.await
|
||||
.expect("replacement manager should build"),
|
||||
);
|
||||
assert!(Arc::ptr_eq(
|
||||
&alpha_startup,
|
||||
&replacement.clients["alpha"].startup_complete
|
||||
));
|
||||
assert!(!Arc::ptr_eq(
|
||||
&old_beta_startup,
|
||||
&replacement.clients["beta"].startup_complete
|
||||
));
|
||||
replacement.take_cancellation_ownership_from(&manager);
|
||||
assert!(!manager.cancel_on_drop.load(Ordering::Acquire));
|
||||
assert!(replacement.cancel_on_drop.load(Ordering::Acquire));
|
||||
|
||||
let retire_task = tokio::spawn(
|
||||
Arc::clone(&manager).shutdown_server_after_readers_release("beta".to_string()),
|
||||
);
|
||||
tokio::task::yield_now().await;
|
||||
assert!(!retire_task.is_finished());
|
||||
assert!(!old_beta_cancel.is_cancelled());
|
||||
drop(manager);
|
||||
retire_task.await.expect("retirement task should finish");
|
||||
assert!(old_beta_cancel.is_cancelled());
|
||||
assert!(!alpha_cancel.is_cancelled());
|
||||
assert!(!replacement.clients["beta"].cancel_token.is_cancelled());
|
||||
assert!(!cancel_token.is_cancelled());
|
||||
replacement.shutdown().await;
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn elicitation_capability_uses_2025_06_18_shape_for_form_only_support() {
|
||||
let capability = Some(ElicitationCapability::default());
|
||||
|
||||
@@ -93,6 +93,7 @@ pub use codex_protocol::protocol::AskForApproval;
|
||||
pub use codex_protocol::protocol::EventMsg;
|
||||
pub use codex_protocol::protocol::InitialHistory;
|
||||
pub use codex_protocol::protocol::McpServerRefreshConfig;
|
||||
pub use codex_protocol::protocol::McpServerUpdateConfig;
|
||||
pub use codex_protocol::protocol::Op;
|
||||
pub use codex_protocol::protocol::SessionConfiguredEvent;
|
||||
pub use codex_protocol::protocol::SessionSource;
|
||||
|
||||
@@ -33,6 +33,7 @@ use codex_protocol::protocol::GuardianAssessmentEvent;
|
||||
use codex_protocol::protocol::GuardianAssessmentStatus;
|
||||
use codex_protocol::protocol::InterAgentCommunication;
|
||||
use codex_protocol::protocol::McpServerRefreshConfig;
|
||||
use codex_protocol::protocol::McpServerUpdateConfig;
|
||||
use codex_protocol::protocol::Op;
|
||||
use codex_protocol::protocol::RealtimeConversationListVoicesResponseEvent;
|
||||
use codex_protocol::protocol::RealtimeVoicesList;
|
||||
@@ -437,6 +438,16 @@ pub async fn refresh_mcp_servers(sess: &Arc<Session>, refresh_config: McpServerR
|
||||
*guard = Some(refresh_config);
|
||||
}
|
||||
|
||||
pub async fn refresh_mcp_server(
|
||||
sess: &Arc<Session>,
|
||||
update: McpServerUpdateConfig,
|
||||
submit_id: String,
|
||||
) {
|
||||
if let Err(error) = sess.refresh_mcp_server_now(update, submit_id).await {
|
||||
warn!(error = %error, "failed to refresh MCP server");
|
||||
}
|
||||
}
|
||||
|
||||
pub async fn reload_user_config(sess: &Arc<Session>) {
|
||||
sess.reload_user_config_layer().await;
|
||||
}
|
||||
@@ -795,6 +806,10 @@ pub(super) async fn submission_loop(
|
||||
refresh_mcp_servers(&sess, config).await;
|
||||
false
|
||||
}
|
||||
Op::RefreshMcpServer { config } => {
|
||||
refresh_mcp_server(&sess, config, sub.id.clone()).await;
|
||||
false
|
||||
}
|
||||
Op::ReloadUserConfig => {
|
||||
reload_user_config(&sess).await;
|
||||
false
|
||||
|
||||
@@ -1,4 +1,5 @@
|
||||
use super::*;
|
||||
use anyhow::Context;
|
||||
use codex_mcp::ElicitationReviewRequest;
|
||||
use codex_mcp::ElicitationReviewer;
|
||||
use codex_mcp::ElicitationReviewerHandle;
|
||||
@@ -430,6 +431,55 @@ impl Session {
|
||||
.await;
|
||||
}
|
||||
|
||||
pub(crate) async fn refresh_mcp_server_now(
|
||||
&self,
|
||||
update: codex_protocol::protocol::McpServerUpdateConfig,
|
||||
submit_id: String,
|
||||
) -> anyhow::Result<()> {
|
||||
let server_name = update.server_name;
|
||||
let configured_server = serde_json::from_value::<McpServerConfig>(update.server)
|
||||
.with_context(|| format!("failed to parse MCP server update for `{server_name}`"))?;
|
||||
let auth = self.services.auth_manager.auth().await;
|
||||
let config = self.get_config().await;
|
||||
let mcp_config = self.runtime_mcp_config(config.as_ref()).await;
|
||||
let mut effective_servers = effective_mcp_servers_from_configured(
|
||||
HashMap::from([(server_name.clone(), configured_server)]),
|
||||
&mcp_config,
|
||||
auth.as_ref(),
|
||||
);
|
||||
let server = effective_servers.remove(&server_name).ok_or_else(|| {
|
||||
anyhow::anyhow!("MCP server update did not resolve server `{server_name}`")
|
||||
})?;
|
||||
let mut auth_entries = compute_auth_statuses(
|
||||
std::iter::once((&server_name, &server)),
|
||||
config.mcp_oauth_credentials_store_mode,
|
||||
config.auth_keyring_backend_kind(),
|
||||
auth.as_ref(),
|
||||
)
|
||||
.await;
|
||||
|
||||
let current_manager = self.services.mcp_connection_manager.load_full();
|
||||
let replacement_manager = Arc::new(
|
||||
current_manager
|
||||
.prepare_server_replacement(
|
||||
server_name.clone(),
|
||||
server,
|
||||
auth_entries.remove(&server_name),
|
||||
submit_id,
|
||||
)
|
||||
.await?,
|
||||
);
|
||||
let superseded_manager = self
|
||||
.services
|
||||
.mcp_connection_manager
|
||||
.swap(Arc::clone(&replacement_manager));
|
||||
replacement_manager.take_cancellation_ownership_from(&superseded_manager);
|
||||
drop(current_manager);
|
||||
|
||||
tokio::spawn(superseded_manager.shutdown_server_after_readers_release(server_name));
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
pub(crate) async fn mcp_startup_cancellation_token(&self) -> CancellationToken {
|
||||
self.services
|
||||
|
||||
@@ -179,7 +179,7 @@ pub struct W3cTraceContext {
|
||||
}
|
||||
|
||||
/// Config payload for refreshing MCP servers.
|
||||
#[derive(Debug, Clone, PartialEq)]
|
||||
#[derive(Clone, PartialEq)]
|
||||
pub struct McpServerRefreshConfig {
|
||||
/// Complete runtime server map after source and thread-scoped resolution.
|
||||
pub mcp_servers: Value,
|
||||
@@ -188,6 +188,34 @@ pub struct McpServerRefreshConfig {
|
||||
pub auth_keyring_backend_kind: Value,
|
||||
}
|
||||
|
||||
impl fmt::Debug for McpServerRefreshConfig {
|
||||
fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
|
||||
formatter
|
||||
.debug_struct("McpServerRefreshConfig")
|
||||
.field("mcp_servers", &"<redacted>")
|
||||
.field("mcp_oauth_credentials_store_mode", &"<redacted>")
|
||||
.field("auth_keyring_backend_kind", &"<redacted>")
|
||||
.finish()
|
||||
}
|
||||
}
|
||||
|
||||
/// Config payload for replacing one MCP server in a running session.
|
||||
#[derive(Clone, PartialEq)]
|
||||
pub struct McpServerUpdateConfig {
|
||||
pub server_name: String,
|
||||
pub server: Value,
|
||||
}
|
||||
|
||||
impl fmt::Debug for McpServerUpdateConfig {
|
||||
fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
|
||||
formatter
|
||||
.debug_struct("McpServerUpdateConfig")
|
||||
.field("server_name", &self.server_name)
|
||||
.field("server", &"<redacted>")
|
||||
.finish()
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, PartialEq)]
|
||||
pub struct ConversationStartParams {
|
||||
/// Whether Codex response handoffs are managed through explicit client append calls.
|
||||
@@ -627,6 +655,9 @@ pub enum Op {
|
||||
/// Request MCP servers to reinitialize and refresh cached tool lists.
|
||||
RefreshMcpServers { config: McpServerRefreshConfig },
|
||||
|
||||
/// Replace one MCP server without reconnecting unrelated servers.
|
||||
RefreshMcpServer { config: McpServerUpdateConfig },
|
||||
|
||||
/// Reload user config layer overrides for the active session.
|
||||
///
|
||||
/// This updates runtime config-derived behavior (for example app
|
||||
@@ -837,6 +868,7 @@ impl Op {
|
||||
Self::RequestPermissionsResponse { .. } => "request_permissions_response",
|
||||
Self::DynamicToolResponse { .. } => "dynamic_tool_response",
|
||||
Self::RefreshMcpServers { .. } => "refresh_mcp_servers",
|
||||
Self::RefreshMcpServer { .. } => "refresh_mcp_server",
|
||||
Self::ReloadUserConfig => "reload_user_config",
|
||||
Self::Compact => "compact",
|
||||
Self::SetThreadMemoryMode { .. } => "set_thread_memory_mode",
|
||||
@@ -4314,6 +4346,23 @@ mod tests {
|
||||
use tempfile::NamedTempFile;
|
||||
use tempfile::TempDir;
|
||||
|
||||
#[test]
|
||||
fn mcp_refresh_debug_output_redacts_server_credentials() {
|
||||
let secret = "sentinel-bearer-token";
|
||||
let refresh = McpServerRefreshConfig {
|
||||
mcp_servers: json!({"server": {"bearer_token": secret}}),
|
||||
mcp_oauth_credentials_store_mode: json!("file"),
|
||||
auth_keyring_backend_kind: json!("default"),
|
||||
};
|
||||
let update = McpServerUpdateConfig {
|
||||
server_name: "server".to_string(),
|
||||
server: json!({"bearer_token": secret}),
|
||||
};
|
||||
|
||||
assert!(!format!("{refresh:?}").contains(secret));
|
||||
assert!(!format!("{update:?}").contains(secret));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn feature_thread_source_serializes_as_its_app_owned_label() -> Result<()> {
|
||||
let source = ThreadSource::Feature("automation".to_string());
|
||||
|
||||
Reference in New Issue
Block a user