agent/codex
2
0
Fork 0
mirror of https://github.com/pchuan98/codex.git synced 2026-07-01 00:31:56 +08:00
Code Issues Packages Projects Releases Wiki Activity

Support OAuth for HTTP MCP servers from selected executor plugins (#28529)

Browse Source 
## Why

#28522 routes selected-plugin HTTP MCP traffic through the owning
executor, but OAuth bootstrap and refresh still used host-local clients.
Executor-only servers therefore cannot complete discovery or login
through the same network boundary as the MCP connection.

## What changed

- adapt `codex_exec_server::HttpClient` to RMCP 1.8's `OAuthHttpClient`
contract
- let RMCP own discovery, dynamic registration, PKCE, token exchange,
and refresh
- route auth status, persisted-token startup, and app-server login
through the server runtime while preserving the existing local discovery
path
- add optional `threadId` to `mcpServer/oauth/login` and echo it in the
completion notification
- implement RMCP's redirect policy and 1 MiB OAuth response limit over
executor HTTP
- cover selected-thread OAuth discovery and login through an
executor-only route

Depends on #28522.
This commit is contained in:
jif
2026-06-25 10:31:17 +01:00
committed by GitHub
parent 6368937939
commit b215961a56
38 changed files with 763 additions and 96 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/Cargo.lock
Generated
-1
View File
@@ -3757,7 +3757,6 @@ dependencies = [
"base64 0.22.1",
"bytes",
"codex-api",
"codex-client",
"codex-config",
"codex-exec-server",
"codex-keyring-store",
codex-rs/app-server-protocol/schema/json/ClientRequest.json
Generated
+6
View File
@@ -1619,6 +1619,12 @@
"null"
]
},
"threadId": {
"type": [
"string",
"null"
]
},
"timeoutSecs": {
"format": "int64",
"type": [
codex-rs/app-server-protocol/schema/json/ServerNotification.json
Generated
+6
View File
@@ -2370,6 +2370,12 @@
},
"success": {
"type": "boolean"
},
"threadId": {
"type": [
"string",
"null"
]
}
},
"required": [
codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.schemas.json
Generated
+12
View File
@@ -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": [
codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.v2.schemas.json
Generated
+12
View File
@@ -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": [
codex-rs/app-server-protocol/schema/json/v2/McpServerOauthLoginCompletedNotification.json
Generated
+6
View File
@@ -12,6 +12,12 @@
},
"success": {
"type": "boolean"
},
"threadId": {
"type": [
"string",
"null"
]
}
},
"required": [
codex-rs/app-server-protocol/schema/json/v2/McpServerOauthLoginParams.json
Generated
+6
View File
@@ -13,6 +13,12 @@
"null"
]
},
"threadId": {
"type": [
"string",
"null"
]
},
"timeoutSecs": {
"format": "int64",
"type": [
codex-rs/app-server-protocol/schema/typescript/v2/McpServerOauthLoginCompletedNotification.ts
Generated
+1 -1
View File
@@ -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, };
codex-rs/app-server-protocol/schema/typescript/v2/McpServerOauthLoginParams.ts
Generated
+1 -1
View File
@@ -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<string> | null, timeoutSecs?: bigint | null, };
export type McpServerOauthLoginParams = { name: string, threadId?: string | null, scopes?: Array<string> | null, timeoutSecs?: bigint | null, };
codex-rs/app-server-protocol/src/protocol/common.rs
+1
View File
@@ -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,
},
codex-rs/app-server-protocol/src/protocol/v2/mcp.rs
+3
View File
@@ -186,6 +186,8 @@ pub struct McpServerRefreshResponse {}
#[ts(export_to = "v2/")]
pub struct McpServerOauthLoginParams {
pub name: String,
#[ts(optional = nullable)]
pub thread_id: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
#[ts(optional = nullable)]
pub scopes: Option<Vec<String>>,
@@ -215,6 +217,7 @@ pub struct McpToolCallProgressNotification {
#[ts(export_to = "v2/")]
pub struct McpServerOauthLoginCompletedNotification {
pub name: String,
pub thread_id: Option<String>,
pub success: bool,
#[serde(default, skip_serializing_if = "Option::is_none")]
#[ts(optional)]
codex-rs/app-server/README.md
+2 -2
View File
@@ -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 `<plugin>@<marketplace>` 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
codex-rs/app-server/src/request_processors.rs
+2 -2
View File
@@ -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;
codex-rs/app-server/src/request_processors/mcp_processor.rs
+40 -8
View File
@@ -113,19 +113,37 @@ impl McpRequestProcessor {
&self,
params: McpServerOauthLoginParams,
) -> Result<McpServerOauthLoginResponse, JSONRPCErrorError> {
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,
},
codex-rs/app-server/src/request_processors/plugins.rs
+1
View File
@@ -1883,6 +1883,7 @@ impl PluginRequestProcessor {
let notification = ServerNotification::McpServerOauthLoginCompleted(
McpServerOauthLoginCompletedNotification {
name: notification_name,
thread_id: None,
success,
error,
},
codex-rs/app-server/tests/suite/v2/executor_mcp.rs
+83 -7
View File
@@ -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;
codex-rs/cli/src/mcp_cmd.rs
+6
View File
@@ -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;
codex-rs/codex-mcp/src/lib.rs
+2
View File
@@ -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;
codex-rs/codex-mcp/src/mcp/auth.rs
+100 -26
View File
@@ -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<dyn HttpClient>,
) -> 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<McpOAuthLoginConfig> {
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<dyn HttpClient>,
) -> Option<Vec<String>> {
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<Vec<String>>,
configured_scopes: Option<Vec<String>>,
@@ -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<String, McpAuthStatusEntry>
where
I: IntoIterator<Item = (&'a String, &'a EffectiveMcpServer)>,
@@ -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<McpAuthStatus> {
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
}
}
}
}
codex-rs/codex-mcp/src/mcp/mod.rs
+4
View File
@@ -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;
codex-rs/codex-mcp/src/runtime.rs
+16
View File
@@ -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<Arc<dyn HttpClient>, String> {
Ok(self
.resolve_server_environment(server_name, config)?
.map_or_else(
|| Arc::new(ReqwestHttpClient) as Arc<dyn HttpClient>,
|environment| environment.get_http_client(),
))
}
}
pub(crate) fn emit_duration(metric: &str, duration: Duration, tags: &[(&str, &str)]) {
codex-rs/core/src/connectors.rs
+4 -1
View File
@@ -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,
codex-rs/core/src/session/mcp.rs
+8 -7
View File
@@ -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();
codex-rs/core/src/session/session.rs
+5
View File
@@ -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)
codex-rs/core/tests/suite/rmcp_client.rs
+2
View File
@@ -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,
};
codex-rs/exec-server-protocol/src/protocol.rs
+14
View File
@@ -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<u64>,
/// 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
codex-rs/exec-server/src/client/reqwest_http_client.rs
+16 -5
View File
@@ -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<u64>) -> Result<reqwest::Client, ExecServerError> {
fn build_client(
timeout_ms: Option<u64>,
redirect_policy: HttpRedirectPolicy,
) -> Result<reqwest::Client, ExecServerError> {
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<HttpRequestResponse, 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, _) = 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<u64>) -> Result<Self, JSONRPCErrorError> {
let client = ReqwestHttpClient::build_client(timeout_ms)
pub(crate) fn new(
timeout_ms: Option<u64>,
redirect_policy: HttpRedirectPolicy,
) -> Result<Self, JSONRPCErrorError> {
let client = ReqwestHttpClient::build_client(timeout_ms, redirect_policy)
.map_err(|error| internal_error(error.to_string()))?;
Ok(Self { client })
}
codex-rs/exec-server/src/lib.rs
+1
View File
@@ -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;
codex-rs/exec-server/src/server/handler.rs
+1 -1
View File
@@ -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 {
codex-rs/exec-server/tests/http_client.rs
+25
View File
@@ -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,
}),
codex-rs/exec-server/tests/http_request.rs
+62
View File
@@ -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,
})?,
codex-rs/rmcp-client/Cargo.toml
-1
View File
@@ -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 }
codex-rs/rmcp-client/src/auth_status.rs
+119 -5
View File
@@ -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<Vec<String>>,
}
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<McpAuthStatus> {
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<HashMap<String, String>>,
env_http_headers: Option<HashMap<String, String>>,
store_mode: OAuthCredentialsStoreMode,
keyring_backend_kind: AuthKeyringBackendKind,
http_client: Arc<dyn HttpClient>,
) -> Result<McpAuthStatus> {
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<HashMap<String, String>>,
env_http_headers: Option<HashMap<String, String>>,
store_mode: OAuthCredentialsStoreMode,
keyring_backend_kind: AuthKeyringBackendKind,
) -> Result<AuthStatusCheck> {
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<Option<StreamableHttpOAuthDiscovery>>,
) -> Result<McpAuthStatus> {
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<HashMap<String, String>>,
env_http_headers: Option<HashMap<String, String>>,
http_client: Arc<dyn HttpClient>,
) -> Result<Option<StreamableHttpOAuthDiscovery>> {
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<dyn HttpClient>,
) -> Result<Option<StreamableHttpOAuthDiscovery>> {
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<Option<StreamableHttpOAuthDiscovery>> {
match authorization_manager.discover_metadata().boxed().await {
Ok(metadata) => Ok(Some(StreamableHttpOAuthDiscovery {
scopes_supported: normalize_scopes(metadata.scopes_supported),
codex-rs/rmcp-client/src/http_client_adapter.rs
+4
View File
@@ -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,
})
codex-rs/rmcp-client/src/lib.rs
+4
View File
@@ -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;
codex-rs/rmcp-client/src/oauth_http_client.rs
+116
View File
@@ -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<dyn HttpClient>,
default_headers: HeaderMap,
}
impl OAuthHttpClientAdapter {
pub(crate) fn new(http_client: Arc<dyn HttpClient>, default_headers: HeaderMap) -> Self {
Self {
http_client,
default_headers,
}
}
async fn execute_request(
&self,
request: HttpRequest,
redirect_policy: OAuthHttpRedirectPolicy,
timeout: Option<Duration>,
) -> Result<HttpResponse, OAuthHttpClientError> {
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::<Result<Vec<_>, 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))
}
}
codex-rs/rmcp-client/src/perform_oauth_login.rs
+66 -17
View File
@@ -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<HashMap<String, String>>,
env_http_headers: Option<HashMap<String, String>>,
http_client: Arc<dyn HttpClient>,
}
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<u16>,
callback_url: Option<&str>,
) -> Result<OauthLoginHandle> {
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<HashMap<String, String>>,
env_http_headers: Option<HashMap<String, String>>,
scopes: &[String],
oauth_client_id: Option<&str>,
oauth_resource: Option<&str>,
timeout_secs: Option<i64>,
callback_port: Option<u16>,
callback_url: Option<&str>,
http_client: Arc<dyn HttpClient>,
) -> Result<OauthLoginHandle> {
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<dyn OAuthHttpClient>,
scopes: &[&str],
redirect_uri: &str,
oauth_client_id: Option<&str>,
) -> Result<OAuthState> {
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"),
codex-rs/rmcp-client/src/rmcp_client.rs
+6 -11
View File
@@ -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<AuthClient<StreamableHttpClientAdapter>>,
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(