Merge pull request #287 from open-webui/dev

0.0.20

.dockerignore

@@ -0,0 +1,72 @@
+########################
+#  Version-control
+########################
+.git
+.gitignore
+.gitattributes
+
+########################
+#  Docker artefacts
+########################
+.dockerignore
+Dockerfile
+docker-compose.yml
+.docker/
+
+########################
+#  Python byte-code / Caches
+########################
+**/__pycache__/
+**/*.py[cod]
+*.so                     # compiled C extensions
+.pytest_cache/
+.mypy_cache/
+.nox/
+.tox/
+
+########################
+#  Virtual-envs & deps
+########################
+.venv/
+venv/
+env/
+*.egg-info/
+*.dist-info/
+build/
+dist/
+pip-wheel-metadata/
+
+########################
+#  Testing & coverage
+########################
+.coverage
+htmlcov/
+nosetests.xml
+coverage.xml
+
+########################
+#  Docs & generated reports
+########################
+docs/_build/
+
+########################
+#  IDE / Editor artefacts
+########################
+.idea/
+.vscode/
+.ropeproject/
+**/*.swp  # Vim swap files
+
+########################
+#  Ops / secrets (keep these out of images!)
+########################
+*.env
+.env*
+*.pem
+*.key
+config.json
+########################
+#  OS junk
+########################
+.DS_Store
+Thumbs.db

.github/workflows/build-release.yml

@@ -34,7 +34,7 @@ jobs:
                   CHANGELOG_ESCAPED=$(echo "$CHANGELOG_CONTENT" | sed ':a;N;$!ba;s/\n/%0A/g')
                   echo "Extracted latest release notes from CHANGELOG.md:" 
                   echo -e "$CHANGELOG_CONTENT" 
-                  echo "::set-output name=content::$CHANGELOG_ESCAPED"
+                  echo "content=$CHANGELOG_ESCAPED" >> "$GITHUB_OUTPUT"
 
             - name: Create GitHub release
               uses: actions/github-script@v7

.github/workflows/publish.yaml

@@ -0,0 +1,38 @@
+# .github/workflows/publish.yml
+name: Publish Python Package
+
+on:
+    push:
+        branches:
+            - main
+    release:
+        types: [published] # Triggers workflow when a GitHub release is published
+
+permissions:
+    contents: read
+
+jobs:
+    publish:
+        name: Build and Publish to PyPI
+        runs-on: ubuntu-latest
+
+        steps:
+            - name: Checkout code
+              uses: actions/checkout@v4
+
+            - name: Set up Python
+              uses: actions/setup-python@v5
+              with:
+                  python-version: "3.11" # or your required version
+
+            - name: Install uv
+              run: pip install uv
+
+            - name: Build package
+              run: uv build --no-sources
+
+            - name: Publish package to PyPI
+              env:
+                  UV_PUBLISH_TOKEN: ${{ secrets.PYPI_API_TOKEN }}
+              run: |
+                  uv publish

.gitignore

@@ -8,4 +8,8 @@ wheels/
 
 # Virtual environments
 .venv
-config.json
+config.json
+.python-version
+.vscode
+
+.DS_Store

.python-version

@@ -1 +0,0 @@
-3.11

CHANGELOG.md

@@ -5,18 +5,56 @@ All notable changes to this project will be documented in this file.
 The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/),
 and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
-# Changelog
+## [0.0.20] - 2026-02-27
 
-All notable changes to this project will be documented in this file.
+### Added
 
-The format is based on Keep a Changelog,
-and this project adheres to Semantic Versioning.
+* 🔁 **MCP Connection Manager with Auto-Reconnect**: Introduced MCPConnectionManager to manage the full lifecycle of MCP client sessions and transports. Connections that drop due to a ClosedResourceError are now automatically re-established and retried transparently, eliminating manual restarts for intermittent server disconnects.
+* 🔄 **Automatic Retry on Closed Sessions**: Tool calls that encounter a ClosedResourceError will now automatically reconnect to the MCP server and retry the call, providing resilient end-to-end execution without user intervention.
+
+### Changed
+
+* 🧩 **Normalized disabledTools Configuration Key**: The disabled tools setting now accepts both camelCase (disabledTools) and snake_case (disabled_tools) in the config file for backwards compatibility and consistency with other configuration keys.
+* 🧹 **Simplified Tool Handler Architecture**: Refactored get_tool_handler to remove the session parameter and nested factory functions in favor of a cleaner design that resolves the session from the request app state at call time, enabling reconnect-aware tool execution.
+* 🛡️ **Guarded Lifespan Startup During Hot Reload**: The reload handler now checks for the presence of an async context manager before awaiting sub-app lifespan startup, preventing errors when lifespan context is unavailable.
+
+### Fixed
+
+* 🪛 **Fixed Erroneous Validation Raise in disabledTools**: Removed a misplaced raise statement in validate_server_config that would always throw an error after validating disabled_tools, even when the configuration was correct.
+* 🧯 **Graceful Handling of asyncio.CancelledError**: Added explicit CancelledError handling during server creation and lifespan startup to properly roll back routes and log errors instead of crashing silently.
+
+## [0.0.19] - 2025-10-14
+
+### Fixed
+
+* 🔁 **Reverted Client Header Forwarding**: Reverted changes introduced in 0.0.18.
+
+## [0.0.18] - 2025-10-14
+
+### Added
+
+* 🧩 **OAuth Support for Streamable HTTP Servers**: Introduced full OAuth 2.0 support with dynamic client registration for 'streamable_http' MCP servers. This enables secure, standards-compliant authentication flows for both user and service clients.
+* 🔧 **Configurable Disabled Tools**: Added 'disabled_tools' option in server configuration, allowing selective disabling of specific tools without modifying code. Useful for managing staged rollouts or limited-access environments.
+* 🧠 **Adjustable Log Level at Runtime**: Logging verbosity can now be adjusted dynamically, giving operators finer control over monitoring and debugging noise without requiring restarts.
+* 🔁 **Client Header Forwarding**: Added automatic forwarding of client HTTP headers to MCP backends—enabling trace propagation, correlation IDs, and richer observability in distributed environments.
+* 🛤️ **Support for Custom Path Prefixes**: Servers can now run under a custom path prefix, allowing flexible deployment within multi-service gateways or reverse proxy environments.
+* 📄 **Root Path Documentation**: Added documentation describing behavior and configuration for root path handling in multi-app or nested setups.
+
+### Changed
+
+* 🧰 **Refactored Connection Timeout Defaults**: Connection timeout is now set to 'None' by default to prevent premature disconnects during long-running or streaming tasks.
+* 🧹 **Removed Deprecated Python Version Tagging**: Dropped explicit Python version tagging; the project now officially supports Python 3.11 and above without manual tagging overhead.
+
+### Fixed
+
+* 🪛 **Improved Hot Reload Lifespan Tracking**: Fixed an issue where sub-applications created during hot reloads were not properly initialized or cleaned up—ensuring stable lifecycle management during dynamic reconfiguration.
+* 🔗 **Symlink Handling in Config Watcher**: Resolved a bug where configuration symlinks were not updating correctly on modification; watcher now tracks and reloads the proper file path automatically.
 
 ## [0.0.17] - 2025-07-22
 
 ### Added
 
-- 🔄 **Hot Reload Support for Configuration Files**: Added `--hot-reload` flag to watch your config file for changes and dynamically reload MCP servers without restarting the application—enabling seamless development workflows and runtime configuration updates.
+- 🔄 **Hot Reload Support for Configuration Files**: Added \`--hot-reload\` flag to watch your config file for changes and dynamically reload MCP servers without restarting the application—enabling seamless development workflows and runtime configuration updates.
 - 🤫 **HTTP Request Filtering for Cleaner Logs**: Added configurable log filtering to reduce noise from frequent HTTP requests, making debugging and monitoring much clearer in production environments.
 
 ### Changed
@@ -115,4 +153,4 @@ and this project adheres to Semantic Versioning.
 
 ### Fixed
 
-- 🧹 **Cleaner Proxy Output**: Dropped None arguments from proxy requests, resulting in reduced clutter and improved interoperability with servers expecting clean inputs—ensuring more reliable downstream performance with MCP tools.
+- 🧹 **Cleaner Proxy Output**: Dropped None arguments from proxy requests, resulting in reduced clutter and improved interoperability with servers expecting clean inputs—ensuring more reliable downstream performance with MCP tools.

OAUTH_GUIDE.md

@@ -0,0 +1,204 @@
+# OAuth 2.1 Support in MCPO
+
+MCPO now supports OAuth 2.1 client-to-server authentication for MCP servers that require it. This enables secure authentication with servers using the Authorization Code flow with refresh tokens.
+
+## Features
+
+- **OAuth 2.1 Authorization Code Flow**: Full support for the modern OAuth 2.1 standard
+- **Automatic Token Refresh**: Tokens are automatically refreshed when they expire
+- **Persistent Token Storage**: Tokens can be stored in memory or on disk
+- **Browser-based Authorization**: Automatically opens browser for user authorization
+- **Loopback Callback Server**: Built-in HTTP server for catching OAuth callbacks
+
+## Configuration
+
+### Basic OAuth Configuration (Recommended)
+
+MCPO defaults to **dynamic client registration** for maximum compatibility. Most OAuth servers only need minimal configuration:
+
+```json
+{
+  "mcpServers": {
+    "my-oauth-server": {
+      "type": "streamable-http",
+      "url": "http://localhost:8000/mcp",
+      "oauth": {
+        "server_url": "http://localhost:8000"
+      }
+    }
+  }
+}
+```
+
+This minimal configuration allows MCPO to:
+- Automatically discover OAuth endpoints from `/.well-known/oauth-authorization-server`
+- Perform dynamic client registration with the server
+- Use server-provided scopes and configuration
+
+### Static Client Configuration (Legacy Servers)
+
+For servers that don't support dynamic client registration, you can specify static client metadata:
+
+```json
+{
+  "mcpServers": {
+    "legacy-oauth-server": {
+      "type": "streamable-http",
+      "url": "http://localhost:8000/mcp", 
+      "oauth": {
+        "server_url": "http://localhost:8000",
+        "storage_type": "file",
+        "client_metadata": {
+          "client_name": "My MCPO Client",
+          "redirect_uris": ["http://localhost:3030/callback"],
+          "grant_types": ["authorization_code", "refresh_token"],
+          "response_types": ["code"]
+        }
+      }
+    }
+  }
+}
+```
+
+> ⚠️ **Important**: Avoid hardcoding `scope`, `authorization_endpoint`, or `token_endpoint` in your configuration. These should be automatically discovered from the server's OAuth metadata during the flow.
+
+## OAuth Configuration Options
+
+### Required Fields
+
+- `server_url`: The base URL of the OAuth server (without `/mcp` endpoint)
+
+### Optional Fields
+
+- `callback_port`: Port for the local callback server (default: 3030)
+- `use_loopback`: Whether to use automatic browser flow (default: true)
+- `storage_type`: Where to store tokens - "memory" or "file" (default: "file")
+- `client_metadata`: OAuth client registration metadata
+
+### Client Metadata Fields (for static registration only)
+
+- `client_name`: Display name for your client
+- `redirect_uris`: List of allowed redirect URIs
+- `grant_types`: OAuth grant types to request (typically `["authorization_code", "refresh_token"]`)
+- `response_types`: OAuth response types to accept (typically `["code"]`)
+- `token_endpoint_auth_method`: How to authenticate with the token endpoint
+
+**Fields handled automatically (don't set these):**
+- `scope`: Automatically provided by the server during dynamic registration
+- `authorization_endpoint`: Discovered from server metadata
+- `token_endpoint`: Discovered from server metadata
+
+## How It Works
+
+1. **Metadata Discovery**: MCPO automatically discovers OAuth endpoints from the server's `/.well-known/oauth-authorization-server` metadata
+2. **Dynamic Registration**: If supported, MCPO registers itself as a client with the OAuth server
+3. **Token Check**: MCPO checks for existing valid tokens for this server
+4. **Authorization Flow**: If no valid tokens exist, MCPO:
+   - Opens your browser to the authorization URL
+   - Starts a local HTTP server to receive the callback
+   - Exchanges the authorization code for tokens
+5. **Token Storage**: Tokens are stored securely:
+   - **File storage**: In `~/.mcpo/tokens/` directory
+   - **Memory storage**: Only for the duration of the session
+6. **Token Refresh**: When tokens expire, MCPO automatically refreshes them using the refresh token
+7. **Authentication**: All requests to the MCP server include the access token
+
+## Storage Types
+
+### File Storage (Recommended)
+- Tokens persist between MCPO restarts
+- Stored in `~/.mcpo/tokens/` with per-server isolation
+- Each server gets its own token file based on a hash of the server name
+
+### Memory Storage
+- Tokens only exist for the current session
+- Lost when MCPO restarts
+- Useful for testing or temporary sessions
+
+## Authorization Flows
+
+### Automatic Browser Flow (Default)
+When `use_loopback` is `true`:
+1. MCPO opens your default browser to the authorization page
+2. After you authorize, you're redirected to `http://localhost:PORT/callback`
+3. MCPO's built-in server captures the authorization code
+4. The browser shows a success message
+
+### Manual Copy/Paste Flow
+When `use_loopback` is `false`:
+1. MCPO prints the authorization URL
+2. You manually open the URL in a browser
+3. After authorization, you copy the full callback URL
+4. You paste it back into MCPO when prompted
+
+## Server Support
+
+OAuth authentication is supported for:
+- ✅ `streamable-http` servers
+- ❌ `sse` servers (not currently supported)
+- ❌ `stdio` servers (OAuth not applicable)
+
+## Security Considerations
+
+1. **Token Storage**: File-based tokens are stored in plaintext. For production use, consider implementing encrypted storage
+2. **Redirect URIs**: Always use `localhost` for development. For production, use HTTPS URLs
+3. **Scopes**: Only request the minimum scopes necessary for your use case
+4. **Token Expiry**: Tokens are automatically refreshed, but expired tokens are not automatically deleted
+
+## Troubleshooting
+
+### "OAuth server_url required"
+Ensure your configuration includes the `server_url` field in the `oauth` section.
+
+### Browser doesn't open
+Check that `use_loopback` is set to `true` and that your system has a default browser configured.
+
+### "No authorization code found"
+If using manual flow, ensure you're copying the complete callback URL including all query parameters.
+
+### Port already in use
+Change the `callback_port` to an available port number.
+
+### Tokens not persisting
+Ensure `storage_type` is set to `"file"` and that MCPO has write permissions to `~/.mcpo/tokens/`.
+
+## Example: Testing OAuth
+
+1. Create a config file with OAuth settings:
+```json
+{
+  "mcpServers": {
+    "test-oauth": {
+      "type": "streamable-http",
+      "url": "http://localhost:8000/mcp",
+      "oauth": {
+        "server_url": "http://localhost:8000",
+        "storage_type": "file"
+      }
+    }
+  }
+}
+```
+
+2. Start MCPO:
+```bash
+mcpo --config config_oauth.json
+```
+
+3. MCPO will:
+   - Open your browser for authorization
+   - Capture the callback
+   - Store tokens
+   - Connect to the MCP server
+
+4. Subsequent connections will reuse the stored tokens
+
+## Implementation Details
+
+The OAuth implementation uses the Python MCP SDK's built-in `OAuthClientProvider` which handles:
+- PKCE (Proof Key for Code Exchange) when required
+- Token refresh logic
+- Authorization header injection
+- Dynamic client registration (if supported by server)
+
+Tokens are bound to individual server instances and isolated from each other, ensuring that each OAuth-enabled server maintains its own authentication session.

README.md

@@ -74,6 +74,18 @@ That’s it. Your MCP tool is now available at http://localhost:8000 with a gene
 
 🤝 **To integrate with Open WebUI after launching the server, check our [docs](https://docs.openwebui.com/openapi-servers/open-webui/).**
 
+
+### 🌐 Serving Under a Subpath (`--root-path`)
+
+If you need to serve mcpo behind a reverse proxy or under a subpath (e.g., `/api/mcpo`), use the `--root-path` argument:
+
+```bash
+mcpo --port 8000 --root-path "/api/mcpo" --api-key "top-secret" -- your_mcp_server_command
+```
+
+All routes will be served under the specified root path, e.g. `http://localhost:8000/api/mcpo/memory`.
+
+
 ### 🔄 Using a Config File
 
 You can serve multiple MCP tools via a single config file that follows the [Claude Desktop](https://modelcontextprotocol.io/quickstart/user) format.
@@ -103,7 +115,8 @@ Example config.json:
     },
     "time": {
       "command": "uvx",
-      "args": ["mcp-server-time", "--local-timezone=America/New_York"]
+      "args": ["mcp-server-time", "--local-timezone=America/New_York"],
+      "disabledTools": ["convert_time"] // Disable specific tools if needed
     },
     "mcp_sse": {
       "type": "sse", // Explicitly define type
@@ -127,6 +140,64 @@ Each tool will be accessible under its own unique route, e.g.:
 
 Each with a dedicated OpenAPI schema and proxy handler. Access full schema UI at: `http://localhost:8000/<tool>/docs`  (e.g. /memory/docs, /time/docs)
 
+### 🔐 OAuth 2.1 Authentication
+
+mcpo supports OAuth 2.1 authentication for MCP servers that require it. The implementation defaults to **dynamic client registration**, so most servers only need minimal configuration:
+
+```json
+{
+  "mcpServers": {
+    "oauth-protected-server": {
+      "type": "streamable-http",
+      "url": "http://localhost:8000/mcp",
+      "oauth": {
+        "server_url": "http://localhost:8000"
+      }
+    }
+  }
+}
+```
+
+#### OAuth Configuration Options
+
+**Basic Options:**
+- `server_url` (required): OAuth server base URL
+- `storage_type`: "file" (persistent) or "memory" (session-only, default: "file")
+- `callback_port`: Local port for OAuth callback (default: 3030)
+- `use_loopback`: Auto-open browser for auth (default: true)
+
+**Advanced Options (rarely needed):**
+For servers that don't support dynamic client registration, you can specify static client metadata:
+
+```json
+{
+  "mcpServers": {
+    "legacy-oauth-server": {
+      "type": "streamable-http", 
+      "url": "http://api.example.com/mcp",
+      "oauth": {
+        "server_url": "http://api.example.com",
+        "client_metadata": {
+          "client_name": "My MCPO Client",
+          "redirect_uris": ["http://localhost:3030/callback"]
+        }
+      }
+    }
+  }
+}
+```
+
+> **Note**: Avoid setting `scope`, `authorization_endpoint`, or `token_endpoint` in the config. These are automatically discovered from the server's OAuth metadata during the dynamic registration flow.
+
+On first connection, mcpo will:
+1. Perform dynamic client registration (if supported)
+2. Open your browser for authorization
+3. Capture the OAuth callback automatically  
+4. Store tokens securely (in `~/.mcpo/tokens/` for file storage)
+5. Use tokens for all subsequent requests
+
+OAuth is supported for `streamable-http` server types. See [OAUTH_GUIDE.md](OAUTH_GUIDE.md) for detailed documentation.
+
 ## 🔧 Requirements
 
 - Python 3.8+

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "mcpo"
-version = "0.0.16"
+version = "0.0.20"
 description = "A simple, secure MCP-to-OpenAPI proxy server"
 authors = [
     { name = "Timothy Jaeryang Baek", email = "tim@openwebui.com" }
@@ -10,8 +10,8 @@ requires-python = ">=3.11"
 dependencies = [
     "click>=8.1.8",
     "fastapi>=0.115.12",
-    "mcp>=1.12.1",
-    "mcp[cli]>=1.12.1",
+    "mcp>=1.17.0",
+    "mcp[cli]>=1.17.0",
     "passlib[bcrypt]>=1.7.4",
     "pydantic>=2.11.1",
     "pyjwt[crypto]>=2.10.1",

src/mcpo/__init__.py

@@ -60,6 +60,9 @@ def main(
     ssl_keyfile: Annotated[
         Optional[str], typer.Option("--ssl-keyfile", "-K", help="SSL keyfile")
     ] = None,
+    root_path: Annotated[
+        Optional[str], typer.Option("--root-path", help="Root path")
+    ] = "",
     path_prefix: Annotated[
         Optional[str], typer.Option("--path-prefix", help="URL prefix")
     ] = None,
@@ -69,6 +72,9 @@ def main(
     hot_reload: Annotated[
         Optional[bool], typer.Option("--hot-reload", help="Enable hot reload for config file changes")
     ] = False,
+    log_level: Annotated[
+        Optional[str], typer.Option("--log-level", help="Set log level (DEBUG, INFO, WARNING, ERROR, CRITICAL)")
+    ] = None,
 ):
     server_command = None
     if not config_path:
@@ -122,6 +128,10 @@ def main(
     if not path_prefix.startswith("/"):
         path_prefix = f"/{path_prefix}"
 
+    # Set LOG_LEVEL environment variable if provided
+    if log_level:
+        os.environ["LOG_LEVEL"] = log_level
+
     # Run your async run function from mcpo.main
     asyncio.run(
         run(
@@ -139,6 +149,7 @@ def main(
             ssl_certfile=ssl_certfile,
             ssl_keyfile=ssl_keyfile,
             path_prefix=path_prefix,
+            root_path=root_path,
             headers=headers,
             hot_reload=hot_reload,
         )

src/mcpo/main.py

@@ -1,11 +1,12 @@
 import asyncio
+import contextlib
 import json
 import logging
 import os
 import signal
 import socket
 from contextlib import AsyncExitStack, asynccontextmanager
-from typing import Optional, Dict, Any
+from typing import Any, Dict, List, Optional
 from urllib.parse import urljoin
 
 import uvicorn
@@ -24,13 +25,15 @@ from mcpo.utils.main import (
     get_tool_handler,
     normalize_server_type,
 )
-from mcpo.utils.main import get_model_fields, get_tool_handler
-from mcpo.utils.auth import get_verify_api_key, APIKeyMiddleware
 from mcpo.utils.config_watcher import ConfigWatcher
+from mcpo.utils.headers import validate_client_header_forwarding_config
+from mcpo.utils.oauth import create_oauth_provider
 
 
 logger = logging.getLogger(__name__)
 
+CONNECTION_TIMEOUT = os.getenv("CONNECTION_TIMEOUT", None)
+
 
 class GracefulShutdown:
     def __init__(self):
@@ -50,13 +53,148 @@ class GracefulShutdown:
         task.add_done_callback(self.tasks.discard)
 
 
+class MCPConnectionManager:
+    """
+    Manages lifecycle of the MCP ClientSession and underlying transport so that
+    we can reconnect transparently when the remote server drops the connection.
+    """
+
+    def __init__(
+        self,
+        *,
+        server_type: str,
+        command: Optional[str],
+        args: List[str],
+        env: Dict[str, str],
+        headers: Optional[Dict[str, str]],
+        connection_timeout: Optional[int],
+        auth_provider: Optional[Any] = None,
+    ):
+        self.server_type = normalize_server_type(server_type)
+        self.command = command
+        self.args = args
+        self.env = env
+        self.headers = headers
+        self.connection_timeout = connection_timeout
+        self.auth_provider = auth_provider
+
+        self._session: Optional[ClientSession] = None
+        self._client_context = None
+        self._lock = asyncio.Lock()
+        self._initialize_lock = asyncio.Lock()
+        self._initialize_result = None
+        self._initialized = False
+
+    @property
+    def current_session(self) -> Optional[ClientSession]:
+        return self._session
+
+    async def get_session(self) -> ClientSession:
+        async with self._lock:
+            if self._session is None:
+                await self._open_session_locked()
+            return self._session
+
+    async def ensure_initialized(self):
+        session = await self.get_session()
+        if self._initialized and self._initialize_result is not None:
+            return session, self._initialize_result
+
+        async with self._initialize_lock:
+            if not self._initialized or self._initialize_result is None:
+                initialize_result = await session.initialize()
+                self._initialize_result = initialize_result
+                self._initialized = True
+            else:
+                initialize_result = self._initialize_result
+
+        return session, initialize_result
+
+    async def reconnect(self):
+        async with self._lock:
+            await self._close_session_locked()
+            await self._open_session_locked()
+        # Run initialize outside of the lock to avoid deadlocks on nested calls
+        return await self.ensure_initialized()
+
+    async def close(self):
+        async with self._lock:
+            await self._close_session_locked()
+
+    async def _open_session_locked(self):
+        client_context = self._create_client_context()
+        try:
+            connection = await client_context.__aenter__()
+        except Exception:
+            # Ensure the context is closed if entering fails
+            with contextlib.suppress(Exception):
+                await client_context.__aexit__(None, None, None)
+            raise
+
+        reader, writer, *_ = connection
+        session = ClientSession(reader, writer)
+        try:
+            await session.__aenter__()
+        except Exception:
+            with contextlib.suppress(Exception):
+                await session.__aexit__(None, None, None)
+            with contextlib.suppress(Exception):
+                await client_context.__aexit__(None, None, None)
+            raise
+
+        self._client_context = client_context
+        self._session = session
+        self._initialized = False
+        self._initialize_result = None
+
+    async def _close_session_locked(self):
+        session, client_context = self._session, self._client_context
+        self._session = None
+        self._client_context = None
+        self._initialized = False
+        self._initialize_result = None
+
+        if session is not None:
+            with contextlib.suppress(Exception):
+                await session.__aexit__(None, None, None)
+
+        if client_context is not None:
+            with contextlib.suppress(Exception):
+                await client_context.__aexit__(None, None, None)
+
+    def _create_client_context(self):
+        if self.server_type == "stdio":
+            server_params = StdioServerParameters(
+                command=self.command,
+                args=self.args,
+                env={**os.environ, **self.env},
+            )
+            return stdio_client(server_params)
+        if self.server_type == "sse":
+            timeout = self.connection_timeout or 900
+            return sse_client(
+                url=self.args[0],
+                sse_read_timeout=timeout,
+                headers=self.headers,
+            )
+        if self.server_type == "streamable-http":
+            return streamablehttp_client(
+                url=self.args[0],
+                headers=self.headers,
+                auth=self.auth_provider,
+            )
+        raise ValueError(f"Unsupported server type: {self.server_type}")
+
+
 def validate_server_config(server_name: str, server_cfg: Dict[str, Any]) -> None:
     """Validate individual server configuration."""
     server_type = server_cfg.get("type")
 
     if normalize_server_type(server_type) in ("sse", "streamable-http"):
         if not server_cfg.get("url"):
-            raise ValueError(f"Server '{server_name}' of type '{server_type}' requires a 'url' field")
+            raise ValueError(
+                f"Server '{server_name}' of type '{server_type}' requires a 'url' field"
+            )
     elif server_cfg.get("command"):
         # stdio server
         if not isinstance(server_cfg["command"], str):
@@ -68,6 +206,17 @@ def validate_server_config(server_name: str, server_cfg: Dict[str, Any]) -> None
         pass
     else:
         raise ValueError(f"Server '{server_name}' must have either 'command' for stdio or 'type' and 'url' for remote servers")
+    
+    # Validate disabledTools (supports camelCase & snake_case for backwards compatibility)
+    disabled_tools = server_cfg.get("disabledTools")
+    if disabled_tools is None:
+        disabled_tools = server_cfg.get("disabled_tools")
+    if disabled_tools is not None:
+        if not isinstance(disabled_tools, list):
+            raise ValueError(f"Server '{server_name}' 'disabledTools' must be a list")
+        for tool_name in disabled_tools:
+            if not isinstance(tool_name, str):
+                raise ValueError(f"Server '{server_name}' 'disabledTools' must contain only strings")
 
 
 def load_config(config_path: str) -> Dict[str, Any]:
@@ -85,6 +234,11 @@ def load_config(config_path: str) -> Dict[str, Any]:
         for server_name, server_cfg in mcp_servers.items():
             validate_server_config(server_name, server_cfg)
 
+            # Validate client header forwarding configuration if present
+            header_config = server_cfg.get("client_header_forwarding", {})
+            if header_config:
+                validate_client_header_forwarding_config(server_name, header_config)
+
         return config_data
     except json.JSONDecodeError as e:
         logger.error(f"Invalid JSON in config file {config_path}: {e}")
@@ -97,9 +251,16 @@ def load_config(config_path: str) -> Dict[str, Any]:
         raise
 
 
-def create_sub_app(server_name: str, server_cfg: Dict[str, Any], cors_allow_origins,
-                   api_key: Optional[str], strict_auth: bool, api_dependency,
-                   connection_timeout, lifespan) -> FastAPI:
+def create_sub_app(
+    server_name: str,
+    server_cfg: Dict[str, Any],
+    cors_allow_origins,
+    api_key: Optional[str],
+    strict_auth: bool,
+    api_dependency,
+    connection_timeout,
+    lifespan,
+) -> FastAPI:
     """Create a sub-application for an MCP server."""
     sub_app = FastAPI(
         title=f"{server_name}",
@@ -129,10 +290,10 @@ def create_sub_app(server_name: str, server_cfg: Dict[str, Any], cors_allow_orig
         sub_app.state.server_type = "sse"
         sub_app.state.args = [server_cfg["url"]]
         sub_app.state.headers = server_cfg.get("headers")
-    elif normalize_server_type(server_config_type) == "streamable-http" and server_cfg.get("url"):
+    elif normalize_server_type(
+        server_config_type
+    ) == "streamable-http" and server_cfg.get("url"):
         url = server_cfg["url"]
-        if not url.endswith("/"):
-            url = f"{url}/"
         sub_app.state.server_type = "streamablehttp"
         sub_app.state.args = [url]
         sub_app.state.headers = server_cfg.get("headers")
@@ -149,42 +310,76 @@ def create_sub_app(server_name: str, server_cfg: Dict[str, Any], cors_allow_orig
     sub_app.state.api_dependency = api_dependency
     sub_app.state.connection_timeout = connection_timeout
 
+    # Store list of tools to be disabled, if present (accept both key styles)
+    disabled_tools = server_cfg.get("disabledTools")
+    if disabled_tools is None:
+        disabled_tools = server_cfg.get("disabled_tools", [])
+    sub_app.state.disabled_tools = disabled_tools
+
+    
+    # Store client header forwarding configuration
+    sub_app.state.client_header_forwarding = server_cfg.get(
+        "client_header_forwarding", {"enabled": False}
+    )
+
+    # Store OAuth configuration if present
+    sub_app.state.oauth_config = server_cfg.get("oauth")
+
+
     return sub_app
 
 
-def mount_config_servers(main_app: FastAPI, config_data: Dict[str, Any],
-                        cors_allow_origins, api_key: Optional[str], strict_auth: bool,
-                        api_dependency, connection_timeout, lifespan, path_prefix: str):
+def mount_config_servers(
+    main_app: FastAPI,
+    config_data: Dict[str, Any],
+    cors_allow_origins,
+    api_key: Optional[str],
+    strict_auth: bool,
+    api_dependency,
+    connection_timeout,
+    lifespan,
+    path_prefix: str,
+):
     """Mount MCP servers from config data."""
     mcp_servers = config_data.get("mcpServers", {})
 
     logger.info("Configuring MCP Servers:")
     for server_name, server_cfg in mcp_servers.items():
         sub_app = create_sub_app(
-            server_name, server_cfg, cors_allow_origins, api_key,
-            strict_auth, api_dependency, connection_timeout, lifespan
+            server_name,
+            server_cfg,
+            cors_allow_origins,
+            api_key,
+            strict_auth,
+            api_dependency,
+            connection_timeout,
+            lifespan,
         )
         main_app.mount(f"{path_prefix}{server_name}", sub_app)
 
 
 def unmount_servers(main_app: FastAPI, path_prefix: str, server_names: list):
     """Unmount specific MCP servers."""
+    active_lifespans = getattr(main_app.state, "active_lifespans", {})
+
     for server_name in server_names:
         mount_path = f"{path_prefix}{server_name}"
-        # Find and remove the mount
-        routes_to_remove = []
-        for route in main_app.router.routes:
-            if hasattr(route, 'path') and route.path == mount_path:
-                routes_to_remove.append(route)
 
-        for route in routes_to_remove:
-            main_app.router.routes.remove(route)
-            logger.info(f"Unmounted server: {server_name}")
+        # Clean up lifespan context if it exists
+        if server_name in active_lifespans:
+            lifespan_context = active_lifespans[server_name]
+            try:
+                # Schedule cleanup of the lifespan context
+                asyncio.create_task(lifespan_context.__aexit__(None, None, None))
+            except Exception as e:
+                logger.warning(f"Error cleaning up lifespan for {server_name}: {e}")
+            finally:
+                del active_lifespans[server_name]
 
 
 async def reload_config_handler(main_app: FastAPI, new_config_data: Dict[str, Any]):
     """Handle config reload by comparing and updating mounted servers."""
-    old_config_data = getattr(main_app.state, 'config_data', {})
+    old_config_data = getattr(main_app.state, "config_data", {})
     backup_routes = list(main_app.router.routes)  # Backup current routes for rollback
 
     try:
@@ -197,13 +392,13 @@ async def reload_config_handler(main_app: FastAPI, new_config_data: Dict[str, An
         servers_to_check = old_servers & new_servers
 
         # Get app configuration from state
-        cors_allow_origins = getattr(main_app.state, 'cors_allow_origins', ["*"])
-        api_key = getattr(main_app.state, 'api_key', None)
-        strict_auth = getattr(main_app.state, 'strict_auth', False)
-        api_dependency = getattr(main_app.state, 'api_dependency', None)
-        connection_timeout = getattr(main_app.state, 'connection_timeout', None)
-        lifespan = getattr(main_app.state, 'lifespan', None)
-        path_prefix = getattr(main_app.state, 'path_prefix', "/")
+        cors_allow_origins = getattr(main_app.state, "cors_allow_origins", ["*"])
+        api_key = getattr(main_app.state, "api_key", None)
+        strict_auth = getattr(main_app.state, "strict_auth", False)
+        api_dependency = getattr(main_app.state, "api_dependency", None)
+        connection_timeout = getattr(main_app.state, "connection_timeout", None)
+        lifespan = getattr(main_app.state, "lifespan", None)
+        path_prefix = getattr(main_app.state, "path_prefix", "/")
 
         # Remove servers that are no longer in config
         if servers_to_remove:
@@ -227,14 +422,57 @@ async def reload_config_handler(main_app: FastAPI, new_config_data: Dict[str, An
         # Add new servers and updated servers
         if servers_to_add:
             logger.info(f"Adding servers: {list(servers_to_add)}")
+
+            # Store lifespan contexts for cleanup
+            if not hasattr(main_app.state, "active_lifespans"):
+                main_app.state.active_lifespans = {}
+
             for server_name in servers_to_add:
                 server_cfg = new_config_data["mcpServers"][server_name]
                 try:
                     sub_app = create_sub_app(
-                        server_name, server_cfg, cors_allow_origins, api_key,
-                        strict_auth, api_dependency, connection_timeout, lifespan
+                        server_name,
+                        server_cfg,
+                        cors_allow_origins,
+                        api_key,
+                        strict_auth,
+                        api_dependency,
+                        connection_timeout,
+                        lifespan,
                     )
                     main_app.mount(f"{path_prefix}{server_name}", sub_app)
+
+                    # Start the lifespan for the new sub-app when available
+                    lifespan_context = None
+                    lifespan_factory = getattr(sub_app.router, "lifespan_context", None)
+                    if lifespan_factory:
+                        lifespan_context = lifespan_factory(sub_app)
+                    if lifespan_context and hasattr(lifespan_context, "__aenter__"):
+                        await lifespan_context.__aenter__()
+                        # Store the context manager for cleanup later
+                        main_app.state.active_lifespans[server_name] = lifespan_context
+                    else:
+                        logger.debug(
+                            "Skipping lifespan startup for server '%s' (no async context manager)",
+                            server_name,
+                        )
+
+                    # Check if connection was successful
+                    is_connected = getattr(sub_app.state, "is_connected", False)
+                    if is_connected:
+                        logger.info(
+                            f"Successfully connected to new server: '{server_name}'"
+                        )
+                    else:
+                        logger.warning(
+                            f"Failed to connect to new server: '{server_name}'"
+                        )
+
+                except asyncio.CancelledError as e:
+                    logger.error(f"Failed to create server '{server_name}' (cancelled): {e}")
+                    # Rollback on failure
+                    main_app.router.routes = backup_routes
+                    raise
                 except Exception as e:
                     logger.error(f"Failed to create server '{server_name}': {e}")
                     # Rollback on failure
@@ -253,11 +491,18 @@ async def reload_config_handler(main_app: FastAPI, new_config_data: Dict[str, An
 
 
 async def create_dynamic_endpoints(app: FastAPI, api_dependency=None):
-    session: ClientSession = app.state.session
-    if not session:
-        raise ValueError("Session is not initialized in the app state.")
+    session_manager: Optional[MCPConnectionManager] = getattr(
+        app.state, "session_manager", None
+    )
 
-    result = await session.initialize()
+    if session_manager:
+        session, result = await session_manager.ensure_initialized()
+        app.state.session = session
+    else:
+        session: Optional[ClientSession] = getattr(app.state, "session", None)
+        if not session:
+            raise ValueError("Session is not initialized in the app state.")
+        result = await session.initialize()
     server_info = getattr(result, "serverInfo", None)
     if server_info:
         app.title = server_info.name or app.title
@@ -272,6 +517,15 @@ async def create_dynamic_endpoints(app: FastAPI, api_dependency=None):
 
     tools_result = await session.list_tools()
     tools = tools_result.tools
+    
+    # Filter out disabled tools
+    disabled_tools = getattr(app.state, "disabled_tools", [])
+    if disabled_tools:
+        original_count = len(tools)
+        tools = [tool for tool in tools if tool.name not in disabled_tools]
+        filtered_count = original_count - len(tools)
+        if filtered_count > 0:
+            logger.info(f"Filtered out {filtered_count} tool(s) for server '{app.title}': {disabled_tools}")
 
     for tool in tools:
         endpoint_name = tool.name
@@ -296,11 +550,16 @@ async def create_dynamic_endpoints(app: FastAPI, api_dependency=None):
                 outputSchema.get("$defs", {}),
             )
 
+        # Get client header forwarding configuration from app state
+        client_header_forwarding_config = getattr(
+            app.state, "client_header_forwarding", {"enabled": False}
+        )
+
         tool_handler = get_tool_handler(
-            session,
             endpoint_name,
             form_model_fields,
             response_model_fields,
+            client_header_forwarding_config,
         )
 
         app.post(
@@ -319,14 +578,16 @@ async def lifespan(app: FastAPI):
     args = getattr(app.state, "args", [])
     args = args if isinstance(args, list) else [args]
     env = getattr(app.state, "env", {})
-    connection_timeout = getattr(app.state, "connection_timeout", 10)
+    connection_timeout = getattr(app.state, "connection_timeout", CONNECTION_TIMEOUT)
     api_dependency = getattr(app.state, "api_dependency", None)
     path_prefix = getattr(app.state, "path_prefix", "/")
 
     # Get shutdown handler from app state
     shutdown_handler = getattr(app.state, "shutdown_handler", None)
 
-    is_main_app = not command and not (server_type in ["sse", "streamable-http"] and args)
+    is_main_app = not command and not (
+        server_type in ["sse", "streamable-http"] and args
+    )
 
     if is_main_app:
         async with AsyncExitStack() as stack:
@@ -353,11 +614,43 @@ async def lifespan(app: FastAPI):
                             f"Connection attempt for '{server_name}' finished, but status is not 'connected'."
                         )
                         failed_servers.append(server_name)
-                except Exception:
+                except asyncio.CancelledError as e:
+                    if shutdown_handler and shutdown_handler.shutdown_event.is_set():
+                        raise
                     logger.error(
-                        f"Failed to establish connection for server: '{server_name}'."
+                        f"Failed to establish connection for server: '{server_name}' - CancelledError: {e}",
+                        exc_info=True,
                     )
                     failed_servers.append(server_name)
+                except Exception as e:
+                    error_class_name = type(e).__name__
+                    if error_class_name == "ExceptionGroup" or (
+                        hasattr(e, "exceptions") and hasattr(e, "message")
+                    ):
+                        logger.error(
+                            f"Failed to establish connection for server: '{server_name}' - Multiple errors occurred:"
+                        )
+                        # Log each individual exception from the group
+                        exceptions = getattr(e, "exceptions", [])
+                        for idx, exc in enumerate(exceptions):
+                            logger.error(
+                                f"  Error {idx + 1}: {type(exc).__name__}: {exc}"
+                            )
+                            # Also log traceback for each exception
+                            if hasattr(exc, "__traceback__"):
+                                import traceback
+
+                                tb_lines = traceback.format_exception(
+                                    type(exc), exc, exc.__traceback__
+                                )
+                                for line in tb_lines:
+                                    logger.debug(f"    {line.rstrip()}")
+                    else:
+                        logger.error(
+                            f"Failed to establish connection for server: '{server_name}' - {type(e).__name__}: {e}",
+                            exc_info=True,
+                        )
+                    failed_servers.append(server_name)
 
             logger.info("\n--- Server Startup Summary ---")
             if successful_servers:
@@ -384,36 +677,63 @@ async def lifespan(app: FastAPI):
         # This is a sub-app's lifespan
         app.state.is_connected = False
         try:
-            if server_type == "stdio":
-                server_params = StdioServerParameters(
-                    command=command,
-                    args=args,
-                    env={**os.environ, **env},
-                )
-                client_context = stdio_client(server_params)
-            elif server_type == "sse":
-                headers = getattr(app.state, "headers", None)
-                client_context = sse_client(
-                    url=args[0],
-                    sse_read_timeout=connection_timeout or 900,
-                    headers=headers,
-                )
-            elif server_type == "streamable-http":
-                headers = getattr(app.state, "headers", None)
-                client_context = streamablehttp_client(url=args[0], headers=headers)
-            else:
-                raise ValueError(f"Unsupported server type: {server_type}")
+            # Check for OAuth configuration
+            oauth_config = getattr(app.state, "oauth_config", None)
+            auth_provider = None
 
-            async with client_context as (reader, writer, *_):
-                async with ClientSession(reader, writer) as session:
-                    app.state.session = session
-                    await create_dynamic_endpoints(app, api_dependency=api_dependency)
-                    app.state.is_connected = True
-                    yield
+            if oauth_config:
+                server_name = app.title
+                logger.info(f"OAuth configuration detected for server: {server_name}")
+                try:
+                    auth_provider = await create_oauth_provider(
+                        server_name=server_name,
+                        oauth_config=oauth_config,
+                        storage_type=oauth_config.get("storage_type", "file"),
+                    )
+                    logger.info(f"OAuth provider created for server: {server_name}")
+                except Exception as e:
+                    logger.error(
+                        f"Failed to create OAuth provider for {server_name}: {e}"
+                    )
+                    raise
+
+            if oauth_config and server_type == "stdio":
+                logger.warning("OAuth not supported for stdio server type")
+            if oauth_config and server_type == "sse":
+                logger.warning("OAuth not supported for SSE server type")
+
+            headers = getattr(app.state, "headers", None)
+            session_manager = MCPConnectionManager(
+                server_type=server_type,
+                command=command,
+                args=args,
+                env=env,
+                headers=headers,
+                connection_timeout=connection_timeout,
+                auth_provider=auth_provider,
+            )
+            app.state.session_manager = session_manager
+
+            session = await session_manager.get_session()
+            app.state.session = session
+            await create_dynamic_endpoints(app, api_dependency=api_dependency)
+            app.state.is_connected = True
+            yield
         except Exception as e:
-            logger.error(f"Failed to connect to MCP server '{app.title}': {e}")
+            # Log the full exception with traceback for debugging
+            logger.error(
+                f"Failed to connect to MCP server '{app.title}': {type(e).__name__}: {e}",
+                exc_info=True,
+            )
             app.state.is_connected = False
-            return
+            # Re-raise the exception so it propagates to the main app's lifespan
+            raise
+        finally:
+            session_manager = getattr(app.state, "session_manager", None)
+            if session_manager:
+                await session_manager.close()
+                app.state.session_manager = None
+            app.state.session = None
 
 
 async def run(
@@ -446,10 +766,20 @@ async def run(
     ssl_certfile = kwargs.get("ssl_certfile")
     ssl_keyfile = kwargs.get("ssl_keyfile")
     path_prefix = kwargs.get("path_prefix") or "/"
+    root_path = kwargs.get("root_path") or ""
+
+    # Configure logging based on LOG_LEVEL environment variable
+    log_level = os.getenv("LOG_LEVEL", "INFO").upper()
+    try:
+        numeric_level = getattr(logging, log_level, None)
+        if not isinstance(numeric_level, int):
+            raise ValueError(f"Invalid log level: {log_level}")
+    except (ValueError, AttributeError):
+        logger.warning(f"Invalid LOG_LEVEL '{log_level}', defaulting to INFO")
+        numeric_level = logging.INFO
 
-    # Configure basic logging
     logging.basicConfig(
-        level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s"
+        level=numeric_level, format="%(asctime)s - %(levelname)s - %(message)s"
     )
 
     # Suppress HTTP request logs
@@ -475,6 +805,7 @@ async def run(
     if ssl_keyfile:
         logger.info(f"  SSL Key File: {ssl_keyfile}")
     logger.info(f"  Path Prefix: {path_prefix}")
+    logger.info(f"  Root Path: {root_path}")
 
     # Create shutdown handler
     shutdown_handler = GracefulShutdown()
@@ -483,6 +814,7 @@ async def run(
         title=name,
         description=description,
         version=version,
+        root_path=root_path,
         ssl_certfile=ssl_certfile,
         ssl_keyfile=ssl_keyfile,
         lifespan=lifespan,
@@ -541,8 +873,15 @@ async def run(
         logger.info(f"Loading MCP server configurations from: {config_path}")
         config_data = load_config(config_path)
         mount_config_servers(
-            main_app, config_data, cors_allow_origins, api_key, strict_auth,
-            api_dependency, connection_timeout, lifespan, path_prefix
+            main_app,
+            config_data,
+            cors_allow_origins,
+            api_key,
+            strict_auth,
+            api_dependency,
+            connection_timeout,
+            lifespan,
+            path_prefix,
         )
 
         # Store config info and app state for hot reload
@@ -571,13 +910,15 @@ async def run(
         config_watcher.start()
 
     logger.info("Uvicorn server starting...")
+    uvicorn_log_level = log_level.lower()
+
     config = uvicorn.Config(
         app=main_app,
         host=host,
         port=port,
         ssl_certfile=ssl_certfile,
         ssl_keyfile=ssl_keyfile,
-        log_level="info",
+        log_level=uvicorn_log_level,
     )
     server = uvicorn.Server(config)
 

src/mcpo/tests/test_hot_reload.py

@@ -42,6 +42,19 @@ def test_validate_server_config_missing_url():
         validate_server_config("test_server", config)
 
 
+def test_validate_server_config_disabled_tools_valid():
+    """Test validation of server configuration with a valid disabledTools."""
+    config = {"command": "echo", "args": ["hello"], "disabledTools": ["search-web"]}
+    validate_server_config("test_server", config)
+
+
+def test_validate_server_config_disabled_tools_invalid():
+    """Test validation fails for an invalid disabledTools."""
+    config = {"command": "echo", "args": ["hello"], "disabledTools": "not-a-list"}
+    with pytest.raises(ValueError, match="'disabledTools' must be a list"):
+        validate_server_config("test_server", config)
+
+
 def test_load_config_valid():
     """Test loading a valid config file."""
     config_data = {

src/mcpo/utils/config_watcher.py

@@ -15,8 +15,10 @@ logger = logging.getLogger(__name__)
 class ConfigChangeHandler(FileSystemEventHandler):
     """Handler for config file changes."""
 
-    def __init__(self, config_path: Path, reload_callback: Callable[[Dict[str, Any]], None], loop: asyncio.AbstractEventLoop):
-        self.config_path = config_path.resolve()  # Resolve to absolute path
+    def __init__(self, origin_config_path: Path, reload_callback: Callable[[Dict[str, Any]], None], loop: asyncio.AbstractEventLoop):
+        self.origin_config_path = origin_config_path
+        self.config_path = origin_config_path.resolve()  # Resolve to absolute path
+        self.is_symlink = origin_config_path.is_symlink()
         self.reload_callback = reload_callback
         self.loop = loop  # Store reference to the main event loop
         self._last_modification = 0
@@ -24,6 +26,12 @@ class ConfigChangeHandler(FileSystemEventHandler):
 
     def on_modified(self, event):
         """Handle file modification events."""
+        if self.is_symlink:
+            self.config_path = self.origin_config_path.resolve()  # Re-resolve to get the latest symlink target
+            logger.info(f"Symlink file modified: {self.config_path}")
+            self._trigger_reload()
+            return
+
         if event.is_directory:
             return
 
@@ -116,7 +124,8 @@ class ConfigWatcher:
     """Watches a config file for changes and triggers reloads."""
 
     def __init__(self, config_path: str, reload_callback: Callable[[Dict[str, Any]], None]):
-        self.config_path = Path(config_path).resolve()
+        self.origin_config_path = Path(config_path)
+        self.config_path = self.origin_config_path.resolve()
         self.reload_callback = reload_callback
         self.observer: Optional[Observer] = None
         self.handler: Optional[ConfigChangeHandler] = None
@@ -135,12 +144,12 @@ class ConfigWatcher:
             logger.error("No running event loop found, cannot start config watcher")
             return
 
-        self.handler = ConfigChangeHandler(self.config_path, self.reload_callback, self.loop)
+        self.handler = ConfigChangeHandler(self.origin_config_path, self.reload_callback, self.loop)
         self.observer = Observer()
 
         # Watch the directory containing the config file
-        watch_dir = self.config_path.parent
-        logger.debug(f"Watching directory: {watch_dir} for file: {self.config_path}")
+        watch_dir = self.origin_config_path.parent
+        logger.info(f"Watching directory: {watch_dir} for file: {self.config_path}")
         self.observer.schedule(self.handler, str(watch_dir), recursive=False)
 
         self.observer.start()

src/mcpo/utils/headers.py

@@ -0,0 +1,99 @@
+import logging
+import re
+from typing import Dict, List, Optional, Any
+from fastapi import Request
+
+logger = logging.getLogger(__name__)
+
+
+def validate_client_header_forwarding_config(server_name: str, config: Dict[str, Any]) -> None:
+    """Validate client header forwarding configuration for a server."""
+    if not isinstance(config, dict):
+        raise ValueError(f"Server '{server_name}' client_header_forwarding must be a dictionary")
+    
+    enabled = config.get("enabled", False)
+    if not isinstance(enabled, bool):
+        raise ValueError(f"Server '{server_name}' client_header_forwarding.enabled must be a boolean")
+    
+    if not enabled:
+        return  # No further validation needed if disabled
+    
+    whitelist = config.get("whitelist", [])
+    blacklist = config.get("blacklist", [])
+    
+    if whitelist and not isinstance(whitelist, list):
+        raise ValueError(f"Server '{server_name}' client_header_forwarding.whitelist must be a list")
+    
+    if blacklist and not isinstance(blacklist, list):
+        raise ValueError(f"Server '{server_name}' client_header_forwarding.blacklist must be a list")
+    
+    debug_headers = config.get("debug_headers", False)
+    if not isinstance(debug_headers, bool):
+        raise ValueError(f"Server '{server_name}' client_header_forwarding.debug_headers must be a boolean")
+
+
+def match_header_pattern(header_name: str, patterns: List[str]) -> bool:
+    """Check if header name matches any of the given patterns."""
+    for pattern in patterns:
+        if pattern == "*":
+            return True
+        if pattern.endswith("*"):
+            # Wildcard pattern like "X-User-*"
+            prefix = pattern[:-1]
+            if header_name.startswith(prefix):
+                return True
+        elif pattern == header_name:
+            return True
+    return False
+
+
+def filter_headers(
+    request_headers: Dict[str, str], 
+    whitelist: List[str],
+    blacklist: List[str],
+    debug_headers: bool = False
+) -> Dict[str, str]:
+    """Filter request headers based on whitelist and blacklist."""
+    filtered_headers = {}
+    
+    for header_name, header_value in request_headers.items():
+        # Skip if in blacklist
+        if blacklist and match_header_pattern(header_name, blacklist):
+            if debug_headers:
+                logger.debug(f"Header '{header_name}' blocked by blacklist")
+            continue
+        
+        # Include if in whitelist (or no whitelist specified)
+        if not whitelist or match_header_pattern(header_name, whitelist):
+            filtered_headers[header_name] = header_value
+            if debug_headers:
+                logger.debug(f"Header '{header_name}' forwarded")
+        elif debug_headers:
+            logger.debug(f"Header '{header_name}' not in whitelist")
+    
+    return filtered_headers
+
+
+def process_headers_for_server(
+    request: Request,
+    header_config: Dict[str, Any]
+) -> Dict[str, str]:
+    """Process and filter headers for a specific MCP server."""
+    if not header_config.get("enabled", False):
+        return {}
+    
+    # Convert FastAPI headers to dict
+    request_headers = dict(request.headers)
+    
+    # Get configuration values
+    whitelist = header_config.get("whitelist", [])
+    blacklist = header_config.get("blacklist", [])
+    debug_headers = header_config.get("debug_headers", False)
+    
+    # Filter headers based on whitelist/blacklist
+    filtered_headers = filter_headers(request_headers, whitelist, blacklist, debug_headers)
+    
+    if debug_headers:
+        logger.debug(f"Final forwarded headers: {list(filtered_headers.keys())}")
+    
+    return filtered_headers

src/mcpo/utils/main.py

@@ -1,10 +1,12 @@
+import logging
 import json
 import traceback
 from typing import Any, Dict, ForwardRef, List, Optional, Type, Union
-import logging
-from fastapi import HTTPException
 
-from mcp import ClientSession, types
+from anyio import ClosedResourceError
+from fastapi import HTTPException, Request
+
+from mcp import types
 from mcp.types import (
     CallToolResult,
     PARSE_ERROR,
@@ -19,6 +21,8 @@ from mcp.shared.exceptions import McpError
 from pydantic import Field, create_model
 from pydantic.fields import FieldInfo
 
+from mcpo.utils.headers import process_headers_for_server
+
 MCP_ERROR_TO_HTTP_STATUS = {
     PARSE_ERROR: 400,
     INVALID_REQUEST: 400,
@@ -29,12 +33,14 @@ MCP_ERROR_TO_HTTP_STATUS = {
 
 logger = logging.getLogger(__name__)
 
+
 def normalize_server_type(server_type: str) -> str:
     """Normalize server_type to a standard value."""
     if server_type in ["streamable_http", "streamablehttp", "streamable-http"]:
         return "streamable-http"
     return server_type
 
+
 def process_tool_response(result: CallToolResult) -> list:
     """Universal response processor for all tool endpoints"""
     response = []
@@ -57,8 +63,8 @@ def process_tool_response(result: CallToolResult) -> list:
 
 
 def name_needs_alias(name: str) -> bool:
-    """Check if a field name needs aliasing (for now if it starts with '__')."""
-    return name.startswith("__")
+    """Check if a field name needs aliasing (if it starts with '_')."""
+    return name.startswith("_")
 
 
 def generate_alias_name(original_name: str, existing_names: set) -> str:
@@ -66,7 +72,7 @@ def generate_alias_name(original_name: str, existing_names: set) -> str:
     Generate an alias field name by stripping unwanted chars, and avoiding conflicts with existing names.
 
     Args:
-        original_name: The original field name (should start with '__')
+        original_name: The original field name (should start with '_')
         existing_names: Set of existing names to avoid conflicts with
 
     Returns:
@@ -148,7 +154,12 @@ def _process_schema_property(
             temp_schema = dict(prop_schema)
             temp_schema["type"] = type_option
             type_hint, _ = _process_schema_property(
-                _model_cache, temp_schema, model_name_prefix, prop_name, False, schema_defs=schema_defs
+                _model_cache,
+                temp_schema,
+                model_name_prefix,
+                prop_name,
+                False,
+                schema_defs=schema_defs,
             )
             type_hints.append(type_hint)
 
@@ -266,11 +277,50 @@ def get_model_fields(form_model_name, properties, required_fields, schema_defs=N
 
 
 def get_tool_handler(
-    session,
     endpoint_name,
     form_model_fields,
     response_model_fields=None,
+    client_header_forwarding_config=None,
 ):
+    async def call_tool_with_reconnect(
+        request: Request, arguments: Dict[str, Any]
+    ) -> CallToolResult:
+        session_manager = getattr(request.app.state, "session_manager", None)
+
+        async def _invoke(session):
+            return await session.call_tool(endpoint_name, arguments=arguments)
+
+        if session_manager:
+            try:
+                session, _ = await session_manager.ensure_initialized()
+            except ClosedResourceError:
+                logger.warning(
+                    "Session closed while initializing '%s'; attempting reconnect",
+                    endpoint_name,
+                )
+                session, _ = await session_manager.reconnect()
+            request.app.state.session = session
+
+            try:
+                return await _invoke(session)
+            except ClosedResourceError:
+                logger.warning(
+                    "Session closed during call to '%s'; attempting reconnect",
+                    endpoint_name,
+                )
+                session, _ = await session_manager.reconnect()
+                request.app.state.session = session
+                return await _invoke(session)
+
+        session = getattr(request.app.state, "session", None)
+        if not session:
+            raise HTTPException(
+                status_code=500,
+                detail={"message": "MCP session is not available"},
+            )
+
+        return await _invoke(session)
+
     if form_model_fields:
         FormModel = create_model(f"{endpoint_name}_form_model", **form_model_fields)
         ResponseModel = (
@@ -279,114 +329,121 @@ def get_tool_handler(
             else Any
         )
 
-        def make_endpoint_func(
-            endpoint_name: str, FormModel, session: ClientSession
-        ):  # Parameterized endpoint
-            async def tool(form_data: FormModel) -> Union[ResponseModel, Any]:
-                args = form_data.model_dump(exclude_none=True, by_alias=True)
-                logger.info(f"Calling endpoint: {endpoint_name}, with args: {args}")
-                try:
-                    result = await session.call_tool(endpoint_name, arguments=args)
+        async def tool(
+            form_data: FormModel, request: Request
+        ) -> Union[ResponseModel, Any]:
+            args = form_data.model_dump(exclude_none=True, by_alias=True)
 
-                    if result.isError:
-                        error_message = "Unknown tool execution error"
-                        error_data = None  # Initialize error_data
-                        if result.content:
-                            if isinstance(result.content[0], types.TextContent):
-                                error_message = result.content[0].text
-                        detail = {"message": error_message}
-                        if error_data is not None:
-                            detail["data"] = error_data
-                        raise HTTPException(
-                            status_code=500,
-                            detail=detail,
-                        )
+            forwarded_headers = {}
+            if (
+                client_header_forwarding_config
+                and client_header_forwarding_config.get("enabled", False)
+            ):
+                forwarded_headers = process_headers_for_server(
+                    request, client_header_forwarding_config
+                )
 
-                    response_data = process_tool_response(result)
-                    final_response = (
-                        response_data[0] if len(response_data) == 1 else response_data
-                    )
-                    return final_response
+            meta = {}
+            if forwarded_headers:
+                meta["headers"] = forwarded_headers
 
-                except McpError as e:
-                    logger.info(
-                        f"MCP Error calling {endpoint_name}: {traceback.format_exc()}"
-                    )
-                    status_code = MCP_ERROR_TO_HTTP_STATUS.get(e.error.code, 500)
-                    raise HTTPException(
-                        status_code=status_code,
-                        detail=(
-                            {"message": e.error.message, "data": e.error.data}
-                            if e.error.data is not None
-                            else {"message": e.error.message}
-                        ),
-                    )
-                except Exception as e:
-                    logger.info(
-                        f"Unexpected error calling {endpoint_name}: {traceback.format_exc()}"
-                    )
-                    raise HTTPException(
-                        status_code=500,
-                        detail={"message": "Unexpected error", "error": str(e)},
-                    )
+            logger.info(f"Calling endpoint: {endpoint_name}, with args: {args}")
+            try:
+                result = await call_tool_with_reconnect(request, args)
 
-            return tool
+                if result.isError:
+                    error_message = "Unknown tool execution error"
+                    error_data = None
+                    if result.content and isinstance(
+                        result.content[0], types.TextContent
+                    ):
+                        error_message = result.content[0].text
+                    detail = {"message": error_message}
+                    if error_data is not None:
+                        detail["data"] = error_data
+                    raise HTTPException(status_code=500, detail=detail)
 
-        tool_handler = make_endpoint_func(endpoint_name, FormModel, session)
-    else:
+                response_data = process_tool_response(result)
+                final_response = (
+                    response_data[0] if len(response_data) == 1 else response_data
+                )
+                return final_response
 
-        def make_endpoint_func_no_args(
-            endpoint_name: str, session: ClientSession
-        ):  # Parameterless endpoint
-            async def tool():  # No parameters
-                logger.info(f"Calling endpoint: {endpoint_name}, with no args")
-                try:
-                    result = await session.call_tool(
-                        endpoint_name, arguments={}
-                    )  # Empty dict
+            except McpError as e:
+                logger.info(
+                    f"MCP Error calling {endpoint_name}: {traceback.format_exc()}"
+                )
+                status_code = MCP_ERROR_TO_HTTP_STATUS.get(e.error.code, 500)
+                raise HTTPException(
+                    status_code=status_code,
+                    detail=(
+                        {"message": e.error.message, "data": e.error.data}
+                        if e.error.data is not None
+                        else {"message": e.error.message}
+                    ),
+                )
+            except Exception as e:
+                logger.info(
+                    f"Unexpected error calling {endpoint_name}: {traceback.format_exc()}"
+                )
+                raise HTTPException(
+                    status_code=500,
+                    detail={"message": "Unexpected error", "error": str(e)},
+                )
 
-                    if result.isError:
-                        error_message = "Unknown tool execution error"
-                        if result.content:
-                            if isinstance(result.content[0], types.TextContent):
-                                error_message = result.content[0].text
-                        detail = {"message": error_message}
-                        raise HTTPException(
-                            status_code=500,
-                            detail=detail,
-                        )
+        return tool
 
-                    response_data = process_tool_response(result)
-                    final_response = (
-                        response_data[0] if len(response_data) == 1 else response_data
-                    )
-                    return final_response
+    async def tool(request: Request):
+        forwarded_headers = {}
+        if (
+            client_header_forwarding_config
+            and client_header_forwarding_config.get("enabled", False)
+        ):
+            forwarded_headers = process_headers_for_server(
+                request, client_header_forwarding_config
+            )
 
-                except McpError as e:
-                    logger.info(
-                        f"MCP Error calling {endpoint_name}: {traceback.format_exc()}"
-                    )
-                    status_code = MCP_ERROR_TO_HTTP_STATUS.get(e.error.code, 500)
-                    # Propagate the error received from MCP as an HTTP exception
-                    raise HTTPException(
-                        status_code=status_code,
-                        detail=(
-                            {"message": e.error.message, "data": e.error.data}
-                            if e.error.data is not None
-                            else {"message": e.error.message}
-                        ),
-                    )
-                except Exception as e:
-                    logger.info(
-                        f"Unexpected error calling {endpoint_name}: {traceback.format_exc()}"
-                    )
-                    raise HTTPException(
-                        status_code=500,
-                        detail={"message": "Unexpected error", "error": str(e)},
-                    )
+        meta = {}
+        if forwarded_headers:
+            meta["headers"] = forwarded_headers
 
-            return tool
+        logger.info(f"Calling endpoint: {endpoint_name}, with no args")
+        try:
+            result = await call_tool_with_reconnect(request, {})
 
-        tool_handler = make_endpoint_func_no_args(endpoint_name, session)
+            if result.isError:
+                error_message = "Unknown tool execution error"
+                if result.content and isinstance(result.content[0], types.TextContent):
+                    error_message = result.content[0].text
+                detail = {"message": error_message}
+                raise HTTPException(status_code=500, detail=detail)
 
-    return tool_handler
+            response_data = process_tool_response(result)
+            final_response = (
+                response_data[0] if len(response_data) == 1 else response_data
+            )
+            return final_response
+
+        except McpError as e:
+            logger.info(
+                f"MCP Error calling {endpoint_name}: {traceback.format_exc()}"
+            )
+            status_code = MCP_ERROR_TO_HTTP_STATUS.get(e.error.code, 500)
+            raise HTTPException(
+                status_code=status_code,
+                detail=(
+                    {"message": e.error.message, "data": e.error.data}
+                    if e.error.data is not None
+                    else {"message": e.error.message}
+                ),
+            )
+        except Exception as e:
+            logger.info(
+                f"Unexpected error calling {endpoint_name}: {traceback.format_exc()}"
+            )
+            raise HTTPException(
+                status_code=500,
+                detail={"message": "Unexpected error", "error": str(e)},
+            )
+
+    return tool

src/mcpo/utils/oauth.py

@@ -0,0 +1,294 @@
+import asyncio
+import json
+import logging
+import os
+import hashlib
+from typing import Optional, Dict, Any, Tuple
+from pathlib import Path
+import webbrowser
+import time
+import threading
+from http.server import BaseHTTPRequestHandler, HTTPServer
+from urllib.parse import parse_qs, urlparse
+
+from mcp.client.auth import OAuthClientProvider, TokenStorage
+from mcp.shared.auth import OAuthClientInformationFull, OAuthClientMetadata, OAuthToken
+from pydantic import AnyUrl
+
+logger = logging.getLogger(__name__)
+
+def _load_callback_html(status: str, title: str, heading: str, message: str, action_text: str) -> str:
+    """Load and render the OAuth callback HTML template"""
+    template_path = Path(__file__).parent / "oauth_callback.html"
+    
+    try:
+        logger.debug(f"Loading OAuth template from: {template_path}")
+        logger.debug(f"Template exists: {template_path.exists()}")
+        
+        with open(template_path, 'r', encoding='utf-8') as f:
+            template = f.read()
+        
+        logger.debug(f"Template loaded, length: {len(template)}")
+        
+        # Define icons and status classes
+        icon = "✓" if status == "success" else "✕"
+        status_class = status
+        
+        # Replace template variables
+        html = template.format(
+            title=title,
+            status_class=status_class,
+            icon=icon,
+            heading=heading,
+            message=message,
+            action_text=action_text
+        )
+        
+        logger.debug("Template rendered successfully")
+        return html
+    except Exception as e:
+        logger.error(f"Failed to load OAuth callback template from {template_path}: {e}")
+        logger.error(f"Template path exists: {template_path.exists()}")
+        # Fallback to simple HTML
+        return f"<html><body><h2>{heading}</h2><p>{message}</p></body></html>"
+
+class InMemoryTokenStorage(TokenStorage):
+    """Simple in-memory token storage per server instance"""
+    def __init__(self, server_name: str):
+        self.server_name = server_name
+        self.tokens: Optional[OAuthToken] = None
+        self.client_info: Optional[OAuthClientInformationFull] = None
+        
+    async def get_tokens(self) -> Optional[OAuthToken]:
+        return self.tokens
+        
+    async def set_tokens(self, tokens: OAuthToken) -> None:
+        self.tokens = tokens
+        logger.info(f"OAuth tokens stored for server: {self.server_name}")
+        
+    async def get_client_info(self) -> Optional[OAuthClientInformationFull]:
+        return self.client_info
+        
+    async def set_client_info(self, info: OAuthClientInformationFull) -> None:
+        self.client_info = info
+
+
+class FileTokenStorage(TokenStorage):
+    """File-based token storage with per-server isolation"""
+    def __init__(self, server_name: str, storage_dir: str = None):
+        self.server_name = server_name
+        # Use hash of server name to avoid filesystem issues
+        safe_name = hashlib.md5(server_name.encode()).hexdigest()[:8]
+        if storage_dir is None:
+            storage_dir = os.path.expanduser("~/.mcpo/tokens")
+        self.storage_dir = Path(storage_dir)
+        self.storage_dir.mkdir(parents=True, exist_ok=True)
+        self.token_file = self.storage_dir / f"{safe_name}_tokens.json"
+        self.client_file = self.storage_dir / f"{safe_name}_client.json"
+        
+    async def get_tokens(self) -> Optional[OAuthToken]:
+        if self.token_file.exists():
+            try:
+                with open(self.token_file, 'r') as f:
+                    data = json.load(f)
+                    return OAuthToken.model_validate(data)
+            except Exception as e:
+                logger.error(f"Failed to load tokens for {self.server_name}: {e}")
+        return None
+        
+    async def set_tokens(self, tokens: OAuthToken) -> None:
+        try:
+            with open(self.token_file, 'w') as f:
+                json.dump(tokens.model_dump(mode='json'), f)
+            logger.info(f"OAuth tokens persisted for server: {self.server_name}")
+        except Exception as e:
+            logger.error(f"Failed to save tokens for {self.server_name}: {e}")
+            
+    async def get_client_info(self) -> Optional[OAuthClientInformationFull]:
+        if self.client_file.exists():
+            try:
+                with open(self.client_file, 'r') as f:
+                    data = json.load(f)
+                    return OAuthClientInformationFull.model_validate(data)
+            except Exception as e:
+                logger.error(f"Failed to load client info for {self.server_name}: {e}")
+        return None
+        
+    async def set_client_info(self, info: OAuthClientInformationFull) -> None:
+        try:
+            with open(self.client_file, 'w') as f:
+                json.dump(info.model_dump(mode='json'), f)
+        except Exception as e:
+            logger.error(f"Failed to save client info for {self.server_name}: {e}")
+
+
+class CallbackHandler(BaseHTTPRequestHandler):
+    """HTTP handler for OAuth callbacks"""
+    def __init__(self, request, client_address, server, data):
+        self.data = data
+        super().__init__(request, client_address, server)
+        
+    def do_GET(self):
+        q = parse_qs(urlparse(self.path).query)
+        if "code" in q:
+            self.data["authorization_code"] = q["code"][0]
+            self.data["state"] = q.get("state", [None])[0]
+            self.send_response(200)
+            self.send_header('Content-Type', 'text/html; charset=utf-8')
+            self.end_headers()
+            
+            html = _load_callback_html(
+                status="success",
+                title="Authorization Successful - MCPO",
+                heading="Authorization Successful!",
+                message="Your OAuth authorization was completed successfully. The application can now access the requested resources.",
+                action_text="You can safely close this browser tab and return to your application."
+            )
+            self.wfile.write(html.encode('utf-8'))
+            
+        elif "error" in q:
+            error_desc = q.get("error_description", [q["error"][0]])[0]
+            self.data["error"] = q["error"][0]
+            self.send_response(400)
+            self.send_header('Content-Type', 'text/html; charset=utf-8')
+            self.end_headers()
+            
+            html = _load_callback_html(
+                status="error",
+                title="Authorization Failed - MCPO",
+                heading="Authorization Failed",
+                message=f"The OAuth authorization process encountered an error: {error_desc}",
+                action_text="Please close this tab and check the application logs for more details."
+            )
+            self.wfile.write(html.encode('utf-8'))
+            
+        else:
+            self.send_response(404)
+            self.send_header('Content-Type', 'text/html; charset=utf-8')
+            self.end_headers()
+            
+    def log_message(self, *_):
+        pass  # Suppress request logs
+
+
+class CallbackServer:
+    """Local HTTP server for OAuth callbacks"""
+    def __init__(self, port: int = 3030):
+        self.port = port
+        self.server = None
+        self.thread = None
+        self.data = {"authorization_code": None, "state": None, "error": None}
+        
+    def _handler(self):
+        data = self.data
+        class H(CallbackHandler):
+            def __init__(self, req, addr, srv):
+                super().__init__(req, addr, srv, data)
+        return H
+        
+    def start(self):
+        self.server = HTTPServer(("localhost", self.port), self._handler())
+        self.thread = threading.Thread(target=self.server.serve_forever, daemon=True)
+        self.thread.start()
+        logger.info(f"OAuth callback server listening on http://localhost:{self.port}/callback")
+        
+    def stop(self):
+        if self.server:
+            self.server.shutdown()
+            self.server.server_close()
+        if self.thread:
+            self.thread.join(timeout=1)
+            
+    def wait_code(self, timeout: int = 300) -> str:
+        start_time = time.time()
+        while time.time() - start_time < timeout:
+            if self.data["authorization_code"]:
+                return self.data["authorization_code"]
+            if self.data["error"]:
+                raise RuntimeError(f"OAuth error: {self.data['error']}")
+            time.sleep(0.05)
+        raise TimeoutError("No OAuth callback received within timeout")
+        
+    def state(self) -> Optional[str]:
+        return self.data["state"]
+
+
+async def create_oauth_provider(
+    server_name: str,
+    oauth_config: Dict[str, Any],
+    storage_type: str = "file"
+) -> OAuthClientProvider:
+    """Create an OAuth provider for a server"""
+    
+    # Extract OAuth configuration
+    server_url = oauth_config.get("server_url")
+    if not server_url:
+        raise ValueError(f"OAuth server_url required for {server_name}")
+        
+    # Build client metadata - only set defaults for required fields
+    metadata_dict = oauth_config.get("client_metadata", {})
+    if not metadata_dict.get("client_name"):
+        metadata_dict["client_name"] = f"MCPO Client for {server_name}"
+    if not metadata_dict.get("redirect_uris"):
+        callback_port = oauth_config.get("callback_port", 3030)
+        metadata_dict["redirect_uris"] = [f"http://localhost:{callback_port}/callback"]
+    if not metadata_dict.get("grant_types"):
+        metadata_dict["grant_types"] = ["authorization_code", "refresh_token"]
+    if not metadata_dict.get("response_types"):
+        metadata_dict["response_types"] = ["code"]
+        
+    # Don't set scope by default - let dynamic registration handle it
+    # Don't set authorization_endpoint or token_endpoint - these are discovered
+        
+    # Convert redirect_uris to AnyUrl
+    redirect_uris = [AnyUrl(uri) for uri in metadata_dict["redirect_uris"]]
+    metadata_dict["redirect_uris"] = redirect_uris
+    
+    client_metadata = OAuthClientMetadata.model_validate(metadata_dict)
+    
+    # Choose storage backend
+    if storage_type == "memory":
+        storage = InMemoryTokenStorage(server_name)
+    else:
+        storage = FileTokenStorage(server_name)
+    
+    # Setup callback handling
+    use_loopback = oauth_config.get("use_loopback", True)
+    callback_port = oauth_config.get("callback_port", 3030)
+    
+    if use_loopback:
+        # Loopback server for automatic callback handling
+        callback_server = CallbackServer(callback_port)
+        
+        async def redirect_handler(url: str) -> None:
+            logger.info(f"Opening browser for OAuth: {url}")
+            webbrowser.open(url)
+            
+        async def callback_handler() -> Tuple[str, Optional[str]]:
+            callback_server.start()
+            try:
+                code = callback_server.wait_code()
+                return code, callback_server.state()
+            finally:
+                callback_server.stop()
+    else:
+        # Manual copy/paste flow
+        async def redirect_handler(url: str) -> None:
+            print(f"\n\nPlease visit this URL to authorize:\n{url}\n")
+            
+        async def callback_handler() -> Tuple[str, Optional[str]]:
+            callback_url = input("Paste the callback URL here: ")
+            q = parse_qs(urlparse(callback_url).query)
+            code = q.get("code", [None])[0]
+            state = q.get("state", [None])[0]
+            if not code:
+                raise ValueError("No authorization code found in callback URL")
+            return code, state
+    
+    return OAuthClientProvider(
+        server_url=server_url,
+        client_metadata=client_metadata,
+        storage=storage,
+        redirect_handler=redirect_handler,
+        callback_handler=callback_handler,
+    )

src/mcpo/utils/oauth_callback.html

@@ -0,0 +1,133 @@
+<!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>
+        * {{
+            margin: 0;
+            padding: 0;
+            box-sizing: border-box;
+        }}
+        
+        body {{
+            font-family: -apple-system, BlinkMacSystemFont, 'Segoe UI', 'Roboto', 'Helvetica Neue', Arial, sans-serif;
+            background: #ffffff;
+            min-height: 100vh;
+            display: flex;
+            align-items: center;
+            justify-content: center;
+            padding: 20px;
+        }}
+        
+        .container {{
+            background: white;
+            border-radius: 12px;
+            box-shadow: 0 10px 30px rgba(0, 0, 0, 0.1);
+            padding: 40px;
+            text-align: center;
+            max-width: 480px;
+            width: 100%;
+            animation: slideUp 0.5s ease-out;
+        }}
+        
+        @keyframes slideUp {{
+            from {{
+                opacity: 0;
+                transform: translateY(20px);
+            }}
+            to {{
+                opacity: 1;
+                transform: translateY(0);
+            }}
+        }}
+        
+        .icon {{
+            width: 64px;
+            height: 64px;
+            margin: 0 auto 24px;
+            border-radius: 50%;
+            display: flex;
+            align-items: center;
+            justify-content: center;
+            font-size: 28px;
+        }}
+        
+        .success .icon {{
+            background: #3b82f6;
+            color: white;
+        }}
+        
+        .error .icon {{
+            background: #ef4444;
+            color: white;
+        }}
+        
+        h1 {{
+            font-size: 24px;
+            font-weight: 600;
+            margin-bottom: 12px;
+            color: #1f2937;
+        }}
+        
+        .success h1 {{
+            color: #3b82f6;
+        }}
+        
+        .error h1 {{
+            color: #ef4444;
+        }}
+        
+        p {{
+            font-size: 16px;
+            color: #6b7280;
+            line-height: 1.5;
+            margin-bottom: 24px;
+        }}
+        
+        .action-text {{
+            font-size: 14px;
+            color: #9ca3af;
+            background: #f9fafb;
+            padding: 12px 16px;
+            border-radius: 8px;
+            border-left: 4px solid #e5e7eb;
+        }}
+        
+        .success .action-text {{
+            border-left-color: #3b82f6;
+            background: #eff6ff;
+            color: #1e40af;
+        }}
+        
+        .error .action-text {{
+            border-left-color: #ef4444;
+            background: #fef2f2;
+            color: #991b1b;
+        }}
+        
+        @media (max-width: 480px) {{
+            .container {{
+                padding: 30px 20px;
+            }}
+            
+            h1 {{
+                font-size: 20px;
+            }}
+            
+            p {{
+                font-size: 15px;
+            }}
+        }}
+    </style>
+</head>
+<body>
+    <div class="container {status_class}">
+        <div class="icon">{icon}</div>
+        <h1>{heading}</h1>
+        <p>{message}</p>
+        <div class="action-text">{action_text}</div>
+    </div>
+</body>
+</html>