From 2c2fff2b431d0428f16ff1ffa6cd83ee4888b86a Mon Sep 17 00:00:00 2001 From: Felipe Coury Date: Thu, 28 May 2026 15:36:05 -0300 Subject: [PATCH] fix(core): await selected skill mcp dependencies --- codex-rs/core/src/mcp_skill_dependencies.rs | 89 ++++++++++----- codex-rs/core/src/session/turn.rs | 73 +++++++----- codex-rs/core/tests/suite/skills.rs | 116 ++++++++++++++++++++ 3 files changed, 221 insertions(+), 57 deletions(-) diff --git a/codex-rs/core/src/mcp_skill_dependencies.rs b/codex-rs/core/src/mcp_skill_dependencies.rs index 936f1b5fb5..092e446589 100644 --- a/codex-rs/core/src/mcp_skill_dependencies.rs +++ b/codex-rs/core/src/mcp_skill_dependencies.rs @@ -37,49 +37,54 @@ pub(crate) async fn maybe_prompt_and_install_mcp_dependencies( cancellation_token: &CancellationToken, mentioned_skills: &[SkillMetadata], elicitation_reviewer: Option, -) { +) -> HashSet { + if mentioned_skills.is_empty() { + return HashSet::new(); + } + let originator_value = originator().value; - if !is_first_party_originator(originator_value.as_str()) { - // Only support first-party clients for now. - return; - } - let config = turn_context.config.clone(); - if mentioned_skills.is_empty() - || !config - .features - .enabled(codex_features::Feature::SkillMcpDependencyInstall) - { - return; - } - let installed = sess .services .mcp_manager .configured_servers(config.as_ref()) .await; + let mut dependency_servers = + collect_configured_mcp_dependency_server_names(mentioned_skills, &installed); + if !is_first_party_originator(originator_value.as_str()) + || !config + .features + .enabled(codex_features::Feature::SkillMcpDependencyInstall) + { + // Existing skill dependencies remain explicit even if auto-install is unavailable. + return dependency_servers; + } + let missing = collect_missing_mcp_dependencies(mentioned_skills, &installed); if missing.is_empty() { - return; + return dependency_servers; } let unprompted_missing = filter_prompted_mcp_dependencies(sess, &missing).await; if unprompted_missing.is_empty() { - return; + return dependency_servers; } if should_install_mcp_dependencies(sess, turn_context, &unprompted_missing, cancellation_token) .await { - maybe_install_mcp_dependencies( - sess, - turn_context, - config.as_ref(), - mentioned_skills, - elicitation_reviewer, - ) - .await; + dependency_servers.extend( + maybe_install_mcp_dependencies( + sess, + turn_context, + config.as_ref(), + mentioned_skills, + elicitation_reviewer, + ) + .await, + ); } + dependency_servers } pub(crate) async fn maybe_install_mcp_dependencies( @@ -88,27 +93,27 @@ pub(crate) async fn maybe_install_mcp_dependencies( config: &crate::config::Config, mentioned_skills: &[SkillMetadata], elicitation_reviewer: Option, -) { +) -> HashSet { if mentioned_skills.is_empty() || !config .features .enabled(codex_features::Feature::SkillMcpDependencyInstall) { - return; + return HashSet::new(); } let codex_home = config.codex_home.clone(); let installed = sess.services.mcp_manager.configured_servers(config).await; let missing = collect_missing_mcp_dependencies(mentioned_skills, &installed); if missing.is_empty() { - return; + return HashSet::new(); } let mut servers = match load_global_mcp_servers(&codex_home).await { Ok(servers) => servers, Err(err) => { warn!("failed to load MCP servers while installing skill dependencies: {err}"); - return; + return HashSet::new(); } }; @@ -124,7 +129,7 @@ pub(crate) async fn maybe_install_mcp_dependencies( } if !updated { - return; + return HashSet::new(); } if let Err(err) = ConfigEditsBuilder::new(&codex_home) @@ -133,9 +138,13 @@ pub(crate) async fn maybe_install_mcp_dependencies( .await { warn!("failed to persist MCP dependencies for mentioned skills: {err}"); - return; + return HashSet::new(); } + let added_server_names = added + .iter() + .map(|(name, _)| name.clone()) + .collect::>(); for (name, server_config) in added { let oauth_config = match oauth_login_support(&server_config.transport).await { McpOAuthLoginSupport::Supported(config) => config, @@ -207,6 +216,7 @@ pub(crate) async fn maybe_install_mcp_dependencies( elicitation_reviewer, ) .await; + added_server_names } async fn should_install_mcp_dependencies( @@ -411,6 +421,25 @@ fn mcp_dependency_to_server_config( Err(format!("unsupported transport {transport}")) } +fn collect_configured_mcp_dependency_server_names( + mentioned_skills: &[SkillMetadata], + installed: &HashMap, +) -> HashSet { + let dependency_keys = mentioned_skills + .iter() + .filter_map(|skill| skill.dependencies.as_ref()) + .flat_map(|dependencies| dependencies.tools.iter()) + .filter(|tool| tool.r#type.eq_ignore_ascii_case("mcp")) + .filter_map(|tool| canonical_mcp_dependency_key(tool).ok()) + .collect::>(); + + installed + .iter() + .filter(|(name, config)| dependency_keys.contains(&canonical_mcp_server_key(name, config))) + .map(|(name, _)| name.clone()) + .collect() +} + fn collect_missing_mcp_dependencies( mentioned_skills: &[SkillMetadata], installed: &HashMap, diff --git a/codex-rs/core/src/session/turn.rs b/codex-rs/core/src/session/turn.rs index 281413d73b..56a4d7ee07 100644 --- a/codex-rs/core/src/session/turn.rs +++ b/codex-rs/core/src/session/turn.rs @@ -509,32 +509,13 @@ async fn build_skills_and_plugins( { explicitly_requested_mcp_servers.insert(CODEX_APPS_MCP_SERVER_NAME.to_string()); } - if !explicitly_requested_mcp_servers.is_empty() { - let mut explicitly_requested_mcp_servers = explicitly_requested_mcp_servers - .into_iter() - .collect::>(); - explicitly_requested_mcp_servers.sort(); - let readiness = { - let mcp_connection_manager = sess.services.mcp_connection_manager.read().await; - mcp_connection_manager.wait_for_servers_ready(&explicitly_requested_mcp_servers) - }; - let failures = match readiness.or_cancel(cancellation_token).await { - Ok(failures) => failures, - Err(_) => return None, - }; - for failure in failures { - sess.send_event( - turn_context, - EventMsg::Warning(WarningEvent { - message: format!( - "MCP dependency `{}` requested for this turn is unavailable: {}", - failure.server, failure.error - ), - }), - ) - .await; - } - } + wait_for_explicit_mcp_servers( + sess, + turn_context, + cancellation_token, + explicitly_requested_mcp_servers, + ) + .await?; let mcp_tools = if turn_context.apps_enabled() || !mentioned_plugins.is_empty() { // Plugin mentions need raw MCP/app inventory even when app tools // are normally hidden so we can describe the plugin's currently @@ -579,7 +560,7 @@ async fn build_skills_and_plugins( &skills_outcome.disabled_paths, &connector_slug_counts, ); - maybe_prompt_and_install_mcp_dependencies( + let skill_dependency_servers = maybe_prompt_and_install_mcp_dependencies( sess, turn_context, cancellation_token, @@ -587,6 +568,13 @@ async fn build_skills_and_plugins( Some(sess.mcp_elicitation_reviewer()), ) .await; + wait_for_explicit_mcp_servers( + sess, + turn_context, + cancellation_token, + skill_dependency_servers, + ) + .await?; let SkillInjections { items: skill_injections, @@ -701,6 +689,37 @@ async fn build_extension_turn_input_items( Some(items) } +async fn wait_for_explicit_mcp_servers( + sess: &Session, + turn_context: &TurnContext, + cancellation_token: &CancellationToken, + servers: HashSet, +) -> Option<()> { + if servers.is_empty() { + return Some(()); + } + let mut servers = servers.into_iter().collect::>(); + servers.sort(); + let readiness = { + let mcp_connection_manager = sess.services.mcp_connection_manager.read().await; + mcp_connection_manager.wait_for_servers_ready(&servers) + }; + let failures = readiness.or_cancel(cancellation_token).await.ok()?; + for failure in failures { + sess.send_event( + turn_context, + EventMsg::Warning(WarningEvent { + message: format!( + "MCP dependency `{}` requested for this turn is unavailable: {}", + failure.server, failure.error + ), + }), + ) + .await; + } + Some(()) +} + async fn track_turn_resolved_config_analytics( sess: &Session, turn_context: &TurnContext, diff --git a/codex-rs/core/tests/suite/skills.rs b/codex-rs/core/tests/suite/skills.rs index e21668fa84..b239fd9982 100644 --- a/codex-rs/core/tests/suite/skills.rs +++ b/codex-rs/core/tests/suite/skills.rs @@ -2,6 +2,8 @@ #![allow(clippy::unwrap_used, clippy::expect_used)] use anyhow::Result; +use codex_config::types::McpServerConfig; +use codex_config::types::McpServerTransportConfig; use codex_exec_server::CreateDirectoryOptions; use codex_exec_server::ExecutorFileSystem; use codex_protocol::models::PermissionProfile; @@ -9,6 +11,7 @@ use codex_protocol::protocol::AskForApproval; use codex_protocol::protocol::Op; use codex_protocol::user_input::UserInput; use codex_utils_absolute_path::AbsolutePathBuf; +use core_test_support::apps_test_server::AppsTestServer; use core_test_support::responses::ev_assistant_message; use core_test_support::responses::ev_completed; use core_test_support::responses::ev_response_created; @@ -19,6 +22,7 @@ use core_test_support::skip_if_no_network; use core_test_support::test_codex::test_codex; use core_test_support::test_codex::turn_permission_fields; use std::sync::Arc; +use std::time::Duration; async fn write_repo_skill( cwd: AbsolutePathBuf, @@ -127,3 +131,115 @@ async fn user_turn_includes_skill_instructions() -> Result<()> { Ok(()) } + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn selected_skill_waits_for_configured_mcp_dependency_startup() -> Result<()> { + skip_if_no_network!(Ok(())); + + let server = start_mock_server().await; + AppsTestServer::mount_with_connector_name_and_tools_list_delay( + &server, + "Dependency", + Some(Duration::from_secs(/*secs*/ 2)), + ) + .await?; + let dependency_url = format!("{}/api/codex/apps", server.uri()); + let skill_dependency_url = dependency_url.clone(); + let mut builder = test_codex() + .with_workspace_setup(move |cwd, fs| { + let skill_dependency_url = skill_dependency_url; + async move { + write_repo_skill(cwd.clone(), Arc::clone(&fs), "demo", "demo skill", "body") + .await?; + let agents_dir = cwd.join(".agents/skills/demo/agents"); + fs.create_directory( + &agents_dir, + CreateDirectoryOptions { recursive: true }, + /*sandbox*/ None, + ) + .await?; + let metadata = format!( + "dependencies:\n tools:\n - type: \"mcp\"\n value: \"dependency\"\n transport: \"streamable_http\"\n url: \"{skill_dependency_url}\"\n" + ); + fs.write_file( + &agents_dir.join("openai.yaml"), + metadata.into_bytes(), + /*sandbox*/ None, + ) + .await?; + Ok(()) + } + }) + .with_config(move |config| { + let mut servers = config.mcp_servers.get().clone(); + servers.insert( + "dependency".to_string(), + McpServerConfig { + transport: McpServerTransportConfig::StreamableHttp { + url: dependency_url, + bearer_token_env_var: None, + http_headers: None, + env_http_headers: None, + }, + environment_id: "local".to_string(), + enabled: true, + required: false, + supports_parallel_tool_calls: false, + disabled_reason: None, + startup_timeout_sec: Some(Duration::from_secs(/*secs*/ 10)), + tool_timeout_sec: None, + default_tools_approval_mode: None, + enabled_tools: None, + disabled_tools: None, + scopes: None, + oauth: None, + oauth_resource: None, + tools: Default::default(), + }, + ); + config + .mcp_servers + .set(servers) + .expect("test mcp servers should accept any configuration"); + }); + let test = builder.build_with_remote_env(&server).await?; + let skill_path = test + .config + .cwd + .join(".agents/skills/demo/SKILL.md") + .canonicalize() + .unwrap_or_else(|_| test.config.cwd.join(".agents/skills/demo/SKILL.md")) + .to_path_buf(); + let mock = mount_sse_once( + &server, + sse(vec![ev_response_created("resp-1"), ev_completed("resp-1")]), + ) + .await; + + test.codex + .submit(Op::UserInput { + items: vec![UserInput::Skill { + name: "demo".to_string(), + path: skill_path, + }], + environments: None, + final_output_json_schema: None, + responsesapi_client_metadata: None, + additional_context: Default::default(), + thread_settings: Default::default(), + }) + .await?; + core_test_support::wait_for_event(test.codex.as_ref(), |event| { + matches!(event, codex_protocol::protocol::EventMsg::TurnComplete(_)) + }) + .await; + + assert!( + mock.single_request() + .tool_by_name("mcp__dependency", "calendar_create_event") + .is_some(), + "expected selected skill MCP dependency tool on the first turn" + ); + + Ok(()) +}