mirror of
https://github.com/openai/codex.git
synced 2026-11-02 19:43:24 +00:00
Route hook MCP calls through current connections (#39331)
## What changed - Execute hook-triggered MCP calls through the runtime's latest connection set instead of preparing a separate catalog-bound call. - Keep hook calls from waiting for server startup or reconnecting, and reject disconnected servers immediately. - Apply the shorter of the hook-requested timeout and the server's configured tool timeout. ## Testing - Update connection manager tests to cover non-waiting calls and disabled tool filters. GitOrigin-RevId: 3166b2d1e26d9ae2b8b3a648b4d0897cc260eb93
This commit is contained in:
@@ -53,6 +53,7 @@ use crate::tools::filter_tools;
|
||||
use anyhow::Context;
|
||||
use anyhow::Result;
|
||||
use anyhow::anyhow;
|
||||
use anyhow::bail;
|
||||
use codex_config::McpServerTransportConfig;
|
||||
use codex_diagnostics::Gauge;
|
||||
use codex_diagnostics::GaugeGuard;
|
||||
@@ -862,6 +863,8 @@ impl McpConnectionSet {
|
||||
tool: &str,
|
||||
arguments: Option<serde_json::Value>,
|
||||
meta: Option<serde_json::Value>,
|
||||
requested_timeout: Option<Duration>,
|
||||
wait_for_server: bool,
|
||||
) -> Result<CallToolResult> {
|
||||
let view = self
|
||||
.servers
|
||||
@@ -872,14 +875,32 @@ impl McpConnectionSet {
|
||||
"tool '{tool}' is disabled for MCP server '{server}'"
|
||||
));
|
||||
}
|
||||
let client = view
|
||||
.connection
|
||||
.client()
|
||||
.await
|
||||
.context("failed to get client")?;
|
||||
let client = if wait_for_server {
|
||||
view.connection
|
||||
.client()
|
||||
.await
|
||||
.context("failed to get client")?
|
||||
} else {
|
||||
let client = view
|
||||
.connection
|
||||
.client
|
||||
.ready_client()
|
||||
.ok_or_else(|| anyhow!("MCP server '{server}' is not connected"))?;
|
||||
if client.client.is_closed().await {
|
||||
bail!("MCP server '{server}' is not connected");
|
||||
}
|
||||
client
|
||||
};
|
||||
|
||||
let effective_timeout = match (view.tool_timeout, requested_timeout) {
|
||||
(Some(server_timeout), Some(requested_timeout)) => {
|
||||
Some(server_timeout.min(requested_timeout))
|
||||
}
|
||||
(server_timeout, requested_timeout) => server_timeout.or(requested_timeout),
|
||||
};
|
||||
let result: rmcp::model::CallToolResult = client
|
||||
.client
|
||||
.call_tool(tool.to_string(), arguments, meta, view.tool_timeout)
|
||||
.call_tool(tool.to_string(), arguments, meta, effective_timeout)
|
||||
.await
|
||||
.with_context(|| format!("tool call failed for `{server}/{tool}`"))?;
|
||||
|
||||
|
||||
@@ -325,82 +325,6 @@ impl McpConnectionSet {
|
||||
)
|
||||
}
|
||||
|
||||
/// Captures an exact live client and filtered tool without starting or reconnecting a server.
|
||||
///
|
||||
/// The prepared call retains the current catalog revision so subsequent execution cannot
|
||||
/// silently drift to a different connection, tool catalog, or managed configuration.
|
||||
pub(crate) async fn prepare_connected_call(
|
||||
self: &Arc<Self>,
|
||||
config: Arc<crate::McpConfig>,
|
||||
server: &str,
|
||||
tool: &str,
|
||||
) -> Option<PreparedMcpCall> {
|
||||
// Only consider servers and tools allowed by the current configuration.
|
||||
let view = self.servers.get(server)?;
|
||||
if !view.tool_filter.allows(tool) {
|
||||
return None;
|
||||
}
|
||||
|
||||
// Reuse an existing connection without starting or reconnecting the server.
|
||||
let mut client = view.connection.client.ready_client()?;
|
||||
if client.client.is_closed().await {
|
||||
return None;
|
||||
}
|
||||
client.tool_timeout = view.tool_timeout;
|
||||
|
||||
// Snapshot the revision before reading tools so concurrent refreshes invalidate the call.
|
||||
let revision = *self.tool_catalog_revision.read().await;
|
||||
|
||||
// Codex Apps catalogs can refresh independently of their MCP connection.
|
||||
let server_tools = if server == CODEX_APPS_MCP_SERVER_NAME {
|
||||
self.codex_apps_tools_override
|
||||
.read()
|
||||
.await
|
||||
.clone()
|
||||
.unwrap_or_else(|| client.tools.clone())
|
||||
} else {
|
||||
client.tools.clone()
|
||||
};
|
||||
|
||||
// Filter and annotate tools using the same rules as the rest of the MCP runtime.
|
||||
let server_tools = filter_tools(server_tools, &view.tool_filter);
|
||||
let server_tools = if server == CODEX_APPS_MCP_SERVER_NAME {
|
||||
prepare_codex_apps_tools_for_model(server_tools, &self.tool_plugin_provenance)
|
||||
} else {
|
||||
crate::rmcp_client::prepare_regular_mcp_tools_for_model(
|
||||
server_tools,
|
||||
&self.tool_plugin_provenance,
|
||||
)
|
||||
};
|
||||
|
||||
// Look up the requested tool and attach its server's policy metadata.
|
||||
let tool_info = server_tools
|
||||
.into_iter()
|
||||
.find(|tool_info| tool_info.tool.name.as_ref() == tool)
|
||||
.map(|tool_info| Self::with_server_metadata(tool_info, &view.metadata))?;
|
||||
|
||||
// Reject tools from connectors that an administrator disabled in managed requirements.
|
||||
if server == CODEX_APPS_MCP_SERVER_NAME
|
||||
&& tool_info
|
||||
.connector_id
|
||||
.as_deref()
|
||||
.and_then(|connector_id| {
|
||||
config
|
||||
.config_layer_stack
|
||||
.requirements_toml()
|
||||
.apps
|
||||
.as_ref()?
|
||||
.apps
|
||||
.get(connector_id)
|
||||
})
|
||||
.is_some_and(|requirements| requirements.enabled == Some(false))
|
||||
{
|
||||
return None;
|
||||
}
|
||||
|
||||
self.prepare_call(&tool_info, Arc::new(client), config, revision)
|
||||
}
|
||||
|
||||
fn prepare_call(
|
||||
self: &Arc<Self>,
|
||||
tool_info: &ToolInfo,
|
||||
|
||||
@@ -22,20 +22,12 @@ use crate::tools::ToolFilter;
|
||||
use crate::tools::ToolInfo;
|
||||
use crate::tools::filter_tools;
|
||||
use crate::tools::normalize_tools_for_model_with_prefix;
|
||||
use codex_config::AbsolutePathBuf;
|
||||
use codex_config::AppRequirementToml;
|
||||
use codex_config::AppToolApproval;
|
||||
use codex_config::AppsRequirementsToml;
|
||||
use codex_config::CONFIG_TOML_FILE;
|
||||
use codex_config::ConfigLayerStack;
|
||||
use codex_config::ConfigRequirements;
|
||||
use codex_config::ConfigRequirementsToml;
|
||||
use codex_config::Constrained;
|
||||
use codex_config::McpServerAuth;
|
||||
use codex_config::McpServerConfig;
|
||||
use codex_config::McpServerEnvVar;
|
||||
use codex_config::McpServerToolConfig;
|
||||
use codex_config::TomlValue;
|
||||
use codex_config::types::AuthKeyringBackendKind;
|
||||
use codex_config::types::OAuthCredentialsStoreMode;
|
||||
use codex_connectors::ConnectorRuntimeContext;
|
||||
@@ -79,7 +71,6 @@ use rmcp::model::ServerCapabilities;
|
||||
use rmcp::model::ServerInfo;
|
||||
use rmcp::model::Tool;
|
||||
use rmcp::service::RequestContext;
|
||||
use std::collections::BTreeMap;
|
||||
use std::collections::HashMap;
|
||||
use std::collections::HashSet;
|
||||
use std::io;
|
||||
@@ -2646,7 +2637,7 @@ async fn list_all_tools_applies_legacy_mcp_prefix_by_default() {
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn prepare_connected_call_fails_immediately_until_server_startup_finishes() {
|
||||
async fn call_tool_requires_connection_without_waiting_for_startup() {
|
||||
let client = create_test_managed_client(vec![create_test_tool("docs", "search")]).await;
|
||||
let (client, startup_started, release_startup) = create_gated_async_managed_client(client);
|
||||
let startup_client = client.clone();
|
||||
@@ -2658,16 +2649,22 @@ async fn prepare_connected_call_fails_immediately_until_server_startup_finishes(
|
||||
/*prefix_mcp_tool_names*/ true,
|
||||
);
|
||||
manager.insert_test_client("docs", client);
|
||||
let manager = Arc::new(manager);
|
||||
let config = Arc::new(crate::mcp::tests::test_mcp_config(std::env::temp_dir()));
|
||||
|
||||
let pending_call = tokio::time::timeout(
|
||||
Duration::from_millis(50),
|
||||
manager.prepare_connected_call(Arc::clone(&config), "docs", "search"),
|
||||
manager.call_tool(
|
||||
"docs",
|
||||
"search",
|
||||
/*arguments*/ None,
|
||||
/*meta*/ None,
|
||||
Some(Duration::from_secs(5)),
|
||||
/*wait_for_server*/ false,
|
||||
),
|
||||
)
|
||||
.await
|
||||
.expect("ready-only preparation must not wait for pending server startup");
|
||||
assert!(pending_call.is_none());
|
||||
.expect("ready-only invocation must not wait for pending server startup")
|
||||
.expect_err("pending server must not accept ready-only calls");
|
||||
assert!(pending_call.to_string().contains("not connected"));
|
||||
|
||||
let startup = tokio::spawn(async move { startup_client.client().await });
|
||||
startup_started.await.expect("server startup should begin");
|
||||
@@ -2677,21 +2674,22 @@ async fn prepare_connected_call_fails_immediately_until_server_startup_finishes(
|
||||
.expect("startup task should finish")
|
||||
.expect("server startup should succeed");
|
||||
|
||||
let prepared_call = manager
|
||||
.prepare_connected_call(config, "docs", "search")
|
||||
let ready_error = manager
|
||||
.call_tool(
|
||||
"docs",
|
||||
"search",
|
||||
/*arguments*/ None,
|
||||
/*meta*/ None,
|
||||
Some(Duration::from_secs(5)),
|
||||
/*wait_for_server*/ false,
|
||||
)
|
||||
.await
|
||||
.expect("ready server should prepare its exact configured tool");
|
||||
assert_eq!(
|
||||
(
|
||||
prepared_call.server_name(),
|
||||
prepared_call.tool_info().tool.name.as_ref(),
|
||||
),
|
||||
("docs", "search")
|
||||
);
|
||||
.expect_err("ready server should reach the uninitialized test transport");
|
||||
assert!(format!("{ready_error:#}").contains("MCP client not initialized"));
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn prepare_connected_call_respects_server_tool_filters() {
|
||||
async fn connected_call_respects_server_tool_filters() {
|
||||
let client = create_ready_async_managed_client(vec![create_test_tool("docs", "search")]).await;
|
||||
client.client().await.expect("server should be ready");
|
||||
let approval_policy = Constrained::allow_any(AskForApproval::OnRequest);
|
||||
@@ -2710,135 +2708,18 @@ async fn prepare_connected_call_respects_server_tool_filters() {
|
||||
.disabled
|
||||
.insert("search".to_string());
|
||||
|
||||
assert!(
|
||||
Arc::new(manager)
|
||||
.prepare_connected_call(
|
||||
Arc::new(crate::mcp::tests::test_mcp_config(std::env::temp_dir())),
|
||||
"docs",
|
||||
"search",
|
||||
)
|
||||
.await
|
||||
.is_none()
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn prepare_connected_call_captures_revision_before_codex_apps_catalog() {
|
||||
let client = create_ready_async_managed_client(vec![create_test_tool(
|
||||
CODEX_APPS_MCP_SERVER_NAME,
|
||||
"search",
|
||||
)])
|
||||
.await;
|
||||
client.client().await.expect("server should be ready");
|
||||
let approval_policy = Constrained::allow_any(AskForApproval::OnRequest);
|
||||
let permission_profile = Constrained::allow_any(PermissionProfile::default());
|
||||
let mut manager = McpConnectionSet::new_uninitialized(
|
||||
&approval_policy,
|
||||
&permission_profile,
|
||||
/*prefix_mcp_tool_names*/ true,
|
||||
);
|
||||
manager.insert_test_client(CODEX_APPS_MCP_SERVER_NAME, client);
|
||||
let manager = Arc::new(manager);
|
||||
let config = Arc::new(crate::mcp::tests::test_mcp_config(std::env::temp_dir()));
|
||||
|
||||
let catalog_override = manager.codex_apps_tools_override.write().await;
|
||||
let mut preparation =
|
||||
Box::pin(manager.prepare_connected_call(config, CODEX_APPS_MCP_SERVER_NAME, "search"));
|
||||
|
||||
assert!(
|
||||
preparation.as_mut().now_or_never().is_none(),
|
||||
"preparation should wait for the Codex Apps catalog"
|
||||
);
|
||||
*manager
|
||||
.tool_catalog_revision
|
||||
.try_write()
|
||||
.expect("catalog lookup must not hold the revision lock across an await") += 1;
|
||||
|
||||
drop(catalog_override);
|
||||
let prepared_call = preparation
|
||||
.await
|
||||
.expect("ready server should prepare its exact configured tool");
|
||||
let error = prepared_call
|
||||
.call(
|
||||
/*arguments*/ None, /*meta*/ None, /*timeout*/ None,
|
||||
let filtered_error = manager
|
||||
.call_tool(
|
||||
"docs",
|
||||
"search",
|
||||
/*arguments*/ None,
|
||||
/*meta*/ None,
|
||||
Some(Duration::from_secs(5)),
|
||||
/*wait_for_server*/ false,
|
||||
)
|
||||
.await
|
||||
.expect_err("catalog refresh during lookup must invalidate the prepared call");
|
||||
assert!(error.to_string().contains("catalog changed"));
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn prepare_connected_call_rejects_only_administrator_disabled_connectors() {
|
||||
let mut administrator_disabled_tool =
|
||||
create_test_tool(CODEX_APPS_MCP_SERVER_NAME, "administrator_disabled");
|
||||
administrator_disabled_tool.connector_id = Some("administrator_disabled".to_string());
|
||||
let mut user_disabled_tool = create_test_tool(CODEX_APPS_MCP_SERVER_NAME, "user_disabled");
|
||||
user_disabled_tool.connector_id = Some("user_disabled".to_string());
|
||||
let client =
|
||||
create_ready_async_managed_client(vec![administrator_disabled_tool, user_disabled_tool])
|
||||
.await;
|
||||
client.client().await.expect("server should be ready");
|
||||
|
||||
let approval_policy = Constrained::allow_any(AskForApproval::OnRequest);
|
||||
let permission_profile = Constrained::allow_any(PermissionProfile::default());
|
||||
let mut manager = McpConnectionSet::new_uninitialized(
|
||||
&approval_policy,
|
||||
&permission_profile,
|
||||
/*prefix_mcp_tool_names*/ true,
|
||||
);
|
||||
manager.insert_test_client(CODEX_APPS_MCP_SERVER_NAME, client);
|
||||
let manager = Arc::new(manager);
|
||||
|
||||
let requirements = ConfigRequirementsToml {
|
||||
apps: Some(AppsRequirementsToml {
|
||||
apps: BTreeMap::from([(
|
||||
"administrator_disabled".to_string(),
|
||||
AppRequirementToml {
|
||||
enabled: Some(false),
|
||||
tools: None,
|
||||
},
|
||||
)]),
|
||||
}),
|
||||
..Default::default()
|
||||
};
|
||||
let config_layer_stack =
|
||||
ConfigLayerStack::new(Vec::new(), ConfigRequirements::default(), requirements)
|
||||
.expect("config layer stack");
|
||||
let user_config: TomlValue = serde_json::from_value(serde_json::json!({
|
||||
"apps": {
|
||||
"user_disabled": {
|
||||
"enabled": false,
|
||||
"tools": {
|
||||
"user_disabled": { "enabled": false }
|
||||
}
|
||||
}
|
||||
}
|
||||
}))
|
||||
.expect("user app configuration");
|
||||
let config_toml_path = AbsolutePathBuf::try_from(std::env::temp_dir().join(CONFIG_TOML_FILE))
|
||||
.expect("absolute config path");
|
||||
let mut config = crate::mcp::tests::test_mcp_config(std::env::temp_dir());
|
||||
config.config_layer_stack = config_layer_stack
|
||||
.with_user_config(&config_toml_path, user_config)
|
||||
.expect("user app configuration should be valid");
|
||||
let config = Arc::new(config);
|
||||
|
||||
assert!(
|
||||
manager
|
||||
.prepare_connected_call(
|
||||
Arc::clone(&config),
|
||||
CODEX_APPS_MCP_SERVER_NAME,
|
||||
"administrator_disabled",
|
||||
)
|
||||
.await
|
||||
.is_none()
|
||||
);
|
||||
assert!(
|
||||
manager
|
||||
.prepare_connected_call(config, CODEX_APPS_MCP_SERVER_NAME, "user_disabled")
|
||||
.await
|
||||
.is_some()
|
||||
);
|
||||
.expect_err("disabled tools should not be callable");
|
||||
assert!(filtered_error.to_string().contains("disabled"));
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
|
||||
@@ -45,7 +45,6 @@ use tokio_util::sync::CancellationToken;
|
||||
|
||||
use crate::McpConfig;
|
||||
use crate::binding::McpBinding;
|
||||
use crate::binding::PreparedMcpCall;
|
||||
use crate::connection_manager::McpConnectionSet;
|
||||
use crate::elicitation::ElicitationLifecycle;
|
||||
use crate::elicitation::ElicitationRequestRouter;
|
||||
@@ -409,24 +408,6 @@ impl McpRuntime {
|
||||
Self::binding_from_published_runtime(current, /*required_servers*/ &[]).await
|
||||
}
|
||||
|
||||
/// Prepares an exact MCP tool call only when its server is already connected.
|
||||
///
|
||||
/// Missing, disconnected, filtered, or uncataloged tools return `None` immediately rather
|
||||
/// than starting a server, waiting for startup, initiating OAuth, or reconnecting. Successful
|
||||
/// calls capture the live client, effective tool policy, and current catalog revision together.
|
||||
pub async fn prepare_call_if_connected(
|
||||
&self,
|
||||
server: &str,
|
||||
tool: &str,
|
||||
) -> Option<PreparedMcpCall> {
|
||||
let current = self.current.load_full();
|
||||
let config = Arc::clone(current.config.as_ref()?);
|
||||
current
|
||||
.connections
|
||||
.prepare_connected_call(config, server, tool)
|
||||
.await
|
||||
}
|
||||
|
||||
/// Returns the latest published configuration without waiting for clients.
|
||||
pub fn current_config(&self) -> Option<Arc<McpConfig>> {
|
||||
self.current.load().config.clone()
|
||||
@@ -481,9 +462,18 @@ impl McpRuntime {
|
||||
tool: &str,
|
||||
arguments: Option<serde_json::Value>,
|
||||
meta: Option<serde_json::Value>,
|
||||
requested_timeout: Option<Duration>,
|
||||
wait_for_server: bool,
|
||||
) -> anyhow::Result<CallToolResult> {
|
||||
self.latest_connections()
|
||||
.call_tool(server, tool, arguments, meta)
|
||||
.call_tool(
|
||||
server,
|
||||
tool,
|
||||
arguments,
|
||||
meta,
|
||||
requested_timeout,
|
||||
wait_for_server,
|
||||
)
|
||||
.await
|
||||
}
|
||||
|
||||
|
||||
@@ -775,7 +775,10 @@ impl CodexThread {
|
||||
self.session
|
||||
.services
|
||||
.mcp_runtime
|
||||
.latest_call_tool(server, tool, arguments, meta)
|
||||
.latest_call_tool(
|
||||
server, tool, arguments, meta, /*requested_timeout*/ None,
|
||||
/*wait_for_server*/ true,
|
||||
)
|
||||
.await
|
||||
}
|
||||
|
||||
|
||||
@@ -1,6 +1,5 @@
|
||||
use std::sync::Arc;
|
||||
|
||||
use anyhow::Context;
|
||||
use anyhow::bail;
|
||||
use codex_hooks::HookMcpCall;
|
||||
use codex_hooks::HookMcpExecutor;
|
||||
@@ -19,22 +18,15 @@ pub(crate) struct CoreHookMcpExecutor {
|
||||
impl HookMcpExecutor for CoreHookMcpExecutor {
|
||||
fn execute(&self, call: HookMcpCall) -> BoxFuture<'_, anyhow::Result<String>> {
|
||||
async move {
|
||||
let prepared_call = self
|
||||
let result = self
|
||||
.runtime
|
||||
.prepare_call_if_connected(&call.server, &call.tool)
|
||||
.await
|
||||
.with_context(|| {
|
||||
format!(
|
||||
"MCP server `{}` or tool `{}` is not connected and available",
|
||||
call.server, call.tool
|
||||
)
|
||||
})?;
|
||||
|
||||
let result = prepared_call
|
||||
.call(
|
||||
.latest_call_tool(
|
||||
&call.server,
|
||||
&call.tool,
|
||||
Some(Value::Object(call.input)),
|
||||
Some(serde_json::json!({ "threadId": self.thread_id.to_string() })),
|
||||
Some(call.timeout),
|
||||
/*wait_for_server*/ false,
|
||||
)
|
||||
.await?;
|
||||
let text = result
|
||||
|
||||
@@ -3594,6 +3594,8 @@ impl Session {
|
||||
Some(serde_json::json!({
|
||||
"threadId": self.thread_id().to_string(),
|
||||
})),
|
||||
/*requested_timeout*/ None,
|
||||
/*wait_for_server*/ true,
|
||||
)
|
||||
.await
|
||||
.ok()
|
||||
|
||||
@@ -138,6 +138,8 @@ async fn hosted_plugin_runtime_ps_mcp_tool_calls_use_current_auth_manager_token(
|
||||
"starts_at": "2026-06-18T12:00:00Z",
|
||||
})),
|
||||
/*meta*/ None,
|
||||
/*requested_timeout*/ None,
|
||||
/*wait_for_server*/ true,
|
||||
)
|
||||
.await?;
|
||||
assert_eq!(tool_result.is_error, Some(false));
|
||||
|
||||
Reference in New Issue
Block a user