OAuth2 access token never refreshes after startup — every from_openapi tool 401s permanently once it expires

[deity3005]

Symptom

Once the OAuth2 access token expires (~1hr TTL), every tool generated from an OpenAPI spec (Voice, Messaging, Lookup, Insights, TFV, End-user-management — essentially the whole tool surface) starts returning 401 permanently, for the rest of the process's life. Restarting the server is the only recovery. This happened to us twice in one session: once at cold start, once about an hour into active use.

Calling setCredentials mid-session does not help either — set_credentials_flow's own docstring already documents this:

Note: This updates the token in config, but tools created at startup already have their httpx client headers set. For full effect, the user should add credentials to their MCP server config and restart the server.

Root cause

In src/servers.py::_create_server, every mounted spec gets one httpx.AsyncClient built once at server startup, with the bearer token baked in as a static header at that moment:

token = config.get("BW_ACCESS_TOKEN")
if token:
    headers["Authorization"] = f"Bearer {token}"
client = AsyncClient(base_url=base_url, headers=headers, ...)

There's no refresh logic anywhere in this path — the header is frozen at construction time and never revisited.

Proposed fix

httpx.Auth.async_auth_flow is the sanctioned mechanism for exactly this (see httpx.DigestAuth for the built-in reference implementation of the same "yield request, inspect response, yield a modified retry" pattern). I added a RefreshingBearerAuth(httpx.Auth) class that:

• Reads config["BW_ACCESS_TOKEN"] live on every request instead of a frozen header — as a side effect, this also fixes the setCredentials limitation quoted above, since setCredentials and every mounted client's auth now share the same live config dict.
• On a 401, re-authenticates via the same client_credentials grant used at startup (oauth.get_oauth_token), updates config["BW_ACCESS_TOKEN"], and transparently retries the request once.
• Uses a shared asyncio.Lock (one per credential, passed through create_bandwidth_mcp → _create_server) so concurrent 401s across multiple in-flight requests/specs dedupe to a single token refresh rather than each independently hitting the token endpoint.
• Fails closed: if there's no client_id/secret to refresh with, or the refresh grant itself fails, the original 401 is surfaced rather than raising a different error or retrying forever.

Verification

• Added test/test_refreshing_auth.py (5 tests: healthy-token no-op, refresh-and-retry on 401, no-op when nothing to refresh with, refresh-failure surfaces the original 401, concurrent 401s dedupe to one refresh) plus updated test/test_servers.py for the new auth wiring. Full suite: 122 passed.
• Live-verified against the real API: seeded a client with a deliberately garbage/expired token plus valid client_id/secret, called GET /api/v2/accounts/{id}/applications, and got a 200 with real data — confirmed via the same call the config's BW_ACCESS_TOKEN had been replaced with a freshly-minted, valid token. No restart, no manual setCredentials call.

Patch

Full diff below (also happy to open a PR if that's preferred — forking is blocked under our org's policy, so this is the fastest way we could share it):

diff --git a/src/refreshing_auth.py b/src/refreshing_auth.py
new file mode 100644
index 0000000..1ecdb79
--- /dev/null
+++ b/src/refreshing_auth.py
@@ -0,0 +1,70 @@
+"""Bearer-token auth that transparently refreshes on a 401.
+
+Previously the OAuth2 access token was baked into a static Authorization
+header at client-construction time (see servers.py history). Bandwidth's
+tokens run ~1hr, so once one expired, every tool call from that client
+401'd for the rest of the process's life — a full restart was the only
+recovery, and `setCredentials` mid-session had no effect on already-mounted
+clients (see the docstring on set_credentials_flow).
+
+httpx.Auth.async_auth_flow is the sanctioned mechanism for this: it's a
+generator that yields a request, receives the response, and can yield a
+second (modified) request for httpx to send transparently — the caller
+never sees the intermediate 401.
+"""
+
+import asyncio
+
+import httpx
+
+
+class RefreshingBearerAuth(httpx.Auth):
+    """Bearer auth that re-authenticates via client_credentials on a 401.
+
+    `config` is the same mutable dict shared across every mounted spec's
+    client (and with setCredentials/clearCredentials), so a refresh
+    triggered by one client's request is immediately visible to every
+    other client's next request too — no cross-client push needed.
+
+    `lock` should be the SAME asyncio.Lock instance across all clients
+    sharing one credential, so concurrent 401s don't each independently
+    hit the token endpoint.
+    """
+
+    requires_response_body = True
+
+    def __init__(self, config: dict, lock: asyncio.Lock):
+        self._config = config
+        self._lock = lock
+
+    async def async_auth_flow(self, request: httpx.Request):
+        token = self._config.get("BW_ACCESS_TOKEN", "")
+        request.headers["Authorization"] = f"Bearer {token}"
+        response = yield request
+
+        if response.status_code != 401:
+            return
+
+        await response.aread()
+
+        async with self._lock:
+            # Another request may have already refreshed while we waited for
+            # the lock — only refresh if the token is still what we sent.
+            if self._config.get("BW_ACCESS_TOKEN", "") == token:
+                client_id = self._config.get("BW_CLIENT_ID")
+                client_secret = self._config.get("BW_CLIENT_SECRET")
+                if not client_id or not client_secret:
+                    return  # nothing to refresh with — surface the original 401
+                from oauth import get_oauth_token
+
+                try:
+                    token_data = await get_oauth_token(client_id, client_secret)
+                except Exception:
+                    return  # refresh failed — surface the original 401
+                self._config["BW_ACCESS_TOKEN"] = token_data["access_token"]
+
+        refreshed = self._config.get("BW_ACCESS_TOKEN", "")
+        if refreshed == token:
+            return  # refresh didn't actually change anything — don't loop
+        request.headers["Authorization"] = f"Bearer {refreshed}"
+        yield request
diff --git a/src/servers.py b/src/servers.py
index 2dca3ae..b8a8b4e 100644
--- a/src/servers.py
+++ b/src/servers.py
@@ -1,3 +1,5 @@
+import asyncio
+
 from pathlib import Path
 
 import specs
@@ -6,6 +8,7 @@ from fastmcp import FastMCP
 from httpx import AsyncClient
 from typing import Dict, List, Optional, Callable, Any
 
+from refreshing_auth import RefreshingBearerAuth
 from server_utils import (
     add_resources,
     create_route_map_fn,
@@ -48,6 +51,7 @@ async def _create_server(
     url: str,
     route_map_fn: Optional[Callable] = None,
     config: Optional[Dict[str, Any]] = None,
+    auth_lock: Optional[asyncio.Lock] = None,
 ) -> FastMCP:
     """Create an MCP server from the provided spec URL and credentials.
 
@@ -57,6 +61,13 @@ async def _create_server(
     """
     if config is None:
         config = {}
+    if auth_lock is None:
+        # Callers that share one credential across multiple _create_server
+        # calls (create_bandwidth_mcp) should pass one shared lock so
+        # concurrent 401s don't each independently hit the token endpoint.
+        # A fresh lock here is still correct for standalone/test use, just
+        # not shared.
+        auth_lock = asyncio.Lock()
     spec_object = await fetch_openapi_spec(url)
 
     if "servers" not in spec_object or not spec_object["servers"]:
@@ -69,9 +80,10 @@ async def _create_server(
     base_url = spec_object["servers"][0]["url"]
 
     headers = {"User-Agent": "Bandwidth MCP Server"}
-    token = config.get("BW_ACCESS_TOKEN")
-    if token:
-        headers["Authorization"] = f"Bearer {token}"
+    # Auth is handled by RefreshingBearerAuth below, not a static header —
+    # it reads config["BW_ACCESS_TOKEN"] live on every request, so a token
+    # refreshed after a 401 (or set later via setCredentials) takes effect
+    # immediately instead of requiring a server restart.
 
     async def _ensure_content_type(request):
         """Workaround for FastMCP from_openapi: when the OpenAPI spec declares
@@ -97,6 +109,7 @@ async def _create_server(
         headers=headers,
         follow_redirects=True,
         event_hooks={"request": [_ensure_content_type]},
+        auth=RefreshingBearerAuth(config, auth_lock),
     )
 
     mcp = FastMCP.from_openapi(
@@ -132,6 +145,10 @@ async def create_bandwidth_mcp(
     if config is None:
         config = {}
     route_map_fn = create_route_map_fn(enabled_tools, excluded_tools)
+    # One lock shared across every mounted spec's client — they all refresh
+    # the same underlying credential, so concurrent 401s across specs
+    # shouldn't each independently hit the token endpoint.
+    auth_lock = asyncio.Lock()
 
     for api_name, api_info in api_server_info.items():
         try:
@@ -146,6 +163,7 @@ async def create_bandwidth_mcp(
                 api_info["url"],
                 route_map_fn=spec_route_map_fn,
                 config=config,
+                auth_lock=auth_lock,
             )
             mcp.mount(server)
         except Exception as e:
diff --git a/test/test_refreshing_auth.py b/test/test_refreshing_auth.py
new file mode 100644
index 0000000..3c0936b
--- /dev/null
+++ b/test/test_refreshing_auth.py
@@ -0,0 +1,162 @@
+import asyncio
+
+import httpx
+import pytest
+from pytest_httpx import HTTPXMock
+
+from refreshing_auth import RefreshingBearerAuth  # bare, matching servers.py's own import
+
+TOKEN_URL = "https://api.bandwidth.com/api/v1/oauth2/token"
+TARGET_URL = "https://api.bandwidth.com/v2/some-resource"
+
+
+@pytest.mark.asyncio
+async def test_no_refresh_on_success(httpx_mock: HTTPXMock):
+    """A healthy token should never trigger a refresh."""
+    httpx_mock.add_response(url=TARGET_URL, status_code=200, json={"ok": True})
+
+    config = {"BW_ACCESS_TOKEN": "good-token"}
+    async with httpx.AsyncClient(auth=RefreshingBearerAuth(config, asyncio.Lock())) as client:
+        response = await client.get(TARGET_URL)
+
+    assert response.status_code == 200
+    requests = httpx_mock.get_requests()
+    assert len(requests) == 1
+    assert requests[0].headers["Authorization"] == "Bearer good-token"
+
+
+@pytest.mark.asyncio
+async def test_refresh_and_retry_on_401(httpx_mock: HTTPXMock):
+    """A 401 should trigger a fresh client_credentials grant and a transparent retry."""
+    httpx_mock.add_response(
+        url=TARGET_URL, status_code=401, json={"message": "expired"}
+    )
+    httpx_mock.add_response(url=TARGET_URL, status_code=200, json={"ok": True})
+    httpx_mock.add_response(
+        url=TOKEN_URL,
+        status_code=200,
+        json={"access_token": _fake_jwt(), "token_type": "Bearer"},
+    )
+
+    config = {
+        "BW_ACCESS_TOKEN": "stale-token",
+        "BW_CLIENT_ID": "client-id",
+        "BW_CLIENT_SECRET": "client-secret",
+    }
+    # httpx's auth_flow protocol (see httpx.DigestAuth for the canonical
+    # example) mutates the same Request object in place across retries — the
+    # correct, sanctioned pattern, and what RefreshingBearerAuth does too. But
+    # that means pytest_httpx's captured Request objects (stored by reference)
+    # reflect the object's FINAL mutated state, not its state at actual send
+    # time — so asserting per-request header values via get_requests() after
+    # the fact is unreliable. Snapshot the header at the moment each request
+    # is actually sent instead, via a request event hook.
+    sent_auth_headers = []
+
+    async def _record_auth_header(request):
+        sent_auth_headers.append(request.headers.get("authorization"))
+
+    async with httpx.AsyncClient(
+        auth=RefreshingBearerAuth(config, asyncio.Lock()),
+        event_hooks={"request": [_record_auth_header]},
+    ) as client:
+        response = await client.get(TARGET_URL)
+
+    assert response.status_code == 200
+    assert config["BW_ACCESS_TOKEN"] == _fake_jwt()
+
+    assert sent_auth_headers == [
+        "Bearer stale-token",
+        f"Bearer {_fake_jwt()}",
+    ], "first attempt should use the stale token, the retry should use the refreshed one"
+
+    token_requests = [r for r in httpx_mock.get_requests() if str(r.url) == TOKEN_URL]
+    assert len(token_requests) == 1
+
+
+@pytest.mark.asyncio
+async def test_refresh_skipped_without_client_credentials_surfaces_original_401(
+    httpx_mock: HTTPXMock,
+):
+    """No client_id/secret to refresh with — surface the original 401, don't loop."""
+    httpx_mock.add_response(url=TARGET_URL, status_code=401, json={"message": "expired"})
+
+    config = {"BW_ACCESS_TOKEN": "stale-token"}
+    async with httpx.AsyncClient(auth=RefreshingBearerAuth(config, asyncio.Lock())) as client:
+        response = await client.get(TARGET_URL)
+
+    assert response.status_code == 401
+    target_requests = [r for r in httpx_mock.get_requests() if r.url == TARGET_URL]
+    assert len(target_requests) == 1, "should not retry when there's nothing to refresh with"
+
+
+@pytest.mark.asyncio
+async def test_refresh_failure_surfaces_original_401(httpx_mock: HTTPXMock):
+    """If the refresh grant itself fails, surface the original 401 rather than raising."""
+    httpx_mock.add_response(url=TARGET_URL, status_code=401, json={"message": "expired"})
+    httpx_mock.add_response(url=TOKEN_URL, status_code=401, json={"error": "invalid_client"})
+
+    config = {
+        "BW_ACCESS_TOKEN": "stale-token",
+        "BW_CLIENT_ID": "bad-client-id",
+        "BW_CLIENT_SECRET": "bad-client-secret",
+    }
+    async with httpx.AsyncClient(auth=RefreshingBearerAuth(config, asyncio.Lock())) as client:
+        response = await client.get(TARGET_URL)
+
+    assert response.status_code == 401
+    assert config["BW_ACCESS_TOKEN"] == "stale-token", "should not overwrite a good value with a failed refresh"
+
+
+@pytest.mark.asyncio
+async def test_concurrent_401s_dedupe_refresh(httpx_mock: HTTPXMock):
+    """Several requests 401ing on the same stale token should trigger exactly one refresh.
+
+    Uses a stateful callback (real Authorization header determines the
+    response) rather than a fixed FIFO queue of canned responses — under
+    real concurrency, some of the 5 requests may not even hit a 401 at all
+    if another request's refresh has already landed by the time they're
+    sent, so a rigid "5 401s then 5 200s" script doesn't correctly model a
+    real backend.
+    """
+    fresh_token = _fake_jwt()
+
+    def _target_callback(request: httpx.Request) -> httpx.Response:
+        if request.headers.get("authorization") == f"Bearer {fresh_token}":
+            return httpx.Response(200, json={"ok": True})
+        return httpx.Response(401, json={"message": "expired"})
+
+    httpx_mock.add_callback(_target_callback, url=TARGET_URL, is_reusable=True)
+    httpx_mock.add_response(
+        url=TOKEN_URL,
+        status_code=200,
+        json={"access_token": fresh_token, "token_type": "Bearer"},
+    )
+
+    config = {
+        "BW_ACCESS_TOKEN": "stale-token",
+        "BW_CLIENT_ID": "client-id",
+        "BW_CLIENT_SECRET": "client-secret",
+    }
+    lock = asyncio.Lock()
+    async with httpx.AsyncClient(auth=RefreshingBearerAuth(config, lock)) as client:
+        responses = await asyncio.gather(*[client.get(TARGET_URL) for _ in range(5)])
+
+    assert all(r.status_code == 200 for r in responses)
+    assert config["BW_ACCESS_TOKEN"] == fresh_token
+    token_requests = [r for r in httpx_mock.get_requests() if str(r.url) == TOKEN_URL]
+    assert len(token_requests) == 1, "concurrent 401s should dedupe to a single refresh"
+
+
+def _fake_jwt() -> str:
+    """A structurally-valid (unsigned) JWT so get_oauth_token's claim-decoding doesn't choke.
+
+    header.payload.signature — only the payload needs to be real JSON.
+    """
+    import base64
+    import json
+
+    def _b64(obj: dict) -> str:
+        return base64.urlsafe_b64encode(json.dumps(obj).encode()).decode().rstrip("=")
+
+    return f"{_b64({'alg': 'none'})}.{_b64({'accounts': ['5001741']})}.sig"
diff --git a/test/test_servers.py b/test/test_servers.py
index c20e4ef..6db0fd3 100644
--- a/test/test_servers.py
+++ b/test/test_servers.py
@@ -2,6 +2,7 @@ import pytest
 from fastmcp import FastMCP
 from pytest_httpx import HTTPXMock
 from utils import create_mock, tool_map, server_client
+from refreshing_auth import RefreshingBearerAuth  # bare, matching servers.py's own import
 from src.servers import create_bandwidth_mcp, _create_server
 
 
@@ -99,9 +100,13 @@ async def test_individual_mcp_server_creation(
     assert (
         client.base_url == expected_base_url
     ), f"Expected base URL '{expected_base_url}', got '{client.base_url}'"
-    assert (
-        client.headers["Authorization"] == expected_auth_header
-    ), f"Expected auth header '{expected_auth_header}', got '{client.headers['Authorization']}'"
+    # Auth is no longer a static header (that's exactly the bug this fixes —
+    # a static header can never pick up a refreshed token). It's handled by
+    # RefreshingBearerAuth, which reads config["BW_ACCESS_TOKEN"] live on
+    # every request. See test_refreshing_auth.py for behavior coverage.
+    assert isinstance(client.auth, RefreshingBearerAuth)
+    assert client.auth._config is config
+    assert "Authorization" not in client.headers
 
 
 @pytest.mark.asyncio