[BUG] MCP OAuth Integration Fails on Production Deployments with step=start_error

Same experience.

Any update here? Anyone able to suggest a workaround? Thanks in advance!

We are seeing this as well. Strangely claude cli works flawlessly and only claude desktop/web have the issue.

I'm also hitting this error. It's really frustrating since there's no detail on where it's going sideways. It's working fine with MCP Inspector but not in Claude Desktop

I'd highly recommend testing with Cloudflare's AI Playground if you're running into issues with Claude Desktop and Web not working when the CLI is. I was in this situation a few weeks ago, and there were several things I think were causing issues with Claude desktop which I was able to also encounter in Cloudflare's playground, but Cloudflare gave significantly more helpful error messages. Once I got it working in the playground, the remaining error I ran into at least had to do with how Claude was calling the registration endpoint for Dynamic Client Registration. It was always sending the registration call with token_endpoint_auth_method set to client_secret_post, which was failing, and since I was using a third-party IDP, I wasn't seeing that call at all, so it just looked like it failed to complete the OAuth flow.

Thanks Michelle -^ I wasn't able to use AI Playground, but I did run it through MCP-Remote again and was able to identify the problem - Entra was throwing an error based on how my callback was setup. I reclassified the callback URL and then it connected fine in MCP-Remote and on Claude Desktop.

Hi guys, I successfully got Claude Desktop (Custom Connectors) to connect to a remote MCP server with Entra OAuth authentication. The key was implementing RFC 8414 OAuth discovery endpoints (/.well-known/oauth-authorization-server, etc.) directly in my FastAPI server alongside the FastMCP endpoints. (I have a FastAPI handling Oauth on the same server as the MCP.)

I'm not an OAuth expert, so I can't vouch for the security, but it's working for my own personal MCP.

🤞can you share your endpoints on here with sanitized info? I tried this with no success a while back when I was trying to get this to work. Hoping I am missing something you have. Greatly appreciated!

I'm hesitant to be excited bc I had the full flow working with mcp inspector. I had to hard code a client id for the DCR endpoint to fit our security posture.

@cwise89 this is what i tested with

import asyncio
import logging
import os
from contextlib import asynccontextmanager
from typing import Any, Dict, Optional
from urllib.parse import urlencode, parse_qs, urlparse

import httpx
from fastapi import FastAPI, Request, HTTPException, Depends, status
from fastapi.responses import HTMLResponse, RedirectResponse, JSONResponse
from fastapi.middleware.cors import CORSMiddleware
from starlette.middleware.sessions import SessionMiddleware
from fastmcp import FastMCP

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

# Environment variables - Replace with your own values
AZURE_CLIENT_ID = os.getenv("AZURE_CLIENT_ID", "your-azure-client-id")
AZURE_CLIENT_SECRET = os.getenv("AZURE_CLIENT_SECRET", "your-azure-client-secret")
AZURE_TENANT_ID = os.getenv("AZURE_TENANT_ID", "your-azure-tenant-id")
AZURE_REDIRECT_URI = os.getenv("AZURE_REDIRECT_URI", "https://your-domain.com/auth/callback")
AUTHORIZED_USER_EMAIL = os.getenv("AUTHORIZED_USER_EMAIL", "user@example.com")
SESSION_SECRET_KEY = os.getenv("SESSION_SECRET_KEY", "generate-a-secure-random-key")

# Azure OAuth endpoints
AZURE_AUTH_ENDPOINT = f"https://login.microsoftonline.com/{AZURE_TENANT_ID}/oauth2/v2.0/authorize"
AZURE_TOKEN_ENDPOINT = f"https://login.microsoftonline.com/{AZURE_TENANT_ID}/oauth2/v2.0/token"
AZURE_USERINFO_ENDPOINT = "https://graph.microsoft.com/v1.0/me"

# Replace with your domain
SERVER_DOMAIN = os.getenv("SERVER_DOMAIN", "https://your-domain.com")

# Create MCP server
mcp = FastMCP("MCP OAuth Server")

# Create the MCP's ASGI app
mcp_app = mcp.http_app(path='/mcp')

# Create FastAPI app with combined lifespan
@asynccontextmanager
async def combined_lifespan(app: FastAPI):
    async with mcp_app.lifespan(app):
        yield

# Create FastAPI app
app = FastAPI(title="MCP OAuth Server", lifespan=combined_lifespan)

# Add session middleware
app.add_middleware(SessionMiddleware, secret_key=SESSION_SECRET_KEY)

# Add CORS middleware
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

def get_current_user(request: Request) -> Optional[Dict[str, Any]]:
    """Get current authenticated user from session"""
    return request.session.get("user")

def require_auth(request: Request) -> Dict[str, Any]:
    """Require authentication, raise 401 if not authenticated"""
    user = get_current_user(request)
    if not user:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Authentication required"
        )
    return user

@app.get("/")
async def root(request: Request):
    """Root endpoint - show login status"""
    user = get_current_user(request)
    if user:
        return HTMLResponse(f"""
        <html>
            <body>
                <h1>MCP OAuth Server</h1>
                <p>Authenticated as: {user.get('email', 'Unknown')}</p>
                <p>User: {user.get('name', 'Unknown')}</p>
                <a href="/auth/logout">Logout</a>
                <br><br>
                <h2>MCP Server Status</h2>
                <p>MCP Server is running and ready to accept connections.</p>
                <p>WebSocket endpoint: <code>ws://localhost:8000/ws</code></p>
            </body>
        </html>
        """)
    else:
        return HTMLResponse(f"""
        <html>
            <body>
                <h1>MCP OAuth Server</h1>
                <p>Please authenticate to continue.</p>
                <a href="/auth/login">Login with Microsoft</a>
            </body>
        </html>
        """)

@app.get("/auth/login")
async def login():
    """Initiate OAuth flow"""
    params = {
        "client_id": AZURE_CLIENT_ID,
        "response_type": "code",
        "redirect_uri": AZURE_REDIRECT_URI,
        "scope": "openid email profile User.Read",
        "state": "random_state_string",
    }
    auth_url = f"{AZURE_AUTH_ENDPOINT}?{urlencode(params)}"
    return RedirectResponse(url=auth_url)

@app.get("/auth/callback")
async def auth_callback(request: Request, code: str = None, error: str = None, state: str = None):
    """Handle OAuth callback"""
    if error:
        logger.error(f"OAuth error: {error}")
        raise HTTPException(status_code=400, detail=f"OAuth error: {error}")
    
    if not code:
        raise HTTPException(status_code=400, detail="Missing authorization code")

    try:
        # Exchange code for token
        token_data = {
            "client_id": AZURE_CLIENT_ID,
            "client_secret": AZURE_CLIENT_SECRET,
            "code": code,
            "grant_type": "authorization_code",
            "redirect_uri": AZURE_REDIRECT_URI,
        }

        async with httpx.AsyncClient() as client:
            token_response = await client.post(AZURE_TOKEN_ENDPOINT, data=token_data)
            token_response.raise_for_status()
            tokens = token_response.json()

            # Get user info
            headers = {"Authorization": f"Bearer {tokens['access_token']}"}
            user_response = await client.get(AZURE_USERINFO_ENDPOINT, headers=headers)
            user_response.raise_for_status()
            user_data = user_response.json()

            # Check if user is authorized
            user_email = user_data.get("mail") or user_data.get("userPrincipalName")
            if user_email != AUTHORIZED_USER_EMAIL:
                raise HTTPException(
                    status_code=403, 
                    detail=f"Unauthorized user: {user_email}"
                )

            # Store user in session and generate access token for MCP
            user_info = {
                "id": user_data.get("id"),
                "name": user_data.get("displayName"),
                "email": user_email,
                "access_token": tokens["access_token"],
            }
            request.session["user"] = user_info

            # Generate authorization code for OAuth flow
            oauth_code = f"oauth-code-{hash(user_email)}-{hash(state or 'default')}"
            request.session["oauth_code"] = oauth_code
            request.session["oauth_user_email"] = user_email

            logger.info(f"User authenticated: {user_email}")
            
            # Check if this is an OAuth flow (from Claude Desktop)
            oauth_params = request.session.get("oauth_params")
            if oauth_params and state and state.startswith("oauth_flow_"):
                # For OAuth flow, redirect back to Claude Desktop
                original_state = state.replace("oauth_flow_", "") if state.startswith("oauth_flow_") else ""
                redirect_uri = oauth_params.get("redirect_uri")
                
                if redirect_uri and redirect_uri != "urn:ietf:wg:oauth:2.0:oob":
                    # Redirect to Claude Desktop callback
                    callback_url = f"{redirect_uri}?code={oauth_code}&state={original_state}"
                    return RedirectResponse(url=callback_url)
                else:
                    # Out-of-band flow - show code to user
                    return HTMLResponse(f"""
                    <html>
                        <body>
                            <h1>✅ Authentication Successful</h1>
                            <p>Authorization Code: <code>{oauth_code}</code></p>
                            <p>Please copy this code and paste it into Claude Desktop.</p>
                        </body>
                    </html>
                    """)
            else:
                # Regular web login
                return RedirectResponse(url="/")

    except Exception as e:
        logger.error(f"Authentication error: {e}")
        raise HTTPException(status_code=500, detail="Authentication failed")

@app.get("/auth/logout")
async def logout(request: Request):
    """Logout user"""
    request.session.clear()
    return RedirectResponse(url="/")

@app.get("/auth/status")
async def auth_status(request: Request):
    """Check authentication status"""
    user = get_current_user(request)
    return {
        "authenticated": user is not None,
        "user": user
    }

# OAuth 2.0 Discovery Endpoints for Claude Desktop
@app.get("/.well-known/oauth-authorization-server")
async def oauth_authorization_server():
    """OAuth 2.0 Authorization Server Metadata"""
    return {
        "issuer": SERVER_DOMAIN,
        "authorization_endpoint": f"{SERVER_DOMAIN}/oauth/authorize",
        "token_endpoint": f"{SERVER_DOMAIN}/oauth/token",
        "userinfo_endpoint": f"{SERVER_DOMAIN}/oauth/userinfo",
        "registration_endpoint": f"{SERVER_DOMAIN}/register",
        "introspection_endpoint": f"{SERVER_DOMAIN}/oauth/introspect",
        "revocation_endpoint": f"{SERVER_DOMAIN}/oauth/revoke",
        "response_types_supported": ["code"],
        "grant_types_supported": ["authorization_code", "client_credentials"],
        "scopes_supported": ["openid", "profile", "email", "mcp"],
        "token_endpoint_auth_methods_supported": ["client_secret_basic", "client_secret_post"],
        "code_challenge_methods_supported": ["S256"],
        "subject_types_supported": ["public"]
    }

@app.get("/.well-known/oauth-protected-resource")
async def oauth_protected_resource():
    """OAuth 2.0 Protected Resource Metadata"""
    return {
        "resource": f"{SERVER_DOMAIN}/api/mcp",
        "authorization_servers": [SERVER_DOMAIN],
        "scopes_supported": ["mcp", "openid"],
        "bearer_methods_supported": ["header", "query"],
        "resource_documentation": SERVER_DOMAIN
    }

@app.post("/register")
async def register_client():
    """OAuth 2.0 Dynamic Client Registration"""
    # For Claude Desktop, we'll return a pre-configured client
    return {
        "client_id": "claude-desktop",
        "client_secret": "claude-desktop-secret",
        "client_name": "Claude Desktop",
        "redirect_uris": ["https://claude.ai/api/mcp/auth_callback", "urn:ietf:wg:oauth:2.0:oob"],
        "grant_types": ["authorization_code"],
        "response_types": ["code"],
        "scope": "openid profile email mcp",
        "token_endpoint_auth_method": "client_secret_post"
    }

# OAuth endpoints for Claude Desktop
@app.get("/oauth/authorize")
async def oauth_authorize(
    request: Request,
    response_type: str = None,
    client_id: str = None,
    redirect_uri: str = None,
    scope: str = None,
    state: str = None,
    code_challenge: str = None,
    code_challenge_method: str = None
):
    """OAuth 2.0 Authorization Endpoint"""
    logger.info(f"OAuth authorize request - client_id: {client_id}, redirect_uri: {redirect_uri}")
    
    # Store OAuth parameters in session for later use
    request.session["oauth_params"] = {
        "response_type": response_type,
        "client_id": client_id,
        "redirect_uri": redirect_uri,
        "scope": scope,
        "state": state,
        "code_challenge": code_challenge,
        "code_challenge_method": code_challenge_method
    }
    
    # Redirect to Azure Entra ID for authentication
    params = {
        "client_id": AZURE_CLIENT_ID,
        "response_type": "code",
        "redirect_uri": AZURE_REDIRECT_URI,
        "scope": "openid email profile User.Read",
        "state": f"oauth_flow_{state}" if state else "oauth_flow",
    }
    auth_url = f"{AZURE_AUTH_ENDPOINT}?{urlencode(params)}"
    return RedirectResponse(url=auth_url)

@app.post("/oauth/token")
async def oauth_token_endpoint(request: Request):
    """OAuth 2.0 Token Endpoint for Claude Desktop"""
    # Get form data
    form_data = await request.form()
    grant_type = form_data.get("grant_type")
    code = form_data.get("code")
    client_id = form_data.get("client_id")
    client_secret = form_data.get("client_secret")
    
    logger.info(f"OAuth token request - grant_type: {grant_type}, client_id: {client_id}, code: {code}")
    
    if grant_type == "authorization_code":
        # Validate client credentials
        if client_id != "claude-desktop" or client_secret != "claude-desktop-secret":
            raise HTTPException(status_code=401, detail="Invalid client credentials")
        
        # In a real implementation, you'd validate the authorization code
        # For now, generate a valid token
        access_token = f"mcp-token-{AUTHORIZED_USER_EMAIL}-{hash(code)}"
        return {
            "access_token": access_token,
            "token_type": "Bearer", 
            "expires_in": 3600,
            "scope": "mcp openid profile email",
            "id_token": "dummy-id-token"
        }
    
    raise HTTPException(status_code=400, detail="Unsupported grant type")

# Keep the old endpoint for compatibility
@app.post("/auth/token")
async def legacy_token_endpoint(request: Request):
    """Legacy OAuth 2.0 Token Endpoint"""
    return await oauth_token_endpoint(request)

@app.get("/oauth/userinfo")
async def oauth_userinfo_endpoint(request: Request):
    """OAuth 2.0 UserInfo Endpoint"""
    # Check for Bearer token in Authorization header
    auth_header = request.headers.get("authorization", "")
    if not auth_header.startswith("Bearer "):
        raise HTTPException(status_code=401, detail="Invalid token")
    
    # In a real implementation, you'd validate the token
    # Return user info for the authorized user
    return {
        "sub": AUTHORIZED_USER_EMAIL,
        "email": AUTHORIZED_USER_EMAIL,
        "name": "Authorized User",
        "preferred_username": "user"
    }

@app.post("/oauth/introspect")
async def oauth_introspect_endpoint(request: Request):
    """OAuth 2.0 Token Introspection"""
    return {
        "active": True,
        "scope": "mcp openid profile email",
        "client_id": "claude-desktop",
        "username": AUTHORIZED_USER_EMAIL,
        "exp": 1735689600  # Example expiration timestamp
    }

# Legacy endpoints for compatibility
@app.get("/auth/userinfo")
async def userinfo_endpoint(request: Request):
    """Legacy OAuth 2.0 UserInfo Endpoint"""
    return await oauth_userinfo_endpoint(request)

@app.post("/auth/introspect")
async def introspect_endpoint(request: Request):
    """Legacy OAuth 2.0 Token Introspection"""
    return await oauth_introspect_endpoint(request)

# MCP Tools - Example implementations
@mcp.tool()
def get_weather(city: str) -> str:
    """Get the weather for a city (mock implementation)"""
    return f"The weather in {city} is sunny and 22°C"

@mcp.tool()
def calculate(expression: str) -> str:
    """Calculate a mathematical expression safely"""
    try:
        # Simple safe calculator
        allowed_chars = "0123456789+-*/.() "
        if not all(c in allowed_chars for c in expression):
            return "Error: Invalid characters in expression"
        
        result = eval(expression)
        return str(result)
    except Exception as e:
        return f"Error: {e}"

@mcp.tool()
def get_user_info() -> str:
    """Get current authenticated user information"""
    # Note: In a real implementation, you'd need to pass the request context
    # This is a simplified version for demonstration
    return f"Authenticated user: {AUTHORIZED_USER_EMAIL}"

# Authentication middleware for MCP endpoints
@app.middleware("http")
async def check_mcp_auth(request: Request, call_next):
    """Check authentication for MCP endpoints"""
    # Check if request is for MCP endpoints
    if request.url.path.startswith("/api/mcp"):
        logger.info(f"MCP endpoint access attempt: {request.url.path}")
        
        # Check for Bearer token first (Claude Desktop)
        auth_header = request.headers.get("authorization", "")
        if auth_header.startswith("Bearer "):
            token = auth_header.split(" ")[1]
            # Simple token validation - in production, you'd validate properly
            if token.startswith("mcp-token-"):
                logger.info(f"Authenticated MCP access via Bearer token")
                response = await call_next(request)
                return response
        
        # Check if user is authenticated via session (web browser)
        try:
            user = request.session.get("user") if hasattr(request, 'session') else None
            claude_authorized = request.session.get("claude_authorized") if hasattr(request, 'session') else None
            
            if not user and not claude_authorized:
                logger.warning("Unauthenticated MCP access attempt")
                return JSONResponse(
                    status_code=401,
                    content={
                        "error": "Authentication required",
                        "message": f"Please authenticate at {SERVER_DOMAIN}/auth/login before accessing MCP endpoints",
                        "auth_url": f"{SERVER_DOMAIN}/auth/login"
                    }
                )
            
            logger.info(f"Authenticated MCP access by user: {user.get('email') if user else 'claude-desktop'}")
        except Exception as e:
            logger.error(f"Session check error: {e}")
            return JSONResponse(
                status_code=401,
                content={
                    "error": "Authentication required",
                    "message": f"Please authenticate at {SERVER_DOMAIN}/auth/login before accessing MCP endpoints",
                    "auth_url": f"{SERVER_DOMAIN}/auth/login"
                }
            )
    
    response = await call_next(request)
    return response

# Mount MCP server to FastAPI
app.mount("/api", mcp_app)

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000, log_level="info")

🤞can you share your endpoints on here with sanitized info? I tried this with no success a while back when I was trying to get this to work. Hoping I am missing something you have. Greatly appreciated!

I will sanitise my server both (oauth and mcp ) over the weekend and share - my great frustration (Claude and I very nearly have gone out separate ways ) was the discovery end points - they were not as I expected - my nginx logs proved to be a life saver -

@cwise89 this is what i tested with `` import asyncio import logging import os from contextlib import asynccontextmanager from typing import Any, Dict, Optional from urllib.parse import urlencode, parse_qs, urlparse import httpx from fastapi import FastAPI, Request, HTTPException, Depends, status from fastapi.responses import HTMLResponse, RedirectResponse, JSONResponse from fastapi.middleware.cors import CORSMiddleware from starlette.middleware.sessions import SessionMiddleware from fastmcp import FastMCP logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) # Environment variables - Replace with your own values AZURE_CLIENT_ID = os.getenv("AZURE_CLIENT_ID", "your-azure-client-id") AZURE_CLIENT_SECRET = os.getenv("AZURE_CLIENT_SECRET", "your-azure-client-secret") AZURE_TENANT_ID = os.getenv("AZURE_TENANT_ID", "your-azure-tenant-id") AZURE_REDIRECT_URI = os.getenv("AZURE_REDIRECT_URI", "https://your-domain.com/auth/callback") AUTHORIZED_USER_EMAIL = os.getenv("AUTHORIZED_USER_EMAIL", "user@example.com") SESSION_SECRET_KEY = os.getenv("SESSION_SECRET_KEY", "generate-a-secure-random-key") # Azure OAuth endpoints AZURE_AUTH_ENDPOINT = f"https://login.microsoftonline.com/{AZURE_TENANT_ID}/oauth2/v2.0/authorize" AZURE_TOKEN_ENDPOINT = f"https://login.microsoftonline.com/{AZURE_TENANT_ID}/oauth2/v2.0/token" AZURE_USERINFO_ENDPOINT = "https://graph.microsoft.com/v1.0/me" # Replace with your domain SERVER_DOMAIN = os.getenv("SERVER_DOMAIN", "https://your-domain.com") # Create MCP server mcp = FastMCP("MCP OAuth Server") # Create the MCP's ASGI app mcp_app = mcp.http_app(path='/mcp') # Create FastAPI app with combined lifespan @asynccontextmanager async def combined_lifespan(app: FastAPI): async with mcp_app.lifespan(app): yield # Create FastAPI app app = FastAPI(title="MCP OAuth Server", lifespan=combined_lifespan) # Add session middleware app.add_middleware(SessionMiddleware, secret_key=SESSION_SECRET_KEY) # Add CORS middleware app.add_middleware( CORSMiddleware, allow_origins=["*"], allow_credentials=True, allow_methods=["*"], allow_headers=["*"], ) def get_current_user(request: Request) -> Optional[Dict[str, Any]]: """Get current authenticated user from session""" return request.session.get("user") def require_auth(request: Request) -> Dict[str, Any]: """Require authentication, raise 401 if not authenticated""" user = get_current_user(request) if not user: raise HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="Authentication required" ) return user @app.get("/") async def root(request: Request): """Root endpoint - show login status""" user = get_current_user(request) if user: return HTMLResponse(f""" <html> <body> <h1>MCP OAuth Server</h1> <p>Authenticated as: {user.get('email', 'Unknown')}</p> <p>User: {user.get('name', 'Unknown')}</p> <a href="/auth/logout">Logout</a> <br><br> <h2>MCP Server Status</h2> <p>MCP Server is running and ready to accept connections.</p> <p>WebSocket endpoint: <code>ws://localhost:8000/ws</code></p> </body> </html> """) else: return HTMLResponse(f""" <html> <body> <h1>MCP OAuth Server</h1> <p>Please authenticate to continue.</p> <a href="/auth/login">Login with Microsoft</a> </body> </html> """) @app.get("/auth/login") async def login(): """Initiate OAuth flow""" params = { "client_id": AZURE_CLIENT_ID, "response_type": "code", "redirect_uri": AZURE_REDIRECT_URI, "scope": "openid email profile User.Read", "state": "random_state_string", } auth_url = f"{AZURE_AUTH_ENDPOINT}?{urlencode(params)}" return RedirectResponse(url=auth_url) @app.get("/auth/callback") async def auth_callback(request: Request, code: str = None, error: str = None, state: str = None): """Handle OAuth callback""" if error: logger.error(f"OAuth error: {error}") raise HTTPException(status_code=400, detail=f"OAuth error: {error}") if not code: raise HTTPException(status_code=400, detail="Missing authorization code") try: # Exchange code for token token_data = { "client_id": AZURE_CLIENT_ID, "client_secret": AZURE_CLIENT_SECRET, "code": code, "grant_type": "authorization_code", "redirect_uri": AZURE_REDIRECT_URI, } async with httpx.AsyncClient() as client: token_response = await client.post(AZURE_TOKEN_ENDPOINT, data=token_data) token_response.raise_for_status() tokens = token_response.json() # Get user info headers = {"Authorization": f"Bearer {tokens['access_token']}"} user_response = await client.get(AZURE_USERINFO_ENDPOINT, headers=headers) user_response.raise_for_status() user_data = user_response.json() # Check if user is authorized user_email = user_data.get("mail") or user_data.get("userPrincipalName") if user_email != AUTHORIZED_USER_EMAIL: raise HTTPException( status_code=403, detail=f"Unauthorized user: {user_email}" ) # Store user in session and generate access token for MCP user_info = { "id": user_data.get("id"), "name": user_data.get("displayName"), "email": user_email, "access_token": tokens["access_token"], } request.session["user"] = user_info # Generate authorization code for OAuth flow oauth_code = f"oauth-code-{hash(user_email)}-{hash(state or 'default')}" request.session["oauth_code"] = oauth_code request.session["oauth_user_email"] = user_email logger.info(f"User authenticated: {user_email}") # Check if this is an OAuth flow (from Claude Desktop) oauth_params = request.session.get("oauth_params") if oauth_params and state and state.startswith("oauth_flow_"): # For OAuth flow, redirect back to Claude Desktop original_state = state.replace("oauth_flow_", "") if state.startswith("oauth_flow_") else "" redirect_uri = oauth_params.get("redirect_uri") if redirect_uri and redirect_uri != "urn:ietf:wg:oauth:2.0:oob": # Redirect to Claude Desktop callback callback_url = f"{redirect_uri}?code={oauth_code}&state={original_state}" return RedirectResponse(url=callback_url) else: # Out-of-band flow - show code to user return HTMLResponse(f""" <html> <body> <h1>✅ Authentication Successful</h1> <p>Authorization Code: <code>{oauth_code}</code></p> <p>Please copy this code and paste it into Claude Desktop.</p> </body> </html> """) else: # Regular web login return RedirectResponse(url="/") except Exception as e: logger.error(f"Authentication error: {e}") raise HTTPException(status_code=500, detail="Authentication failed") @app.get("/auth/logout") async def logout(request: Request): """Logout user""" request.session.clear() return RedirectResponse(url="/") @app.get("/auth/status") async def auth_status(request: Request): """Check authentication status""" user = get_current_user(request) return { "authenticated": user is not None, "user": user } # OAuth 2.0 Discovery Endpoints for Claude Desktop @app.get("/.well-known/oauth-authorization-server") async def oauth_authorization_server(): """OAuth 2.0 Authorization Server Metadata""" return { "issuer": SERVER_DOMAIN, "authorization_endpoint": f"{SERVER_DOMAIN}/oauth/authorize", "token_endpoint": f"{SERVER_DOMAIN}/oauth/token", "userinfo_endpoint": f"{SERVER_DOMAIN}/oauth/userinfo", "registration_endpoint": f"{SERVER_DOMAIN}/register", "introspection_endpoint": f"{SERVER_DOMAIN}/oauth/introspect", "revocation_endpoint": f"{SERVER_DOMAIN}/oauth/revoke", "response_types_supported": ["code"], "grant_types_supported": ["authorization_code", "client_credentials"], "scopes_supported": ["openid", "profile", "email", "mcp"], "token_endpoint_auth_methods_supported": ["client_secret_basic", "client_secret_post"], "code_challenge_methods_supported": ["S256"], "subject_types_supported": ["public"] } @app.get("/.well-known/oauth-protected-resource") async def oauth_protected_resource(): """OAuth 2.0 Protected Resource Metadata""" return { "resource": f"{SERVER_DOMAIN}/api/mcp", "authorization_servers": [SERVER_DOMAIN], "scopes_supported": ["mcp", "openid"], "bearer_methods_supported": ["header", "query"], "resource_documentation": SERVER_DOMAIN } @app.post("/register") async def register_client(): """OAuth 2.0 Dynamic Client Registration""" # For Claude Desktop, we'll return a pre-configured client return { "client_id": "claude-desktop", "client_secret": "claude-desktop-secret", "client_name": "Claude Desktop", "redirect_uris": ["https://claude.ai/api/mcp/auth_callback", "urn:ietf:wg:oauth:2.0:oob"], "grant_types": ["authorization_code"], "response_types": ["code"], "scope": "openid profile email mcp", "token_endpoint_auth_method": "client_secret_post" } # OAuth endpoints for Claude Desktop @app.get("/oauth/authorize") async def oauth_authorize( request: Request, response_type: str = None, client_id: str = None, redirect_uri: str = None, scope: str = None, state: str = None, code_challenge: str = None, code_challenge_method: str = None ): """OAuth 2.0 Authorization Endpoint""" logger.info(f"OAuth authorize request - client_id: {client_id}, redirect_uri: {redirect_uri}") # Store OAuth parameters in session for later use request.session["oauth_params"] = { "response_type": response_type, "client_id": client_id, "redirect_uri": redirect_uri, "scope": scope, "state": state, "code_challenge": code_challenge, "code_challenge_method": code_challenge_method } # Redirect to Azure Entra ID for authentication params = { "client_id": AZURE_CLIENT_ID, "response_type": "code", "redirect_uri": AZURE_REDIRECT_URI, "scope": "openid email profile User.Read", "state": f"oauth_flow_{state}" if state else "oauth_flow", } auth_url = f"{AZURE_AUTH_ENDPOINT}?{urlencode(params)}" return RedirectResponse(url=auth_url) @app.post("/oauth/token") async def oauth_token_endpoint(request: Request): """OAuth 2.0 Token Endpoint for Claude Desktop""" # Get form data form_data = await request.form() grant_type = form_data.get("grant_type") code = form_data.get("code") client_id = form_data.get("client_id") client_secret = form_data.get("client_secret") logger.info(f"OAuth token request - grant_type: {grant_type}, client_id: {client_id}, code: {code}") if grant_type == "authorization_code": # Validate client credentials if client_id != "claude-desktop" or client_secret != "claude-desktop-secret": raise HTTPException(status_code=401, detail="Invalid client credentials") # In a real implementation, you'd validate the authorization code # For now, generate a valid token access_token = f"mcp-token-{AUTHORIZED_USER_EMAIL}-{hash(code)}" return { "access_token": access_token, "token_type": "Bearer", "expires_in": 3600, "scope": "mcp openid profile email", "id_token": "dummy-id-token" } raise HTTPException(status_code=400, detail="Unsupported grant type") # Keep the old endpoint for compatibility @app.post("/auth/token") async def legacy_token_endpoint(request: Request): """Legacy OAuth 2.0 Token Endpoint""" return await oauth_token_endpoint(request) @app.get("/oauth/userinfo") async def oauth_userinfo_endpoint(request: Request): """OAuth 2.0 UserInfo Endpoint""" # Check for Bearer token in Authorization header auth_header = request.headers.get("authorization", "") if not auth_header.startswith("Bearer "): raise HTTPException(status_code=401, detail="Invalid token") # In a real implementation, you'd validate the token # Return user info for the authorized user return { "sub": AUTHORIZED_USER_EMAIL, "email": AUTHORIZED_USER_EMAIL, "name": "Authorized User", "preferred_username": "user" } @app.post("/oauth/introspect") async def oauth_introspect_endpoint(request: Request): """OAuth 2.0 Token Introspection""" return { "active": True, "scope": "mcp openid profile email", "client_id": "claude-desktop", "username": AUTHORIZED_USER_EMAIL, "exp": 1735689600 # Example expiration timestamp } # Legacy endpoints for compatibility @app.get("/auth/userinfo") async def userinfo_endpoint(request: Request): """Legacy OAuth 2.0 UserInfo Endpoint""" return await oauth_userinfo_endpoint(request) @app.post("/auth/introspect") async def introspect_endpoint(request: Request): """Legacy OAuth 2.0 Token Introspection""" return await oauth_introspect_endpoint(request) # MCP Tools - Example implementations @mcp.tool() def get_weather(city: str) -> str: """Get the weather for a city (mock implementation)""" return f"The weather in {city} is sunny and 22°C" @mcp.tool() def calculate(expression: str) -> str: """Calculate a mathematical expression safely""" try: # Simple safe calculator allowed_chars = "0123456789+-*/.() " if not all(c in allowed_chars for c in expression): return "Error: Invalid characters in expression" result = eval(expression) return str(result) except Exception as e: return f"Error: {e}" @mcp.tool() def get_user_info() -> str: """Get current authenticated user information""" # Note: In a real implementation, you'd need to pass the request context # This is a simplified version for demonstration return f"Authenticated user: {AUTHORIZED_USER_EMAIL}" # Authentication middleware for MCP endpoints @app.middleware("http") async def check_mcp_auth(request: Request, call_next): """Check authentication for MCP endpoints""" # Check if request is for MCP endpoints if request.url.path.startswith("/api/mcp"): logger.info(f"MCP endpoint access attempt: {request.url.path}") # Check for Bearer token first (Claude Desktop) auth_header = request.headers.get("authorization", "") if auth_header.startswith("Bearer "): token = auth_header.split(" ")[1] # Simple token validation - in production, you'd validate properly if token.startswith("mcp-token-"): logger.info(f"Authenticated MCP access via Bearer token") response = await call_next(request) return response # Check if user is authenticated via session (web browser) try: user = request.session.get("user") if hasattr(request, 'session') else None claude_authorized = request.session.get("claude_authorized") if hasattr(request, 'session') else None if not user and not claude_authorized: logger.warning("Unauthenticated MCP access attempt") return JSONResponse( status_code=401, content={ "error": "Authentication required", "message": f"Please authenticate at {SERVER_DOMAIN}/auth/login before accessing MCP endpoints", "auth_url": f"{SERVER_DOMAIN}/auth/login" } ) logger.info(f"Authenticated MCP access by user: {user.get('email') if user else 'claude-desktop'}") except Exception as e: logger.error(f"Session check error: {e}") return JSONResponse( status_code=401, content={ "error": "Authentication required", "message": f"Please authenticate at {SERVER_DOMAIN}/auth/login before accessing MCP endpoints", "auth_url": f"{SERVER_DOMAIN}/auth/login" } ) response = await call_next(request) return response # Mount MCP server to FastAPI app.mount("/api", mcp_app) if __name__ == "__main__": import uvicorn uvicorn.run(app, host="0.0.0.0", port=8000, log_level="info") ``

This worked! Thank you so much!!

Closing this because it's related to Claude Desktop, this repo is for Claude Code!

This issue has been automatically locked since it was closed and has not had any activity for 7 days. If you're experiencing a similar issue, please file a new issue and reference this one if it's relevant.