tessl/pypi-mcp

Model Context Protocol SDK for building MCP servers and clients in Python

—

Pending

Overview

Eval results

Files

Authentication & Security

Name: tessl/pypi-mcp
Author: tessl

OAuth 2.0 authentication, bearer token verification, transport security settings, and middleware for protecting MCP server endpoints. The authentication system provides secure access control for MCP servers with support for various OAuth flows and token verification strategies.

Capabilities

Authentication Settings

Configuration classes for OAuth 2.0 authentication and security settings.

class AuthSettings:
    def __init__(
        self,
        client_id: str,
        client_secret: str,
        authorization_endpoint: str,
        token_endpoint: str,
        scopes: list[str] | None = None,
        **kwargs
    ):
        """
        OAuth 2.0 authentication configuration.

        Parameters:
        - client_id: OAuth client identifier
        - client_secret: OAuth client secret
        - authorization_endpoint: OAuth authorization URL
        - token_endpoint: OAuth token exchange URL
        - scopes: Required OAuth scopes
        - **kwargs: Additional OAuth parameters
        """

class ClientRegistrationOptions:
    def __init__(
        self,
        registration_endpoint: str,
        client_name: str,
        redirect_uris: list[str],
        **kwargs
    ):
        """
        OAuth client registration options.

        Parameters:
        - registration_endpoint: Client registration URL
        - client_name: Human-readable client name
        - redirect_uris: List of valid redirect URIs
        - **kwargs: Additional registration parameters
        """

class RevocationOptions:
    def __init__(
        self,
        revocation_endpoint: str,
        revocation_endpoint_auth_method: str = "client_secret_basic",
        **kwargs
    ):
        """
        OAuth token revocation options.

        Parameters:
        - revocation_endpoint: Token revocation URL
        - revocation_endpoint_auth_method: Authentication method for revocation
        - **kwargs: Additional revocation parameters
        """

Token Verification

Interfaces and implementations for verifying OAuth tokens and bearer tokens.

class TokenVerifier:
    async def verify_token(self, token: str) -> dict[str, Any]:
        """
        Verify an OAuth token and return claims.

        Parameters:
        - token: Bearer token to verify

        Returns:
        Dictionary containing token claims and user information
        """

class ProviderTokenVerifier(TokenVerifier):
    def __init__(
        self,
        auth_settings: AuthSettings,
        jwks_url: str | None = None,
        **kwargs
    ):
        """
        Provider-based token verifier using OAuth introspection.

        Parameters:
        - auth_settings: OAuth authentication settings
        - jwks_url: JSON Web Key Set URL for token validation
        - **kwargs: Additional verification options
        """

    async def verify_token(self, token: str) -> dict[str, Any]:
        """Verify token using OAuth provider introspection."""

class JWTTokenVerifier(TokenVerifier):
    def __init__(
        self,
        public_key: str,
        algorithm: str = "RS256",
        audience: str | None = None,
        **kwargs
    ):
        """
        JWT token verifier for local token validation.

        Parameters:
        - public_key: Public key for JWT verification
        - algorithm: JWT signing algorithm
        - audience: Expected token audience
        - **kwargs: Additional JWT validation options
        """

    async def verify_token(self, token: str) -> dict[str, Any]:
        """Verify JWT token locally."""

Authentication Middleware

Middleware components for enforcing authentication requirements and managing authentication context.

class BearerAuthBackend:
    def __init__(
        self,
        token_verifier: TokenVerifier,
        scheme_name: str = "Bearer",
        **kwargs
    ):
        """
        Bearer token authentication backend.

        Parameters:
        - token_verifier: Token verification implementation
        - scheme_name: Authentication scheme name
        - **kwargs: Additional backend options
        """

    async def authenticate(self, request) -> dict[str, Any] | None:
        """
        Authenticate request using bearer token.

        Parameters:
        - request: HTTP request object

        Returns:
        User information dictionary or None if authentication fails
        """

class RequireAuthMiddleware:
    def __init__(
        self,
        auth_backend: BearerAuthBackend,
        excluded_paths: list[str] | None = None,
        **kwargs
    ):
        """
        Middleware requiring authentication for protected endpoints.

        Parameters:
        - auth_backend: Authentication backend
        - excluded_paths: Paths that don't require authentication
        - **kwargs: Additional middleware options
        """

class AuthContextMiddleware:
    def __init__(self, **kwargs):
        """
        Middleware for providing authentication context to handlers.

        Parameters:
        - **kwargs: Middleware configuration options
        """

Transport Security

Security settings for protecting transport connections and preventing common attacks.

class TransportSecuritySettings:
    def __init__(
        self,
        dns_rebinding_protection: bool = True,
        allowed_origins: list[str] | None = None,
        require_https: bool = True,
        **kwargs
    ):
        """
        Transport security configuration.

        Parameters:
        - dns_rebinding_protection: Enable DNS rebinding attack protection
        - allowed_origins: List of allowed origins for CORS
        - require_https: Require HTTPS for security-sensitive operations
        - **kwargs: Additional security options
        """

Usage Examples

Basic OAuth Authentication

from mcp.server import FastMCP
from mcp.server.auth import AuthSettings, ProviderTokenVerifier

# Configure OAuth authentication
auth_settings = AuthSettings(
    client_id="your-client-id",
    client_secret="your-client-secret",
    authorization_endpoint="https://auth.example.com/oauth/authorize",
    token_endpoint="https://auth.example.com/oauth/token",
    scopes=["read", "write"]
)

# Create token verifier
token_verifier = ProviderTokenVerifier(
    auth_settings=auth_settings,
    jwks_url="https://auth.example.com/.well-known/jwks.json"
)

# Create authenticated server
app = FastMCP(
    "secure-server",
    auth_server_provider=auth_settings,
    token_verifier=token_verifier
)

@app.tool()
async def protected_tool(ctx, data: str) -> str:
    """Tool requiring authentication."""
    user_id = ctx.user_id
    if not user_id:
        raise ValueError("Authentication required")
    
    return f"Processed data for user {user_id}: {data}"

# Run with authentication
if __name__ == "__main__":
    import asyncio
    asyncio.run(app.run_sse(port=8443))

JWT Token Verification

from mcp.server import FastMCP, Context
from mcp.server.auth import JWTTokenVerifier

# JWT verification with public key
public_key = """-----BEGIN PUBLIC KEY-----
MIIBIjANBgkqhkiG9w0BAQEFAAOCAQ8AMIIBCgKCAQEA...
-----END PUBLIC KEY-----"""

token_verifier = JWTTokenVerifier(
    public_key=public_key,
    algorithm="RS256",
    audience="mcp-server"
)

app = FastMCP("jwt-server", token_verifier=token_verifier)

@app.tool()
async def user_info(ctx: Context) -> dict:
    """Get authenticated user information."""
    if not ctx.user_id:
        raise ValueError("JWT authentication required")
    
    # Access token claims from context
    return {
        "user_id": ctx.user_id,
        "request_id": ctx.request_id,
        "authenticated": True
    }

if __name__ == "__main__":
    import asyncio
    asyncio.run(app.run_sse(port=8000))

Custom Authentication Middleware

from mcp.server import FastMCP
from mcp.server.auth import BearerAuthBackend, RequireAuthMiddleware, TokenVerifier

class CustomTokenVerifier(TokenVerifier):
    """Custom token verification logic."""
    
    async def verify_token(self, token: str) -> dict[str, Any]:
        # Custom token validation
        if token.startswith("valid_"):
            user_id = token.replace("valid_", "")
            return {
                "user_id": user_id,
                "roles": ["user"],
                "permissions": ["read", "write"]
            }
        else:
            raise ValueError("Invalid token")

# Setup authentication chain
token_verifier = CustomTokenVerifier()
auth_backend = BearerAuthBackend(token_verifier)
auth_middleware = RequireAuthMiddleware(
    auth_backend,
    excluded_paths=["/health", "/public"]
)

app = FastMCP("custom-auth-server")

# Apply middleware
app.add_middleware(auth_middleware)

@app.get("/health")
async def health_check():
    """Public health check endpoint."""
    return {"status": "healthy"}

@app.tool()
async def secure_operation(ctx, operation: str) -> str:
    """Protected tool requiring authentication."""
    user_id = ctx.user_id
    return f"User {user_id} performed: {operation}"

if __name__ == "__main__":
    import asyncio
    asyncio.run(app.run_sse(port=8000))

Transport Security Configuration

from mcp.server import FastMCP
from mcp.server.transport_security import TransportSecuritySettings

# Configure transport security
security_settings = TransportSecuritySettings(
    dns_rebinding_protection=True,
    allowed_origins=[
        "https://app.example.com",
        "https://admin.example.com"
    ],
    require_https=True
)

app = FastMCP(
    "secure-transport-server",
    transport_security=security_settings
)

@app.tool()
async def sensitive_data() -> dict:
    """Tool handling sensitive data."""
    return {
        "data": "sensitive information",
        "security": "protected by transport security"
    }

if __name__ == "__main__":
    import asyncio
    # Run on HTTPS with security settings
    asyncio.run(app.run_sse(
        host="0.0.0.0", 
        port=8443,
        ssl_certfile="cert.pem",
        ssl_keyfile="key.pem"
    ))

Client-Side Authentication

import asyncio
from mcp import ClientSession, sse_client
from mcp.client.auth import ClientAuth

async def authenticated_client():
    # Configure client authentication
    auth = ClientAuth(token="valid_user123")
    headers = await auth.get_headers()
    
    # Connect with authentication headers
    async with sse_client(
        "https://localhost:8443/mcp",
        headers=headers
    ) as (read, write):
        async with ClientSession(read, write) as session:
            await session.initialize()
            
            # Call authenticated tools
            result = await session.call_tool("secure_operation", {
                "operation": "data_processing"
            })
            print(f"Authenticated result: {result}")

asyncio.run(authenticated_client())

Install with Tessl CLI