diff --git a/codex-rs/Cargo.lock b/codex-rs/Cargo.lock index 2c50f268f..bc5953803 100644 --- a/codex-rs/Cargo.lock +++ b/codex-rs/Cargo.lock @@ -3757,7 +3757,6 @@ dependencies = [ "base64 0.22.1", "bytes", "codex-api", - "codex-client", "codex-config", "codex-exec-server", "codex-keyring-store", diff --git a/codex-rs/app-server-protocol/schema/json/ClientRequest.json b/codex-rs/app-server-protocol/schema/json/ClientRequest.json index 68c242504..0027988c4 100644 --- a/codex-rs/app-server-protocol/schema/json/ClientRequest.json +++ b/codex-rs/app-server-protocol/schema/json/ClientRequest.json @@ -1619,6 +1619,12 @@ "null" ] }, + "threadId": { + "type": [ + "string", + "null" + ] + }, "timeoutSecs": { "format": "int64", "type": [ diff --git a/codex-rs/app-server-protocol/schema/json/ServerNotification.json b/codex-rs/app-server-protocol/schema/json/ServerNotification.json index a3d07c24a..18435f6f2 100644 --- a/codex-rs/app-server-protocol/schema/json/ServerNotification.json +++ b/codex-rs/app-server-protocol/schema/json/ServerNotification.json @@ -2370,6 +2370,12 @@ }, "success": { "type": "boolean" + }, + "threadId": { + "type": [ + "string", + "null" + ] } }, "required": [ diff --git a/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.schemas.json b/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.schemas.json index 4d417ffb2..33981b5b5 100644 --- a/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.schemas.json +++ b/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.schemas.json @@ -12026,6 +12026,12 @@ }, "success": { "type": "boolean" + }, + "threadId": { + "type": [ + "string", + "null" + ] } }, "required": [ @@ -12050,6 +12056,12 @@ "null" ] }, + "threadId": { + "type": [ + "string", + "null" + ] + }, "timeoutSecs": { "format": "int64", "type": [ diff --git a/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.v2.schemas.json b/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.v2.schemas.json index 0df923a0f..5a7cf80e8 100644 --- a/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.v2.schemas.json +++ b/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.v2.schemas.json @@ -8430,6 +8430,12 @@ }, "success": { "type": "boolean" + }, + "threadId": { + "type": [ + "string", + "null" + ] } }, "required": [ @@ -8454,6 +8460,12 @@ "null" ] }, + "threadId": { + "type": [ + "string", + "null" + ] + }, "timeoutSecs": { "format": "int64", "type": [ diff --git a/codex-rs/app-server-protocol/schema/json/v2/McpServerOauthLoginCompletedNotification.json b/codex-rs/app-server-protocol/schema/json/v2/McpServerOauthLoginCompletedNotification.json index 35efd2baf..6204ff672 100644 --- a/codex-rs/app-server-protocol/schema/json/v2/McpServerOauthLoginCompletedNotification.json +++ b/codex-rs/app-server-protocol/schema/json/v2/McpServerOauthLoginCompletedNotification.json @@ -12,6 +12,12 @@ }, "success": { "type": "boolean" + }, + "threadId": { + "type": [ + "string", + "null" + ] } }, "required": [ diff --git a/codex-rs/app-server-protocol/schema/json/v2/McpServerOauthLoginParams.json b/codex-rs/app-server-protocol/schema/json/v2/McpServerOauthLoginParams.json index 4370f444b..de66dcefd 100644 --- a/codex-rs/app-server-protocol/schema/json/v2/McpServerOauthLoginParams.json +++ b/codex-rs/app-server-protocol/schema/json/v2/McpServerOauthLoginParams.json @@ -13,6 +13,12 @@ "null" ] }, + "threadId": { + "type": [ + "string", + "null" + ] + }, "timeoutSecs": { "format": "int64", "type": [ diff --git a/codex-rs/app-server-protocol/schema/typescript/v2/McpServerOauthLoginCompletedNotification.ts b/codex-rs/app-server-protocol/schema/typescript/v2/McpServerOauthLoginCompletedNotification.ts index 592860ae3..cfa660309 100644 --- a/codex-rs/app-server-protocol/schema/typescript/v2/McpServerOauthLoginCompletedNotification.ts +++ b/codex-rs/app-server-protocol/schema/typescript/v2/McpServerOauthLoginCompletedNotification.ts @@ -2,4 +2,4 @@ // This file was generated by [ts-rs](https://github.com/Aleph-Alpha/ts-rs). Do not edit this file manually. -export type McpServerOauthLoginCompletedNotification = { name: string, success: boolean, error?: string, }; +export type McpServerOauthLoginCompletedNotification = { name: string, threadId: string | null, success: boolean, error?: string, }; diff --git a/codex-rs/app-server-protocol/schema/typescript/v2/McpServerOauthLoginParams.ts b/codex-rs/app-server-protocol/schema/typescript/v2/McpServerOauthLoginParams.ts index a61c30460..ff088b8d9 100644 --- a/codex-rs/app-server-protocol/schema/typescript/v2/McpServerOauthLoginParams.ts +++ b/codex-rs/app-server-protocol/schema/typescript/v2/McpServerOauthLoginParams.ts @@ -2,4 +2,4 @@ // This file was generated by [ts-rs](https://github.com/Aleph-Alpha/ts-rs). Do not edit this file manually. -export type McpServerOauthLoginParams = { name: string, scopes?: Array | null, timeoutSecs?: bigint | null, }; +export type McpServerOauthLoginParams = { name: string, threadId?: string | null, scopes?: Array | null, timeoutSecs?: bigint | null, }; diff --git a/codex-rs/app-server-protocol/src/protocol/common.rs b/codex-rs/app-server-protocol/src/protocol/common.rs index 51a37fbaa..3120007e5 100644 --- a/codex-rs/app-server-protocol/src/protocol/common.rs +++ b/codex-rs/app-server-protocol/src/protocol/common.rs @@ -1913,6 +1913,7 @@ mod tests { request_id: request_id(), params: v2::McpServerOauthLoginParams { name: "server-a".to_string(), + thread_id: None, scopes: None, timeout_secs: None, }, diff --git a/codex-rs/app-server-protocol/src/protocol/v2/mcp.rs b/codex-rs/app-server-protocol/src/protocol/v2/mcp.rs index cdfeb524f..ca0e124d5 100644 --- a/codex-rs/app-server-protocol/src/protocol/v2/mcp.rs +++ b/codex-rs/app-server-protocol/src/protocol/v2/mcp.rs @@ -186,6 +186,8 @@ pub struct McpServerRefreshResponse {} #[ts(export_to = "v2/")] pub struct McpServerOauthLoginParams { pub name: String, + #[ts(optional = nullable)] + pub thread_id: Option, #[serde(default, skip_serializing_if = "Option::is_none")] #[ts(optional = nullable)] pub scopes: Option>, @@ -215,6 +217,7 @@ pub struct McpToolCallProgressNotification { #[ts(export_to = "v2/")] pub struct McpServerOauthLoginCompletedNotification { pub name: String, + pub thread_id: Option, pub success: bool, #[serde(default, skip_serializing_if = "Option::is_none")] #[ts(optional)] diff --git a/codex-rs/app-server/README.md b/codex-rs/app-server/README.md index 981e8556d..397fa3d58 100644 --- a/codex-rs/app-server/README.md +++ b/codex-rs/app-server/README.md @@ -229,7 +229,7 @@ Example with notification opt-out: - `skills/config/write` — write user-level skill config by name or absolute path. - `plugin/install` — install a plugin from a discovered marketplace entry, rejecting marketplace entries marked unavailable for install, install MCPs if any, and return the effective plugin auth policy plus any apps that still need auth (**under development; do not call from production clients yet**). - `plugin/uninstall` — uninstall a local plugin by `pluginId` in `@` form by removing its cached files and clearing its user-level config entry, or uninstall a remote ChatGPT plugin by backend `pluginId` by forwarding the uninstall to the ChatGPT plugin backend and removing any downloaded remote-plugin cache (**under development; do not call from production clients yet**). -- `mcpServer/oauth/login` — start an OAuth login for a configured MCP server; returns an `authorization_url` and later emits `mcpServer/oauthLogin/completed` once the browser flow finishes. +- `mcpServer/oauth/login` — start an OAuth login for a configured MCP server; pass `threadId` to resolve servers from that thread's selected plugins and executor, and receive an `authorization_url` followed by `mcpServer/oauthLogin/completed` once the browser flow finishes. - `tool/requestUserInput` — prompt the user with 1–3 short questions for a tool call and return their answers (experimental). - `config/mcpServer/reload` — reload MCP server config from disk and queue a refresh for loaded threads (applied on each thread's next active turn); returns `{}`. Use this after editing `config.toml` without restarting the server. - `mcpServerStatus/list` — enumerate configured MCP servers with their tools, auth status, server info, plus resources/resource templates for `full` detail; supports optional `threadId` and cursor+limit pagination. If `threadId` is omitted, the server reads from the latest global config directly. If `detail` is omitted, the server defaults to `full`. @@ -1934,7 +1934,7 @@ Codex supports these authentication modes. The current mode is surfaced in `acco - `account/workspaceMessages/read` — fetch active workspace messages, including workspace notification headlines when available. - `account/rateLimits/updated` (notify) — emitted whenever a user's ChatGPT rate limits change. This is a sparse rolling update; merge available values into the most recent `account/rateLimits/read` response or refetch that snapshot. - `account/sendAddCreditsNudgeEmail` — ask ChatGPT to email the workspace owner about depleted credits or a reached usage limit. -- `mcpServer/oauthLogin/completed` (notify) — emitted after a `mcpServer/oauth/login` flow finishes for a server; payload includes `{ name, success, error? }`. +- `mcpServer/oauthLogin/completed` (notify) — emitted after a `mcpServer/oauth/login` flow finishes for a server; payload includes `{ name, threadId, success, error? }`. - `mcpServer/startupStatus/updated` (notify) — emitted when a configured MCP server's startup status changes; payload includes `{ threadId, name, status, error }`, where `threadId` is the owning thread when startup is thread-scoped and `null` when it is app-scoped, and `status` is `starting`, `ready`, `failed`, or `cancelled`. ### 1) Check auth state diff --git a/codex-rs/app-server/src/request_processors.rs b/codex-rs/app-server/src/request_processors.rs index d73a41706..9b56957a5 100644 --- a/codex-rs/app-server/src/request_processors.rs +++ b/codex-rs/app-server/src/request_processors.rs @@ -380,7 +380,7 @@ use codex_mcp::McpRuntimeContext; use codex_mcp::McpServerStatusSnapshot; use codex_mcp::McpSnapshotDetail; use codex_mcp::collect_mcp_server_status_snapshot_with_detail; -use codex_mcp::discover_supported_scopes; +use codex_mcp::discover_supported_scopes_with_http_client; use codex_mcp::read_mcp_resource as read_mcp_resource_without_thread; use codex_mcp::resolve_oauth_scopes; use codex_memories_write::clear_memory_roots_contents; @@ -428,7 +428,7 @@ use codex_protocol::protocol::USER_MESSAGE_BEGIN; use codex_protocol::protocol::W3cTraceContext; use codex_protocol::user_input::MAX_USER_INPUT_TEXT_CHARS; use codex_protocol::user_input::UserInput as CoreInputItem; -use codex_rmcp_client::perform_oauth_login_return_url; +use codex_rmcp_client::perform_oauth_login_return_url_with_http_client; use codex_rollout::is_persisted_rollout_item; use codex_rollout::state_db::StateDbHandle; use codex_rollout::state_db::reconcile_rollout; diff --git a/codex-rs/app-server/src/request_processors/mcp_processor.rs b/codex-rs/app-server/src/request_processors/mcp_processor.rs index 08d464598..18b49ede3 100644 --- a/codex-rs/app-server/src/request_processors/mcp_processor.rs +++ b/codex-rs/app-server/src/request_processors/mcp_processor.rs @@ -113,19 +113,37 @@ impl McpRequestProcessor { &self, params: McpServerOauthLoginParams, ) -> Result { - let config = self.load_latest_config(/*fallback_cwd*/ None).await?; let McpServerOauthLoginParams { name, + thread_id, scopes, timeout_secs, } = params; let auth = self.auth_manager.auth().await; - let effective_servers = self - .thread_manager - .mcp_manager() - .effective_servers(&config, auth.as_ref()) - .await; + let (config, mcp_config) = match thread_id.as_deref() { + Some(thread_id) => { + let (_, thread) = self.load_thread(thread_id).await?; + let thread_config = thread.config().await; + let config = self + .config_manager + .load_latest_config_for_thread(thread_config.as_ref()) + .await + .map_err(|err| internal_error(format!("failed to reload config: {err}")))?; + let mcp_config = thread.runtime_mcp_config(&config).await; + (config, mcp_config) + } + None => { + let config = self.load_latest_config(/*fallback_cwd*/ None).await?; + let mcp_config = self + .thread_manager + .mcp_manager() + .runtime_config(&config) + .await; + (config, mcp_config) + } + }; + let effective_servers = codex_mcp::effective_mcp_servers(&mcp_config, auth.as_ref()); let Some(server) = effective_servers .get(&name) .and_then(codex_mcp::EffectiveMcpServer::configured_config) @@ -149,15 +167,26 @@ impl McpRequestProcessor { } }; + let runtime_context = McpRuntimeContext::new( + self.thread_manager.environment_manager(), + config.cwd.to_path_buf(), + ); + let http_client = runtime_context + .resolve_http_client(&name, server) + .map_err(|err| { + internal_error(format!("failed to resolve MCP server runtime: {err}")) + })?; + let discovered_scopes = if scopes.is_none() && server.scopes.is_none() { - discover_supported_scopes(&server.transport).await + discover_supported_scopes_with_http_client(&server.transport, Arc::clone(&http_client)) + .await } else { None }; let resolved_scopes = resolve_oauth_scopes(scopes, server.scopes.clone(), discovered_scopes); - let handle = perform_oauth_login_return_url( + let handle = perform_oauth_login_return_url_with_http_client( &name, &url, config.mcp_oauth_credentials_store_mode, @@ -170,11 +199,13 @@ impl McpRequestProcessor { timeout_secs, config.mcp_oauth_callback_port, config.mcp_oauth_callback_url.as_deref(), + http_client, ) .await .map_err(|err| internal_error(format!("failed to login to MCP server '{name}': {err}")))?; let authorization_url = handle.authorization_url().to_string(); let notification_name = name.clone(); + let notification_thread_id = thread_id; let outgoing = Arc::clone(&self.outgoing); tokio::spawn(async move { @@ -186,6 +217,7 @@ impl McpRequestProcessor { let notification = ServerNotification::McpServerOauthLoginCompleted( McpServerOauthLoginCompletedNotification { name: notification_name, + thread_id: notification_thread_id, success, error, }, diff --git a/codex-rs/app-server/src/request_processors/plugins.rs b/codex-rs/app-server/src/request_processors/plugins.rs index 4aa8352d1..d123f0379 100644 --- a/codex-rs/app-server/src/request_processors/plugins.rs +++ b/codex-rs/app-server/src/request_processors/plugins.rs @@ -1883,6 +1883,7 @@ impl PluginRequestProcessor { let notification = ServerNotification::McpServerOauthLoginCompleted( McpServerOauthLoginCompletedNotification { name: notification_name, + thread_id: None, success, error, }, diff --git a/codex-rs/app-server/tests/suite/v2/executor_mcp.rs b/codex-rs/app-server/tests/suite/v2/executor_mcp.rs index 55e5f2dd4..f93af6821 100644 --- a/codex-rs/app-server/tests/suite/v2/executor_mcp.rs +++ b/codex-rs/app-server/tests/suite/v2/executor_mcp.rs @@ -2,10 +2,14 @@ use anyhow::Result; use app_test_support::TestAppServer; use app_test_support::to_response; use app_test_support::write_mock_responses_config_toml; +use axum::Json; use axum::Router; +use axum::routing::get; use codex_app_server_protocol::CapabilityRootLocation; use codex_app_server_protocol::ListMcpServerStatusParams; use codex_app_server_protocol::ListMcpServerStatusResponse; +use codex_app_server_protocol::McpServerOauthLoginCompletedNotification; +use codex_app_server_protocol::McpServerOauthLoginResponse; use codex_app_server_protocol::McpServerToolCallParams; use codex_app_server_protocol::McpServerToolCallResponse; use codex_app_server_protocol::RequestId; @@ -45,6 +49,8 @@ const DEFAULT_READ_TIMEOUT: Duration = Duration::from_secs(20); const EXECUTOR_HTTP_MCP_URL: &str = "http://executor-only.invalid/mcp"; const HTTP_MCP_SERVER_NAME: &str = "executor_http"; const MCP_SERVER_NAME: &str = "executor_demo"; +const OAUTH_MCP_SERVER_NAME: &str = "executor_oauth"; +const EXECUTOR_OAUTH_MCP_URL: &str = "http://oauth-only.invalid/oauth-mcp"; const EXECUTOR_ENV_NAME: &str = "MCP_EXECUTOR_MARKER"; const EXECUTOR_ENV_VALUE: &str = "executor-only"; const EXECUTOR_ID: &str = "executor-1"; @@ -56,12 +62,35 @@ async fn selected_executor_plugin_exposes_its_mcps_only_to_that_thread() -> Resu let responses_server = responses::start_mock_server().await; let http_listener = TcpListener::bind("127.0.0.1:0").await?; let http_addr = http_listener.local_addr()?; + let http_server_config = StreamableHttpServerConfig::default() + .with_allowed_hosts(["executor-only.invalid", "oauth-only.invalid"]); let http_mcp_service = StreamableHttpService::new( || Ok(ExecutorHttpMcpServer), Arc::new(LocalSessionManager::default()), - StreamableHttpServerConfig::default().with_allowed_hosts(["executor-only.invalid"]), + http_server_config.clone(), ); - let http_router = Router::new().nest_service("/mcp", http_mcp_service); + let oauth_mcp_service = StreamableHttpService::new( + || Ok(ExecutorHttpMcpServer), + Arc::new(LocalSessionManager::default()), + http_server_config, + ); + let oauth_metadata = json!({ + "authorization_endpoint": "https://oauth-only.invalid/authorize", + "token_endpoint": "https://oauth-only.invalid/token", + "scopes_supported": ["read", "write"], + "response_types_supported": ["code"], + "code_challenge_methods_supported": ["S256"], + }); + let http_router = Router::new() + .route( + "/.well-known/oauth-authorization-server/oauth-mcp", + get(move || { + let metadata = oauth_metadata.clone(); + async move { Json(metadata) } + }), + ) + .nest_service("/mcp", http_mcp_service) + .nest_service("/oauth-mcp", oauth_mcp_service); let http_server_handle = tokio::spawn(async move { let _ = axum::serve(http_listener, http_router).await; }); @@ -117,6 +146,12 @@ HTTP_PROXY = {http_proxy} "url": EXECUTOR_HTTP_MCP_URL, "environment_id": "local", "startup_timeout_sec": 10, + }, + (OAUTH_MCP_SERVER_NAME): { + "url": EXECUTOR_OAUTH_MCP_URL, + "environment_id": "local", + "oauth": {"clientId": "executor-oauth-client"}, + "startup_timeout_sec": 10, } } }))?, @@ -158,6 +193,44 @@ startup_timeout_sec = 10 ) .await??; + let request_id = app_server + .send_raw_request( + "mcpServer/oauth/login", + Some(json!({ + "name": OAUTH_MCP_SERVER_NAME, + "threadId": selected_thread.clone(), + "timeoutSecs": 1, + })), + ) + .await?; + let response = timeout( + DEFAULT_READ_TIMEOUT, + app_server.read_stream_until_response_message(RequestId::Integer(request_id)), + ) + .await??; + let response: McpServerOauthLoginResponse = to_response(response)?; + assert!( + response + .authorization_url + .starts_with("https://oauth-only.invalid/authorize?") + ); + assert!( + response + .authorization_url + .contains("client_id=executor-oauth-client") + ); + let notification = timeout( + DEFAULT_READ_TIMEOUT, + app_server.read_stream_until_notification_message("mcpServer/oauthLogin/completed"), + ) + .await??; + let completed: McpServerOauthLoginCompletedNotification = + serde_json::from_value(notification.params.expect("notification params"))?; + assert_eq!( + completed.thread_id.as_deref(), + Some(selected_thread.as_str()) + ); + let namespace = format!("mcp__{MCP_SERVER_NAME}"); let response_mock = responses::mount_sse_sequence( &responses_server, @@ -269,15 +342,18 @@ startup_timeout_sec = 10 .iter() .any(|name| name == HTTP_MCP_SERVER_NAME) ); + assert!( + selected_server_names + .iter() + .any(|name| name == OAUTH_MCP_SERVER_NAME) + ); let unselected_thread = start_thread(&mut app_server, /*selected_capability_roots*/ None).await?; let unselected_server_names = mcp_server_names(&mut app_server, unselected_thread).await?; - assert!( - unselected_server_names - .iter() - .all(|name| { name != MCP_SERVER_NAME && name != HTTP_MCP_SERVER_NAME }) - ); + assert!(unselected_server_names.iter().all(|name| { + name != MCP_SERVER_NAME && name != HTTP_MCP_SERVER_NAME && name != OAUTH_MCP_SERVER_NAME + })); http_server_handle.abort(); let _ = http_server_handle.await; diff --git a/codex-rs/cli/src/mcp_cmd.rs b/codex-rs/cli/src/mcp_cmd.rs index 15f0b0bb4..586f9c184 100644 --- a/codex-rs/cli/src/mcp_cmd.rs +++ b/codex-rs/cli/src/mcp_cmd.rs @@ -18,8 +18,10 @@ use codex_core::config::edit::ConfigEditsBuilder; use codex_core::config::find_codex_home; use codex_core::config::load_global_mcp_servers; use codex_core_plugins::PluginsManager; +use codex_exec_server::EnvironmentManager; use codex_login::AuthManager; use codex_mcp::McpOAuthLoginSupport; +use codex_mcp::McpRuntimeContext; use codex_mcp::ResolvedMcpOAuthScopes; use codex_mcp::compute_auth_statuses; use codex_mcp::discover_supported_scopes; @@ -558,6 +560,10 @@ async fn run_list(config_overrides: &CliConfigOverrides, list_args: ListArgs) -> config.mcp_oauth_credentials_store_mode, config.auth_keyring_backend_kind(), auth.as_ref(), + &McpRuntimeContext::new( + Arc::new(EnvironmentManager::without_environments()), + config.cwd.to_path_buf(), + ), ) .await; diff --git a/codex-rs/codex-mcp/src/lib.rs b/codex-rs/codex-mcp/src/lib.rs index b4d397eac..98efdc1b5 100644 --- a/codex-rs/codex-mcp/src/lib.rs +++ b/codex-rs/codex-mcp/src/lib.rs @@ -61,7 +61,9 @@ pub use mcp::McpOAuthScopesSource; pub use mcp::ResolvedMcpOAuthScopes; pub use mcp::compute_auth_statuses; pub use mcp::discover_supported_scopes; +pub use mcp::discover_supported_scopes_with_http_client; pub use mcp::oauth_login_support; +pub use mcp::oauth_login_support_with_http_client; pub use mcp::resolve_oauth_scopes; pub use mcp::should_retry_without_scopes; diff --git a/codex-rs/codex-mcp/src/mcp/auth.rs b/codex-rs/codex-mcp/src/mcp/auth.rs index d19766af1..111f04163 100644 --- a/codex-rs/codex-mcp/src/mcp/auth.rs +++ b/codex-rs/codex-mcp/src/mcp/auth.rs @@ -1,4 +1,5 @@ use std::collections::HashMap; +use std::sync::Arc; use anyhow::Result; use codex_config::McpServerAuth; @@ -6,15 +7,19 @@ use codex_config::McpServerConfig; use codex_config::McpServerTransportConfig; use codex_config::types::AuthKeyringBackendKind; use codex_config::types::OAuthCredentialsStoreMode; +use codex_exec_server::HttpClient; use codex_login::CodexAuth; use codex_protocol::protocol::McpAuthStatus; use codex_rmcp_client::OAuthProviderError; use codex_rmcp_client::determine_streamable_http_auth_status; +use codex_rmcp_client::determine_streamable_http_auth_status_with_http_client; use codex_rmcp_client::discover_streamable_http_oauth; +use codex_rmcp_client::discover_streamable_http_oauth_with_http_client; use futures::FutureExt; use futures::future::join_all; use tracing::warn; +use crate::runtime::McpRuntimeContext; use crate::server::EffectiveMcpServer; #[derive(Debug, Clone)] @@ -53,6 +58,50 @@ pub struct McpAuthStatusEntry { } pub async fn oauth_login_support(transport: &McpServerTransportConfig) -> McpOAuthLoginSupport { + let Some(mut config) = oauth_login_candidate(transport) else { + return McpOAuthLoginSupport::Unsupported; + }; + match discover_streamable_http_oauth( + &config.url, + config.http_headers.clone(), + config.env_http_headers.clone(), + ) + .await + { + Ok(Some(discovery)) => { + config.discovered_scopes = discovery.scopes_supported; + McpOAuthLoginSupport::Supported(config) + } + Ok(None) => McpOAuthLoginSupport::Unsupported, + Err(err) => McpOAuthLoginSupport::Unknown(err), + } +} + +pub async fn oauth_login_support_with_http_client( + transport: &McpServerTransportConfig, + http_client: Arc, +) -> McpOAuthLoginSupport { + let Some(mut config) = oauth_login_candidate(transport) else { + return McpOAuthLoginSupport::Unsupported; + }; + match discover_streamable_http_oauth_with_http_client( + &config.url, + config.http_headers.clone(), + config.env_http_headers.clone(), + http_client, + ) + .await + { + Ok(Some(discovery)) => { + config.discovered_scopes = discovery.scopes_supported; + McpOAuthLoginSupport::Supported(config) + } + Ok(None) => McpOAuthLoginSupport::Unsupported, + Err(err) => McpOAuthLoginSupport::Unknown(err), + } +} + +fn oauth_login_candidate(transport: &McpServerTransportConfig) -> Option { let McpServerTransportConfig::StreamableHttp { url, bearer_token_env_var, @@ -60,24 +109,17 @@ pub async fn oauth_login_support(transport: &McpServerTransportConfig) -> McpOAu env_http_headers, } = transport else { - return McpOAuthLoginSupport::Unsupported; + return None; }; - if bearer_token_env_var.is_some() { - return McpOAuthLoginSupport::Unsupported; - } - - match discover_streamable_http_oauth(url, http_headers.clone(), env_http_headers.clone()).await - { - Ok(Some(discovery)) => McpOAuthLoginSupport::Supported(McpOAuthLoginConfig { - url: url.clone(), - http_headers: http_headers.clone(), - env_http_headers: env_http_headers.clone(), - discovered_scopes: discovery.scopes_supported, - }), - Ok(None) => McpOAuthLoginSupport::Unsupported, - Err(err) => McpOAuthLoginSupport::Unknown(err), + return None; } + Some(McpOAuthLoginConfig { + url: url.clone(), + http_headers: http_headers.clone(), + env_http_headers: env_http_headers.clone(), + discovered_scopes: None, + }) } pub async fn discover_supported_scopes( @@ -89,6 +131,16 @@ pub async fn discover_supported_scopes( } } +pub async fn discover_supported_scopes_with_http_client( + transport: &McpServerTransportConfig, + http_client: Arc, +) -> Option> { + match oauth_login_support_with_http_client(transport, http_client).await { + McpOAuthLoginSupport::Supported(config) => config.discovered_scopes, + McpOAuthLoginSupport::Unsupported | McpOAuthLoginSupport::Unknown(_) => None, + } +} + pub fn resolve_oauth_scopes( explicit_scopes: Option>, configured_scopes: Option>, @@ -133,6 +185,7 @@ pub async fn compute_auth_statuses<'a, I>( store_mode: OAuthCredentialsStoreMode, keyring_backend_kind: AuthKeyringBackendKind, auth: Option<&CodexAuth>, + runtime_context: &McpRuntimeContext, ) -> HashMap where I: IntoIterator, @@ -140,6 +193,7 @@ where let futures = servers.into_iter().map(|(name, server)| { let name = name.clone(); let config = server.configured_config().cloned(); + let runtime_context = runtime_context.clone(); let has_runtime_auth = config .as_ref() .is_some_and(|config| matches!(&config.auth, McpServerAuth::ChatGpt)) @@ -162,6 +216,7 @@ where store_mode, keyring_backend_kind, has_runtime_auth, + &runtime_context, ) .await { @@ -193,6 +248,7 @@ async fn compute_auth_status( store_mode: OAuthCredentialsStoreMode, keyring_backend_kind: AuthKeyringBackendKind, has_runtime_auth: bool, + runtime_context: &McpRuntimeContext, ) -> Result { if !config.enabled { return Ok(McpAuthStatus::Unsupported); @@ -210,17 +266,35 @@ async fn compute_auth_status( http_headers, env_http_headers, } => { - determine_streamable_http_auth_status( - server_name, - url, - bearer_token_env_var.as_deref(), - http_headers.clone(), - env_http_headers.clone(), - store_mode, - keyring_backend_kind, - ) - .boxed() - .await + if config.is_local_environment() { + determine_streamable_http_auth_status( + server_name, + url, + bearer_token_env_var.as_deref(), + http_headers.clone(), + env_http_headers.clone(), + store_mode, + keyring_backend_kind, + ) + .boxed() + .await + } else { + let http_client = runtime_context + .resolve_http_client(server_name, config) + .map_err(anyhow::Error::msg)?; + determine_streamable_http_auth_status_with_http_client( + server_name, + url, + bearer_token_env_var.as_deref(), + http_headers.clone(), + env_http_headers.clone(), + store_mode, + keyring_backend_kind, + http_client, + ) + .boxed() + .await + } } } } diff --git a/codex-rs/codex-mcp/src/mcp/mod.rs b/codex-rs/codex-mcp/src/mcp/mod.rs index f2c8f1d02..4550afa79 100644 --- a/codex-rs/codex-mcp/src/mcp/mod.rs +++ b/codex-rs/codex-mcp/src/mcp/mod.rs @@ -5,7 +5,9 @@ pub use auth::McpOAuthScopesSource; pub use auth::ResolvedMcpOAuthScopes; pub use auth::compute_auth_statuses; pub use auth::discover_supported_scopes; +pub use auth::discover_supported_scopes_with_http_client; pub use auth::oauth_login_support; +pub use auth::oauth_login_support_with_http_client; pub use auth::resolve_oauth_scopes; pub use auth::should_retry_without_scopes; @@ -309,6 +311,7 @@ pub async fn read_mcp_resource( config.mcp_oauth_credentials_store_mode, config.auth_keyring_backend_kind, auth, + &runtime_context, ) .await; let (tx_event, rx_event) = unbounded(); @@ -378,6 +381,7 @@ pub async fn collect_mcp_server_status_snapshot_with_detail( config.mcp_oauth_credentials_store_mode, config.auth_keyring_backend_kind, auth, + &runtime_context, ) .await; diff --git a/codex-rs/codex-mcp/src/runtime.rs b/codex-rs/codex-mcp/src/runtime.rs index be9d50d3f..b9b769ffb 100644 --- a/codex-rs/codex-mcp/src/runtime.rs +++ b/codex-rs/codex-mcp/src/runtime.rs @@ -11,6 +11,8 @@ use std::time::Duration; use codex_exec_server::Environment; use codex_exec_server::EnvironmentManager; +use codex_exec_server::HttpClient; +use codex_exec_server::ReqwestHttpClient; use codex_protocol::models::PermissionProfile; use codex_utils_path_uri::PathUri; use serde::Deserialize; @@ -81,6 +83,20 @@ impl McpRuntimeContext { config.environment_id )) } + + /// Resolves the HTTP capability owned by the server's configured environment. + pub fn resolve_http_client( + &self, + server_name: &str, + config: &codex_config::McpServerConfig, + ) -> Result, String> { + Ok(self + .resolve_server_environment(server_name, config)? + .map_or_else( + || Arc::new(ReqwestHttpClient) as Arc, + |environment| environment.get_http_client(), + )) + } } pub(crate) fn emit_duration(metric: &str, duration: Duration, tags: &[(&str, &str)]) { diff --git a/codex-rs/core/src/connectors.rs b/codex-rs/core/src/connectors.rs index 79978da99..76b4a8483 100644 --- a/codex-rs/core/src/connectors.rs +++ b/codex-rs/core/src/connectors.rs @@ -246,11 +246,14 @@ pub async fn list_accessible_connectors_from_mcp_tools_with_mcp_manager( }); } + let runtime_context = + McpRuntimeContext::new(Arc::clone(&environment_manager), config.cwd.to_path_buf()); let auth_status_entries = compute_auth_statuses( mcp_servers.iter(), config.mcp_oauth_credentials_store_mode, config.auth_keyring_backend_kind(), auth.as_ref(), + &runtime_context, ) .await; @@ -270,7 +273,7 @@ pub async fn list_accessible_connectors_from_mcp_tools_with_mcp_manager( PermissionProfile::default(), // Connector discovery is threadless. Use an actually configured env if // one exists, but do not reintroduce the old hidden-local fallback. - McpRuntimeContext::new(environment_manager, config.cwd.to_path_buf()), + runtime_context, config.codex_home.to_path_buf(), codex_apps_tools_cache_key(auth.as_ref()), mcp_config.prefix_mcp_tool_names, diff --git a/codex-rs/core/src/session/mcp.rs b/codex-rs/core/src/session/mcp.rs index 5a115b917..c060ef090 100644 --- a/codex-rs/core/src/session/mcp.rs +++ b/codex-rs/core/src/session/mcp.rs @@ -271,13 +271,6 @@ impl Session { let tool_plugin_provenance = codex_mcp::tool_plugin_provenance(&mcp_config); let mcp_servers = effective_mcp_servers_from_configured(mcp_servers, &mcp_config, auth.as_ref()); - let auth_statuses = compute_auth_statuses( - mcp_servers.iter(), - store_mode, - keyring_backend_kind, - auth.as_ref(), - ) - .await; let environment_manager = self.services.turn_environments.environment_manager(); // TODO(anp): Migrate MCP runtime cwd plumbing to PathUri so foreign environment cwd // values can be used without falling back to the legacy host cwd. @@ -291,6 +284,14 @@ impl Session { turn_context.cwd.to_path_buf() }); let mcp_runtime_context = McpRuntimeContext::new(environment_manager, cwd); + let auth_statuses = compute_auth_statuses( + mcp_servers.iter(), + store_mode, + keyring_backend_kind, + auth.as_ref(), + &mcp_runtime_context, + ) + .await; let mcp_startup_cancellation_token = { let mut guard = self.services.mcp_startup_cancellation_token.lock().await; guard.cancel(); diff --git a/codex-rs/core/src/session/session.rs b/codex-rs/core/src/session/session.rs index 5988834ee..e43a26bbe 100644 --- a/codex-rs/core/src/session/session.rs +++ b/codex-rs/core/src/session/session.rs @@ -648,6 +648,10 @@ impl Session { let config_for_mcp = Arc::clone(&config); let mcp_manager_for_mcp = Arc::clone(&mcp_manager); let mcp_thread_init_for_startup = &mcp_thread_init; + let mcp_runtime_context_for_auth = McpRuntimeContext::new( + Arc::clone(&environment_manager), + session_configuration.cwd().to_path_buf(), + ); let auth_and_mcp_fut = async move { let auth = auth_manager_clone.auth().await; let mcp_config = mcp_manager_for_mcp @@ -660,6 +664,7 @@ impl Session { config_for_mcp.mcp_oauth_credentials_store_mode, config_for_mcp.auth_keyring_backend_kind(), auth.as_ref(), + &mcp_runtime_context_for_auth, ) .await; (auth, mcp_servers, auth_statuses, tool_plugin_provenance) diff --git a/codex-rs/core/tests/suite/rmcp_client.rs b/codex-rs/core/tests/suite/rmcp_client.rs index a4d56f8cc..fc1a7df18 100644 --- a/codex-rs/core/tests/suite/rmcp_client.rs +++ b/codex-rs/core/tests/suite/rmcp_client.rs @@ -24,6 +24,7 @@ use codex_config::types::McpServerTransportConfig; use codex_core::config::Config; use codex_exec_server::CreateDirectoryOptions; use codex_exec_server::Environment; +use codex_exec_server::HttpRedirectPolicy; use codex_exec_server::HttpRequestParams; use codex_login::CodexAuth; use codex_mcp::MCP_SANDBOX_STATE_META_CAPABILITY; @@ -2934,6 +2935,7 @@ async fn wait_for_remote_streamable_http_server( headers: Vec::new(), body: None, timeout_ms: Some(remaining.as_millis().clamp(1, 1_000) as u64), + redirect_policy: HttpRedirectPolicy::Follow, request_id: "buffered-request".to_string(), stream_response: false, }; diff --git a/codex-rs/exec-server-protocol/src/protocol.rs b/codex-rs/exec-server-protocol/src/protocol.rs index 133654dc0..491fc8cf9 100644 --- a/codex-rs/exec-server-protocol/src/protocol.rs +++ b/codex-rs/exec-server-protocol/src/protocol.rs @@ -419,6 +419,17 @@ pub struct HttpHeader { pub value: String, } +/// Redirect behavior for an executor-side HTTP request. +#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] +pub enum HttpRedirectPolicy { + /// Follow redirects using the HTTP client's normal limits. + #[default] + Follow, + /// Return the redirect response without following its location. + Stop, +} + /// Executor-side HTTP request envelope. /// /// This intentionally stays transport-shaped rather than MCP-shaped so callers @@ -443,6 +454,9 @@ pub struct HttpRequestParams { /// millisecond deadline. #[serde(default, skip_serializing_if = "Option::is_none")] pub timeout_ms: Option, + /// Whether the executor should follow HTTP redirects. + #[serde(default)] + pub redirect_policy: HttpRedirectPolicy, /// Caller-chosen stream id for `http/request/bodyDelta` notifications. /// /// The id must remain unique on a connection until the terminal body delta diff --git a/codex-rs/exec-server/src/client/reqwest_http_client.rs b/codex-rs/exec-server/src/client/reqwest_http_client.rs index 9de35315e..2fcb7c7e6 100644 --- a/codex-rs/exec-server/src/client/reqwest_http_client.rs +++ b/codex-rs/exec-server/src/client/reqwest_http_client.rs @@ -24,6 +24,7 @@ use super::response_body_stream::send_body_delta; use crate::HttpClient; use crate::client::ExecServerError; use crate::protocol::HttpHeader; +use crate::protocol::HttpRedirectPolicy; use crate::protocol::HttpRequestBodyDeltaNotification; use crate::protocol::HttpRequestParams; use crate::protocol::HttpRequestResponse; @@ -50,13 +51,20 @@ pub(crate) struct ReqwestHttpRequestRunner { } impl ReqwestHttpClient { - fn build_client(timeout_ms: Option) -> Result { + fn build_client( + timeout_ms: Option, + redirect_policy: HttpRedirectPolicy, + ) -> Result { let builder = match timeout_ms { None => reqwest::Client::builder(), Some(timeout_ms) => { reqwest::Client::builder().timeout(Duration::from_millis(timeout_ms)) } }; + let builder = match redirect_policy { + HttpRedirectPolicy::Follow => builder, + HttpRedirectPolicy::Stop => builder.redirect(reqwest::redirect::Policy::none()), + }; build_reqwest_client_with_custom_ca(builder) .map_err(|error| ExecServerError::HttpRequest(error.to_string())) } @@ -68,7 +76,7 @@ impl HttpClient for ReqwestHttpClient { params: HttpRequestParams, ) -> BoxFuture<'_, Result> { async move { - let runner = ReqwestHttpRequestRunner::new(params.timeout_ms) + let runner = ReqwestHttpRequestRunner::new(params.timeout_ms, params.redirect_policy) .map_err(|error| ExecServerError::HttpRequest(error.message))?; let (response, _) = runner .run(HttpRequestParams { @@ -87,7 +95,7 @@ impl HttpClient for ReqwestHttpClient { params: HttpRequestParams, ) -> BoxFuture<'_, Result<(HttpRequestResponse, HttpResponseBodyStream), ExecServerError>> { async move { - let runner = ReqwestHttpRequestRunner::new(params.timeout_ms) + let runner = ReqwestHttpRequestRunner::new(params.timeout_ms, params.redirect_policy) .map_err(|error| ExecServerError::HttpRequest(error.message))?; let (response, pending_stream) = runner .run(HttpRequestParams { @@ -111,8 +119,11 @@ impl HttpClient for ReqwestHttpClient { } impl ReqwestHttpRequestRunner { - pub(crate) fn new(timeout_ms: Option) -> Result { - let client = ReqwestHttpClient::build_client(timeout_ms) + pub(crate) fn new( + timeout_ms: Option, + redirect_policy: HttpRedirectPolicy, + ) -> Result { + let client = ReqwestHttpClient::build_client(timeout_ms, redirect_policy) .map_err(|error| internal_error(error.to_string()))?; Ok(Self { client }) } diff --git a/codex-rs/exec-server/src/lib.rs b/codex-rs/exec-server/src/lib.rs index 69d4c07df..5b2bfa7e5 100644 --- a/codex-rs/exec-server/src/lib.rs +++ b/codex-rs/exec-server/src/lib.rs @@ -123,6 +123,7 @@ pub use protocol::FsWalkResponse; pub use protocol::FsWriteFileParams; pub use protocol::FsWriteFileResponse; pub use protocol::HttpHeader; +pub use protocol::HttpRedirectPolicy; pub use protocol::HttpRequestBodyDeltaNotification; pub use protocol::HttpRequestParams; pub use protocol::HttpRequestResponse; diff --git a/codex-rs/exec-server/src/server/handler.rs b/codex-rs/exec-server/src/server/handler.rs index 561cf0ff6..02858ae88 100644 --- a/codex-rs/exec-server/src/server/handler.rs +++ b/codex-rs/exec-server/src/server/handler.rs @@ -211,7 +211,7 @@ impl ExecServerHandler { if stream_response { self.reserve_http_body_stream(&http_request_id).await?; } - let response = ReqwestHttpRequestRunner::new(params.timeout_ms)? + let response = ReqwestHttpRequestRunner::new(params.timeout_ms, params.redirect_policy)? .run(params) .await; if response.is_err() && stream_response { diff --git a/codex-rs/exec-server/tests/http_client.rs b/codex-rs/exec-server/tests/http_client.rs index 248564b49..b3f0182c6 100644 --- a/codex-rs/exec-server/tests/http_client.rs +++ b/codex-rs/exec-server/tests/http_client.rs @@ -6,6 +6,7 @@ use anyhow::Result; use anyhow::bail; use codex_exec_server::ExecServerClient; use codex_exec_server::HttpHeader; +use codex_exec_server::HttpRedirectPolicy; use codex_exec_server::HttpRequestBodyDeltaNotification; use codex_exec_server::HttpRequestParams; use codex_exec_server::HttpRequestResponse; @@ -63,6 +64,7 @@ async fn http_request_forces_buffered_request_params() -> Result<()> { headers: Vec::new(), body: None, timeout_ms: None, + redirect_policy: HttpRedirectPolicy::Follow, request_id: "ignored-stream-id".to_string(), stream_response: false, } @@ -91,6 +93,7 @@ async fn http_request_forces_buffered_request_params() -> Result<()> { headers: Vec::new(), body: None, timeout_ms: None, + redirect_policy: HttpRedirectPolicy::Follow, request_id: "ignored-stream-id".to_string(), stream_response: true, }), @@ -131,6 +134,7 @@ async fn http_response_body_stream_uses_generated_ids_and_receives_ordered_delta }], body: None, timeout_ms: None, + redirect_policy: HttpRedirectPolicy::Follow, request_id: "http-1".to_string(), stream_response: true, } @@ -186,6 +190,7 @@ async fn http_response_body_stream_uses_generated_ids_and_receives_ordered_delta headers: Vec::new(), body: None, timeout_ms: None, + redirect_policy: HttpRedirectPolicy::Follow, request_id: "http-2".to_string(), stream_response: true, } @@ -215,6 +220,7 @@ async fn http_response_body_stream_uses_generated_ids_and_receives_ordered_delta }], body: None, timeout_ms: None, + redirect_policy: HttpRedirectPolicy::Follow, request_id: "caller-stream-id".to_string(), stream_response: false, }), @@ -253,6 +259,7 @@ async fn http_response_body_stream_uses_generated_ids_and_receives_ordered_delta headers: Vec::new(), body: None, timeout_ms: None, + redirect_policy: HttpRedirectPolicy::Follow, request_id: "caller-stream-id".to_string(), stream_response: false, }), @@ -289,6 +296,7 @@ async fn http_response_body_stream_drops_queued_terminal_before_next_generated_i headers: Vec::new(), body: None, timeout_ms: None, + redirect_policy: HttpRedirectPolicy::Follow, request_id: "http-1".to_string(), stream_response: true, } @@ -322,6 +330,7 @@ async fn http_response_body_stream_drops_queued_terminal_before_next_generated_i headers: Vec::new(), body: None, timeout_ms: None, + redirect_policy: HttpRedirectPolicy::Follow, request_id: "http-2".to_string(), stream_response: true, } @@ -348,6 +357,7 @@ async fn http_response_body_stream_drops_queued_terminal_before_next_generated_i headers: Vec::new(), body: None, timeout_ms: None, + redirect_policy: HttpRedirectPolicy::Follow, request_id: "caller-stream-id".to_string(), stream_response: false, }), @@ -372,6 +382,7 @@ async fn http_response_body_stream_drops_queued_terminal_before_next_generated_i headers: Vec::new(), body: None, timeout_ms: None, + redirect_policy: HttpRedirectPolicy::Follow, request_id: "caller-stream-id".to_string(), stream_response: false, }; @@ -411,6 +422,7 @@ async fn http_response_body_stream_ignores_late_deltas_after_cancelled_request() headers: Vec::new(), body: None, timeout_ms: None, + redirect_policy: HttpRedirectPolicy::Follow, request_id: "http-1".to_string(), stream_response: true, } @@ -430,6 +442,7 @@ async fn http_response_body_stream_ignores_late_deltas_after_cancelled_request() headers: Vec::new(), body: None, timeout_ms: None, + redirect_policy: HttpRedirectPolicy::Follow, request_id: "http-2".to_string(), stream_response: true, } @@ -474,6 +487,7 @@ async fn http_response_body_stream_ignores_late_deltas_after_cancelled_request() headers: Vec::new(), body: None, timeout_ms: None, + redirect_policy: HttpRedirectPolicy::Follow, request_id: "caller-stream-id".to_string(), stream_response: false, }) @@ -495,6 +509,7 @@ async fn http_response_body_stream_ignores_late_deltas_after_cancelled_request() headers: Vec::new(), body: None, timeout_ms: None, + redirect_policy: HttpRedirectPolicy::Follow, request_id: "caller-stream-id".to_string(), stream_response: false, }), @@ -541,6 +556,7 @@ async fn http_response_body_stream_ignores_late_deltas_after_drop() -> Result<() headers: Vec::new(), body: None, timeout_ms: None, + redirect_policy: HttpRedirectPolicy::Follow, request_id: "http-1".to_string(), stream_response: true, } @@ -580,6 +596,7 @@ async fn http_response_body_stream_ignores_late_deltas_after_drop() -> Result<() headers: Vec::new(), body: None, timeout_ms: None, + redirect_policy: HttpRedirectPolicy::Follow, request_id: "http-2".to_string(), stream_response: true, } @@ -615,6 +632,7 @@ async fn http_response_body_stream_ignores_late_deltas_after_drop() -> Result<() headers: Vec::new(), body: None, timeout_ms: None, + redirect_policy: HttpRedirectPolicy::Follow, request_id: "caller-stream-id".to_string(), stream_response: false, }), @@ -647,6 +665,7 @@ async fn http_response_body_stream_ignores_late_deltas_after_drop() -> Result<() headers: Vec::new(), body: None, timeout_ms: None, + redirect_policy: HttpRedirectPolicy::Follow, request_id: "caller-stream-id".to_string(), stream_response: false, }), @@ -691,6 +710,7 @@ async fn http_response_body_stream_fails_when_transport_disconnects() -> Result< headers: Vec::new(), body: None, timeout_ms: None, + redirect_policy: HttpRedirectPolicy::Follow, request_id: "http-1".to_string(), stream_response: true, } @@ -717,6 +737,7 @@ async fn http_response_body_stream_fails_when_transport_disconnects() -> Result< headers: Vec::new(), body: None, timeout_ms: None, + redirect_policy: HttpRedirectPolicy::Follow, request_id: "caller-stream-id".to_string(), stream_response: false, }), @@ -759,6 +780,7 @@ async fn http_response_body_stream_reports_disconnect_when_queue_is_full() -> Re headers: Vec::new(), body: None, timeout_ms: None, + redirect_policy: HttpRedirectPolicy::Follow, request_id: "http-1".to_string(), stream_response: true, } @@ -796,6 +818,7 @@ async fn http_response_body_stream_reports_disconnect_when_queue_is_full() -> Re headers: Vec::new(), body: None, timeout_ms: None, + redirect_policy: HttpRedirectPolicy::Follow, request_id: "caller-stream-id".to_string(), stream_response: false, }), @@ -852,6 +875,7 @@ async fn http_response_body_stream_reports_backpressure_truncation() -> Result<( headers: Vec::new(), body: None, timeout_ms: None, + redirect_policy: HttpRedirectPolicy::Follow, request_id: "http-1".to_string(), stream_response: true, } @@ -894,6 +918,7 @@ async fn http_response_body_stream_reports_backpressure_truncation() -> Result<( headers: Vec::new(), body: None, timeout_ms: None, + redirect_policy: HttpRedirectPolicy::Follow, request_id: "caller-stream-id".to_string(), stream_response: false, }), diff --git a/codex-rs/exec-server/tests/http_request.rs b/codex-rs/exec-server/tests/http_request.rs index a4734bf92..1edacc4b1 100644 --- a/codex-rs/exec-server/tests/http_request.rs +++ b/codex-rs/exec-server/tests/http_request.rs @@ -7,6 +7,7 @@ use std::io::ErrorKind; use std::time::Duration; use codex_exec_server::HttpHeader; +use codex_exec_server::HttpRedirectPolicy; use codex_exec_server::HttpRequestBodyDeltaNotification; use codex_exec_server::HttpRequestParams; use codex_exec_server::HttpRequestResponse; @@ -63,6 +64,7 @@ async fn exec_server_http_request_buffers_response_body() -> anyhow::Result<()> }], body: Some(b"request-body".to_vec().into()), timeout_ms: Some(5_000), + redirect_policy: HttpRedirectPolicy::Follow, request_id: "buffered-request".to_string(), stream_response: false, })?, @@ -108,6 +110,61 @@ async fn exec_server_http_request_buffers_response_body() -> anyhow::Result<()> Ok(()) } +/// What this tests: OAuth callers can inspect redirect responses without the +/// executor following the Location header. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn exec_server_http_request_can_stop_at_redirects() -> anyhow::Result<()> { + let mut server = exec_server().await?; + initialize_exec_server(&mut server).await?; + + let listener = TcpListener::bind("127.0.0.1:0").await?; + let base_url = format!("http://{}", listener.local_addr()?); + let http_request_id = server + .send_request( + "http/request", + serde_json::to_value(HttpRequestParams { + method: "GET".to_string(), + url: format!("{base_url}/redirect"), + headers: Vec::new(), + body: None, + timeout_ms: Some(5_000), + redirect_policy: HttpRedirectPolicy::Stop, + request_id: "redirect-request".to_string(), + stream_response: false, + })?, + ) + .await?; + + let captured = accept_http_request(&listener).await?; + assert_eq!(captured.request_line, "GET /redirect HTTP/1.1"); + respond_with_status_and_headers( + captured.stream, + "302 Found", + &[("location", &format!("{base_url}/final"))], + b"redirect", + ) + .await?; + + let response: HttpRequestResponse = wait_for_response(&mut server, http_request_id).await?; + assert_eq!( + ( + response.status, + response_header(&response.headers, "location"), + response.body.into_inner(), + ), + (302, Some(format!("{base_url}/final")), b"redirect".to_vec(),) + ); + assert!( + timeout(Duration::from_millis(100), listener.accept()) + .await + .is_err(), + "redirect target should not be requested" + ); + + server.shutdown().await?; + Ok(()) +} + /// What this tests: a real exec-server websocket `http/request` can return /// response headers immediately and stream the response body as ordered /// `http/request/bodyDelta` notifications. @@ -132,6 +189,7 @@ async fn exec_server_http_request_streams_response_body_notifications() -> anyho }], body: None, timeout_ms: Some(5_000), + redirect_policy: HttpRedirectPolicy::Follow, request_id: "stream-1".to_string(), stream_response: true, })?, @@ -218,6 +276,7 @@ async fn exec_server_http_request_rejects_duplicate_stream_request_ids() -> anyh headers: Vec::new(), body: None, timeout_ms: None, + redirect_policy: HttpRedirectPolicy::Follow, request_id: "stream-dup".to_string(), stream_response: true, })?, @@ -241,6 +300,7 @@ async fn exec_server_http_request_rejects_duplicate_stream_request_ids() -> anyh headers: Vec::new(), body: None, timeout_ms: None, + redirect_policy: HttpRedirectPolicy::Follow, request_id: "stream-dup".to_string(), stream_response: true, })?, @@ -295,6 +355,7 @@ async fn exec_server_http_request_honors_optional_timeout() -> anyhow::Result<() headers: Vec::new(), body: None, timeout_ms: None, + redirect_policy: HttpRedirectPolicy::Follow, request_id: "buffered-request".to_string(), stream_response: false, })?, @@ -320,6 +381,7 @@ async fn exec_server_http_request_honors_optional_timeout() -> anyhow::Result<() headers: Vec::new(), body: None, timeout_ms: Some(10), + redirect_policy: HttpRedirectPolicy::Follow, request_id: "buffered-request".to_string(), stream_response: false, })?, diff --git a/codex-rs/rmcp-client/Cargo.toml b/codex-rs/rmcp-client/Cargo.toml index 44f3eb139..517f7270b 100644 --- a/codex-rs/rmcp-client/Cargo.toml +++ b/codex-rs/rmcp-client/Cargo.toml @@ -16,7 +16,6 @@ axum = { workspace = true, default-features = false, features = [ ] } base64 = { workspace = true } codex-api = { workspace = true } -codex-client = { workspace = true } codex-config = { workspace = true } codex-exec-server = { workspace = true } codex-keyring-store = { workspace = true } diff --git a/codex-rs/rmcp-client/src/auth_status.rs b/codex-rs/rmcp-client/src/auth_status.rs index b2bd834d5..5b8378e67 100644 --- a/codex-rs/rmcp-client/src/auth_status.rs +++ b/codex-rs/rmcp-client/src/auth_status.rs @@ -1,7 +1,9 @@ use std::collections::HashMap; +use std::sync::Arc; use std::time::Duration; use anyhow::Result; +use codex_exec_server::HttpClient; use codex_protocol::protocol::McpAuthStatus; use futures::FutureExt; use reqwest::Client; @@ -13,6 +15,7 @@ use tracing::debug; use crate::oauth::StoredOAuthTokenStatus; use crate::oauth::oauth_token_status; +use crate::oauth_http_client::OAuthHttpClientAdapter; use crate::utils::apply_default_headers; use crate::utils::build_default_headers; use codex_config::types::AuthKeyringBackendKind; @@ -25,6 +28,11 @@ pub struct StreamableHttpOAuthDiscovery { pub scopes_supported: Option>, } +enum AuthStatusCheck { + Complete(McpAuthStatus), + Discover(HeaderMap), +} + /// Determine the authentication status for a streamable HTTP MCP server. pub async fn determine_streamable_http_auth_status( server_name: &str, @@ -35,24 +43,100 @@ pub async fn determine_streamable_http_auth_status( store_mode: OAuthCredentialsStoreMode, keyring_backend_kind: AuthKeyringBackendKind, ) -> Result { + let default_headers = match auth_status_before_discovery( + server_name, + url, + bearer_token_env_var, + http_headers, + env_http_headers, + store_mode, + keyring_backend_kind, + )? { + AuthStatusCheck::Complete(status) => return Ok(status), + AuthStatusCheck::Discover(default_headers) => default_headers, + }; + + determine_auth_status_from_discovery( + server_name, + url, + discover_streamable_http_oauth_with_headers(url, &default_headers).await, + ) +} + +/// Determine authentication status while routing OAuth discovery through the +/// provided HTTP client. +#[allow(clippy::too_many_arguments)] +pub async fn determine_streamable_http_auth_status_with_http_client( + server_name: &str, + url: &str, + bearer_token_env_var: Option<&str>, + http_headers: Option>, + env_http_headers: Option>, + store_mode: OAuthCredentialsStoreMode, + keyring_backend_kind: AuthKeyringBackendKind, + http_client: Arc, +) -> Result { + let default_headers = match auth_status_before_discovery( + server_name, + url, + bearer_token_env_var, + http_headers, + env_http_headers, + store_mode, + keyring_backend_kind, + )? { + AuthStatusCheck::Complete(status) => return Ok(status), + AuthStatusCheck::Discover(default_headers) => default_headers, + }; + determine_auth_status_from_discovery( + server_name, + url, + discover_streamable_http_oauth_with_headers_and_http_client( + url, + default_headers, + http_client, + ) + .await, + ) +} + +fn auth_status_before_discovery( + server_name: &str, + url: &str, + bearer_token_env_var: Option<&str>, + http_headers: Option>, + env_http_headers: Option>, + store_mode: OAuthCredentialsStoreMode, + keyring_backend_kind: AuthKeyringBackendKind, +) -> Result { if bearer_token_env_var.is_some() { - return Ok(McpAuthStatus::BearerToken); + return Ok(AuthStatusCheck::Complete(McpAuthStatus::BearerToken)); } let default_headers = build_default_headers(http_headers, env_http_headers)?; if default_headers.contains_key(AUTHORIZATION) { - return Ok(McpAuthStatus::BearerToken); + return Ok(AuthStatusCheck::Complete(McpAuthStatus::BearerToken)); } match oauth_token_status(server_name, url, store_mode, keyring_backend_kind)? { - StoredOAuthTokenStatus::Usable => return Ok(McpAuthStatus::OAuth), + StoredOAuthTokenStatus::Usable => { + return Ok(AuthStatusCheck::Complete(McpAuthStatus::OAuth)); + } StoredOAuthTokenStatus::AuthorizationRequired => { - return Ok(McpAuthStatus::NotLoggedIn); + return Ok(AuthStatusCheck::Complete(McpAuthStatus::NotLoggedIn)); } StoredOAuthTokenStatus::Missing => {} } - match discover_streamable_http_oauth_with_headers(url, &default_headers).await { + Ok(AuthStatusCheck::Discover(default_headers)) +} + +fn determine_auth_status_from_discovery( + server_name: &str, + url: &str, + discovery: Result>, +) -> Result { + match discovery { Ok(Some(_)) => Ok(McpAuthStatus::NotLoggedIn), Ok(None) => Ok(McpAuthStatus::Unsupported), Err(error) => { @@ -82,6 +166,17 @@ pub async fn discover_streamable_http_oauth( discover_streamable_http_oauth_with_headers(url, &default_headers).await } +pub async fn discover_streamable_http_oauth_with_http_client( + url: &str, + http_headers: Option>, + env_http_headers: Option>, + http_client: Arc, +) -> Result> { + let default_headers = build_default_headers(http_headers, env_http_headers)?; + discover_streamable_http_oauth_with_headers_and_http_client(url, default_headers, http_client) + .await +} + async fn discover_streamable_http_oauth_with_headers( url: &str, default_headers: &HeaderMap, @@ -92,6 +187,25 @@ async fn discover_streamable_http_oauth_with_headers( let client = apply_default_headers(builder, default_headers).build()?; let mut authorization_manager = AuthorizationManager::new(url).await?; authorization_manager.with_client(client)?; + discover_streamable_http_oauth_with_manager(&authorization_manager).await +} + +async fn discover_streamable_http_oauth_with_headers_and_http_client( + url: &str, + default_headers: HeaderMap, + http_client: Arc, +) -> Result> { + let authorization_manager = AuthorizationManager::new_with_oauth_http_client( + url, + Arc::new(OAuthHttpClientAdapter::new(http_client, default_headers)), + ) + .await?; + discover_streamable_http_oauth_with_manager(&authorization_manager).await +} + +async fn discover_streamable_http_oauth_with_manager( + authorization_manager: &AuthorizationManager, +) -> Result> { match authorization_manager.discover_metadata().boxed().await { Ok(metadata) => Ok(Some(StreamableHttpOAuthDiscovery { scopes_supported: normalize_scopes(metadata.scopes_supported), diff --git a/codex-rs/rmcp-client/src/http_client_adapter.rs b/codex-rs/rmcp-client/src/http_client_adapter.rs index a9cdc7b88..19befb623 100644 --- a/codex-rs/rmcp-client/src/http_client_adapter.rs +++ b/codex-rs/rmcp-client/src/http_client_adapter.rs @@ -16,6 +16,7 @@ use codex_api::SharedAuthProvider; use codex_exec_server::ExecServerError; use codex_exec_server::HttpClient; use codex_exec_server::HttpHeader; +use codex_exec_server::HttpRedirectPolicy; use codex_exec_server::HttpRequestParams; use codex_exec_server::HttpResponseBodyStream; use futures::StreamExt; @@ -135,6 +136,7 @@ impl StreamableHttpClient for StreamableHttpClientAdapter { headers: protocol_headers(&headers), body: Some(body.into()), timeout_ms: None, + redirect_policy: HttpRedirectPolicy::Follow, request_id: "buffered-request".to_string(), stream_response: true, }) @@ -261,6 +263,7 @@ impl StreamableHttpClient for StreamableHttpClientAdapter { headers: protocol_headers(&headers), body: None, timeout_ms: None, + redirect_policy: HttpRedirectPolicy::Follow, request_id: "buffered-request".to_string(), stream_response: false, }) @@ -330,6 +333,7 @@ impl StreamableHttpClient for StreamableHttpClientAdapter { headers: protocol_headers(&headers), body: None, timeout_ms: None, + redirect_policy: HttpRedirectPolicy::Follow, request_id: "buffered-request".to_string(), stream_response: true, }) diff --git a/codex-rs/rmcp-client/src/lib.rs b/codex-rs/rmcp-client/src/lib.rs index e1ee18c75..e1041fdaf 100644 --- a/codex-rs/rmcp-client/src/lib.rs +++ b/codex-rs/rmcp-client/src/lib.rs @@ -5,6 +5,7 @@ mod http_client_adapter; mod in_process_transport; mod logging_client_handler; mod oauth; +mod oauth_http_client; mod perform_oauth_login; mod program_resolver; mod rmcp_client; @@ -13,7 +14,9 @@ mod utils; pub use auth_status::StreamableHttpOAuthDiscovery; pub use auth_status::determine_streamable_http_auth_status; +pub use auth_status::determine_streamable_http_auth_status_with_http_client; pub use auth_status::discover_streamable_http_oauth; +pub use auth_status::discover_streamable_http_oauth_with_http_client; pub use auth_status::supports_oauth_login; pub use codex_protocol::protocol::McpAuthStatus; pub use in_process_transport::InProcessTransportFactory; @@ -26,6 +29,7 @@ pub use perform_oauth_login::OAuthProviderError; pub use perform_oauth_login::OauthLoginHandle; pub use perform_oauth_login::perform_oauth_login; pub use perform_oauth_login::perform_oauth_login_return_url; +pub use perform_oauth_login::perform_oauth_login_return_url_with_http_client; pub use perform_oauth_login::perform_oauth_login_silent; pub use rmcp::model::ElicitationAction; pub use rmcp_client::Elicitation; diff --git a/codex-rs/rmcp-client/src/oauth_http_client.rs b/codex-rs/rmcp-client/src/oauth_http_client.rs new file mode 100644 index 000000000..44cf63cb7 --- /dev/null +++ b/codex-rs/rmcp-client/src/oauth_http_client.rs @@ -0,0 +1,116 @@ +use std::sync::Arc; +use std::sync::atomic::AtomicU64; +use std::sync::atomic::Ordering; +use std::time::Duration; + +use codex_exec_server::HttpClient; +use codex_exec_server::HttpHeader; +use codex_exec_server::HttpRedirectPolicy; +use codex_exec_server::HttpRequestParams; +use oauth2::HttpRequest; +use oauth2::HttpResponse; +use reqwest::header::HeaderMap; +use rmcp::transport::auth::OAuthHttpClient; +use rmcp::transport::auth::OAuthHttpClientError; +use rmcp::transport::auth::OAuthHttpClientFuture; +use rmcp::transport::auth::OAuthHttpRedirectPolicy; +use rmcp::transport::auth::OAuthHttpRequest; + +const MAX_OAUTH_HTTP_RESPONSE_BODY_BYTES: usize = 1024 * 1024; +static NEXT_OAUTH_REQUEST_ID: AtomicU64 = AtomicU64::new(0); + +#[derive(Clone)] +pub(crate) struct OAuthHttpClientAdapter { + http_client: Arc, + default_headers: HeaderMap, +} + +impl OAuthHttpClientAdapter { + pub(crate) fn new(http_client: Arc, default_headers: HeaderMap) -> Self { + Self { + http_client, + default_headers, + } + } + + async fn execute_request( + &self, + request: HttpRequest, + redirect_policy: OAuthHttpRedirectPolicy, + timeout: Option, + ) -> Result { + let redirect_policy = match redirect_policy { + OAuthHttpRedirectPolicy::Follow => HttpRedirectPolicy::Follow, + OAuthHttpRedirectPolicy::Stop => HttpRedirectPolicy::Stop, + _ => { + return Err(OAuthHttpClientError::new( + "unsupported OAuth HTTP redirect policy", + )); + } + }; + let (parts, body) = request.into_parts(); + let mut headers = self.default_headers.clone(); + for name in parts.headers.keys() { + headers.remove(name); + } + headers.extend(parts.headers); + let headers = headers + .iter() + .map(|(name, value)| { + Ok(HttpHeader { + name: name.as_str().to_string(), + value: value + .to_str() + .map_err(|error| OAuthHttpClientError::new(error.to_string()))? + .to_string(), + }) + }) + .collect::, OAuthHttpClientError>>()?; + let timeout_ms = timeout.map(|timeout| { + u64::try_from(timeout.as_millis()) + .unwrap_or(u64::MAX) + .max(1) + }); + let request_id = NEXT_OAUTH_REQUEST_ID.fetch_add(1, Ordering::Relaxed); + let (response, mut body_stream) = self + .http_client + .http_request_stream(HttpRequestParams { + method: parts.method.to_string(), + url: parts.uri.to_string(), + headers, + body: (!body.is_empty()).then_some(body.into()), + timeout_ms, + redirect_policy, + request_id: format!("oauth-request-{request_id}"), + stream_response: true, + }) + .await + .map_err(|error| OAuthHttpClientError::new(error.to_string()))?; + let mut body = Vec::new(); + while let Some(chunk) = body_stream + .recv() + .await + .map_err(|error| OAuthHttpClientError::new(error.to_string()))? + { + if chunk.len() > MAX_OAUTH_HTTP_RESPONSE_BODY_BYTES - body.len() { + return Err(OAuthHttpClientError::new(format!( + "OAuth HTTP response body exceeds {MAX_OAUTH_HTTP_RESPONSE_BODY_BYTES} bytes" + ))); + } + body.extend_from_slice(&chunk); + } + let mut builder = oauth2::http::Response::builder().status(response.status); + for header in response.headers { + builder = builder.header(header.name, header.value); + } + builder + .body(body) + .map_err(|error| OAuthHttpClientError::new(error.to_string())) + } +} + +impl OAuthHttpClient for OAuthHttpClientAdapter { + fn execute(&self, request: OAuthHttpRequest) -> OAuthHttpClientFuture<'_> { + Box::pin(self.execute_request(request.request, request.redirect_policy, request.timeout)) + } +} diff --git a/codex-rs/rmcp-client/src/perform_oauth_login.rs b/codex-rs/rmcp-client/src/perform_oauth_login.rs index 8c72f320b..7a9d242bc 100644 --- a/codex-rs/rmcp-client/src/perform_oauth_login.rs +++ b/codex-rs/rmcp-client/src/perform_oauth_login.rs @@ -9,11 +9,13 @@ use anyhow::anyhow; use anyhow::bail; use base64::Engine; use base64::engine::general_purpose::URL_SAFE_NO_PAD; -use reqwest::ClientBuilder; +use codex_exec_server::HttpClient; +use codex_exec_server::ReqwestHttpClient; use reqwest::Url; use rmcp::transport::AuthorizationManager; use rmcp::transport::AuthorizationSession; use rmcp::transport::auth::OAuthClientConfig; +use rmcp::transport::auth::OAuthHttpClient; use rmcp::transport::auth::OAuthState; use sha2::Digest; use sha2::Sha256; @@ -26,15 +28,16 @@ use urlencoding::decode; use crate::StoredOAuthTokens; use crate::WrappedOAuthTokenResponse; use crate::oauth::compute_expires_at_millis; +use crate::oauth_http_client::OAuthHttpClientAdapter; use crate::save_oauth_tokens; -use crate::utils::apply_default_headers; use crate::utils::build_default_headers; use codex_config::types::AuthKeyringBackendKind; use codex_config::types::OAuthCredentialsStoreMode; -struct OauthHeaders { +struct OAuthHttpContext { http_headers: Option>, env_http_headers: Option>, + http_client: Arc, } struct CallbackServerGuard { @@ -154,16 +157,17 @@ async fn perform_oauth_login_with_browser_output( callback_url: Option<&str>, emit_browser_url: bool, ) -> Result<()> { - let headers = OauthHeaders { + let http_context = OAuthHttpContext { http_headers, env_http_headers, + http_client: Arc::new(ReqwestHttpClient), }; OauthLoginFlow::new( server_name, server_url, store_mode, keyring_backend_kind, - headers, + http_context, scopes, oauth_client_id, oauth_resource, @@ -192,16 +196,51 @@ pub async fn perform_oauth_login_return_url( callback_port: Option, callback_url: Option<&str>, ) -> Result { - let headers = OauthHeaders { + perform_oauth_login_return_url_with_http_client( + server_name, + server_url, + store_mode, + keyring_backend_kind, http_headers, env_http_headers, + scopes, + oauth_client_id, + oauth_resource, + timeout_secs, + callback_port, + callback_url, + Arc::new(ReqwestHttpClient), + ) + .await +} + +#[allow(clippy::too_many_arguments)] +pub async fn perform_oauth_login_return_url_with_http_client( + server_name: &str, + server_url: &str, + store_mode: OAuthCredentialsStoreMode, + keyring_backend_kind: AuthKeyringBackendKind, + http_headers: Option>, + env_http_headers: Option>, + scopes: &[String], + oauth_client_id: Option<&str>, + oauth_resource: Option<&str>, + timeout_secs: Option, + callback_port: Option, + callback_url: Option<&str>, + http_client: Arc, +) -> Result { + let http_context = OAuthHttpContext { + http_headers, + env_http_headers, + http_client, }; let flow = OauthLoginFlow::new( server_name, server_url, store_mode, keyring_backend_kind, - headers, + http_context, scopes, oauth_client_id, oauth_resource, @@ -456,7 +495,7 @@ impl OauthLoginFlow { server_url: &str, store_mode: OAuthCredentialsStoreMode, keyring_backend_kind: AuthKeyringBackendKind, - headers: OauthHeaders, + http_context: OAuthHttpContext, scopes: &[String], oauth_client_id: Option<&str>, oauth_resource: Option<&str>, @@ -487,17 +526,18 @@ impl OauthLoginFlow { let (tx, rx) = oneshot::channel(); spawn_callback_server(server, tx, callback_path); - let OauthHeaders { + let OAuthHttpContext { http_headers, env_http_headers, - } = headers; + http_client, + } = http_context; let default_headers = build_default_headers(http_headers, env_http_headers)?; - let http_client = apply_default_headers(ClientBuilder::new(), &default_headers).build()?; + let oauth_http_client = Arc::new(OAuthHttpClientAdapter::new(http_client, default_headers)); let scope_refs: Vec<&str> = scopes.iter().map(String::as_str).collect(); let oauth_state = start_authorization( server_url, - http_client, + oauth_http_client, &scope_refs, &redirect_uri, oauth_client_id, @@ -620,22 +660,23 @@ impl OauthLoginFlow { async fn start_authorization( server_url: &str, - http_client: reqwest::Client, + http_client: Arc, scopes: &[&str], redirect_uri: &str, oauth_client_id: Option<&str>, ) -> Result { let Some(oauth_client_id) = oauth_client_id.filter(|client_id| !client_id.trim().is_empty()) else { - let mut oauth_state = OAuthState::new(server_url, Some(http_client)).await?; + let mut oauth_state = + OAuthState::new_with_oauth_http_client(server_url, http_client).await?; oauth_state .start_authorization(scopes, redirect_uri, Some("Codex")) .await?; return Ok(oauth_state); }; - let mut auth_manager = AuthorizationManager::new(server_url).await?; - auth_manager.with_client(http_client)?; + let mut auth_manager = + AuthorizationManager::new_with_oauth_http_client(server_url, http_client).await?; let metadata = auth_manager.discover_metadata().await?; auth_manager.set_metadata(metadata); auth_manager.configure_client( @@ -668,15 +709,20 @@ fn append_query_param(url: &str, key: &str, value: Option<&str>) -> String { #[cfg(test)] mod tests { + use std::sync::Arc; + use axum::Json; use axum::Router; use axum::routing::get; + use codex_exec_server::ReqwestHttpClient; use pretty_assertions::assert_eq; use reqwest::Url; + use reqwest::header::HeaderMap; use serde_json::json; use tokio::net::TcpListener; use super::CallbackOutcome; + use super::OAuthHttpClientAdapter; use super::OAuthProviderError; use super::append_callback_id_to_redirect_uri; use super::append_query_param; @@ -727,7 +773,10 @@ mod tests { let base_url = spawn_oauth_metadata_server().await; let oauth_state = start_authorization( &format!("{base_url}/mcp"), - reqwest::Client::new(), + Arc::new(OAuthHttpClientAdapter::new( + Arc::new(ReqwestHttpClient), + HeaderMap::new(), + )), &[], "http://127.0.0.1/callback", Some("eci-prd-pub-codex-123"), diff --git a/codex-rs/rmcp-client/src/rmcp_client.rs b/codex-rs/rmcp-client/src/rmcp_client.rs index 93abf7d11..c6527990f 100644 --- a/codex-rs/rmcp-client/src/rmcp_client.rs +++ b/codex-rs/rmcp-client/src/rmcp_client.rs @@ -11,7 +11,6 @@ use std::time::Instant; use anyhow::Result; use anyhow::anyhow; use codex_api::SharedAuthProvider; -use codex_client::maybe_build_rustls_client_config_with_custom_ca; use codex_config::types::AuthKeyringBackendKind; use codex_config::types::McpServerEnvVar; use codex_exec_server::HttpClient; @@ -68,11 +67,11 @@ use crate::in_process_transport::InProcessTransportFactory; use crate::load_oauth_tokens; use crate::oauth::OAuthPersistor; use crate::oauth::StoredOAuthTokens; +use crate::oauth_http_client::OAuthHttpClientAdapter; use crate::stdio_server_launcher::StdioServerCommand; use crate::stdio_server_launcher::StdioServerLauncher; use crate::stdio_server_launcher::StdioServerProcessHandle; use crate::stdio_server_launcher::StdioServerTransport; -use crate::utils::apply_default_headers; use crate::utils::build_default_headers; use codex_config::types::OAuthCredentialsStoreMode; @@ -1166,16 +1165,12 @@ async fn create_oauth_transport_and_runtime( StreamableHttpClientTransport>, OAuthPersistor, )> { - let mut builder = apply_default_headers(reqwest::Client::builder(), &default_headers); - if let Some(tls_config) = maybe_build_rustls_client_config_with_custom_ca()? { - builder = builder.tls_backend_preconfigured(tls_config.as_ref().clone()); - } - let oauth_metadata_client = builder.build()?; - // TODO(aibrahim): teach OAuth bootstrap and refresh to use the same - // shared HTTP client abstraction instead of always creating the local - // reqwest metadata client here. + let oauth_http_client = Arc::new(OAuthHttpClientAdapter::new( + http_client.clone(), + default_headers.clone(), + )); let mut oauth_state = - OAuthState::new(url.to_string(), Some(oauth_metadata_client.clone())).await?; + OAuthState::new_with_oauth_http_client(url.to_string(), oauth_http_client).await?; oauth_state .set_credentials(