mirror of
https://github.com/openai/codex.git
synced 2026-10-21 05:22:21 +00:00
Cache tool catalogs for streamable HTTP MCP servers (#37970)
## Why Subagents should be able to use known HTTP MCP tool definitions without opening a connection until a tool is actually called. ## What changed - Add streamable HTTP configurations to the process-scoped MCP tool catalog cache when their authentication identity can be derived safely. - Fingerprint transport settings, relevant environment variables, protocol mode, plugin status, and client capabilities so catalogs are reused only across equivalent connections. - Keep OAuth and other dynamically resolved credential configurations out of the shared cache. ## Testing - Add an integration test proving that a subagent receives cached HTTP tools without reinitializing the server, then starts its own connection when it calls one. GitOrigin-RevId: bb2787e29e9b71993def31aafd1f0d8f1b728d26
This commit is contained in:
@@ -462,8 +462,12 @@ impl McpConnectionSet {
|
||||
&configured_config,
|
||||
&runtime_context,
|
||||
environment.as_ref(),
|
||||
&client_elicitation_capability,
|
||||
&client_mcp_extensions,
|
||||
(&client_elicitation_capability, &client_mcp_extensions),
|
||||
Some((
|
||||
&connection_identity,
|
||||
protocol_mode,
|
||||
server.is_agent_plugin(),
|
||||
)),
|
||||
)
|
||||
} else {
|
||||
None
|
||||
|
||||
@@ -1993,8 +1993,11 @@ async fn tool_catalog_cache_sanitizes_tools_and_tracks_environment_generation()
|
||||
&config,
|
||||
&runtime_context,
|
||||
Some(environment),
|
||||
&ElicitationCapability::default(),
|
||||
&ClientMcpExtensions::default(),
|
||||
(
|
||||
&ElicitationCapability::default(),
|
||||
&ClientMcpExtensions::default(),
|
||||
),
|
||||
/*connection_identity*/ None,
|
||||
)
|
||||
.expect("cache context")
|
||||
};
|
||||
@@ -2056,8 +2059,11 @@ fn tool_catalog_cache_bypasses_remote_sourced_environment_variables() {
|
||||
&config,
|
||||
&runtime_context,
|
||||
/*resolved_environment*/ None,
|
||||
&ElicitationCapability::default(),
|
||||
&ClientMcpExtensions::default(),
|
||||
(
|
||||
&ElicitationCapability::default(),
|
||||
&ClientMcpExtensions::default()
|
||||
),
|
||||
/*connection_identity*/ None,
|
||||
)
|
||||
.is_none()
|
||||
);
|
||||
@@ -2374,8 +2380,11 @@ async fn capture_binding_shares_optional_startup_grace_across_connection_sets()
|
||||
&server_config,
|
||||
&runtime_context,
|
||||
/*resolved_environment*/ None,
|
||||
&ElicitationCapability::default(),
|
||||
&ClientMcpExtensions::default(),
|
||||
(
|
||||
&ElicitationCapability::default(),
|
||||
&ClientMcpExtensions::default(),
|
||||
),
|
||||
/*connection_identity*/ None,
|
||||
)
|
||||
.expect("shared pending MCP catalog");
|
||||
|
||||
@@ -3579,12 +3588,6 @@ async fn no_local_runtime_fails_local_stdio_but_keeps_local_http_server() {
|
||||
|
||||
assert!(manager.contains_server("stdio"));
|
||||
assert!(manager.contains_server("http"));
|
||||
assert!(
|
||||
manager
|
||||
.test_client("http")
|
||||
.tool_catalog_cache_context
|
||||
.is_none()
|
||||
);
|
||||
assert!(
|
||||
!manager
|
||||
.wait_for_server_ready("stdio", Duration::from_millis(10))
|
||||
|
||||
@@ -12,6 +12,7 @@ use std::sync::atomic::AtomicU64;
|
||||
use std::sync::atomic::Ordering;
|
||||
use std::time::Duration;
|
||||
|
||||
use codex_config::McpServerAuth;
|
||||
use codex_config::McpServerConfig;
|
||||
use codex_config::McpServerTransportConfig;
|
||||
use codex_exec_server::Environment;
|
||||
@@ -22,8 +23,11 @@ use sha1::Digest;
|
||||
use sha1::Sha1;
|
||||
use tokio::time::Instant;
|
||||
|
||||
use crate::McpProtocolMode;
|
||||
use crate::McpRuntimeContext;
|
||||
use crate::ToolInfo;
|
||||
use crate::server::McpServerConnectionIdentity;
|
||||
use crate::server::has_explicit_http_authorization;
|
||||
|
||||
const TOOL_CATALOG_CACHE_CAPACITY: usize = 32;
|
||||
const TOOL_CATALOG_CACHE_TTL: Duration = Duration::from_secs(30 * 60);
|
||||
@@ -78,16 +82,16 @@ impl McpToolCatalogCache {
|
||||
config: &McpServerConfig,
|
||||
runtime_context: &McpRuntimeContext,
|
||||
resolved_environment: Option<&Arc<Environment>>,
|
||||
client_elicitation_capability: &ElicitationCapability,
|
||||
client_mcp_extensions: &ClientMcpExtensions,
|
||||
client_context: (&ElicitationCapability, &ClientMcpExtensions),
|
||||
connection_identity: Option<(&McpServerConnectionIdentity, McpProtocolMode, bool)>,
|
||||
) -> Option<McpToolCatalogCacheContext> {
|
||||
let identity = ToolCatalogIdentity::new(
|
||||
server_name,
|
||||
config,
|
||||
runtime_context,
|
||||
resolved_environment,
|
||||
client_elicitation_capability,
|
||||
client_mcp_extensions,
|
||||
client_context,
|
||||
connection_identity,
|
||||
)?;
|
||||
let entry = lock_unpoisoned(&self.entries)
|
||||
.get_or_insert(identity, || Arc::new(ToolCatalogCacheEntry::default()))
|
||||
@@ -218,14 +222,11 @@ impl ToolCatalogIdentity {
|
||||
config: &McpServerConfig,
|
||||
runtime_context: &McpRuntimeContext,
|
||||
environment: Option<&Arc<Environment>>,
|
||||
client_elicitation_capability: &ElicitationCapability,
|
||||
client_mcp_extensions: &ClientMcpExtensions,
|
||||
client_context: (&ElicitationCapability, &ClientMcpExtensions),
|
||||
connection_identity: Option<(&McpServerConnectionIdentity, McpProtocolMode, bool)>,
|
||||
) -> Option<Self> {
|
||||
let transport = ToolCatalogTransportIdentity::new(
|
||||
config,
|
||||
client_elicitation_capability,
|
||||
client_mcp_extensions,
|
||||
)?;
|
||||
let transport =
|
||||
ToolCatalogTransportIdentity::new(config, client_context, connection_identity)?;
|
||||
Some(Self {
|
||||
server_name: server_name.to_string(),
|
||||
transport,
|
||||
@@ -242,14 +243,71 @@ impl ToolCatalogIdentity {
|
||||
#[derive(PartialEq, Eq, Hash)]
|
||||
enum ToolCatalogTransportIdentity {
|
||||
Stdio { fingerprint: [u8; 20] },
|
||||
StreamableHttp { fingerprint: [u8; 20] },
|
||||
}
|
||||
|
||||
impl ToolCatalogTransportIdentity {
|
||||
fn new(
|
||||
config: &McpServerConfig,
|
||||
client_elicitation_capability: &ElicitationCapability,
|
||||
client_mcp_extensions: &ClientMcpExtensions,
|
||||
client_context: (&ElicitationCapability, &ClientMcpExtensions),
|
||||
connection_identity: Option<(&McpServerConnectionIdentity, McpProtocolMode, bool)>,
|
||||
) -> Option<Self> {
|
||||
let (client_elicitation_capability, client_mcp_extensions) = client_context;
|
||||
if let McpServerTransportConfig::StreamableHttp {
|
||||
url,
|
||||
bearer_token_env_var,
|
||||
http_headers,
|
||||
env_http_headers,
|
||||
} = &config.transport
|
||||
{
|
||||
let (connection_identity, protocol_mode, agent_plugin) = connection_identity?;
|
||||
if config.oauth.is_some()
|
||||
|| config.scopes.is_some()
|
||||
|| config.oauth_resource.is_some()
|
||||
|| (matches!(config.auth, McpServerAuth::ChatGpt)
|
||||
&& !has_explicit_http_authorization(config))
|
||||
|| (!has_explicit_http_authorization(config)
|
||||
&& connection_identity.oauth_credentials().ok()?.is_some())
|
||||
{
|
||||
return None;
|
||||
}
|
||||
|
||||
let mut hasher = Sha1::new();
|
||||
hasher.update(
|
||||
serde_json::to_vec(&(
|
||||
url,
|
||||
bearer_token_env_var,
|
||||
http_headers
|
||||
.as_ref()
|
||||
.map(|headers| headers.iter().collect::<BTreeMap<_, _>>()),
|
||||
env_http_headers
|
||||
.as_ref()
|
||||
.map(|headers| headers.iter().collect::<BTreeMap<_, _>>()),
|
||||
&config.auth,
|
||||
&config.environment_id,
|
||||
agent_plugin,
|
||||
protocol_mode.preferred_protocol_version().as_str(),
|
||||
client_elicitation_capability,
|
||||
client_mcp_extensions.iter().collect::<BTreeMap<_, _>>(),
|
||||
))
|
||||
.ok()?,
|
||||
);
|
||||
let mut env_vars = bearer_token_env_var
|
||||
.iter()
|
||||
.chain(env_http_headers.iter().flat_map(|headers| headers.values()))
|
||||
.collect::<Vec<_>>();
|
||||
env_vars.sort_unstable();
|
||||
env_vars.dedup();
|
||||
for name in env_vars {
|
||||
hasher.update(name.as_bytes());
|
||||
let mut value_hasher = DefaultHasher::new();
|
||||
std::env::var_os(name).hash(&mut value_hasher);
|
||||
hasher.update(value_hasher.finish().to_le_bytes());
|
||||
}
|
||||
return Some(Self::StreamableHttp {
|
||||
fingerprint: hasher.finalize().into(),
|
||||
});
|
||||
}
|
||||
let McpServerTransportConfig::Stdio {
|
||||
command,
|
||||
args,
|
||||
@@ -258,7 +316,6 @@ impl ToolCatalogTransportIdentity {
|
||||
cwd,
|
||||
} = &config.transport
|
||||
else {
|
||||
// HTTP catalogs need a canonical resolved-auth identity before they can be shared.
|
||||
return None;
|
||||
};
|
||||
if env_vars
|
||||
|
||||
@@ -18,6 +18,7 @@ use codex_protocol::protocol::SessionSource;
|
||||
use codex_protocol::protocol::SubAgentSource;
|
||||
use codex_protocol::user_input::UserInput;
|
||||
use codex_utils_path_uri::PathUri;
|
||||
use core_test_support::apps_test_server::AppsTestServer;
|
||||
use core_test_support::responses;
|
||||
use core_test_support::responses::ResponseMock;
|
||||
use core_test_support::responses::mount_sse_once;
|
||||
@@ -245,6 +246,106 @@ async fn mcp_calls_stay_bound_to_each_thread() -> anyhow::Result<()> {
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
|
||||
async fn cached_http_mcp_starts_lazily_for_subagents() -> anyhow::Result<()> {
|
||||
skip_if_no_network!(Ok(()));
|
||||
|
||||
let responses_server = responses::start_mock_server().await;
|
||||
let (http_server, startup_control) =
|
||||
AppsTestServer::mount_with_startup_control(&responses_server).await?;
|
||||
let server_url = format!("{}/api/codex/ps/mcp", http_server.chatgpt_base_url);
|
||||
let fixture = test_codex()
|
||||
.with_model_info_override("gpt-5.4", |model| model.supports_search_tool = false)
|
||||
.with_config(move |config| {
|
||||
config.permissions.approval_policy = Constrained::allow_any(AskForApproval::Never);
|
||||
config
|
||||
.permissions
|
||||
.set_permission_profile(PermissionProfile::Disabled)
|
||||
.expect("test config should allow disabled permissions");
|
||||
let mut servers = config.mcp_servers.get().clone();
|
||||
servers.insert(
|
||||
SERVER_NAME.to_string(),
|
||||
serde_json::from_value(json!({
|
||||
"url": server_url,
|
||||
"http_headers": { "Authorization": "Bearer cached-http-test-token" },
|
||||
"enabled_tools": ["calendar_create_event"],
|
||||
"startup_timeout_sec": 10,
|
||||
}))
|
||||
.expect("HTTP MCP server configuration"),
|
||||
);
|
||||
config
|
||||
.mcp_servers
|
||||
.set(servers)
|
||||
.expect("test MCP server configuration");
|
||||
})
|
||||
.build_with_auto_env(&responses_server)
|
||||
.await?;
|
||||
wait_for_mcp_server(&fixture.codex, SERVER_NAME).await?;
|
||||
assert_eq!(startup_control.initialize_attempts(), 1);
|
||||
|
||||
let NewThread {
|
||||
thread: subagent, ..
|
||||
} = fixture
|
||||
.thread_manager
|
||||
.start_thread(StartThreadOptions {
|
||||
session_source: Some(SessionSource::SubAgent(SubAgentSource::ThreadSpawn {
|
||||
parent_thread_id: fixture.session_configured.thread_id,
|
||||
depth: 1,
|
||||
agent_path: None,
|
||||
agent_nickname: None,
|
||||
agent_role: None,
|
||||
})),
|
||||
..StartThreadOptions::new(fixture.config.clone())
|
||||
})
|
||||
.await?;
|
||||
assert_eq!(startup_control.initialize_attempts(), 1);
|
||||
|
||||
let call_id = "http-call";
|
||||
let call_response = mount_sse_once(
|
||||
&responses_server,
|
||||
responses::sse(vec![
|
||||
responses::ev_response_created(call_id),
|
||||
responses::ev_function_call_with_namespace(
|
||||
call_id,
|
||||
NAMESPACE,
|
||||
"calendar_create_event",
|
||||
r#"{"title":"cached","starts_at":"2026-01-01T00:00:00Z"}"#,
|
||||
),
|
||||
responses::ev_completed(call_id),
|
||||
]),
|
||||
)
|
||||
.await;
|
||||
let completion_response = mount_sse_once(
|
||||
&responses_server,
|
||||
responses::sse(vec![
|
||||
responses::ev_assistant_message("http-call-message", "done"),
|
||||
responses::ev_completed("http-call-done"),
|
||||
]),
|
||||
)
|
||||
.await;
|
||||
subagent
|
||||
.submit(user_turn("Call the cached HTTP tool."))
|
||||
.await?;
|
||||
wait_for_event(&subagent, |event| {
|
||||
matches!(event, EventMsg::TurnComplete(_))
|
||||
})
|
||||
.await;
|
||||
let request = call_response.single_request();
|
||||
assert!(
|
||||
request
|
||||
.tool_by_name(NAMESPACE, "calendar_create_event")
|
||||
.is_some()
|
||||
);
|
||||
let output = completion_response.function_call_output_text(call_id);
|
||||
assert!(output.is_some());
|
||||
assert_eq!(startup_control.initialize_attempts(), 2);
|
||||
|
||||
fixture.codex.shutdown_and_wait().await?;
|
||||
subagent.shutdown_and_wait().await?;
|
||||
responses_server.verify().await;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
|
||||
async fn cached_mcp_startup_is_eager_for_root_and_lazy_for_subagents() -> anyhow::Result<()> {
|
||||
skip_if_wine_exec!(
|
||||
|
||||
Reference in New Issue
Block a user