diff --git a/docs/clients/auth/oauth.mdx b/docs/clients/auth/oauth.mdx
index 25804adc3f..84fbe2164d 100644
--- a/docs/clients/auth/oauth.mdx
+++ b/docs/clients/auth/oauth.mdx
@@ -55,6 +55,8 @@ You don't need to pass `mcp_url` when using `OAuth` with `Client(auth=...)` —
- **`scopes`** (`str | list[str]`, optional): OAuth scopes to request. Can be space-separated string or list of strings
- **`client_name`** (`str`, optional): Client name for dynamic registration. Defaults to `"FastMCP Client"`
+- **`client_id`** (`str`, optional): Pre-registered OAuth client ID. When provided, skips Dynamic Client Registration entirely. See [Pre-Registered Clients](#pre-registered-clients)
+- **`client_secret`** (`str`, optional): OAuth client secret for pre-registered clients. Optional — public clients that rely on PKCE can omit this
- **`client_metadata_url`** (`str`, optional): URL-based client identity (CIMD). See [CIMD Authentication](/clients/auth/cimd) for details
- **`token_storage`** (`AsyncKeyValue`, optional): Storage backend for persisting OAuth tokens. Defaults to in-memory storage (tokens lost on restart). See [Token Storage](#token-storage) for encrypted storage options
- **`additional_client_metadata`** (`dict[str, Any]`, optional): Extra metadata for client registration
@@ -74,7 +76,7 @@ The client first checks the configured `token_storage` backend for existing, val
If no valid tokens exist, the client attempts to discover the OAuth server's endpoints using a well-known URI (e.g., `/.well-known/oauth-authorization-server`) based on the `mcp_url`.
-If the OAuth server supports it and the client isn't already registered (or credentials aren't cached), the client performs dynamic client registration according to RFC 7591. Alternatively, if a `client_metadata_url` is configured and the server supports CIMD, the client uses its metadata URL as its identity instead of registering.
+If a `client_id` is provided, the client uses those pre-registered credentials directly and skips this step entirely. Otherwise, if a `client_metadata_url` is configured and the server supports CIMD, the client uses its metadata URL as its identity. As a fallback, the client performs Dynamic Client Registration (RFC 7591) if the server supports it.
A temporary local HTTP server is started on an available port (or the port specified via `callback_port`). This server's address (e.g., `http://127.0.0.1:/callback`) acts as the `redirect_uri` for the OAuth flow.
@@ -152,3 +154,33 @@ async with Client(
```
See the [CIMD Authentication](/clients/auth/cimd) page for complete documentation on creating, hosting, and validating CIMD documents.
+
+## Pre-Registered Clients
+
+
+
+Some OAuth servers don't support Dynamic Client Registration — the MCP spec explicitly makes DCR optional. If your client has been pre-registered with the server (you already have a `client_id` and optionally a `client_secret`), you can provide them directly to skip DCR entirely.
+
+```python
+from fastmcp import Client
+from fastmcp.client.auth import OAuth
+
+async with Client(
+ "https://mcp-server.example.com/mcp",
+ auth=OAuth(
+ client_id="my-registered-client-id",
+ client_secret="my-client-secret",
+ ),
+) as client:
+ await client.ping()
+```
+
+Public clients that rely on PKCE for security can omit `client_secret`:
+
+```python
+oauth = OAuth(client_id="my-public-client-id")
+```
+
+
+When using pre-registered credentials, the client will not attempt Dynamic Client Registration. If the server rejects the credentials, the error is surfaced immediately rather than falling back to DCR.
+
diff --git a/src/fastmcp/client/auth/oauth.py b/src/fastmcp/client/auth/oauth.py
index 9fc90b4e89..a4d1e9c772 100644
--- a/src/fastmcp/client/auth/oauth.py
+++ b/src/fastmcp/client/auth/oauth.py
@@ -154,7 +154,12 @@ def __init__(
additional_client_metadata: dict[str, Any] | None = None,
callback_port: int | None = None,
httpx_client_factory: McpHttpClientFactory | None = None,
+ # Alternative to dynamic client registration:
+ # --- Clients host a static JSON document at an HTTPS URL ---
client_metadata_url: str | None = None,
+ # --- OR clients provide full client information ---
+ client_id: str | None = None,
+ client_secret: str | None = None,
):
"""
Initialize OAuth client provider for an MCP server.
@@ -173,6 +178,9 @@ def __init__(
provided, this URL is used as the client_id instead of performing
Dynamic Client Registration. Must be an HTTPS URL with a non-root
path (e.g. "https://myapp.example.com/oauth/client.json").
+ client_id: Pre-registered OAuth client ID. When provided, skips dynamic
+ client registration and uses these static credentials instead.
+ client_secret: OAuth client secret (optional, used with client_id)
"""
# Store config for deferred binding if mcp_url not yet known
self._scopes = scopes
@@ -181,6 +189,9 @@ def __init__(
self._additional_client_metadata = additional_client_metadata
self._callback_port = callback_port
self._client_metadata_url = client_metadata_url
+ self._client_id = client_id
+ self._client_secret = client_secret
+ self._static_client_info = None
self.httpx_client_factory = httpx_client_factory or httpx.AsyncClient
self._bound = False
@@ -218,6 +229,23 @@ def _bind(self, mcp_url: str) -> None:
**(self._additional_client_metadata or {}),
)
+ if self._client_id:
+ # Create the full static client info directly which will avoid DCR.
+ # Spread client_metadata so redirect_uris, grant_types, response_types,
+ # scope, etc. are included — servers may validate these fields.
+ metadata = client_metadata.model_dump(exclude_none=True)
+ # Default token_endpoint_auth_method based on whether a secret is
+ # provided, unless the caller already set it via additional_client_metadata.
+ if "token_endpoint_auth_method" not in metadata:
+ metadata["token_endpoint_auth_method"] = (
+ "client_secret_post" if self._client_secret else "none"
+ )
+ self._static_client_info = OAuthClientInformationFull(
+ client_id=self._client_id,
+ client_secret=self._client_secret,
+ **metadata,
+ )
+
token_storage = self._token_storage or MemoryStore()
if isinstance(token_storage, MemoryStore):
@@ -230,6 +258,7 @@ def _bind(self, mcp_url: str) -> None:
stacklevel=2,
)
+ # Use full URL for token storage to properly separate tokens per MCP endpoint
self.token_storage_adapter: TokenStorageAdapter = TokenStorageAdapter(
async_key_value=token_storage, server_url=mcp_url
)
@@ -249,10 +278,12 @@ def _bind(self, mcp_url: str) -> None:
async def _initialize(self) -> None:
"""Load stored tokens and client info, properly setting token expiry."""
- # Call parent's _initialize to load tokens and client info
await super()._initialize()
- # If tokens were loaded and have expires_in, update the context's token_expiry_time
+ if self._static_client_info is not None:
+ self.context.client_info = self._static_client_info
+ await self.token_storage_adapter.set_client_info(self._static_client_info)
+
if self.context.current_tokens and self.context.current_tokens.expires_in:
self.context.update_token_expiry(self.context.current_tokens)
@@ -342,6 +373,15 @@ async def async_auth_flow(
break
except ClientNotFoundError:
+ # Static credentials are fixed — retrying won't help. Surface the
+ # error so the user can correct their client_id / client_secret.
+ if self._static_client_info is not None:
+ raise ClientNotFoundError(
+ "OAuth server rejected the static client credentials. "
+ "Verify that the client_id (and client_secret, if provided) "
+ "are correct and that the client is registered with the server."
+ ) from None
+
logger.debug(
"OAuth client not found on server, clearing cache and retrying..."
)
diff --git a/tests/client/auth/test_oauth_static_client.py b/tests/client/auth/test_oauth_static_client.py
new file mode 100644
index 0000000000..c9f17cdbe3
--- /dev/null
+++ b/tests/client/auth/test_oauth_static_client.py
@@ -0,0 +1,274 @@
+"""Tests for OAuth static client registration (pre-registered client_id/client_secret)."""
+
+from unittest.mock import patch
+
+import httpx
+import pytest
+from mcp.shared.auth import OAuthClientInformationFull
+from pydantic import AnyUrl
+
+from fastmcp.client import Client
+from fastmcp.client.auth import OAuth
+from fastmcp.client.auth.oauth import ClientNotFoundError
+from fastmcp.client.transports import StreamableHttpTransport
+from fastmcp.server.auth.auth import ClientRegistrationOptions
+from fastmcp.server.auth.providers.in_memory import InMemoryOAuthProvider
+from fastmcp.server.server import FastMCP
+from fastmcp.utilities.http import find_available_port
+from fastmcp.utilities.tests import HeadlessOAuth, run_server_async
+
+
+class TestStaticClientInfoConstruction:
+ """Static client info should include full metadata from client_metadata."""
+
+ def test_static_client_info_includes_metadata(self):
+ """Static client info should include redirect_uris, grant_types, etc."""
+ oauth = OAuth(
+ mcp_url="https://example.com/mcp",
+ client_id="my-client-id",
+ client_secret="my-secret",
+ scopes=["read", "write"],
+ )
+
+ info = oauth._static_client_info
+ assert info is not None
+ assert info.client_id == "my-client-id"
+ assert info.client_secret == "my-secret"
+ # Metadata fields should be populated from client_metadata
+ assert info.redirect_uris is not None
+ assert len(info.redirect_uris) == 1
+ assert info.grant_types is not None
+ assert "authorization_code" in info.grant_types
+ assert "refresh_token" in info.grant_types
+ assert info.response_types is not None
+ assert "code" in info.response_types
+ assert info.scope == "read write"
+ assert info.token_endpoint_auth_method == "client_secret_post"
+
+ def test_static_client_info_without_secret(self):
+ """Public clients can provide client_id without client_secret."""
+ oauth = OAuth(
+ mcp_url="https://example.com/mcp",
+ client_id="public-client",
+ )
+
+ info = oauth._static_client_info
+ assert info is not None
+ assert info.client_id == "public-client"
+ assert info.client_secret is None
+ assert info.token_endpoint_auth_method == "none"
+ # Metadata should still be present
+ assert info.redirect_uris is not None
+ assert info.grant_types is not None
+
+ def test_no_static_client_info_without_client_id(self):
+ """When no client_id is provided, _static_client_info should be None."""
+ oauth = OAuth(mcp_url="https://example.com/mcp")
+ assert oauth._static_client_info is None
+
+ def test_static_client_info_includes_additional_metadata(self):
+ """Additional client metadata should be included in static client info."""
+ oauth = OAuth(
+ mcp_url="https://example.com/mcp",
+ client_id="my-client",
+ additional_client_metadata={
+ "token_endpoint_auth_method": "client_secret_post"
+ },
+ )
+
+ info = oauth._static_client_info
+ assert info is not None
+ assert info.token_endpoint_auth_method == "client_secret_post"
+
+
+class TestStaticClientInitialize:
+ """_initialize should set context.client_info and persist to storage."""
+
+ async def test_initialize_sets_context_client_info(self):
+ """_initialize should inject static client info into the auth context."""
+ oauth = OAuth(
+ mcp_url="https://example.com/mcp",
+ client_id="my-client",
+ client_secret="my-secret",
+ )
+
+ # Mock the parent _initialize since it needs a real server
+ with patch.object(OAuth.__bases__[0], "_initialize", return_value=None):
+ await oauth._initialize()
+
+ assert oauth.context.client_info is not None
+ assert oauth.context.client_info.client_id == "my-client"
+ assert oauth.context.client_info.client_secret == "my-secret"
+
+ async def test_initialize_persists_static_client_to_storage(self):
+ """Static client info should be persisted to token storage."""
+ oauth = OAuth(
+ mcp_url="https://example.com/mcp",
+ client_id="my-client",
+ client_secret="my-secret",
+ )
+
+ with patch.object(OAuth.__bases__[0], "_initialize", return_value=None):
+ await oauth._initialize()
+
+ # Verify it was persisted to storage
+ stored = await oauth.token_storage_adapter.get_client_info()
+ assert stored is not None
+ assert stored.client_id == "my-client"
+
+ async def test_initialize_without_static_creds_works(self):
+ """_initialize should not error when no static credentials are provided."""
+ oauth = OAuth(mcp_url="https://example.com/mcp")
+
+ with patch.object(OAuth.__bases__[0], "_initialize", return_value=None):
+ # This should not raise AttributeError
+ await oauth._initialize()
+
+ # context.client_info should be whatever the parent set (None by default)
+
+
+class TestStaticClientRetryBehavior:
+ """Retry-on-stale-credentials should short-circuit for static creds."""
+
+ async def test_retry_skipped_with_static_creds(self):
+ """When static creds are rejected, should raise immediately, not retry."""
+ oauth = OAuth(
+ mcp_url="https://example.com/mcp",
+ client_id="bad-client-id",
+ client_secret="bad-secret",
+ )
+
+ # Make the parent auth flow raise ClientNotFoundError
+ async def failing_auth_flow(request):
+ raise ClientNotFoundError("client not found")
+ yield # make it a generator # noqa: E275
+
+ with patch.object(
+ OAuth.__bases__[0], "async_auth_flow", side_effect=failing_auth_flow
+ ):
+ flow = oauth.async_auth_flow(httpx.Request("GET", "https://example.com"))
+ with pytest.raises(ClientNotFoundError, match="static client credentials"):
+ await flow.__anext__()
+
+ async def test_retry_still_works_without_static_creds(self):
+ """Without static creds, the retry behavior should be preserved."""
+ oauth = OAuth(mcp_url="https://example.com/mcp")
+
+ call_count = 0
+
+ async def auth_flow_with_retry(request):
+ nonlocal call_count
+ call_count += 1
+ if call_count == 1:
+ raise ClientNotFoundError("client not found")
+ # Second attempt succeeds
+ yield httpx.Request("GET", "https://example.com")
+
+ with patch.object(
+ OAuth.__bases__[0], "async_auth_flow", side_effect=auth_flow_with_retry
+ ):
+ flow = oauth.async_auth_flow(httpx.Request("GET", "https://example.com"))
+ request = await flow.__anext__()
+ assert request is not None
+ assert call_count == 2
+
+
+class TestStaticClientE2E:
+ """End-to-end tests with a real OAuth server using pre-registered clients."""
+
+ async def test_static_client_with_dcr_disabled(self):
+ """Static client_id should work when the server has DCR disabled."""
+ port = find_available_port()
+ callback_port = find_available_port()
+ issuer_url = f"http://127.0.0.1:{port}"
+
+ provider = InMemoryOAuthProvider(
+ base_url=issuer_url,
+ client_registration_options=ClientRegistrationOptions(
+ enabled=False, # DCR disabled
+ valid_scopes=["read", "write"],
+ ),
+ )
+
+ server = FastMCP("TestServer", auth=provider)
+
+ @server.tool
+ def greet(name: str) -> str:
+ return f"Hello, {name}!"
+
+ # Pre-register a client directly in the provider.
+ # The redirect_uri must match what the OAuth client will use.
+ pre_registered = OAuthClientInformationFull(
+ client_id="pre-registered-client",
+ client_secret="pre-registered-secret",
+ redirect_uris=[AnyUrl(f"http://localhost:{callback_port}/callback")],
+ grant_types=["authorization_code", "refresh_token"],
+ response_types=["code"],
+ token_endpoint_auth_method="client_secret_post",
+ scope="read write",
+ )
+ await provider.register_client(pre_registered)
+
+ async with run_server_async(server, port=port, transport="http") as url:
+ oauth = HeadlessOAuth(
+ mcp_url=url,
+ client_id="pre-registered-client",
+ client_secret="pre-registered-secret",
+ scopes=["read", "write"],
+ callback_port=callback_port,
+ )
+
+ async with Client(
+ transport=StreamableHttpTransport(url),
+ auth=oauth,
+ ) as client:
+ assert await client.ping()
+ tools = await client.list_tools()
+ assert any(t.name == "greet" for t in tools)
+
+ async def test_static_client_with_dcr_enabled(self):
+ """Static client_id should also work when DCR is enabled (skips DCR)."""
+ port = find_available_port()
+ callback_port = find_available_port()
+ issuer_url = f"http://127.0.0.1:{port}"
+
+ provider = InMemoryOAuthProvider(
+ base_url=issuer_url,
+ client_registration_options=ClientRegistrationOptions(
+ enabled=True,
+ valid_scopes=["read"],
+ ),
+ )
+
+ server = FastMCP("TestServer", auth=provider)
+
+ @server.tool
+ def add(a: int, b: int) -> int:
+ return a + b
+
+ pre_registered = OAuthClientInformationFull(
+ client_id="my-app",
+ client_secret="my-secret",
+ redirect_uris=[AnyUrl(f"http://localhost:{callback_port}/callback")],
+ grant_types=["authorization_code", "refresh_token"],
+ response_types=["code"],
+ token_endpoint_auth_method="client_secret_post",
+ scope="read",
+ )
+ await provider.register_client(pre_registered)
+
+ async with run_server_async(server, port=port, transport="http") as url:
+ oauth = HeadlessOAuth(
+ mcp_url=url,
+ client_id="my-app",
+ client_secret="my-secret",
+ scopes=["read"],
+ callback_port=callback_port,
+ )
+
+ async with Client(
+ transport=StreamableHttpTransport(url),
+ auth=oauth,
+ ) as client:
+ result = await client.call_tool("add", {"a": 3, "b": 4})
+ assert result.data == 7