github-featured/modelcontextprotocol--python-sdk
2
0
Fork 0
Code Issues Pull Requests Deployments Models Agent Notes Packages Projects Releases Wiki Activity

Remove github from auth examples (#1011)

Browse Source
This commit is contained in:
Inna Harper
2025-06-24 15:43:08 +01:00
committed by GitHub
parent ecc916516d
commit ce007de66d
6 changed files with 377 additions and 503 deletions
Download Patch File Download Diff File Expand all files Collapse all files
examples/servers/simple-auth/README.md
+2 -16
View File
@@ -4,20 +4,6 @@ This example demonstrates OAuth 2.0 authentication with the Model Context Protoc
---
## Setup Requirements
**Create a GitHub OAuth App:**
- Go to GitHub Settings > Developer settings > OAuth Apps > New OAuth App
- **Authorization callback URL:** `http://localhost:9000/github/callback`
- Note down your **Client ID** and **Client Secret**
**Set environment variables:**
```bash
export MCP_GITHUB_CLIENT_ID="your_client_id_here"
export MCP_GITHUB_CLIENT_SECRET="your_client_secret_here"
```
---
## Running the Servers
@@ -33,9 +19,8 @@ uv run mcp-simple-auth-as --port=9000
**What it provides:**
- OAuth 2.0 flows (registration, authorization, token exchange)
- GitHub OAuth integration for user authentication
- Simple credential-based authentication (no external provider needed)
- Token introspection endpoint for Resource Servers (`/introspect`)
- User data proxy endpoint (`/github/user`)
---
@@ -90,6 +75,7 @@ curl http://localhost:9000/.well-known/oauth-authorization-server
}
```
## Legacy MCP Server as Authorization Server (Backwards Compatibility)
For backwards compatibility with older MCP implementations, a legacy server is provided that acts as an Authorization Server (following the old spec where MCP servers could optionally provide OAuth):
examples/servers/simple-auth/mcp_simple_auth/auth_server.py
+40 -87
View File
@@ -7,8 +7,6 @@ Can be replaced with enterprise authorization servers like Auth0, Entra ID, etc.
NOTE: this is a simplified example for demonstration purposes.
This is not a production-ready implementation.
Usage:
python -m mcp_simple_auth.auth_server --port=9000
"""
import asyncio
@@ -20,14 +18,14 @@ from pydantic import AnyHttpUrl, BaseModel
from starlette.applications import Starlette
from starlette.exceptions import HTTPException
from starlette.requests import Request
from starlette.responses import JSONResponse, RedirectResponse, Response
from starlette.responses import JSONResponse, Response
from starlette.routing import Route
from uvicorn import Config, Server
from mcp.server.auth.routes import cors_middleware, create_auth_routes
from mcp.server.auth.settings import AuthSettings, ClientRegistrationOptions
from .github_oauth_provider import GitHubOAuthProvider, GitHubOAuthSettings
from .simple_auth_provider import SimpleAuthSettings, SimpleOAuthProvider
logger = logging.getLogger(__name__)
@@ -39,60 +37,64 @@ class AuthServerSettings(BaseModel):
host: str = "localhost"
port: int = 9000
server_url: AnyHttpUrl = AnyHttpUrl("http://localhost:9000")
github_callback_path: str = "http://localhost:9000/github/callback"
auth_callback_path: str = "http://localhost:9000/login/callback"
class GitHubProxyAuthProvider(GitHubOAuthProvider):
class SimpleAuthProvider(SimpleOAuthProvider):
"""
Authorization Server provider that proxies GitHub OAuth.
Authorization Server provider with simple demo authentication.
This provider:
1. Issues MCP tokens after GitHub authentication
1. Issues MCP tokens after simple credential authentication
2. Stores token state for introspection by Resource Servers
3. Maps MCP tokens to GitHub tokens for API access
"""
def __init__(self, github_settings: GitHubOAuthSettings, github_callback_path: str):
super().__init__(github_settings, github_callback_path)
def __init__(self, auth_settings: SimpleAuthSettings, auth_callback_path: str, server_url: str):
super().__init__(auth_settings, auth_callback_path, server_url)
def create_authorization_server(server_settings: AuthServerSettings, github_settings: GitHubOAuthSettings) -> Starlette:
def create_authorization_server(server_settings: AuthServerSettings, auth_settings: SimpleAuthSettings) -> Starlette:
"""Create the Authorization Server application."""
oauth_provider = GitHubProxyAuthProvider(github_settings, server_settings.github_callback_path)
oauth_provider = SimpleAuthProvider(
auth_settings, server_settings.auth_callback_path, str(server_settings.server_url)
)
auth_settings = AuthSettings(
mcp_auth_settings = AuthSettings(
issuer_url=server_settings.server_url,
client_registration_options=ClientRegistrationOptions(
enabled=True,
valid_scopes=[github_settings.mcp_scope],
default_scopes=[github_settings.mcp_scope],
valid_scopes=[auth_settings.mcp_scope],
default_scopes=[auth_settings.mcp_scope],
),
required_scopes=[github_settings.mcp_scope],
required_scopes=[auth_settings.mcp_scope],
resource_server_url=None,
)
# Create OAuth routes
routes = create_auth_routes(
provider=oauth_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,
issuer_url=mcp_auth_settings.issuer_url,
service_documentation_url=mcp_auth_settings.service_documentation_url,
client_registration_options=mcp_auth_settings.client_registration_options,
revocation_options=mcp_auth_settings.revocation_options,
)
# Add GitHub callback route
async def github_callback_handler(request: Request) -> Response:
"""Handle GitHub OAuth callback."""
code = request.query_params.get("code")
# Add login page route (GET)
async def login_page_handler(request: Request) -> Response:
"""Show login form."""
state = request.query_params.get("state")
if not state:
raise HTTPException(400, "Missing state parameter")
return await oauth_provider.get_login_page(state)
if not code or not state:
raise HTTPException(400, "Missing code or state parameter")
routes.append(Route("/login", endpoint=login_page_handler, methods=["GET"]))
redirect_uri = await oauth_provider.handle_github_callback(code, state)
return RedirectResponse(url=redirect_uri, status_code=302)
# Add login callback route (POST)
async def login_callback_handler(request: Request) -> Response:
"""Handle simple authentication callback."""
return await oauth_provider.handle_login_callback(request)
routes.append(Route("/github/callback", endpoint=github_callback_handler, methods=["GET"]))
routes.append(Route("/login/callback", endpoint=login_callback_handler, methods=["POST"]))
# Add token introspection endpoint (RFC 7662) for Resource Servers
async def introspect_handler(request: Request) -> Response:
@@ -112,7 +114,6 @@ def create_authorization_server(server_settings: AuthServerSettings, github_sett
if not access_token:
return JSONResponse({"active": False})
# Return token info for Resource Server
return JSONResponse(
{
"active": True,
@@ -133,39 +134,12 @@ def create_authorization_server(server_settings: AuthServerSettings, github_sett
)
)
# Add GitHub user info endpoint (for Resource Server to fetch user data)
async def github_user_handler(request: Request) -> Response:
"""
Proxy endpoint to get GitHub user info using stored GitHub tokens.
Resource Servers call this with MCP tokens to get GitHub user data
without exposing GitHub tokens to clients.
"""
# Extract Bearer token
auth_header = request.headers.get("authorization", "")
if not auth_header.startswith("Bearer "):
return JSONResponse({"error": "unauthorized"}, status_code=401)
mcp_token = auth_header[7:]
# Get GitHub user info using the provider method
user_info = await oauth_provider.get_github_user_info(mcp_token)
return JSONResponse(user_info)
routes.append(
Route(
"/github/user",
endpoint=cors_middleware(github_user_handler, ["GET", "OPTIONS"]),
methods=["GET", "OPTIONS"],
)
)
return Starlette(routes=routes)
async def run_server(server_settings: AuthServerSettings, github_settings: GitHubOAuthSettings):
async def run_server(server_settings: AuthServerSettings, auth_settings: SimpleAuthSettings):
"""Run the Authorization Server."""
auth_server = create_authorization_server(server_settings, github_settings)
auth_server = create_authorization_server(server_settings, auth_settings)
config = Config(
auth_server,
@@ -175,22 +149,7 @@ async def run_server(server_settings: AuthServerSettings, github_settings: GitHu
)
server = Server(config)
logger.info("=" * 80)
logger.info("MCP AUTHORIZATION SERVER")
logger.info("=" * 80)
logger.info(f"Server URL: {server_settings.server_url}")
logger.info("Endpoints:")
logger.info(f" - OAuth Metadata: {server_settings.server_url}/.well-known/oauth-authorization-server")
logger.info(f" - Client Registration: {server_settings.server_url}/register")
logger.info(f" - Authorization: {server_settings.server_url}/authorize")
logger.info(f" - Token Exchange: {server_settings.server_url}/token")
logger.info(f" - Token Introspection: {server_settings.server_url}/introspect")
logger.info(f" - GitHub Callback: {server_settings.server_url}/github/callback")
logger.info(f" - GitHub User Proxy: {server_settings.server_url}/github/user")
logger.info("")
logger.info("Resource Servers should use /introspect to validate tokens")
logger.info("Configure GitHub App callback URL: " + server_settings.github_callback_path)
logger.info("=" * 80)
logger.info(f"🚀 MCP Authorization Server running on {server_settings.server_url}")
await server.serve()
@@ -203,18 +162,12 @@ def main(port: int) -> int:
This server handles OAuth flows and can be used by multiple Resource Servers.
Environment variables needed:
- MCP_GITHUB_CLIENT_ID: GitHub OAuth Client ID
- MCP_GITHUB_CLIENT_SECRET: GitHub OAuth Client Secret
Uses simple hardcoded credentials for demo purposes.
"""
logging.basicConfig(level=logging.INFO)
# Load GitHub settings from environment variables
github_settings = GitHubOAuthSettings()
# Validate required fields
if not github_settings.github_client_id or not github_settings.github_client_secret:
raise ValueError("GitHub credentials not provided")
# Load simple auth settings
auth_settings = SimpleAuthSettings()
# Create server settings
host = "localhost"
@@ -223,10 +176,10 @@ def main(port: int) -> int:
host=host,
port=port,
server_url=AnyHttpUrl(server_url),
github_callback_path=f"{server_url}/github/callback",
auth_callback_path=f"{server_url}/login",
)
asyncio.run(run_server(server_settings, github_settings))
asyncio.run(run_server(server_settings, auth_settings))
return 0
examples/servers/simple-auth/mcp_simple_auth/github_oauth_provider.py
-266
View File
@@ -1,266 +0,0 @@
"""
Shared GitHub OAuth provider for MCP servers.
This module contains the common GitHub OAuth functionality used by both
the standalone authorization server and the legacy combined server.
NOTE: this is a simplified example for demonstration purposes.
This is not a production-ready implementation.
"""
import logging
import secrets
import time
from typing import Any
from pydantic import AnyHttpUrl
from pydantic_settings import BaseSettings, SettingsConfigDict
from starlette.exceptions import HTTPException
from mcp.server.auth.provider import (
AccessToken,
AuthorizationCode,
AuthorizationParams,
OAuthAuthorizationServerProvider,
RefreshToken,
construct_redirect_uri,
)
from mcp.shared._httpx_utils import create_mcp_http_client
from mcp.shared.auth import OAuthClientInformationFull, OAuthToken
logger = logging.getLogger(__name__)
class GitHubOAuthSettings(BaseSettings):
"""Common GitHub OAuth settings."""
model_config = SettingsConfigDict(env_prefix="MCP_")
# GitHub OAuth settings - MUST be provided via environment variables
github_client_id: str | None = None
github_client_secret: str | None = None
# GitHub OAuth URLs
github_auth_url: str = "https://github.com/login/oauth/authorize"
github_token_url: str = "https://github.com/login/oauth/access_token"
mcp_scope: str = "user"
github_scope: str = "read:user"
class GitHubOAuthProvider(OAuthAuthorizationServerProvider):
"""
OAuth provider that uses GitHub as the identity provider.
This provider handles the OAuth flow by:
1. Redirecting users to GitHub for authentication
2. Exchanging GitHub tokens for MCP tokens
3. Maintaining token mappings for API access
"""
def __init__(self, settings: GitHubOAuthSettings, github_callback_url: str):
self.settings = settings
self.github_callback_url = github_callback_url
self.clients: dict[str, OAuthClientInformationFull] = {}
self.auth_codes: dict[str, AuthorizationCode] = {}
self.tokens: dict[str, AccessToken] = {}
self.state_mapping: dict[str, dict[str, str | None]] = {}
# Maps MCP tokens to GitHub tokens
self.token_mapping: dict[str, str] = {}
async def get_client(self, client_id: str) -> OAuthClientInformationFull | None:
"""Get OAuth client information."""
return self.clients.get(client_id)
async def register_client(self, client_info: OAuthClientInformationFull):
"""Register a new OAuth client."""
self.clients[client_info.client_id] = client_info
async def authorize(self, client: OAuthClientInformationFull, params: AuthorizationParams) -> str:
"""Generate an authorization URL for GitHub OAuth flow."""
state = params.state or secrets.token_hex(16)
# Store state mapping for callback
self.state_mapping[state] = {
"redirect_uri": str(params.redirect_uri),
"code_challenge": params.code_challenge,
"redirect_uri_provided_explicitly": str(params.redirect_uri_provided_explicitly),
"client_id": client.client_id,
"resource": params.resource, # RFC 8707
}
# Build GitHub authorization URL
auth_url = (
f"{self.settings.github_auth_url}"
f"?client_id={self.settings.github_client_id}"
f"&redirect_uri={self.github_callback_url}"
f"&scope={self.settings.github_scope}"
f"&state={state}"
)
return auth_url
async def handle_github_callback(self, code: str, state: str) -> str:
"""Handle GitHub OAuth callback and return redirect URI."""
state_data = self.state_mapping.get(state)
if not state_data:
raise HTTPException(400, "Invalid state parameter")
redirect_uri = state_data["redirect_uri"]
code_challenge = state_data["code_challenge"]
redirect_uri_provided_explicitly = state_data["redirect_uri_provided_explicitly"] == "True"
client_id = state_data["client_id"]
resource = state_data.get("resource") # RFC 8707
# These are required values from our own state mapping
assert redirect_uri is not None
assert code_challenge is not None
assert client_id is not None
# Exchange code for token with GitHub
async with create_mcp_http_client() as client:
response = await client.post(
self.settings.github_token_url,
data={
"client_id": self.settings.github_client_id,
"client_secret": self.settings.github_client_secret,
"code": code,
"redirect_uri": self.github_callback_url,
},
headers={"Accept": "application/json"},
)
if response.status_code != 200:
raise HTTPException(400, "Failed to exchange code for token")
data = response.json()
if "error" in data:
raise HTTPException(400, data.get("error_description", data["error"]))
github_token = data["access_token"]
# Create MCP authorization code
new_code = f"mcp_{secrets.token_hex(16)}"
auth_code = AuthorizationCode(
code=new_code,
client_id=client_id,
redirect_uri=AnyHttpUrl(redirect_uri),
redirect_uri_provided_explicitly=redirect_uri_provided_explicitly,
expires_at=time.time() + 300,
scopes=[self.settings.mcp_scope],
code_challenge=code_challenge,
resource=resource, # RFC 8707
)
self.auth_codes[new_code] = auth_code
# Store GitHub token with MCP client_id
self.tokens[github_token] = AccessToken(
token=github_token,
client_id=client_id,
scopes=[self.settings.github_scope],
expires_at=None,
)
del self.state_mapping[state]
return construct_redirect_uri(redirect_uri, code=new_code, state=state)
async def load_authorization_code(
self, client: OAuthClientInformationFull, authorization_code: str
) -> AuthorizationCode | None:
"""Load an authorization code."""
return self.auth_codes.get(authorization_code)
async def exchange_authorization_code(
self, client: OAuthClientInformationFull, authorization_code: AuthorizationCode
) -> OAuthToken:
"""Exchange authorization code for tokens."""
if authorization_code.code not in self.auth_codes:
raise ValueError("Invalid authorization code")
# Generate MCP access token
mcp_token = f"mcp_{secrets.token_hex(32)}"
# Store MCP token
self.tokens[mcp_token] = AccessToken(
token=mcp_token,
client_id=client.client_id,
scopes=authorization_code.scopes,
expires_at=int(time.time()) + 3600,
resource=authorization_code.resource, # RFC 8707
)
# Find GitHub token for this client
github_token = next(
(
token
for token, data in self.tokens.items()
if (token.startswith("ghu_") or token.startswith("gho_")) and data.client_id == client.client_id
),
None,
)
# Store mapping between MCP token and GitHub token
if github_token:
self.token_mapping[mcp_token] = github_token
del self.auth_codes[authorization_code.code]
return OAuthToken(
access_token=mcp_token,
token_type="Bearer",
expires_in=3600,
scope=" ".join(authorization_code.scopes),
)
async def load_access_token(self, token: str) -> AccessToken | None:
"""Load and validate an access token."""
access_token = self.tokens.get(token)
if not access_token:
return None
# Check if expired
if access_token.expires_at and access_token.expires_at < time.time():
del self.tokens[token]
return None
return access_token
async def load_refresh_token(self, client: OAuthClientInformationFull, refresh_token: str) -> RefreshToken | None:
"""Load a refresh token - not supported in this example."""
return None
async def exchange_refresh_token(
self,
client: OAuthClientInformationFull,
refresh_token: RefreshToken,
scopes: list[str],
) -> OAuthToken:
"""Exchange refresh token - not supported in this example."""
raise NotImplementedError("Refresh tokens not supported")
async def revoke_token(self, token: str, token_type_hint: str | None = None) -> None:
"""Revoke a token."""
if token in self.tokens:
del self.tokens[token]
async def get_github_user_info(self, mcp_token: str) -> dict[str, Any]:
"""Get GitHub user info using MCP token."""
github_token = self.token_mapping.get(mcp_token)
if not github_token:
raise ValueError("No GitHub token found for MCP token")
async with create_mcp_http_client() as client:
response = await client.get(
"https://api.github.com/user",
headers={
"Authorization": f"Bearer {github_token}",
"Accept": "application/vnd.github.v3+json",
},
)
if response.status_code != 200:
raise ValueError(f"GitHub API error: {response.status_code}")
return response.json()
examples/servers/simple-auth/mcp_simple_auth/legacy_as_server.py
+50 -63
View File
@@ -7,11 +7,9 @@ Used for backwards compatibility testing with the new split AS/RS architecture.
NOTE: this is a simplified example for demonstration purposes.
This is not a production-ready implementation.
Usage:
python -m mcp_simple_auth.legacy_as_server --port=8002
"""
import datetime
import logging
from typing import Any, Literal
@@ -19,97 +17,91 @@ import click
from pydantic import AnyHttpUrl, BaseModel
from starlette.exceptions import HTTPException
from starlette.requests import Request
from starlette.responses import RedirectResponse, Response
from starlette.responses import Response
from mcp.server.auth.middleware.auth_context import get_access_token
from mcp.server.auth.settings import AuthSettings, ClientRegistrationOptions
from mcp.server.fastmcp.server import FastMCP
from .github_oauth_provider import GitHubOAuthProvider, GitHubOAuthSettings
from .simple_auth_provider import SimpleAuthSettings, SimpleOAuthProvider
logger = logging.getLogger(__name__)
class ServerSettings(BaseModel):
"""Settings for the simple GitHub MCP server."""
"""Settings for the simple auth MCP server."""
# Server settings
host: str = "localhost"
port: int = 8000
server_url: AnyHttpUrl = AnyHttpUrl("http://localhost:8000")
github_callback_path: str = "http://localhost:8000/github/callback"
auth_callback_path: str = "http://localhost:8000/login/callback"
class SimpleGitHubOAuthProvider(GitHubOAuthProvider):
"""GitHub OAuth provider for legacy MCP server."""
class LegacySimpleOAuthProvider(SimpleOAuthProvider):
"""Simple OAuth provider for legacy MCP server."""
def __init__(self, github_settings: GitHubOAuthSettings, github_callback_path: str):
super().__init__(github_settings, github_callback_path)
def __init__(self, auth_settings: SimpleAuthSettings, auth_callback_path: str, server_url: str):
super().__init__(auth_settings, auth_callback_path, server_url)
def create_simple_mcp_server(server_settings: ServerSettings, github_settings: GitHubOAuthSettings) -> FastMCP:
"""Create a simple FastMCP server with GitHub OAuth."""
oauth_provider = SimpleGitHubOAuthProvider(github_settings, server_settings.github_callback_path)
def create_simple_mcp_server(server_settings: ServerSettings, auth_settings: SimpleAuthSettings) -> FastMCP:
"""Create a simple FastMCP server with simple authentication."""
oauth_provider = LegacySimpleOAuthProvider(
auth_settings, server_settings.auth_callback_path, str(server_settings.server_url)
)
auth_settings = AuthSettings(
mcp_auth_settings = AuthSettings(
issuer_url=server_settings.server_url,
client_registration_options=ClientRegistrationOptions(
enabled=True,
valid_scopes=[github_settings.mcp_scope],
default_scopes=[github_settings.mcp_scope],
valid_scopes=[auth_settings.mcp_scope],
default_scopes=[auth_settings.mcp_scope],
),
required_scopes=[github_settings.mcp_scope],
required_scopes=[auth_settings.mcp_scope],
# No resource_server_url parameter in legacy mode
resource_server_url=None,
)
app = FastMCP(
name="Simple GitHub MCP Server",
instructions="A simple MCP server with GitHub OAuth authentication",
name="Simple Auth MCP Server",
instructions="A simple MCP server with simple credential authentication",
auth_server_provider=oauth_provider,
host=server_settings.host,
port=server_settings.port,
debug=True,
auth=auth_settings,
auth=mcp_auth_settings,
)
@app.custom_route("/github/callback", methods=["GET"])
async def github_callback_handler(request: Request) -> Response:
"""Handle GitHub OAuth callback."""
code = request.query_params.get("code")
@app.custom_route("/login", methods=["GET"])
async def login_page_handler(request: Request) -> Response:
"""Show login form."""
state = request.query_params.get("state")
if not state:
raise HTTPException(400, "Missing state parameter")
return await oauth_provider.get_login_page(state)
if not code or not state:
raise HTTPException(400, "Missing code or state parameter")
redirect_uri = await oauth_provider.handle_github_callback(code, state)
return RedirectResponse(status_code=302, url=redirect_uri)
def get_github_token() -> str:
"""Get the GitHub token for the authenticated user."""
access_token = get_access_token()
if not access_token:
raise ValueError("Not authenticated")
# Get GitHub token from mapping
github_token = oauth_provider.token_mapping.get(access_token.token)
if not github_token:
raise ValueError("No GitHub token found for user")
return github_token
@app.custom_route("/login/callback", methods=["POST"])
async def login_callback_handler(request: Request) -> Response:
"""Handle simple authentication callback."""
return await oauth_provider.handle_login_callback(request)
@app.tool()
async def get_user_profile() -> dict[str, Any]:
"""Get the authenticated user's GitHub profile information.
This is the only tool in our simple example. It requires the 'user' scope.
async def get_time() -> dict[str, Any]:
"""
access_token = get_access_token()
if not access_token:
raise ValueError("Not authenticated")
Get the current server time.
return await oauth_provider.get_github_user_info(access_token.token)
This tool demonstrates that system information can be protected
by OAuth authentication. User must be authenticated to access it.
"""
now = datetime.datetime.now()
return {
"current_time": now.isoformat(),
"timezone": "UTC", # Simplified for demo
"timestamp": now.timestamp(),
"formatted": now.strftime("%Y-%m-%d %H:%M:%S"),
}
return app
@@ -123,15 +115,10 @@ def create_simple_mcp_server(server_settings: ServerSettings, github_settings: G
help="Transport protocol to use ('sse' or 'streamable-http')",
)
def main(port: int, transport: Literal["sse", "streamable-http"]) -> int:
"""Run the simple GitHub MCP server."""
"""Run the simple auth MCP server."""
logging.basicConfig(level=logging.INFO)
# Load GitHub settings from environment variables
github_settings = GitHubOAuthSettings()
# Validate required fields
if not github_settings.github_client_id or not github_settings.github_client_secret:
raise ValueError("GitHub credentials not provided")
auth_settings = SimpleAuthSettings()
# Create server settings
host = "localhost"
server_url = f"http://{host}:{port}"
@@ -139,11 +126,11 @@ def main(port: int, transport: Literal["sse", "streamable-http"]) -> int:
host=host,
port=port,
server_url=AnyHttpUrl(server_url),
github_callback_path=f"{server_url}/github/callback",
auth_callback_path=f"{server_url}/login",
)
mcp_server = create_simple_mcp_server(server_settings, github_settings)
logger.info(f"Starting server with {transport} transport")
mcp_server = create_simple_mcp_server(server_settings, auth_settings)
logger.info(f"🚀 MCP Legacy Server running on {server_url}")
mcp_server.run(transport=transport)
return 0
examples/servers/simple-auth/mcp_simple_auth/server.py
+15 -71
View File
@@ -4,19 +4,18 @@ MCP Resource Server with Token Introspection.
This server validates tokens via Authorization Server introspection and serves MCP resources.
Demonstrates RFC 9728 Protected Resource Metadata for AS/RS separation.
Usage:
python -m mcp_simple_auth.server --port=8001 --auth-server=http://localhost:9000
NOTE: this is a simplified example for demonstration purposes.
This is not a production-ready implementation.
"""
import datetime
import logging
from typing import Any, Literal
import click
import httpx
from pydantic import AnyHttpUrl
from pydantic_settings import BaseSettings, SettingsConfigDict
from mcp.server.auth.middleware.auth_context import get_access_token
from mcp.server.auth.settings import AuthSettings
from mcp.server.fastmcp.server import FastMCP
@@ -38,7 +37,7 @@ class ResourceServerSettings(BaseSettings):
# Authorization Server settings
auth_server_url: AnyHttpUrl = AnyHttpUrl("http://localhost:9000")
auth_server_introspection_endpoint: str = "http://localhost:9000/introspect"
auth_server_github_user_endpoint: str = "http://localhost:9000/github/user"
# No user endpoint needed - we get user data from token introspection
# MCP settings
mcp_scope: str = "user"
@@ -83,60 +82,22 @@ def create_resource_server(settings: ResourceServerSettings) -> FastMCP:
),
)
async def get_github_user_data() -> dict[str, Any]:
"""
Get GitHub user data via Authorization Server proxy endpoint.
This avoids exposing GitHub tokens to the Resource Server.
The Authorization Server handles the GitHub API call and returns the data.
"""
access_token = get_access_token()
if not access_token:
raise ValueError("Not authenticated")
# Call Authorization Server's GitHub proxy endpoint
async with httpx.AsyncClient() as client:
response = await client.get(
settings.auth_server_github_user_endpoint,
headers={
"Authorization": f"Bearer {access_token.token}",
},
)
if response.status_code != 200:
raise ValueError(f"GitHub user data fetch failed: {response.status_code} - {response.text}")
return response.json()
@app.tool()
async def get_user_profile() -> dict[str, Any]:
async def get_time() -> dict[str, Any]:
"""
Get the authenticated user's GitHub profile information.
Get the current server time.
This tool requires the 'user' scope and demonstrates how Resource Servers
can access user data without directly handling GitHub tokens.
This tool demonstrates that system information can be protected
by OAuth authentication. User must be authenticated to access it.
"""
return await get_github_user_data()
@app.tool()
async def get_user_info() -> dict[str, Any]:
"""
Get information about the currently authenticated user.
Returns token and scope information from the Resource Server's perspective.
"""
access_token = get_access_token()
if not access_token:
raise ValueError("Not authenticated")
now = datetime.datetime.now()
return {
"authenticated": True,
"client_id": access_token.client_id,
"scopes": access_token.scopes,
"token_expires_at": access_token.expires_at,
"token_type": "Bearer",
"resource_server": str(settings.server_url),
"authorization_server": str(settings.auth_server_url),
"current_time": now.isoformat(),
"timezone": "UTC", # Simplified for demo
"timestamp": now.timestamp(),
"formatted": now.strftime("%Y-%m-%d %H:%M:%S"),
}
return app
@@ -182,7 +143,6 @@ def main(port: int, auth_server: str, transport: Literal["sse", "streamable-http
server_url=AnyHttpUrl(server_url),
auth_server_url=auth_server_url,
auth_server_introspection_endpoint=f"{auth_server}/introspect",
auth_server_github_user_endpoint=f"{auth_server}/github/user",
oauth_strict=oauth_strict,
)
except ValueError as e:
@@ -193,24 +153,8 @@ def main(port: int, auth_server: str, transport: Literal["sse", "streamable-http
try:
mcp_server = create_resource_server(settings)
logger.info("=" * 80)
logger.info("📦 MCP RESOURCE SERVER")
logger.info("=" * 80)
logger.info(f"🌐 Server URL: {settings.server_url}")
logger.info(f"🔑 Authorization Server: {settings.auth_server_url}")
logger.info("📋 Endpoints:")
logger.info(f" ┌─ Protected Resource Metadata: {settings.server_url}/.well-known/oauth-protected-resource")
mcp_path = "sse" if transport == "sse" else "mcp"
logger.info(f" ├─ MCP Protocol: {settings.server_url}/{mcp_path}")
logger.info(f" └─ Token Introspection: {settings.auth_server_introspection_endpoint}")
logger.info("")
logger.info("🛠️ Available Tools:")
logger.info(" ├─ get_user_profile() - Get GitHub user profile")
logger.info(" └─ get_user_info() - Get authentication status")
logger.info("")
logger.info("🔍 Tokens validated via Authorization Server introspection")
logger.info("📱 Clients discover Authorization Server via Protected Resource Metadata")
logger.info("=" * 80)
logger.info(f"🚀 MCP Resource Server running on {settings.server_url}")
logger.info(f"🔑 Using Authorization Server: {settings.auth_server_url}")
# Run the server - this should block and keep running
mcp_server.run(transport=transport)
examples/servers/simple-auth/mcp_simple_auth/simple_auth_provider.py
+270
View File
@@ -0,0 +1,270 @@
"""
Simple OAuth provider for MCP servers.
This module contains a basic OAuth implementation using hardcoded user credentials
for demonstration purposes. No external authentication provider is required.
NOTE: this is a simplified example for demonstration purposes.
This is not a production-ready implementation.
"""
import logging
import secrets
import time
from typing import Any
from pydantic import AnyHttpUrl
from pydantic_settings import BaseSettings, SettingsConfigDict
from starlette.exceptions import HTTPException
from starlette.requests import Request
from starlette.responses import HTMLResponse, RedirectResponse, Response
from mcp.server.auth.provider import (
AccessToken,
AuthorizationCode,
AuthorizationParams,
OAuthAuthorizationServerProvider,
RefreshToken,
construct_redirect_uri,
)
from mcp.shared.auth import OAuthClientInformationFull, OAuthToken
logger = logging.getLogger(__name__)
class SimpleAuthSettings(BaseSettings):
"""Simple OAuth settings for demo purposes."""
model_config = SettingsConfigDict(env_prefix="MCP_")
# Demo user credentials
demo_username: str = "demo_user"
demo_password: str = "demo_password"
# MCP OAuth scope
mcp_scope: str = "user"
class SimpleOAuthProvider(OAuthAuthorizationServerProvider):
"""
Simple OAuth provider for demo purposes.
This provider handles the OAuth flow by:
1. Providing a simple login form for demo credentials
2. Issuing MCP tokens after successful authentication
3. Maintaining token state for introspection
"""
def __init__(self, settings: SimpleAuthSettings, auth_callback_url: str, server_url: str):
self.settings = settings
self.auth_callback_url = auth_callback_url
self.server_url = server_url
self.clients: dict[str, OAuthClientInformationFull] = {}
self.auth_codes: dict[str, AuthorizationCode] = {}
self.tokens: dict[str, AccessToken] = {}
self.state_mapping: dict[str, dict[str, str | None]] = {}
# Store authenticated user information
self.user_data: dict[str, dict[str, Any]] = {}
async def get_client(self, client_id: str) -> OAuthClientInformationFull | None:
"""Get OAuth client information."""
return self.clients.get(client_id)
async def register_client(self, client_info: OAuthClientInformationFull):
"""Register a new OAuth client."""
self.clients[client_info.client_id] = client_info
async def authorize(self, client: OAuthClientInformationFull, params: AuthorizationParams) -> str:
"""Generate an authorization URL for simple login flow."""
state = params.state or secrets.token_hex(16)
# Store state mapping for callback
self.state_mapping[state] = {
"redirect_uri": str(params.redirect_uri),
"code_challenge": params.code_challenge,
"redirect_uri_provided_explicitly": str(params.redirect_uri_provided_explicitly),
"client_id": client.client_id,
"resource": params.resource, # RFC 8707
}
# Build simple login URL that points to login page
auth_url = f"{self.auth_callback_url}" f"?state={state}" f"&client_id={client.client_id}"
return auth_url
async def get_login_page(self, state: str) -> HTMLResponse:
"""Generate login page HTML for the given state."""
if not state:
raise HTTPException(400, "Missing state parameter")
# Create simple login form HTML
html_content = f"""
<!DOCTYPE html>
<html>
<head>
<title>MCP Demo Authentication</title>
<style>
body {{ font-family: Arial, sans-serif; max-width: 500px; margin: 0 auto; padding: 20px; }}
.form-group {{ margin-bottom: 15px; }}
input {{ width: 100%; padding: 8px; margin-top: 5px; }}
button {{ background-color: #4CAF50; color: white; padding: 10px 15px; border: none; cursor: pointer; }}
</style>
</head>
<body>
<h2>MCP Demo Authentication</h2>
<p>This is a simplified authentication demo. Use the demo credentials below:</p>
<p><strong>Username:</strong> demo_user<br>
<strong>Password:</strong> demo_password</p>
<form action="{self.server_url.rstrip('/')}/login/callback" method="post">
<input type="hidden" name="state" value="{state}">
<div class="form-group">
<label>Username:</label>
<input type="text" name="username" value="demo_user" required>
</div>
<div class="form-group">
<label>Password:</label>
<input type="password" name="password" value="demo_password" required>
</div>
<button type="submit">Sign In</button>
</form>
</body>
</html>
"""
return HTMLResponse(content=html_content)
async def handle_login_callback(self, request: Request) -> Response:
"""Handle login form submission callback."""
form = await request.form()
username = form.get("username")
password = form.get("password")
state = form.get("state")
if not username or not password or not state:
raise HTTPException(400, "Missing username, password, or state parameter")
# Ensure we have strings, not UploadFile objects
if not isinstance(username, str) or not isinstance(password, str) or not isinstance(state, str):
raise HTTPException(400, "Invalid parameter types")
redirect_uri = await self.handle_simple_callback(username, password, state)
return RedirectResponse(url=redirect_uri, status_code=302)
async def handle_simple_callback(self, username: str, password: str, state: str) -> str:
"""Handle simple authentication callback and return redirect URI."""
state_data = self.state_mapping.get(state)
if not state_data:
raise HTTPException(400, "Invalid state parameter")
redirect_uri = state_data["redirect_uri"]
code_challenge = state_data["code_challenge"]
redirect_uri_provided_explicitly = state_data["redirect_uri_provided_explicitly"] == "True"
client_id = state_data["client_id"]
resource = state_data.get("resource") # RFC 8707
# These are required values from our own state mapping
assert redirect_uri is not None
assert code_challenge is not None
assert client_id is not None
# Validate demo credentials
if username != self.settings.demo_username or password != self.settings.demo_password:
raise HTTPException(401, "Invalid credentials")
# Create MCP authorization code
new_code = f"mcp_{secrets.token_hex(16)}"
auth_code = AuthorizationCode(
code=new_code,
client_id=client_id,
redirect_uri=AnyHttpUrl(redirect_uri),
redirect_uri_provided_explicitly=redirect_uri_provided_explicitly,
expires_at=time.time() + 300,
scopes=[self.settings.mcp_scope],
code_challenge=code_challenge,
resource=resource, # RFC 8707
)
self.auth_codes[new_code] = auth_code
# Store user data
self.user_data[username] = {
"username": username,
"user_id": f"user_{secrets.token_hex(8)}",
"authenticated_at": time.time(),
}
del self.state_mapping[state]
return construct_redirect_uri(redirect_uri, code=new_code, state=state)
async def load_authorization_code(
self, client: OAuthClientInformationFull, authorization_code: str
) -> AuthorizationCode | None:
"""Load an authorization code."""
return self.auth_codes.get(authorization_code)
async def exchange_authorization_code(
self, client: OAuthClientInformationFull, authorization_code: AuthorizationCode
) -> OAuthToken:
"""Exchange authorization code for tokens."""
if authorization_code.code not in self.auth_codes:
raise ValueError("Invalid authorization code")
# Generate MCP access token
mcp_token = f"mcp_{secrets.token_hex(32)}"
# Store MCP token
self.tokens[mcp_token] = AccessToken(
token=mcp_token,
client_id=client.client_id,
scopes=authorization_code.scopes,
expires_at=int(time.time()) + 3600,
resource=authorization_code.resource, # RFC 8707
)
# Store user data mapping for this token
self.user_data[mcp_token] = {
"username": self.settings.demo_username,
"user_id": f"user_{secrets.token_hex(8)}",
"authenticated_at": time.time(),
}
del self.auth_codes[authorization_code.code]
return OAuthToken(
access_token=mcp_token,
token_type="Bearer",
expires_in=3600,
scope=" ".join(authorization_code.scopes),
)
async def load_access_token(self, token: str) -> AccessToken | None:
"""Load and validate an access token."""
access_token = self.tokens.get(token)
if not access_token:
return None
# Check if expired
if access_token.expires_at and access_token.expires_at < time.time():
del self.tokens[token]
return None
return access_token
async def load_refresh_token(self, client: OAuthClientInformationFull, refresh_token: str) -> RefreshToken | None:
"""Load a refresh token - not supported in this example."""
return None
async def exchange_refresh_token(
self,
client: OAuthClientInformationFull,
refresh_token: RefreshToken,
scopes: list[str],
) -> OAuthToken:
"""Exchange refresh token - not supported in this example."""
raise NotImplementedError("Refresh tokens not supported")
async def revoke_token(self, token: str, token_type_hint: str | None = None) -> None:
"""Revoke a token."""
if token in self.tokens:
del self.tokens[token]