mirror of
https://github.com/openai/codex.git
synced 2026-10-27 18:48:37 +00:00
feat(mcp): hide background startup status by default
This commit is contained in:
@@ -386,6 +386,39 @@ impl McpConnectionManager {
|
||||
}
|
||||
}
|
||||
|
||||
/// Returns an owned future that waits only for the requested MCP servers.
|
||||
///
|
||||
/// Constructing this future clones the selected clients, allowing callers to
|
||||
/// release any lock guarding the manager before awaiting startup.
|
||||
pub fn wait_for_servers_ready(
|
||||
&self,
|
||||
server_names: &[String],
|
||||
) -> impl std::future::Future<Output = Vec<McpStartupFailure>> + Send + 'static {
|
||||
let clients = server_names
|
||||
.iter()
|
||||
.map(|server_name| (server_name.clone(), self.clients.get(server_name).cloned()))
|
||||
.collect::<Vec<_>>();
|
||||
async move {
|
||||
let mut failures = Vec::new();
|
||||
for (server_name, client) in clients {
|
||||
let Some(client) = client else {
|
||||
failures.push(McpStartupFailure {
|
||||
server: server_name.clone(),
|
||||
error: format!("MCP server `{server_name}` was not initialized"),
|
||||
});
|
||||
continue;
|
||||
};
|
||||
if let Err(error) = client.client().await {
|
||||
failures.push(McpStartupFailure {
|
||||
server: server_name,
|
||||
error: startup_outcome_error_message(error),
|
||||
});
|
||||
}
|
||||
}
|
||||
failures
|
||||
}
|
||||
}
|
||||
|
||||
pub async fn required_startup_failures(
|
||||
&self,
|
||||
required_servers: &[String],
|
||||
@@ -477,6 +510,37 @@ impl McpConnectionManager {
|
||||
server_infos
|
||||
}
|
||||
|
||||
/// Returns model-visible tools already available from startup cache or a ready client.
|
||||
///
|
||||
/// Pending MCP servers without a cached snapshot are intentionally omitted so
|
||||
/// ordinary model work does not wait on optional background initialization.
|
||||
#[instrument(level = "trace", skip_all, fields(mcp_server_count = self.clients.len()))]
|
||||
pub async fn list_ready_or_cached_tools(&self) -> Vec<ToolInfo> {
|
||||
let mut tools = Vec::new();
|
||||
for (server_name, managed_client) in &self.clients {
|
||||
let has_cached_tool_info_snapshot =
|
||||
managed_client.cached_tool_info_snapshot.is_some();
|
||||
let startup_complete = managed_client
|
||||
.startup_complete
|
||||
.load(std::sync::atomic::Ordering::Acquire);
|
||||
trace!(
|
||||
server_name = %server_name,
|
||||
has_cached_tool_info_snapshot,
|
||||
startup_complete,
|
||||
"checking ready MCP server tools while building nonblocking tool list"
|
||||
);
|
||||
let Some(server_tools) = managed_client.listed_tools_if_ready_or_cached().await else {
|
||||
continue;
|
||||
};
|
||||
tools.extend(
|
||||
server_tools
|
||||
.into_iter()
|
||||
.map(|tool| self.with_server_metadata(tool)),
|
||||
);
|
||||
}
|
||||
normalize_tools_for_model_with_prefix(tools, self.prefix_mcp_tool_names)
|
||||
}
|
||||
|
||||
/// Force-refresh codex apps tools by bypassing the in-process cache.
|
||||
///
|
||||
/// On success, the refreshed tools replace the cache contents and the
|
||||
|
||||
@@ -810,7 +810,7 @@ async fn list_all_tools_uses_cached_tool_info_snapshot_while_client_is_pending()
|
||||
client: pending_client,
|
||||
cached_tool_info_snapshot: Some(startup_tools),
|
||||
cached_server_info: None,
|
||||
startup_complete: Arc::new(std::sync::atomic::AtomicBool::new(false)),
|
||||
startup_complete: Arc::new(std::sync::atomic::AtomicBool::new(/*v*/ false)),
|
||||
tool_plugin_provenance: Arc::new(ToolPluginProvenance::default()),
|
||||
cancel_token: CancellationToken::new(),
|
||||
},
|
||||
@@ -980,6 +980,39 @@ async fn list_all_tools_blocks_while_client_is_pending_without_cached_tool_info_
|
||||
assert!(timeout_result.is_err());
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn list_ready_or_cached_tools_skips_pending_client_without_cached_tool_info_snapshot() {
|
||||
let pending_client = futures::future::pending::<Result<ManagedClient, StartupOutcomeError>>()
|
||||
.boxed()
|
||||
.shared();
|
||||
let approval_policy = Constrained::allow_any(AskForApproval::OnFailure);
|
||||
let permission_profile = Constrained::allow_any(PermissionProfile::default());
|
||||
let mut manager = McpConnectionManager::new_uninitialized(
|
||||
&approval_policy,
|
||||
&permission_profile,
|
||||
/*prefix_mcp_tool_names*/ true,
|
||||
);
|
||||
manager.clients.insert(
|
||||
"optional".to_string(),
|
||||
AsyncManagedClient {
|
||||
client: pending_client,
|
||||
cached_tool_info_snapshot: None,
|
||||
cached_server_info: None,
|
||||
startup_complete: Arc::new(std::sync::atomic::AtomicBool::new(/*v*/ false)),
|
||||
tool_plugin_provenance: Arc::new(ToolPluginProvenance::default()),
|
||||
cancel_token: CancellationToken::new(),
|
||||
},
|
||||
);
|
||||
|
||||
let tools = tokio::time::timeout(
|
||||
Duration::from_millis(10),
|
||||
manager.list_ready_or_cached_tools(),
|
||||
)
|
||||
.await
|
||||
.expect("pending optional client should not block ready tool listing");
|
||||
assert!(tools.is_empty());
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn list_all_tools_does_not_block_when_cached_tool_info_snapshot_is_empty() {
|
||||
let pending_client = futures::future::pending::<Result<ManagedClient, StartupOutcomeError>>()
|
||||
|
||||
@@ -255,58 +255,57 @@ impl AsyncManagedClient {
|
||||
None
|
||||
}
|
||||
|
||||
pub(crate) async fn listed_tools(&self) -> Option<Vec<ToolInfo>> {
|
||||
let annotate_tools = |tools: Vec<ToolInfo>| {
|
||||
let mut tools = tools;
|
||||
for tool in &mut tools {
|
||||
if tool.server_name == CODEX_APPS_MCP_SERVER_NAME {
|
||||
tool.tool = tool_with_model_visible_input_schema(&tool.tool);
|
||||
}
|
||||
|
||||
let plugin_names = match tool.connector_id.as_deref() {
|
||||
Some(connector_id) => self
|
||||
.tool_plugin_provenance
|
||||
.plugin_display_names_for_connector_id(connector_id),
|
||||
None => self
|
||||
.tool_plugin_provenance
|
||||
.plugin_display_names_for_mcp_server_name(tool.server_name.as_str()),
|
||||
};
|
||||
tool.plugin_display_names = plugin_names.to_vec();
|
||||
|
||||
if plugin_names.is_empty() {
|
||||
continue;
|
||||
}
|
||||
|
||||
let plugin_source_note = if plugin_names.len() == 1 {
|
||||
format!("This tool is part of plugin `{}`.", plugin_names[0])
|
||||
} else {
|
||||
format!(
|
||||
"This tool is part of plugins {}.",
|
||||
plugin_names
|
||||
.iter()
|
||||
.map(|plugin_name| format!("`{plugin_name}`"))
|
||||
.collect::<Vec<_>>()
|
||||
.join(", ")
|
||||
)
|
||||
};
|
||||
let description = tool
|
||||
.tool
|
||||
.description
|
||||
.as_deref()
|
||||
.map(str::trim)
|
||||
.unwrap_or("");
|
||||
let annotated_description = if description.is_empty() {
|
||||
plugin_source_note
|
||||
} else if matches!(description.chars().last(), Some('.' | '!' | '?')) {
|
||||
format!("{description} {plugin_source_note}")
|
||||
} else {
|
||||
format!("{description}. {plugin_source_note}")
|
||||
};
|
||||
tool.tool.description = Some(Cow::Owned(annotated_description));
|
||||
fn annotate_tools(&self, mut tools: Vec<ToolInfo>) -> Vec<ToolInfo> {
|
||||
for tool in &mut tools {
|
||||
if tool.server_name == CODEX_APPS_MCP_SERVER_NAME {
|
||||
tool.tool = tool_with_model_visible_input_schema(&tool.tool);
|
||||
}
|
||||
tools
|
||||
};
|
||||
|
||||
let plugin_names = match tool.connector_id.as_deref() {
|
||||
Some(connector_id) => self
|
||||
.tool_plugin_provenance
|
||||
.plugin_display_names_for_connector_id(connector_id),
|
||||
None => self
|
||||
.tool_plugin_provenance
|
||||
.plugin_display_names_for_mcp_server_name(tool.server_name.as_str()),
|
||||
};
|
||||
tool.plugin_display_names = plugin_names.to_vec();
|
||||
|
||||
if plugin_names.is_empty() {
|
||||
continue;
|
||||
}
|
||||
|
||||
let plugin_source_note = if plugin_names.len() == 1 {
|
||||
format!("This tool is part of plugin `{}`.", plugin_names[0])
|
||||
} else {
|
||||
format!(
|
||||
"This tool is part of plugins {}.",
|
||||
plugin_names
|
||||
.iter()
|
||||
.map(|plugin_name| format!("`{plugin_name}`"))
|
||||
.collect::<Vec<_>>()
|
||||
.join(", ")
|
||||
)
|
||||
};
|
||||
let description = tool
|
||||
.tool
|
||||
.description
|
||||
.as_deref()
|
||||
.map(str::trim)
|
||||
.unwrap_or("");
|
||||
let annotated_description = if description.is_empty() {
|
||||
plugin_source_note
|
||||
} else if matches!(description.chars().last(), Some('.' | '!' | '?')) {
|
||||
format!("{description} {plugin_source_note}")
|
||||
} else {
|
||||
format!("{description}. {plugin_source_note}")
|
||||
};
|
||||
tool.tool.description = Some(Cow::Owned(annotated_description));
|
||||
}
|
||||
tools
|
||||
}
|
||||
|
||||
pub(crate) async fn listed_tools(&self) -> Option<Vec<ToolInfo>> {
|
||||
// Keep cache payloads raw; plugin provenance is resolved per-session at read time.
|
||||
let tools = if let Some(startup_tools) = self.cached_tool_info_snapshot_while_initializing()
|
||||
{
|
||||
@@ -317,7 +316,17 @@ impl AsyncManagedClient {
|
||||
Err(_) => self.cached_tool_info_snapshot.clone(),
|
||||
}
|
||||
};
|
||||
tools.map(annotate_tools)
|
||||
tools.map(|tools| self.annotate_tools(tools))
|
||||
}
|
||||
|
||||
pub(crate) async fn listed_tools_if_ready_or_cached(&self) -> Option<Vec<ToolInfo>> {
|
||||
if let Some(startup_tools) = self.cached_tool_info_snapshot_while_initializing() {
|
||||
return Some(self.annotate_tools(startup_tools));
|
||||
}
|
||||
if !self.startup_complete.load(Ordering::Acquire) {
|
||||
return None;
|
||||
}
|
||||
self.listed_tools().await
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
Reference in New Issue
Block a user