[rmcp-client] Refresh OAuth credentials after invalid_token responses

codex-rs/rmcp-client/src/http_client_adapter.rs

@@ -41,6 +41,7 @@ use sse_stream::SseStream;
 mod www_authenticate;
 
 use self::www_authenticate::insufficient_scope_challenge;
+pub(crate) use self::www_authenticate::is_bearer_invalid_token_challenge;
 
 const EVENT_STREAM_MIME_TYPE: &str = "text/event-stream";
 const JSON_MIME_TYPE: &str = "application/json";

codex-rs/rmcp-client/src/http_client_adapter/www_authenticate.rs

@@ -60,6 +60,10 @@ impl BearerChallenge {
             Parameter::Missing | Parameter::Value(_) | Parameter::Invalid => None,
         }
     }
+
+    fn has_error(&self, expected: &str) -> bool {
+        matches!(&self.error, Parameter::Value(error) if error == expected)
+    }
 }
 
 /// Finds a Bearer insufficient-scope challenge among all `WWW-Authenticate`
@@ -80,6 +84,12 @@ pub(super) fn insufficient_scope_challenge(
         })
 }
 
+/// Returns whether a `WWW-Authenticate` field value contains a Bearer
+/// challenge with an `invalid_token` error.
+pub(crate) fn is_bearer_invalid_token_challenge(header: &str) -> bool {
+    parse_bearer_challenge(header, "invalid_token").is_some()
+}
+
 /// Parses a Bearer `WWW-Authenticate` challenge with an `insufficient_scope`
 /// error and extracts its optional required scope.
 ///
@@ -96,6 +106,11 @@ pub(super) fn insufficient_scope_challenge(
 ///
 /// RMCP has related parsing logic, but it is private to that crate.
 fn parse_bearer_insufficient_scope(header: &str) -> Option<BearerInsufficientScope> {
+    parse_bearer_challenge(header, "insufficient_scope")
+        .and_then(BearerChallenge::into_insufficient_scope)
+}
+
+fn parse_bearer_challenge(header: &str, expected_error: &str) -> Option<BearerChallenge> {
     let segments = split_unquoted_segments(header)?;
     let mut bearer_challenge: Option<BearerChallenge> = None;
 
@@ -107,9 +122,8 @@ fn parse_bearer_insufficient_scope(header: &str) -> Option<BearerInsufficientSco
             continue;
         }
 
-        if let Some(challenge) = bearer_challenge
-            .take()
-            .and_then(BearerChallenge::into_insufficient_scope)
+        if let Some(challenge) = bearer_challenge.take()
+            && challenge.has_error(expected_error)
         {
             return Some(challenge);
         }
@@ -124,7 +138,7 @@ fn parse_bearer_insufficient_scope(header: &str) -> Option<BearerInsufficientSco
         }
     }
 
-    bearer_challenge.and_then(BearerChallenge::into_insufficient_scope)
+    bearer_challenge.filter(|challenge| challenge.has_error(expected_error))
 }
 
 fn parse_challenge_start(segment: &str) -> Option<ChallengeStart<'_>> {

codex-rs/rmcp-client/src/http_client_adapter/www_authenticate_tests.rs

@@ -4,6 +4,7 @@ use pretty_assertions::assert_eq;
 use super::BearerInsufficientScope;
 use super::InsufficientScopeChallenge;
 use super::insufficient_scope_challenge;
+use super::is_bearer_invalid_token_challenge;
 use super::parse_bearer_insufficient_scope;
 
 #[test]
@@ -122,3 +123,18 @@ fn selects_bearer_challenge_from_a_later_www_authenticate_field_value() {
         })
     );
 }
+
+#[test]
+fn detects_only_bearer_invalid_token_challenges() {
+    assert!(is_bearer_invalid_token_challenge(
+        r#"Basic realm="example", Bearer error="invalid_token""#
+    ));
+
+    for value in [
+        r#"Bearer realm="example""#,
+        r#"Bearer error="insufficient_scope""#,
+        r#"Basic error="invalid_token""#,
+    ] {
+        assert!(!is_bearer_invalid_token_challenge(value), "{value}");
+    }
+}

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_invalid_token;
 mod perform_oauth_login;
 mod program_resolver;
 mod rmcp_client;
@@ -17,6 +18,7 @@ pub use auth_status::discover_streamable_http_oauth;
 pub use auth_status::supports_oauth_login;
 pub use codex_protocol::protocol::McpAuthStatus;
 pub use in_process_transport::InProcessTransportFactory;
+pub use oauth::OAUTH_REFRESH_REAUTHENTICATION_REQUIRED_ERROR;
 pub use oauth::StoredOAuthTokens;
 pub use oauth::WrappedOAuthTokenResponse;
 pub use oauth::delete_oauth_tokens;

codex-rs/rmcp-client/src/oauth.rs

@@ -19,6 +19,7 @@
 use anyhow::Context;
 use anyhow::Error;
 use anyhow::Result;
+use anyhow::anyhow;
 use codex_config::types::OAuthCredentialsStoreMode;
 use oauth2::AccessToken;
 use oauth2::RefreshToken;
@@ -45,13 +46,17 @@ use tracing::warn;
 
 use codex_keyring_store::DefaultKeyringStore;
 use codex_keyring_store::KeyringStore;
+use rmcp::transport::auth::AuthError;
 use rmcp::transport::auth::AuthorizationManager;
 use tokio::sync::Mutex;
 
 use codex_utils_home_dir::find_codex_home;
 
 const KEYRING_SERVICE: &str = "Codex MCP Credentials";
 const REFRESH_SKEW_MILLIS: u64 = 30_000;
+const MISSING_REFRESH_TOKEN_ERROR: &str = "No refresh token available";
+pub const OAUTH_REFRESH_REAUTHENTICATION_REQUIRED_ERROR: &str =
+    "OAuth refresh token was rejected; reauthentication required";
 
 #[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
 pub struct StoredOAuthTokens {
@@ -345,10 +350,6 @@ impl OAuthPersistor {
         Ok(())
     }
 
-    #[expect(
-        clippy::await_holding_invalid_type,
-        reason = "AuthorizationManager async access must be serialized through its mutex"
-    )]
     pub(crate) async fn refresh_if_needed(&self) -> Result<()> {
         let expires_at = {
             let guard = self.inner.last_credentials.lock().await;
@@ -359,21 +360,43 @@ impl OAuthPersistor {
             return Ok(());
         }
 
+        self.refresh().await
+    }
+
+    pub(crate) async fn force_refresh(&self) -> Result<()> {
+        self.refresh().await
+    }
+
+    #[expect(
+        clippy::await_holding_invalid_type,
+        reason = "AuthorizationManager async access must be serialized through its mutex"
+    )]
+    async fn refresh(&self) -> Result<()> {
         {
             let manager = self.inner.authorization_manager.clone();
             let guard = manager.lock().await;
-            guard.refresh_token().await.with_context(|| {
-                format!(
-                    "failed to refresh OAuth tokens for server {}",
-                    self.inner.server_name
-                )
-            })?;
+            if let Err(error) = guard.refresh_token().await {
+                return Err(refresh_error(error, &self.inner.server_name));
+            }
         }
 
         self.persist_if_needed().await
     }
 }
 
+fn refresh_error(error: AuthError, server_name: &str) -> Error {
+    match error {
+        AuthError::TokenRefreshFailed(message)
+            if message == MISSING_REFRESH_TOKEN_ERROR || message.contains("invalid_grant") =>
+        {
+            anyhow!(OAUTH_REFRESH_REAUTHENTICATION_REQUIRED_ERROR)
+        }
+        error => Error::new(error).context(format!(
+            "failed to refresh OAuth tokens for server {server_name}"
+        )),
+    }
+}
+
 const FALLBACK_FILENAME: &str = ".credentials.json";
 const MCP_SERVER_TYPE: &str = "http";
 

codex-rs/rmcp-client/src/oauth_invalid_token.rs

@@ -0,0 +1,35 @@
+use anyhow::Error;
+use rmcp::service::ClientInitializeError;
+use rmcp::transport::DynamicTransportError;
+use rmcp::transport::streamable_http_client::StreamableHttpError;
+
+use crate::http_client_adapter::StreamableHttpClientAdapterError;
+use crate::http_client_adapter::is_bearer_invalid_token_challenge;
+
+#[derive(Debug, thiserror::Error)]
+#[error("MCP OAuth access token was rejected; credentials refreshed, retry the request")]
+pub(crate) struct RetryRequired;
+
+pub(crate) fn rejected_initialize_request(error: &Error) -> bool {
+    let Some(ClientInitializeError::TransportError { error, context }) = error
+        .chain()
+        .find_map(|source| source.downcast_ref::<ClientInitializeError>())
+    else {
+        return false;
+    };
+
+    context.as_ref() == "send initialize request" && rejected_transport(error)
+}
+
+pub(crate) fn rejected_transport(error: &DynamicTransportError) -> bool {
+    error
+        .error
+        .downcast_ref::<StreamableHttpError<StreamableHttpClientAdapterError>>()
+        .is_some_and(|error| {
+            matches!(
+                error,
+                StreamableHttpError::AuthRequired(auth)
+                    if is_bearer_invalid_token_challenge(&auth.www_authenticate_header)
+            )
+        })
+}

codex-rs/rmcp-client/src/rmcp_client.rs

@@ -41,6 +41,7 @@ use rmcp::model::ReadResourceResult;
 use rmcp::model::RequestId;
 use rmcp::model::ServerResult;
 use rmcp::model::Tool;
+use rmcp::service::ClientInitializeError;
 use rmcp::service::RoleClient;
 use rmcp::service::RunningService;
 use rmcp::service::{self};
@@ -66,6 +67,7 @@ use crate::in_process_transport::InProcessTransportFactory;
 use crate::load_oauth_tokens;
 use crate::oauth::OAuthPersistor;
 use crate::oauth::StoredOAuthTokens;
+use crate::oauth_invalid_token;
 use crate::stdio_server_launcher::StdioServerCommand;
 use crate::stdio_server_launcher::StdioServerLauncher;
 use crate::stdio_server_launcher::StdioServerProcessHandle;
@@ -223,6 +225,10 @@ enum ClientOperationError {
     Timeout { label: String, duration: Duration },
 }
 
+#[derive(Debug, thiserror::Error)]
+#[error("handshaking with MCP server failed: {0}")]
+struct McpHandshakeError(#[source] ClientInitializeError);
+
 pub type Elicitation = CreateElicitationRequestParams;
 
 #[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
@@ -396,9 +402,48 @@ impl RmcpClient {
             }
         };
 
-        let (service, oauth_persistor) =
-            Self::connect_pending_transport(pending_transport, client_service.clone(), timeout)
-                .await?;
+        let startup_oauth_persistor = match &pending_transport {
+            PendingTransport::StreamableHttpWithOAuth {
+                oauth_persistor, ..
+            } => Some(oauth_persistor.clone()),
+            PendingTransport::InProcess { .. }
+            | PendingTransport::Stdio { .. }
+            | PendingTransport::StreamableHttp { .. } => None,
+        };
+        let connect = async {
+            match Self::connect_pending_transport(
+                pending_transport,
+                client_service.clone(),
+                /*timeout*/ None,
+            )
+            .await
+            {
+                Err(error) => {
+                    let Some(oauth_persistor) = startup_oauth_persistor else {
+                        return Err(error);
+                    };
+                    if !oauth_invalid_token::rejected_initialize_request(&error) {
+                        return Err(error);
+                    }
+                    oauth_persistor.force_refresh().await?;
+                    let retry_transport =
+                        Self::create_pending_transport(&self.transport_recipe).await?;
+                    Self::connect_pending_transport(
+                        retry_transport,
+                        client_service.clone(),
+                        /*timeout*/ None,
+                    )
+                    .await
+                }
+                result => result,
+            }
+        };
+        let (service, oauth_persistor) = match timeout {
+            Some(duration) => time::timeout(duration, connect).await.map_err(|_| {
+                anyhow!("timed out handshaking with MCP server after {duration:?}")
+            })??,
+            None => connect.await?,
+        };
 
         let initialize_result_rmcp = service
             .peer()
@@ -596,7 +641,20 @@ impl RmcpClient {
                 }
                 .boxed()
             })
-            .await?;
+            .await;
+        let result = match result {
+            Ok(result) => result,
+            Err(error) => {
+                let Some(oauth_persistor) = self.oauth_persistor().await else {
+                    return Err(error);
+                };
+                if !Self::is_invalid_token_operation_error(&error) {
+                    return Err(error);
+                }
+                oauth_persistor.force_refresh().await?;
+                return Err(oauth_invalid_token::RetryRequired.into());
+            }
+        };
         self.persist_oauth_tokens().await;
         Ok(result)
     }
@@ -849,10 +907,8 @@ impl RmcpClient {
             Some(duration) => time::timeout(duration, transport)
                 .await
                 .map_err(|_| anyhow!("timed out handshaking with MCP server after {duration:?}"))?
-                .map_err(|err| anyhow!("handshaking with MCP server failed: {err}"))?,
-            None => transport
-                .await
-                .map_err(|err| anyhow!("handshaking with MCP server failed: {err}"))?,
+                .map_err(McpHandshakeError)?,
+            None => transport.await.map_err(McpHandshakeError)?,
         };
 
         Ok((Arc::new(service), oauth_persistor))
@@ -941,6 +997,16 @@ impl RmcpClient {
             })
     }
 
+    fn is_invalid_token_operation_error(error: &anyhow::Error) -> bool {
+        let Some(ClientOperationError::Service(rmcp::service::ServiceError::TransportSend(error))) =
+            error.downcast_ref::<ClientOperationError>()
+        else {
+            return false;
+        };
+
+        oauth_invalid_token::rejected_transport(error)
+    }
+
     async fn reinitialize_after_session_expiry(
         &self,
         failed_service: &Arc<RunningService<RoleClient, ElicitationClientService>>,