Merge pull request #478 from jlowin/oauthclient

pyproject.toml

@@ -7,11 +7,12 @@ dependencies = [
     "python-dotenv>=1.1.0",
     "exceptiongroup>=1.2.2",
     "httpx>=0.28.1",
-    "mcp>=1.9.0,<2.0.0",
+    "mcp>=1.9.2,<2.0.0",
     "openapi-pydantic>=0.5.1",
     "rich>=13.9.4",
     "typer>=0.15.2",
     "websockets>=14.0",
+    "authlib>=1.5.2",
 ]
 requires-python = ">=3.10"
 readme = "README.md"

src/fastmcp/client/__init__.py

@@ -11,6 +11,7 @@ from .transports import (
     FastMCPTransport,
     StreamableHttpTransport,
 )
+from .auth import OAuth
 
 __all__ = [
     "Client",
@@ -24,4 +25,5 @@ __all__ = [
     "NpxStdioTransport",
     "FastMCPTransport",
     "StreamableHttpTransport",
+    "OAuth",
 ]

src/fastmcp/client/auth.py

@@ -0,0 +1,394 @@
+from __future__ import annotations
+
+import asyncio
+import json
+import webbrowser
+from pathlib import Path
+from typing import Any, Literal
+from urllib.parse import urljoin, urlparse
+
+import anyio
+import httpx
+from mcp.client.auth import OAuthClientProvider as _MCPOAuthClientProvider
+from mcp.client.auth import TokenStorage
+from mcp.shared.auth import (
+    OAuthClientInformationFull,
+    OAuthClientMetadata,
+)
+from mcp.shared.auth import (
+    OAuthMetadata as _MCPServerOAuthMetadata,
+)
+from mcp.shared.auth import (
+    OAuthToken as OAuthToken,
+)
+from pydantic import AnyHttpUrl, ValidationError
+
+from fastmcp.client.oauth_callback import (
+    create_oauth_callback_server,
+    find_available_port,
+)
+from fastmcp.settings import settings as fastmcp_global_settings
+from fastmcp.utilities.logging import get_logger
+
+__all__ = ["OAuth"]
+
+logger = get_logger(__name__)
+
+
+def default_cache_dir() -> Path:
+    return fastmcp_global_settings.home / "oauth-mcp-client-cache"
+
+
+# Flexible OAuth models for real-world compatibility
+class ServerOAuthMetadata(_MCPServerOAuthMetadata):
+    """
+    More flexible OAuth metadata model that accepts broader ranges of values
+    than the restrictive MCP standard model.
+
+    This handles real-world OAuth servers like PayPal that may support
+    additional methods not in the MCP specification.
+    """
+
+    # Allow any code challenge methods, not just S256
+    code_challenge_methods_supported: list[str] | None = None
+
+    # Allow any token endpoint auth methods
+    token_endpoint_auth_methods_supported: list[str] | None = None
+
+    # Allow any grant types
+    grant_types_supported: list[str] | None = None
+
+    # Allow any response types
+    response_types_supported: list[str] = ["code"]
+
+    # Allow any response modes
+    response_modes_supported: list[str] | None = None
+
+
+class OAuthClientProvider(_MCPOAuthClientProvider):
+    """
+    OAuth client provider with more flexible OAuth metadata discovery.
+
+    This subclass handles real-world OAuth servers that may not conform
+    strictly to the MCP OAuth specification but are still valid OAuth 2.0 servers.
+    """
+
+    async def _discover_oauth_metadata(
+        self, server_url: str
+    ) -> ServerOAuthMetadata | None:
+        """
+        Discover OAuth metadata with flexible validation.
+
+        This is nearly identical to the parent implementation but uses
+        ServerOAuthMetadata instead of the restrictive MCP OAuthMetadata.
+        """
+        # Extract base URL per MCP spec
+        auth_base_url = self._get_authorization_base_url(server_url)
+        url = urljoin(auth_base_url, "/.well-known/oauth-authorization-server")
+
+        from mcp.types import LATEST_PROTOCOL_VERSION
+
+        headers = {"MCP-Protocol-Version": LATEST_PROTOCOL_VERSION}
+
+        async with httpx.AsyncClient() as client:
+            try:
+                response = await client.get(url, headers=headers)
+                if response.status_code == 404:
+                    return None
+                response.raise_for_status()
+                metadata_json = response.json()
+                logger.debug(f"OAuth metadata discovered: {metadata_json}")
+                return ServerOAuthMetadata.model_validate(metadata_json)
+            except Exception:
+                # Retry without MCP header for CORS compatibility
+                try:
+                    response = await client.get(url)
+                    if response.status_code == 404:
+                        return None
+                    response.raise_for_status()
+                    metadata_json = response.json()
+                    logger.debug(
+                        f"OAuth metadata discovered (no MCP header): {metadata_json}"
+                    )
+                    return ServerOAuthMetadata.model_validate(metadata_json)
+                except Exception:
+                    logger.exception("Failed to discover OAuth metadata")
+                    return None
+
+
+class FileTokenStorage(TokenStorage):
+    """
+    File-based token storage implementation for OAuth credentials and tokens.
+    Implements the mcp.client.auth.TokenStorage protocol.
+
+    Each instance is tied to a specific server URL for proper token isolation.
+    """
+
+    def __init__(self, server_url: str, cache_dir: Path | None = None):
+        """Initialize storage for a specific server URL."""
+        self.server_url = server_url
+        self.cache_dir = cache_dir or default_cache_dir()
+        self.cache_dir.mkdir(exist_ok=True, parents=True)
+
+    @staticmethod
+    def get_base_url(url: str) -> str:
+        """Extract the base URL (scheme + host) from a URL."""
+        parsed = urlparse(url)
+        return f"{parsed.scheme}://{parsed.netloc}"
+
+    def get_cache_key(self) -> str:
+        """Generate a safe filesystem key from the server's base URL."""
+        base_url = self.get_base_url(self.server_url)
+        return (
+            base_url.replace("://", "_")
+            .replace(".", "_")
+            .replace("/", "_")
+            .replace(":", "_")
+        )
+
+    def _get_file_path(self, file_type: Literal["client_info", "tokens"]) -> Path:
+        """Get the file path for the specified cache file type."""
+        key = self.get_cache_key()
+        return self.cache_dir / f"{key}_{file_type}.json"
+
+    async def get_tokens(self) -> OAuthToken | None:
+        """Load tokens from file storage."""
+        path = self._get_file_path("tokens")
+
+        try:
+            tokens = OAuthToken.model_validate_json(path.read_text())
+            # now = datetime.datetime.now(datetime.timezone.utc)
+            # if tokens.expires_at is not None and tokens.expires_at <= now:
+            #     logger.debug(f"Token expired for {self.get_base_url(self.server_url)}")
+            #     return None
+            return tokens
+        except (FileNotFoundError, json.JSONDecodeError, ValidationError) as e:
+            logger.debug(
+                f"Could not load tokens for {self.get_base_url(self.server_url)}: {e}"
+            )
+            return None
+
+    async def set_tokens(self, tokens: OAuthToken) -> None:
+        """Save tokens to file storage."""
+        path = self._get_file_path("tokens")
+        path.write_text(tokens.model_dump_json(indent=2))
+        logger.debug(f"Saved tokens for {self.get_base_url(self.server_url)}")
+
+    async def get_client_info(self) -> OAuthClientInformationFull | None:
+        """Load client information from file storage."""
+        path = self._get_file_path("client_info")
+        try:
+            client_info = OAuthClientInformationFull.model_validate_json(
+                path.read_text()
+            )
+            # Check if we have corresponding valid tokens
+            # If no tokens exist, the OAuth flow was incomplete and we should
+            # force a fresh client registration
+            tokens = await self.get_tokens()
+            if tokens is None:
+                logger.debug(
+                    f"No tokens found for client info at {self.get_base_url(self.server_url)}. "
+                    "OAuth flow may have been incomplete. Clearing client info to force fresh registration."
+                )
+                # Clear the incomplete client info
+                client_info_path = self._get_file_path("client_info")
+                client_info_path.unlink(missing_ok=True)
+                return None
+
+            return client_info
+        except (FileNotFoundError, json.JSONDecodeError, ValidationError) as e:
+            logger.debug(
+                f"Could not load client info for {self.get_base_url(self.server_url)}: {e}"
+            )
+            return None
+
+    async def set_client_info(self, client_info: OAuthClientInformationFull) -> None:
+        """Save client information to file storage."""
+        path = self._get_file_path("client_info")
+        path.write_text(client_info.model_dump_json(indent=2))
+        logger.debug(f"Saved client info for {self.get_base_url(self.server_url)}")
+
+    def clear(self) -> None:
+        """Clear all cached data for this server."""
+        file_types: list[Literal["client_info", "tokens"]] = ["client_info", "tokens"]
+        for file_type in file_types:
+            path = self._get_file_path(file_type)
+            path.unlink(missing_ok=True)
+        logger.info(f"Cleared OAuth cache for {self.get_base_url(self.server_url)}")
+
+    @classmethod
+    def clear_all(cls, cache_dir: Path | None = None) -> None:
+        """Clear all cached data for all servers."""
+        cache_dir = cache_dir or default_cache_dir()
+        if not cache_dir.exists():
+            return
+
+        file_types: list[Literal["client_info", "tokens"]] = ["client_info", "tokens"]
+        for file_type in file_types:
+            for file in cache_dir.glob(f"*_{file_type}.json"):
+                file.unlink(missing_ok=True)
+        logger.info("Cleared all OAuth client cache data.")
+
+
+async def discover_oauth_metadata(
+    server_base_url: str, httpx_kwargs: dict[str, Any] | None = None
+) -> _MCPServerOAuthMetadata | None:
+    """
+    Discover OAuth metadata from the server using RFC 8414 well-known endpoint.
+
+    Args:
+        server_base_url: Base URL of the OAuth server (e.g., "https://example.com")
+        httpx_kwargs: Additional kwargs for httpx client
+
+    Returns:
+        OAuth metadata if found, None otherwise
+    """
+    well_known_url = urljoin(server_base_url, "/.well-known/oauth-authorization-server")
+    logger.debug(f"Discovering OAuth metadata from: {well_known_url}")
+
+    async with httpx.AsyncClient(**(httpx_kwargs or {})) as client:
+        try:
+            response = await client.get(well_known_url, timeout=10.0)
+            if response.status_code == 200:
+                logger.debug("Successfully discovered OAuth metadata")
+                return _MCPServerOAuthMetadata.model_validate(response.json())
+            elif response.status_code == 404:
+                logger.debug(
+                    "OAuth metadata not found (404) - server may not require auth"
+                )
+                return None
+            else:
+                logger.warning(f"OAuth metadata request failed: {response.status_code}")
+                return None
+        except (httpx.RequestError, json.JSONDecodeError, ValidationError) as e:
+            logger.debug(f"OAuth metadata discovery failed: {e}")
+            return None
+
+
+async def check_if_auth_required(
+    mcp_url: str, httpx_kwargs: dict[str, Any] | None = None
+) -> bool:
+    """
+    Check if the MCP endpoint requires authentication by making a test request.
+
+    Returns:
+        True if auth appears to be required, False otherwise
+    """
+    async with httpx.AsyncClient(**(httpx_kwargs or {})) as client:
+        try:
+            # Try a simple request to the endpoint
+            response = await client.get(mcp_url, timeout=5.0)
+
+            # If we get 401/403, auth is likely required
+            if response.status_code in (401, 403):
+                return True
+
+            # Check for WWW-Authenticate header
+            if "WWW-Authenticate" in response.headers:
+                return True
+
+            # If we get a successful response, auth may not be required
+            return False
+
+        except httpx.RequestError:
+            # If we can't connect, assume auth might be required
+            return True
+
+
+def OAuth(
+    mcp_url: str,
+    scopes: str | list[str] | None = None,
+    client_name: str = "FastMCP Client",
+    token_storage_cache_dir: Path | None = None,
+    additional_client_metadata: dict[str, Any] | None = None,
+) -> _MCPOAuthClientProvider:
+    """
+    Create an OAuthClientProvider for an MCP server.
+
+    This is intended to be provided to the `auth` parameter of an
+    httpx.AsyncClient (or appropriate FastMCP client/transport instance)
+
+    Args:
+        mcp_url: Full URL to the MCP endpoint (e.g.,
+        "http://host/mcp/sse")
+        scopes: OAuth scopes to request. Can be a
+        space-separated string or a list of strings.
+        client_name: Name for this client during registration
+        token_storage_cache_dir: Directory for FileTokenStorage
+        additional_client_metadata: Extra fields for OAuthClientMetadata
+
+    Returns:
+        OAuthClientProvider
+    """
+    parsed_url = urlparse(mcp_url)
+    server_base_url = f"{parsed_url.scheme}://{parsed_url.netloc}"
+
+    # Setup OAuth client
+    redirect_port = find_available_port()
+    redirect_uri = f"http://127.0.0.1:{redirect_port}/callback"
+
+    if isinstance(scopes, list):
+        scopes = " ".join(scopes)
+
+    client_metadata = OAuthClientMetadata(
+        client_name=client_name,
+        redirect_uris=[AnyHttpUrl(redirect_uri)],
+        grant_types=["authorization_code", "refresh_token"],
+        response_types=["code"],
+        token_endpoint_auth_method="client_secret_post",
+        scope=scopes,
+        **(additional_client_metadata or {}),
+    )
+
+    # Create server-specific token storage
+    storage = FileTokenStorage(
+        server_url=server_base_url, cache_dir=token_storage_cache_dir
+    )
+
+    # Define OAuth handlers
+    async def redirect_handler(authorization_url: str) -> None:
+        """Open browser for authorization."""
+        logger.info(f"OAuth authorization URL: {authorization_url}")
+        webbrowser.open(authorization_url)
+
+    async def callback_handler() -> tuple[str, str | None]:
+        """Handle OAuth callback and return (auth_code, state)."""
+        # Create a future to capture the OAuth response
+        response_future = asyncio.get_running_loop().create_future()
+
+        # Create server with the future
+        server = create_oauth_callback_server(
+            port=redirect_port,
+            server_url=server_base_url,
+            response_future=response_future,
+        )
+
+        # Run server until response is received with timeout logic
+        async with anyio.create_task_group() as tg:
+            tg.start_soon(server.serve)
+            logger.info(
+                f"🎧 OAuth callback server started on http://127.0.0.1:{redirect_port}"
+            )
+
+            TIMEOUT = 300.0  # 5 minute timeout
+            try:
+                with anyio.fail_after(TIMEOUT):
+                    auth_code, state = await response_future
+                    return auth_code, state
+            except TimeoutError:
+                raise TimeoutError(f"OAuth callback timed out after {TIMEOUT} seconds")
+            finally:
+                server.should_exit = True
+                await asyncio.sleep(0.1)  # Allow server to shutdown gracefully
+                tg.cancel_scope.cancel()
+
+    # Create OAuth provider
+    oauth_provider = OAuthClientProvider(
+        server_url=server_base_url,
+        client_metadata=client_metadata,
+        storage=storage,
+        redirect_handler=redirect_handler,
+        callback_handler=callback_handler,
+    )
+
+    return oauth_provider

src/fastmcp/client/client.py

@@ -1,9 +1,10 @@
 import datetime
 from contextlib import AsyncExitStack, asynccontextmanager
 from pathlib import Path
-from typing import Any, Generic, cast, overload
+from typing import Any, Generic, Literal, cast, overload
 
 import anyio
+import httpx
 import mcp.types
 from exceptiongroup import catch
 from mcp import ClientSession
@@ -43,6 +44,7 @@ from .transports import (
 
 __all__ = [
     "Client",
+    "SessionKwargs",
     "RootsHandler",
     "RootsList",
     "LogHandler",
@@ -142,8 +144,11 @@ class Client(Generic[ClientTransportT]):
         progress_handler: ProgressHandler | None = None,
         timeout: datetime.timedelta | float | int | None = None,
         init_timeout: datetime.timedelta | float | int | None = None,
+        auth: httpx.Auth | Literal["oauth"] | str | None = None,
     ):
         self.transport = cast(ClientTransportT, infer_transport(transport))
+        if auth is not None:
+            self.transport._set_auth(auth)
         self._session: ClientSession | None = None
         self._exit_stack: AsyncExitStack | None = None
         self._nesting_counter: int = 0

src/fastmcp/client/oauth_callback.py

@@ -0,0 +1,317 @@
+"""
+OAuth callback server for handling authorization code flows.
+
+This module provides a reusable callback server that can handle OAuth redirects
+and display styled responses to users.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import socket
+from dataclasses import dataclass
+
+from starlette.applications import Starlette
+from starlette.requests import Request
+from starlette.responses import HTMLResponse
+from starlette.routing import Route
+from uvicorn import Config, Server
+
+from fastmcp.utilities.logging import get_logger
+
+logger = get_logger(__name__)
+
+
+def create_callback_html(
+    message: str,
+    is_success: bool = True,
+    title: str = "FastMCP OAuth",
+    server_url: str | None = None,
+) -> str:
+    """Create a styled HTML response for OAuth callbacks."""
+    status_emoji = "✅" if is_success else "❌"
+    status_color = "#10b981" if is_success else "#ef4444"  # emerald-500 / red-500
+
+    # Add server info for success cases
+    server_info = ""
+    if is_success and server_url:
+        server_info = f"""
+            <div class="server-info">
+                Connected to: <strong>{server_url}</strong>
+            </div>
+        """
+
+    return f"""
+    <!DOCTYPE html>
+    <html lang="en">
+    <head>
+        <meta charset="UTF-8">
+        <meta name="viewport" content="width=device-width, initial-scale=1.0">
+        <title>{title}</title>
+        <style>
+            body {{
+                font-family: 'SF Mono', 'Monaco', 'Consolas', 'Roboto Mono', monospace;
+                margin: 0;
+                padding: 0;
+                min-height: 100vh;
+                display: flex;
+                align-items: center;
+                justify-content: center;
+                background: linear-gradient(135deg, #0f0f23 0%, #1a1a2e 25%, #16213e 50%, #0f0f23 100%);
+                color: #e2e8f0;
+                overflow: hidden;
+            }}
+            
+            body::before {{
+                content: '';
+                position: fixed;
+                top: 0;
+                left: 0;
+                width: 100%;
+                height: 100%;
+                background: 
+                    radial-gradient(circle at 20% 80%, rgba(120, 119, 198, 0.1) 0%, transparent 50%),
+                    radial-gradient(circle at 80% 20%, rgba(16, 185, 129, 0.1) 0%, transparent 50%),
+                    radial-gradient(circle at 40% 40%, rgba(14, 165, 233, 0.1) 0%, transparent 50%);
+                pointer-events: none;
+                z-index: -1;
+            }}
+            
+            .container {{
+                background: rgba(30, 41, 59, 0.9);
+                backdrop-filter: blur(10px);
+                border: 1px solid rgba(71, 85, 105, 0.3);
+                padding: 3rem 2rem;
+                border-radius: 1rem;
+                box-shadow: 
+                    0 25px 50px -12px rgba(0, 0, 0, 0.7),
+                    0 0 0 1px rgba(255, 255, 255, 0.05),
+                    inset 0 1px 0 0 rgba(255, 255, 255, 0.1);
+                text-align: center;
+                max-width: 500px;
+                margin: 1rem;
+                position: relative;
+            }}
+            
+            .container::before {{
+                content: '';
+                position: absolute;
+                top: 0;
+                left: 0;
+                right: 0;
+                height: 1px;
+                background: linear-gradient(90deg, transparent, rgba(16, 185, 129, 0.5), transparent);
+            }}
+            
+            .status-icon {{
+                font-size: 4rem;
+                margin-bottom: 1rem;
+                display: block;
+                filter: drop-shadow(0 0 20px currentColor);
+            }}
+            
+            .message {{
+                font-size: 1.25rem;
+                line-height: 1.6;
+                color: {status_color};
+                margin-bottom: 1.5rem;
+                font-weight: 600;
+                text-shadow: 0 0 10px rgba({
+        "16, 185, 129" if is_success else "239, 68, 68"
+    }, 0.3);
+            }}
+            
+            .server-info {{
+                background: rgba(6, 182, 212, 0.1);
+                border: 1px solid rgba(6, 182, 212, 0.3);
+                border-radius: 0.75rem;
+                padding: 1rem;
+                margin: 1rem 0;
+                font-size: 0.9rem;
+                color: #67e8f9;
+                font-family: 'SF Mono', 'Monaco', 'Consolas', 'Roboto Mono', monospace;
+                text-shadow: 0 0 10px rgba(103, 232, 249, 0.3);
+            }}
+            
+            .server-info strong {{
+                color: #22d3ee;
+                font-weight: 700;
+            }}
+            
+            .subtitle {{
+                font-size: 1rem;
+                color: #94a3b8;
+                margin-top: 1rem;
+            }}
+            
+            .close-instruction {{
+                background: rgba(51, 65, 85, 0.8);
+                border: 1px solid rgba(71, 85, 105, 0.4);
+                border-radius: 0.75rem;
+                padding: 1rem;
+                margin-top: 1.5rem;
+                font-size: 0.9rem;
+                color: #cbd5e1;
+                font-family: 'SF Mono', 'Monaco', 'Consolas', 'Roboto Mono', monospace;
+            }}
+            
+            @keyframes glow {{
+                0%, 100% {{ opacity: 1; }}
+                50% {{ opacity: 0.7; }}
+            }}
+            
+            .status-icon {{
+                animation: glow 2s ease-in-out infinite;
+            }}
+        </style>
+    </head>
+    <body>
+        <div class="container">
+            <span class="status-icon">{status_emoji}</span>
+            <div class="message">{message}</div>
+            {server_info}
+            <div class="close-instruction">
+                You can safely close this tab now.
+            </div>
+        </div>
+    </body>
+    </html>
+    """
+
+
+def find_available_port() -> int:
+    """Find an available port by letting the OS assign one."""
+    with socket.socket() as s:
+        s.bind(("127.0.0.1", 0))
+        return s.getsockname()[1]
+
+
+@dataclass
+class CallbackResponse:
+    code: str | None = None
+    state: str | None = None
+    error: str | None = None
+    error_description: str | None = None
+
+    @classmethod
+    def from_dict(cls, data: dict[str, str]) -> CallbackResponse:
+        return cls(**{k: v for k, v in data.items() if k in cls.__annotations__})
+
+    def to_dict(self) -> dict[str, str]:
+        return {k: v for k, v in self.__dict__.items() if v is not None}
+
+
+def create_oauth_callback_server(
+    port: int,
+    callback_path: str = "/callback",
+    server_url: str | None = None,
+    response_future: asyncio.Future | None = None,
+) -> Server:
+    """
+    Create an OAuth callback server.
+
+    Args:
+        port: The port to run the server on
+        callback_path: The path to listen for OAuth redirects on
+        server_url: Optional server URL to display in success messages
+        response_future: Optional future to resolve when OAuth callback is received
+
+    Returns:
+        Configured uvicorn Server instance (not yet running)
+    """
+
+    async def callback_handler(request: Request):
+        """Handle OAuth callback requests with proper HTML responses."""
+        query_params = dict(request.query_params)
+        callback_response = CallbackResponse.from_dict(query_params)
+
+        if callback_response.error:
+            error_desc = callback_response.error_description or "Unknown error"
+
+            # Resolve future with exception if provided
+            if response_future and not response_future.done():
+                response_future.set_exception(
+                    RuntimeError(
+                        f"OAuth error: {callback_response.error} - {error_desc}"
+                    )
+                )
+
+            return HTMLResponse(
+                create_callback_html(
+                    f"FastMCP OAuth Error: {callback_response.error}<br>{error_desc}",
+                    is_success=False,
+                ),
+                status_code=400,
+            )
+
+        if not callback_response.code:
+            # Resolve future with exception if provided
+            if response_future and not response_future.done():
+                response_future.set_exception(
+                    RuntimeError("OAuth callback missing authorization code")
+                )
+
+            return HTMLResponse(
+                create_callback_html(
+                    "FastMCP OAuth Error: No authorization code received",
+                    is_success=False,
+                ),
+                status_code=400,
+            )
+
+        # Success case
+        if response_future and not response_future.done():
+            response_future.set_result(
+                (callback_response.code, callback_response.state)
+            )
+
+        return HTMLResponse(
+            create_callback_html("FastMCP OAuth login complete!", server_url=server_url)
+        )
+
+    app = Starlette(routes=[Route(callback_path, callback_handler)])
+
+    return Server(
+        Config(
+            app=app,
+            host="127.0.0.1",
+            port=port,
+            lifespan="off",
+            log_level="warning",
+        )
+    )
+
+
+if __name__ == "__main__":
+    """Run a test server when executed directly."""
+    import webbrowser
+
+    import uvicorn
+
+    port = find_available_port()
+    print("🎭 OAuth Callback Test Server")
+    print("📍 Test URLs:")
+    print(f"  Success: http://localhost:{port}/callback?code=test123&state=xyz")
+    print(
+        f"  Error:   http://localhost:{port}/callback?error=access_denied&error_description=User%20denied"
+    )
+    print(f"  Missing: http://localhost:{port}/callback")
+    print("🛑 Press Ctrl+C to stop")
+    print()
+
+    # Create test server without future (just for testing HTML responses)
+    server = create_oauth_callback_server(
+        port=port, server_url="https://fastmcp-test-server.example.com"
+    )
+
+    # Open browser to success example
+    webbrowser.open(f"http://localhost:{port}/callback?code=test123&state=xyz")
+
+    # Run with uvicorn directly
+    uvicorn.run(
+        server.config.app,
+        host="127.0.0.1",
+        port=port,
+        log_level="warning",
+        access_log=False,
+    )

src/fastmcp/client/transports.py

@@ -6,10 +6,19 @@ import os
 import shutil
 import sys
 import warnings
-from collections.abc import AsyncIterator
+from collections.abc import AsyncIterator, Callable
 from pathlib import Path
-from typing import TYPE_CHECKING, Any, TypedDict, TypeVar, cast, overload
+from typing import (
+    TYPE_CHECKING,
+    Any,
+    Literal,
+    TypedDict,
+    TypeVar,
+    cast,
+    overload,
+)
 
+import httpx
 from mcp import ClientSession, StdioServerParameters
 from mcp.client.session import (
     ListRootsFnT,
@@ -26,6 +35,7 @@ from mcp.shared.memory import create_connected_server_and_client_session
 from pydantic import AnyUrl
 from typing_extensions import Unpack
 
+from fastmcp.client.auth import OAuth
 from fastmcp.server import FastMCP as FastMCPServer
 from fastmcp.server.dependencies import get_http_headers
 from fastmcp.server.server import FastMCP
@@ -40,6 +50,21 @@ logger = get_logger(__name__)
 # TypeVar for preserving specific ClientTransport subclass types
 ClientTransportT = TypeVar("ClientTransportT", bound="ClientTransport")
 
+__all__ = [
+    "ClientTransport",
+    "SSETransport",
+    "StreamableHttpTransport",
+    "FastMCPServer",
+    "StdioTransport",
+    "PythonStdioTransport",
+    "FastMCPStdioTransport",
+    "NodeStdioTransport",
+    "UvxStdioTransport",
+    "NpxStdioTransport",
+    "FastMCPTransport",
+    "infer_transport",
+]
+
 
 class SessionKwargs(TypedDict, total=False):
     """Keyword arguments for the MCP ClientSession constructor."""
@@ -92,6 +117,10 @@ class ClientTransport(abc.ABC):
         """Close the transport."""
         pass
 
+    def _set_auth(self, auth: httpx.Auth | Literal["oauth"] | str | None):
+        if auth is not None:
+            raise ValueError("This transport does not support auth")
+
 
 class WSTransport(ClientTransport):
     """Transport implementation that connects to an MCP server via WebSockets."""
@@ -131,7 +160,9 @@ class SSETransport(ClientTransport):
         self,
         url: str | AnyUrl,
         headers: dict[str, str] | None = None,
+        auth: httpx.Auth | Literal["oauth"] | str | None = None,
         sse_read_timeout: datetime.timedelta | float | int | None = None,
+        httpx_client_factory: Callable[[], httpx.AsyncClient] | None = None,
     ):
         if isinstance(url, AnyUrl):
             url = str(url)
@@ -139,11 +170,21 @@ class SSETransport(ClientTransport):
             raise ValueError("Invalid HTTP/S URL provided for SSE.")
         self.url = url
         self.headers = headers or {}
+        self._set_auth(auth)
+        self.httpx_client_factory = httpx_client_factory
 
         if isinstance(sse_read_timeout, int | float):
             sse_read_timeout = datetime.timedelta(seconds=sse_read_timeout)
         self.sse_read_timeout = sse_read_timeout
 
+    def _set_auth(self, auth: httpx.Auth | Literal["oauth"] | str | None):
+        if auth == "oauth":
+            auth = OAuth(self.url)
+        elif isinstance(auth, str):
+            self.headers["Authorization"] = auth
+            auth = None
+        self.auth = auth
+
     @contextlib.asynccontextmanager
     async def connect_session(
         self, **session_kwargs: Unpack[SessionKwargs]
@@ -165,7 +206,10 @@ class SSETransport(ClientTransport):
             )
             client_kwargs["timeout"] = read_timeout_seconds.total_seconds()
 
-        async with sse_client(self.url, **client_kwargs) as transport:
+        if self.httpx_client_factory is not None:
+            client_kwargs["httpx_client_factory"] = self.httpx_client_factory
+
+        async with sse_client(self.url, auth=self.auth, **client_kwargs) as transport:
             read_stream, write_stream = transport
             async with ClientSession(
                 read_stream, write_stream, **session_kwargs
@@ -183,7 +227,9 @@ class StreamableHttpTransport(ClientTransport):
         self,
         url: str | AnyUrl,
         headers: dict[str, str] | None = None,
+        auth: httpx.Auth | Literal["oauth"] | str | None = None,
         sse_read_timeout: datetime.timedelta | float | int | None = None,
+        httpx_client_factory: Callable[[], httpx.AsyncClient] | None = None,
     ):
         if isinstance(url, AnyUrl):
             url = str(url)
@@ -191,11 +237,21 @@ class StreamableHttpTransport(ClientTransport):
             raise ValueError("Invalid HTTP/S URL provided for Streamable HTTP.")
         self.url = url
         self.headers = headers or {}
+        self._set_auth(auth)
+        self.httpx_client_factory = httpx_client_factory
 
         if isinstance(sse_read_timeout, int | float):
             sse_read_timeout = datetime.timedelta(seconds=sse_read_timeout)
         self.sse_read_timeout = sse_read_timeout
 
+    def _set_auth(self, auth: httpx.Auth | Literal["oauth"] | str | None):
+        if auth == "oauth":
+            auth = OAuth(self.url)
+        elif isinstance(auth, str):
+            self.headers["Authorization"] = auth
+            auth = None
+        self.auth = auth
+
     @contextlib.asynccontextmanager
     async def connect_session(
         self, **session_kwargs: Unpack[SessionKwargs]
@@ -214,7 +270,14 @@ class StreamableHttpTransport(ClientTransport):
         if session_kwargs.get("read_timeout_seconds", None) is not None:
             client_kwargs["timeout"] = session_kwargs.get("read_timeout_seconds")
 
-        async with streamablehttp_client(self.url, **client_kwargs) as transport:
+        if self.httpx_client_factory is not None:
+            client_kwargs["httpx_client_factory"] = self.httpx_client_factory
+
+        async with streamablehttp_client(
+            self.url,
+            auth=self.auth,
+            **client_kwargs,
+        ) as transport:
             read_stream, write_stream, _ = transport
             async with ClientSession(
                 read_stream, write_stream, **session_kwargs

src/fastmcp/low_level/README.md

@@ -1 +0,0 @@
-Patched low-level objects. When possible, we prefer the official SDK, but we patch bugs here if necessary.

src/fastmcp/server/auth/auth.py

@@ -0,0 +1,38 @@
+from mcp.server.auth.provider import (
+    AccessToken,
+    AuthorizationCode,
+    OAuthAuthorizationServerProvider,
+    RefreshToken,
+)
+from mcp.server.auth.settings import (
+    AuthSettings,
+    ClientRegistrationOptions,
+    RevocationOptions,
+)
+from pydantic import AnyHttpUrl
+
+
+class OAuthProvider(
+    OAuthAuthorizationServerProvider[AuthorizationCode, RefreshToken, AccessToken]
+):
+    def __init__(
+        self,
+        issuer_url: AnyHttpUrl | str,
+        service_documentation_url: AnyHttpUrl | str | None = None,
+        client_registration_options: ClientRegistrationOptions | None = None,
+        revocation_options: RevocationOptions | None = None,
+        required_scopes: list[str] | None = None,
+    ):
+        super().__init__()
+        if isinstance(issuer_url, str):
+            issuer_url = AnyHttpUrl(issuer_url)
+        if isinstance(service_documentation_url, str):
+            service_documentation_url = AnyHttpUrl(service_documentation_url)
+
+        self.settings = AuthSettings(
+            issuer_url=issuer_url,
+            service_documentation_url=service_documentation_url,
+            client_registration_options=client_registration_options,
+            revocation_options=revocation_options,
+            required_scopes=required_scopes,
+        )

src/fastmcp/server/auth/in_memory_provider.py

@@ -0,0 +1,325 @@
+import secrets
+import time
+
+from mcp.server.auth.provider import (
+    AccessToken,
+    AuthorizationCode,
+    AuthorizationParams,
+    AuthorizeError,
+    RefreshToken,
+    TokenError,
+    construct_redirect_uri,
+)
+from mcp.shared.auth import (
+    OAuthClientInformationFull,
+    OAuthToken,
+)
+from pydantic import AnyHttpUrl
+
+from fastmcp.server.auth.auth import (
+    ClientRegistrationOptions,
+    OAuthProvider,
+    RevocationOptions,
+)
+
+# Default expiration times (in seconds)
+DEFAULT_AUTH_CODE_EXPIRY_SECONDS = 5 * 60  # 5 minutes
+DEFAULT_ACCESS_TOKEN_EXPIRY_SECONDS = 60 * 60  # 1 hour
+DEFAULT_REFRESH_TOKEN_EXPIRY_SECONDS = None  # No expiry
+
+
+class InMemoryOAuthProvider(OAuthProvider):
+    """
+    An in-memory OAuth provider for testing purposes.
+    It simulates the OAuth 2.0 flow locally without external calls.
+    """
+
+    def __init__(
+        self,
+        issuer_url: AnyHttpUrl | str | None = None,
+        service_documentation_url: AnyHttpUrl | str | None = None,
+        client_registration_options: ClientRegistrationOptions | None = None,
+        revocation_options: RevocationOptions | None = None,
+        required_scopes: list[str] | None = None,
+    ):
+        super().__init__(
+            issuer_url or "https://example.com",
+            service_documentation_url=service_documentation_url,
+            client_registration_options=client_registration_options,
+            revocation_options=revocation_options,
+            required_scopes=required_scopes,
+        )
+        self.clients: dict[str, OAuthClientInformationFull] = {}
+        self.auth_codes: dict[str, AuthorizationCode] = {}
+        self.access_tokens: dict[str, AccessToken] = {}
+        self.refresh_tokens: dict[str, RefreshToken] = {}
+
+        # For revoking associated tokens
+        self._access_to_refresh_map: dict[
+            str, str
+        ] = {}  # access_token_str -> refresh_token_str
+        self._refresh_to_access_map: dict[
+            str, str
+        ] = {}  # refresh_token_str -> access_token_str
+
+    async def get_client(self, client_id: str) -> OAuthClientInformationFull | None:
+        return self.clients.get(client_id)
+
+    async def register_client(self, client_info: OAuthClientInformationFull) -> None:
+        if client_info.client_id in self.clients:
+            # As per RFC 7591, if client_id is already known, it's an update.
+            # For this simple provider, we'll treat it as re-registration.
+            # A real provider might handle updates or raise errors for conflicts.
+            pass
+        self.clients[client_info.client_id] = client_info
+
+    async def authorize(
+        self, client: OAuthClientInformationFull, params: AuthorizationParams
+    ) -> str:
+        """
+        Simulates user authorization and generates an authorization code.
+        Returns a redirect URI with the code and state.
+        """
+        if client.client_id not in self.clients:
+            raise AuthorizeError(
+                error="unauthorized_client",
+                error_description=f"Client '{client.client_id}' not registered.",
+            )
+
+        # Validate redirect_uri (already validated by AuthorizationHandler, but good practice)
+        try:
+            # OAuthClientInformationFull should have a method like validate_redirect_uri
+            # For this test provider, we assume it's valid if it matches one in client_info
+            # The AuthorizationHandler already does robust validation using client.validate_redirect_uri
+            if params.redirect_uri not in client.redirect_uris:
+                # This check might be too simplistic if redirect_uris can be patterns
+                # or if params.redirect_uri is None and client has a default.
+                # However, the AuthorizationHandler handles the primary validation.
+                pass  # Let's assume AuthorizationHandler did its job.
+        except Exception:  # Replace with specific validation error if client.validate_redirect_uri existed
+            raise AuthorizeError(
+                error="invalid_request", error_description="Invalid redirect_uri."
+            )
+
+        auth_code_value = f"test_auth_code_{secrets.token_hex(16)}"
+        expires_at = time.time() + DEFAULT_AUTH_CODE_EXPIRY_SECONDS
+
+        # Ensure scopes are a list
+        scopes_list = params.scopes if params.scopes is not None else []
+        if client.scope:  # Filter params.scopes against client's registered scopes
+            client_allowed_scopes = set(client.scope.split())
+            scopes_list = [s for s in scopes_list if s in client_allowed_scopes]
+
+        auth_code = AuthorizationCode(
+            code=auth_code_value,
+            client_id=client.client_id,
+            redirect_uri=params.redirect_uri,
+            redirect_uri_provided_explicitly=params.redirect_uri_provided_explicitly,
+            scopes=scopes_list,
+            expires_at=expires_at,
+            code_challenge=params.code_challenge,
+            # code_challenge_method is assumed S256 by the framework
+        )
+        self.auth_codes[auth_code_value] = auth_code
+
+        return construct_redirect_uri(
+            str(params.redirect_uri), code=auth_code_value, state=params.state
+        )
+
+    async def load_authorization_code(
+        self, client: OAuthClientInformationFull, authorization_code: str
+    ) -> AuthorizationCode | None:
+        auth_code_obj = self.auth_codes.get(authorization_code)
+        if auth_code_obj:
+            if auth_code_obj.client_id != client.client_id:
+                return None  # Belongs to a different client
+            if auth_code_obj.expires_at < time.time():
+                del self.auth_codes[authorization_code]  # Expired
+                return None
+            return auth_code_obj
+        return None
+
+    async def exchange_authorization_code(
+        self, client: OAuthClientInformationFull, authorization_code: AuthorizationCode
+    ) -> OAuthToken:
+        # Authorization code should have been validated (existence, expiry, client_id match)
+        # by the TokenHandler calling load_authorization_code before this.
+        # We might want to re-verify or simply trust it's valid.
+
+        if authorization_code.code not in self.auth_codes:
+            raise TokenError(
+                "invalid_grant", "Authorization code not found or already used."
+            )
+
+        # Consume the auth code
+        del self.auth_codes[authorization_code.code]
+
+        access_token_value = f"test_access_token_{secrets.token_hex(32)}"
+        refresh_token_value = f"test_refresh_token_{secrets.token_hex(32)}"
+
+        access_token_expires_at = int(time.time() + DEFAULT_ACCESS_TOKEN_EXPIRY_SECONDS)
+
+        # Refresh token expiry
+        refresh_token_expires_at = None
+        if DEFAULT_REFRESH_TOKEN_EXPIRY_SECONDS is not None:
+            refresh_token_expires_at = int(
+                time.time() + DEFAULT_REFRESH_TOKEN_EXPIRY_SECONDS
+            )
+
+        self.access_tokens[access_token_value] = AccessToken(
+            token=access_token_value,
+            client_id=client.client_id,
+            scopes=authorization_code.scopes,
+            expires_at=access_token_expires_at,
+        )
+        self.refresh_tokens[refresh_token_value] = RefreshToken(
+            token=refresh_token_value,
+            client_id=client.client_id,
+            scopes=authorization_code.scopes,  # Refresh token inherits scopes
+            expires_at=refresh_token_expires_at,
+        )
+
+        self._access_to_refresh_map[access_token_value] = refresh_token_value
+        self._refresh_to_access_map[refresh_token_value] = access_token_value
+
+        return OAuthToken(
+            access_token=access_token_value,
+            token_type="bearer",
+            expires_in=DEFAULT_ACCESS_TOKEN_EXPIRY_SECONDS,
+            refresh_token=refresh_token_value,
+            scope=" ".join(authorization_code.scopes),
+        )
+
+    async def load_refresh_token(
+        self, client: OAuthClientInformationFull, refresh_token: str
+    ) -> RefreshToken | None:
+        token_obj = self.refresh_tokens.get(refresh_token)
+        if token_obj:
+            if token_obj.client_id != client.client_id:
+                return None  # Belongs to different client
+            if token_obj.expires_at is not None and token_obj.expires_at < time.time():
+                self._revoke_internal(
+                    refresh_token_str=token_obj.token
+                )  # Clean up expired
+                return None
+            return token_obj
+        return None
+
+    async def exchange_refresh_token(
+        self,
+        client: OAuthClientInformationFull,
+        refresh_token: RefreshToken,  # This is the RefreshToken object, already loaded
+        scopes: list[str],  # Requested scopes for the new access token
+    ) -> OAuthToken:
+        # Validate scopes: requested scopes must be a subset of original scopes
+        original_scopes = set(refresh_token.scopes)
+        requested_scopes = set(scopes)
+        if not requested_scopes.issubset(original_scopes):
+            raise TokenError(
+                "invalid_scope",
+                "Requested scopes exceed those authorized by the refresh token.",
+            )
+
+        # Invalidate old refresh token and its associated access token (rotation)
+        self._revoke_internal(refresh_token_str=refresh_token.token)
+
+        # Issue new tokens
+        new_access_token_value = f"test_access_token_{secrets.token_hex(32)}"
+        new_refresh_token_value = f"test_refresh_token_{secrets.token_hex(32)}"
+
+        access_token_expires_at = int(time.time() + DEFAULT_ACCESS_TOKEN_EXPIRY_SECONDS)
+
+        # Refresh token expiry
+        refresh_token_expires_at = None
+        if DEFAULT_REFRESH_TOKEN_EXPIRY_SECONDS is not None:
+            refresh_token_expires_at = int(
+                time.time() + DEFAULT_REFRESH_TOKEN_EXPIRY_SECONDS
+            )
+
+        self.access_tokens[new_access_token_value] = AccessToken(
+            token=new_access_token_value,
+            client_id=client.client_id,
+            scopes=scopes,  # Use newly requested (and validated) scopes
+            expires_at=access_token_expires_at,
+        )
+        self.refresh_tokens[new_refresh_token_value] = RefreshToken(
+            token=new_refresh_token_value,
+            client_id=client.client_id,
+            scopes=scopes,  # New refresh token also gets these scopes
+            expires_at=refresh_token_expires_at,
+        )
+
+        self._access_to_refresh_map[new_access_token_value] = new_refresh_token_value
+        self._refresh_to_access_map[new_refresh_token_value] = new_access_token_value
+
+        return OAuthToken(
+            access_token=new_access_token_value,
+            token_type="bearer",
+            expires_in=DEFAULT_ACCESS_TOKEN_EXPIRY_SECONDS,
+            refresh_token=new_refresh_token_value,
+            scope=" ".join(scopes),
+        )
+
+    async def load_access_token(self, token: str) -> AccessToken | None:
+        token_obj = self.access_tokens.get(token)
+        if token_obj:
+            if token_obj.expires_at is not None and token_obj.expires_at < time.time():
+                self._revoke_internal(
+                    access_token_str=token_obj.token
+                )  # Clean up expired
+                return None
+            return token_obj
+        return None
+
+    def _revoke_internal(
+        self, access_token_str: str | None = None, refresh_token_str: str | None = None
+    ):
+        """Internal helper to remove tokens and their associations."""
+        removed_access_token = None
+        removed_refresh_token = None
+
+        if access_token_str:
+            if access_token_str in self.access_tokens:
+                del self.access_tokens[access_token_str]
+                removed_access_token = access_token_str
+
+            # Get associated refresh token
+            associated_refresh = self._access_to_refresh_map.pop(access_token_str, None)
+            if associated_refresh:
+                if associated_refresh in self.refresh_tokens:
+                    del self.refresh_tokens[associated_refresh]
+                    removed_refresh_token = associated_refresh
+                self._refresh_to_access_map.pop(associated_refresh, None)
+
+        if refresh_token_str:
+            if refresh_token_str in self.refresh_tokens:
+                del self.refresh_tokens[refresh_token_str]
+                removed_refresh_token = refresh_token_str
+
+            # Get associated access token
+            associated_access = self._refresh_to_access_map.pop(refresh_token_str, None)
+            if associated_access:
+                if associated_access in self.access_tokens:
+                    del self.access_tokens[associated_access]
+                    removed_access_token = associated_access
+                self._access_to_refresh_map.pop(associated_access, None)
+
+        # Clean up any dangling references if one part of the pair was already gone
+        if removed_access_token and removed_access_token in self._access_to_refresh_map:
+            del self._access_to_refresh_map[removed_access_token]
+        if (
+            removed_refresh_token
+            and removed_refresh_token in self._refresh_to_access_map
+        ):
+            del self._refresh_to_access_map[removed_refresh_token]
+
+    async def revoke_token(
+        self,
+        token: AccessToken | RefreshToken,
+    ) -> None:
+        """Revokes an access or refresh token and its counterpart."""
+        if isinstance(token, AccessToken):
+            self._revoke_internal(access_token_str=token.token)
+        elif isinstance(token, RefreshToken):
+            self._revoke_internal(refresh_token_str=token.token)
+        # If token is not found or already revoked, _revoke_internal does nothing, which is correct.

src/fastmcp/server/http.py

@@ -10,14 +10,7 @@ from mcp.server.auth.middleware.bearer_auth import (
     BearerAuthBackend,
     RequireAuthMiddleware,
 )
-from mcp.server.auth.provider import (
-    AccessTokenT,
-    AuthorizationCodeT,
-    OAuthAuthorizationServerProvider,
-    RefreshTokenT,
-)
 from mcp.server.auth.routes import create_auth_routes
-from mcp.server.auth.settings import AuthSettings
 from mcp.server.lowlevel.server import LifespanResultT
 from mcp.server.sse import SseServerTransport
 from mcp.server.streamable_http_manager import StreamableHTTPSessionManager
@@ -29,6 +22,7 @@ from starlette.responses import Response
 from starlette.routing import BaseRoute, Mount, Route
 from starlette.types import Lifespan, Receive, Scope, Send
 
+from fastmcp.server.auth.auth import OAuthProvider
 from fastmcp.utilities.logging import get_logger
 
 if TYPE_CHECKING:
@@ -75,17 +69,12 @@ class RequestContextMiddleware:
 
 
 def setup_auth_middleware_and_routes(
-    auth_server_provider: OAuthAuthorizationServerProvider[
-        AuthorizationCodeT, RefreshTokenT, AccessTokenT
-    ]
-    | None,
-    auth_settings: AuthSettings | None,
+    auth: OAuthProvider,
 ) -> tuple[list[Middleware], list[BaseRoute], list[str]]:
     """Set up authentication middleware and routes if auth is enabled.
 
     Args:
-        auth_server_provider: The OAuth authorization server provider
-        auth_settings: The auth settings
+        auth: The OAuthProvider authorization server provider
 
     Returns:
         Tuple of (middleware, auth_routes, required_scopes)
@@ -94,31 +83,25 @@ def setup_auth_middleware_and_routes(
     auth_routes: list[BaseRoute] = []
     required_scopes: list[str] = []
 
-    if auth_server_provider:
-        if not auth_settings:
-            raise ValueError(
-                "auth_settings must be provided when auth_server_provider is specified"
-            )
+    middleware = [
+        Middleware(
+            AuthenticationMiddleware,
+            backend=BearerAuthBackend(provider=auth),
+        ),
+        Middleware(AuthContextMiddleware),
+    ]
 
-        middleware = [
-            Middleware(
-                AuthenticationMiddleware,
-                backend=BearerAuthBackend(provider=auth_server_provider),
-            ),
-            Middleware(AuthContextMiddleware),
-        ]
-
-        required_scopes = auth_settings.required_scopes or []
-
-        auth_routes.extend(
-            create_auth_routes(
-                provider=auth_server_provider,
-                issuer_url=auth_settings.issuer_url,
-                service_documentation_url=auth_settings.service_documentation_url,
-                client_registration_options=auth_settings.client_registration_options,
-                revocation_options=auth_settings.revocation_options,
-            )
+    required_scopes = auth.settings.required_scopes or []
+
+    auth_routes.extend(
+        create_auth_routes(
+            provider=auth,
+            issuer_url=auth.settings.issuer_url,
+            service_documentation_url=auth.settings.service_documentation_url,
+            client_registration_options=auth.settings.client_registration_options,
+            revocation_options=auth.settings.revocation_options,
         )
+    )
 
     return middleware, auth_routes, required_scopes
 
@@ -155,11 +138,7 @@ def create_sse_app(
     server: FastMCP[LifespanResultT],
     message_path: str,
     sse_path: str,
-    auth_server_provider: OAuthAuthorizationServerProvider[
-        AuthorizationCodeT, RefreshTokenT, AccessTokenT
-    ]
-    | None = None,
-    auth_settings: AuthSettings | None = None,
+    auth: OAuthProvider | None = None,
     debug: bool = False,
     routes: list[BaseRoute] | None = None,
     middleware: list[Middleware] | None = None,
@@ -170,8 +149,7 @@ def create_sse_app(
         server: The FastMCP server instance
         message_path: Path for SSE messages
         sse_path: Path for SSE connections
-        auth_server_provider: Optional auth provider
-        auth_settings: Optional auth settings
+        auth: Optional auth provider
         debug: Whether to enable debug mode
         routes: Optional list of custom routes
         middleware: Optional list of middleware
@@ -196,15 +174,15 @@ def create_sse_app(
         return Response()
 
     # Get auth middleware and routes
-    auth_middleware, auth_routes, required_scopes = setup_auth_middleware_and_routes(
-        auth_server_provider, auth_settings
-    )
-
-    server_routes.extend(auth_routes)
-    server_middleware.extend(auth_middleware)
 
     # Add SSE routes with or without auth
-    if auth_server_provider:
+    if auth:
+        auth_middleware, auth_routes, required_scopes = (
+            setup_auth_middleware_and_routes(auth)
+        )
+
+        server_routes.extend(auth_routes)
+        server_middleware.extend(auth_middleware)
         # Auth is enabled, wrap endpoints with RequireAuthMiddleware
         server_routes.append(
             Route(
@@ -264,11 +242,7 @@ def create_streamable_http_app(
     server: FastMCP[LifespanResultT],
     streamable_http_path: str,
     event_store: None = None,
-    auth_server_provider: OAuthAuthorizationServerProvider[
-        AuthorizationCodeT, RefreshTokenT, AccessTokenT
-    ]
-    | None = None,
-    auth_settings: AuthSettings | None = None,
+    auth: OAuthProvider | None = None,
     json_response: bool = False,
     stateless_http: bool = False,
     debug: bool = False,
@@ -281,8 +255,7 @@ def create_streamable_http_app(
         server: The FastMCP server instance
         streamable_http_path: Path for StreamableHTTP connections
         event_store: Optional event store for session management
-        auth_server_provider: Optional auth provider
-        auth_settings: Optional auth settings
+        auth: Optional auth provider
         json_response: Whether to use JSON response format
         stateless_http: Whether to use stateless mode (new transport per request)
         debug: Whether to enable debug mode
@@ -331,16 +304,15 @@ def create_streamable_http_app(
                 # Re-raise other RuntimeErrors if they don't match the specific message
                 raise
 
-    # Get auth middleware and routes
-    auth_middleware, auth_routes, required_scopes = setup_auth_middleware_and_routes(
-        auth_server_provider, auth_settings
-    )
+    # Add StreamableHTTP routes with or without auth
+    if auth:
+        auth_middleware, auth_routes, required_scopes = (
+            setup_auth_middleware_and_routes(auth)
+        )
 
-    server_routes.extend(auth_routes)
-    server_middleware.extend(auth_middleware)
+        server_routes.extend(auth_routes)
+        server_middleware.extend(auth_middleware)
 
-    # Add StreamableHTTP routes with or without auth
-    if auth_server_provider:
         # Auth is enabled, wrap endpoint with RequireAuthMiddleware
         server_routes.append(
             Mount(

src/fastmcp/server/server.py

@@ -18,7 +18,6 @@ from typing import TYPE_CHECKING, Any, Generic, Literal
 import anyio
 import httpx
 import uvicorn
-from mcp.server.auth.provider import OAuthAuthorizationServerProvider
 from mcp.server.lowlevel.helper_types import ReadResourceContents
 from mcp.server.lowlevel.server import LifespanResultT, NotificationOptions
 from mcp.server.lowlevel.server import Server as MCPServer
@@ -48,6 +47,7 @@ from fastmcp.prompts import Prompt, PromptManager
 from fastmcp.prompts.prompt import PromptResult
 from fastmcp.resources import Resource, ResourceManager
 from fastmcp.resources.template import ResourceTemplate
+from fastmcp.server.auth.auth import OAuthProvider
 from fastmcp.server.http import (
     StarletteWithLifespan,
     create_sse_app,
@@ -110,8 +110,7 @@ class FastMCP(Generic[LifespanResultT]):
         self,
         name: str | None = None,
         instructions: str | None = None,
-        auth_server_provider: OAuthAuthorizationServerProvider[Any, Any, Any]
-        | None = None,
+        auth: OAuthProvider | None = None,
         lifespan: (
             Callable[
                 [FastMCP[LifespanResultT]],
@@ -186,13 +185,7 @@ class FastMCP(Generic[LifespanResultT]):
             lifespan=_lifespan_wrapper(self, lifespan),
         )
 
-        if (self.settings.auth is not None) != (auth_server_provider is not None):
-            # TODO: after we support separate authorization servers (see
-            raise ValueError(
-                "settings.auth must be specified if and only if auth_server_provider "
-                "is specified"
-            )
-        self._auth_server_provider = auth_server_provider
+        self.auth = auth
 
         # Set up MCP protocol handlers
         self._setup_handlers()
@@ -903,8 +896,7 @@ class FastMCP(Generic[LifespanResultT]):
             server=self,
             message_path=message_path or self.settings.message_path,
             sse_path=path or self.settings.sse_path,
-            auth_server_provider=self._auth_server_provider,
-            auth_settings=self.settings.auth,
+            auth=self.auth,
             debug=self.settings.debug,
             middleware=middleware,
         )
@@ -951,8 +943,7 @@ class FastMCP(Generic[LifespanResultT]):
                 server=self,
                 streamable_http_path=path or self.settings.streamable_http_path,
                 event_store=None,
-                auth_server_provider=self._auth_server_provider,
-                auth_settings=self.settings.auth,
+                auth=self.auth,
                 json_response=self.settings.json_response,
                 stateless_http=self.settings.stateless_http,
                 debug=self.settings.debug,
@@ -963,8 +954,7 @@ class FastMCP(Generic[LifespanResultT]):
                 server=self,
                 message_path=self.settings.message_path,
                 sse_path=path or self.settings.sse_path,
-                auth_server_provider=self._auth_server_provider,
-                auth_settings=self.settings.auth,
+                auth=self.auth,
                 debug=self.settings.debug,
                 middleware=middleware,
             )

src/fastmcp/settings.py

@@ -1,16 +1,13 @@
 from __future__ import annotations as _annotations
 
 import inspect
-from typing import TYPE_CHECKING, Annotated, Literal
+from pathlib import Path
+from typing import Annotated, Literal
 
-from mcp.server.auth.settings import AuthSettings
 from pydantic import Field, model_validator
 from pydantic_settings import BaseSettings, SettingsConfigDict
 from typing_extensions import Self
 
-if TYPE_CHECKING:
-    pass
-
 LOG_LEVEL = Literal["DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"]
 
 DuplicateBehavior = Literal["warn", "error", "replace", "ignore"]
@@ -27,6 +24,8 @@ class Settings(BaseSettings):
         nested_model_default_partial_update=True,
     )
 
+    home: Path = Path.home() / ".fastmcp"
+
     test_mode: bool = False
     log_level: LOG_LEVEL = "INFO"
     enable_rich_tracebacks: Annotated[
@@ -171,8 +170,6 @@ class ServerSettings(BaseSettings):
     # cache settings (for checking mounted servers)
     cache_expiration_seconds: float = 0
 
-    auth: AuthSettings | None = None
-
     # StreamableHTTP settings
     json_response: bool = False
     stateless_http: bool = (

src/fastmcp/utilities/tests.py

@@ -94,7 +94,7 @@ def run_server_in_process(
     proc.start()
 
     # Wait for server to be running
-    max_attempts = 100
+    max_attempts = 10
     attempt = 0
     while attempt < max_attempts and proc.is_alive():
         try:
@@ -102,7 +102,10 @@ def run_server_in_process(
                 s.connect((host, port))
                 break
         except ConnectionRefusedError:
-            time.sleep(0.01)
+            if attempt < 3:
+                time.sleep(0.01)
+            else:
+                time.sleep(0.1)
             attempt += 1
     else:
         raise RuntimeError(f"Server failed to start after {max_attempts} attempts")

tests/client/test_oauth.py

@@ -0,0 +1,284 @@
+import sys
+from collections.abc import Generator
+from unittest.mock import patch
+from urllib.parse import parse_qs, urlparse
+
+import httpx
+import pytest
+import uvicorn
+
+import fastmcp.client.auth  # Import module, not the function directly
+from fastmcp.client import Client
+from fastmcp.client.transports import StreamableHttpTransport
+from fastmcp.server.auth.auth import ClientRegistrationOptions
+from fastmcp.server.auth.in_memory_provider import InMemoryOAuthProvider
+from fastmcp.server.server import FastMCP
+from fastmcp.utilities.tests import run_server_in_process
+
+
+def fastmcp_server(issuer_url: str):
+    """Create a FastMCP server with OAuth authentication."""
+    server = FastMCP(
+        "TestServer",
+        auth=InMemoryOAuthProvider(
+            issuer_url=issuer_url,
+            client_registration_options=ClientRegistrationOptions(enabled=True),
+        ),
+    )
+
+    @server.tool()
+    def add(a: int, b: int) -> int:
+        """Add two numbers together."""
+        return a + b
+
+    @server.resource("resource://test")
+    def get_test_resource() -> str:
+        """Get a test resource."""
+        return "Hello from authenticated resource!"
+
+    return server
+
+
+def run_server(host: str, port: int, transport: str | None = None) -> None:
+    try:
+        # Configure OAuth provider with the actual server URL
+        issuer_url = f"http://{host}:{port}"
+        app = fastmcp_server(issuer_url).http_app()
+        server = uvicorn.Server(
+            config=uvicorn.Config(
+                app=app,
+                host=host,
+                port=port,
+                log_level="error",
+                lifespan="on",
+            )
+        )
+        server.run()
+    except Exception as e:
+        print(f"Server error: {e}")
+        sys.exit(1)
+    sys.exit(0)
+
+
+@pytest.fixture(scope="module")
+def streamable_http_server() -> Generator[str, None, None]:
+    with run_server_in_process(run_server) as url:
+        yield f"{url}/mcp"
+
+
+@pytest.fixture()
+def client_unauthorized(streamable_http_server: str) -> Client:
+    return Client(transport=StreamableHttpTransport(streamable_http_server))
+
+
+class HeadlessOAuthProvider(httpx.Auth):
+    """
+    OAuth provider that bypasses browser interaction for testing.
+
+    This simulates the complete OAuth flow programmatically by:
+    1. Discovering OAuth metadata from the server
+    2. Registering a client
+    3. Getting an authorization code (simulates user approval)
+    4. Exchanging it for an access token
+    5. Adding Bearer token to all requests
+
+    This enables testing OAuth-protected FastMCP servers without
+    requiring browser interaction or external OAuth providers.
+    """
+
+    def __init__(self, mcp_url: str):
+        self.mcp_url = mcp_url
+        parsed_url = urlparse(mcp_url)
+        self.server_base_url = f"{parsed_url.scheme}://{parsed_url.netloc}"
+        self._access_token = None
+
+    async def async_auth_flow(self, request):
+        """httpx.Auth interface - add Bearer token to requests."""
+        if not self._access_token:
+            await self._obtain_token()
+
+        if self._access_token:
+            request.headers["Authorization"] = f"Bearer {self._access_token}"
+
+        yield request
+
+    async def _obtain_token(self):
+        """Get a valid access token by simulating the OAuth flow."""
+        import base64
+        import hashlib
+        import secrets
+
+        from mcp.shared.auth import OAuthClientInformationFull
+        from pydantic import AnyHttpUrl
+
+        # Generate PKCE challenge/verifier
+        code_verifier = (
+            base64.urlsafe_b64encode(secrets.token_bytes(32)).decode().rstrip("=")
+        )
+        code_challenge = (
+            base64.urlsafe_b64encode(hashlib.sha256(code_verifier.encode()).digest())
+            .decode()
+            .rstrip("=")
+        )
+
+        # Create HTTP client to talk to the server
+        async with httpx.AsyncClient() as http_client:
+            # 1. Discover OAuth metadata
+            metadata_url = (
+                f"{self.server_base_url}/.well-known/oauth-authorization-server"
+            )
+            response = await http_client.get(metadata_url)
+            response.raise_for_status()
+            metadata = response.json()
+
+            # 2. Register a client
+            client_info = OAuthClientInformationFull(
+                client_id="test_client_headless",
+                client_secret="test_secret_headless",
+                redirect_uris=[AnyHttpUrl("http://localhost:8080/callback")],
+            )
+
+            register_response = await http_client.post(
+                metadata["registration_endpoint"],
+                json=client_info.model_dump(mode="json"),
+            )
+            register_response.raise_for_status()
+            registered_client = register_response.json()
+
+            # 3. Get authorization code (simulate user approval)
+            auth_params = {
+                "response_type": "code",
+                "client_id": registered_client["client_id"],
+                "redirect_uri": "http://localhost:8080/callback",
+                "code_challenge": code_challenge,
+                "code_challenge_method": "S256",
+                "state": "test_state_headless",
+            }
+
+            auth_response = await http_client.get(
+                metadata["authorization_endpoint"],
+                params=auth_params,
+                follow_redirects=False,
+            )
+
+            # Extract auth code from redirect
+            if auth_response.status_code == 302:
+                redirect_url = auth_response.headers["location"]
+                parsed = urlparse(redirect_url)
+                query_params = parse_qs(parsed.query)
+
+                if "error" in query_params:
+                    error = query_params["error"][0]
+                    error_desc = query_params.get(
+                        "error_description", ["Unknown error"]
+                    )[0]
+                    raise RuntimeError(
+                        f"OAuth authorization failed: {error} - {error_desc}"
+                    )
+
+                auth_code = query_params["code"][0]
+
+                # 4. Exchange auth code for access token
+                token_data = {
+                    "grant_type": "authorization_code",
+                    "client_id": registered_client["client_id"],
+                    "client_secret": registered_client["client_secret"],
+                    "code": auth_code,
+                    "redirect_uri": "http://localhost:8080/callback",
+                    "code_verifier": code_verifier,
+                }
+
+                token_response = await http_client.post(
+                    metadata["token_endpoint"], data=token_data
+                )
+                token_response.raise_for_status()
+                token_info = token_response.json()
+
+                self._access_token = token_info["access_token"]
+            else:
+                raise RuntimeError(f"Authorization failed: {auth_response.status_code}")
+
+
+@pytest.fixture()
+def client_with_headless_oauth(
+    streamable_http_server: str,
+) -> Generator[Client, None, None]:
+    """Client with headless OAuth that bypasses browser interaction."""
+
+    # Patch the OAuth function to return our headless provider
+    def headless_oauth(*args, **kwargs):
+        mcp_url = args[0] if args else kwargs.get("mcp_url", "")
+        if not mcp_url:
+            raise ValueError("mcp_url is required")
+        return HeadlessOAuthProvider(mcp_url)
+
+    with patch("fastmcp.client.auth.OAuth", side_effect=headless_oauth):
+        client = Client(
+            transport=StreamableHttpTransport(streamable_http_server),
+            auth=fastmcp.client.auth.OAuth(mcp_url=streamable_http_server),
+        )
+        yield client
+
+
+async def test_unauthorized(client_unauthorized: Client):
+    """Test that unauthenticated requests are rejected."""
+    with pytest.raises(httpx.HTTPStatusError, match="401 Unauthorized"):
+        async with client_unauthorized:
+            pass
+
+
+async def test_ping(client_with_headless_oauth: Client):
+    """Test that we can ping the server."""
+    async with client_with_headless_oauth:
+        assert await client_with_headless_oauth.ping()
+
+
+async def test_list_tools(client_with_headless_oauth: Client):
+    """Test that we can list tools."""
+    async with client_with_headless_oauth:
+        tools = await client_with_headless_oauth.list_tools()
+        tool_names = [tool.name for tool in tools]
+        assert "add" in tool_names
+
+
+async def test_call_tool(client_with_headless_oauth: Client):
+    """Test that we can call a tool."""
+    async with client_with_headless_oauth:
+        result = await client_with_headless_oauth.call_tool("add", {"a": 5, "b": 3})
+        assert result[0].text == "8"  # type: ignore[attr-defined]
+
+
+async def test_list_resources(client_with_headless_oauth: Client):
+    """Test that we can list resources."""
+    async with client_with_headless_oauth:
+        resources = await client_with_headless_oauth.list_resources()
+        resource_uris = [str(resource.uri) for resource in resources]
+        assert "resource://test" in resource_uris
+
+
+async def test_read_resource(client_with_headless_oauth: Client):
+    """Test that we can read a resource."""
+    async with client_with_headless_oauth:
+        resource = await client_with_headless_oauth.read_resource("resource://test")
+        assert resource[0].text == "Hello from authenticated resource!"  # type: ignore[attr-defined]
+
+
+async def test_oauth_server_metadata_discovery(streamable_http_server: str):
+    """Test that we can discover OAuth metadata from the running server."""
+    parsed_url = urlparse(streamable_http_server)
+    server_base_url = f"{parsed_url.scheme}://{parsed_url.netloc}"
+
+    async with httpx.AsyncClient() as client:
+        # Test OAuth discovery endpoint
+        metadata_url = f"{server_base_url}/.well-known/oauth-authorization-server"
+        response = await client.get(metadata_url)
+        assert response.status_code == 200
+
+        metadata = response.json()
+        assert "authorization_endpoint" in metadata
+        assert "token_endpoint" in metadata
+        assert "registration_endpoint" in metadata
+
+        # The endpoints should be properly formed URLs
+        assert metadata["authorization_endpoint"].startswith(server_base_url)
+        assert metadata["token_endpoint"].startswith(server_base_url)