FastMCP v2.2.7 is out, and with it comes support for adding authentication to FastMCP servers. What this means is that when a MCP client attempts to connect to the MCP server, the standard OAuth Authorization Code flow may be used to authenticate the client before granting them access to the MCP server’s tools and resources. Since Pangea AuthN supports acting as an OAuth server, in this post we’ll be going over how to configure a FastMCP server to authenticate MCP clients with the Pangea AuthN service.
Create a Pangea account at https://pangea.cloud/signup. During the account creation process, an organization (top-level group) and project (individual app) will be created as well. On the "Get started with a common service" dialog, just click on the Skip button to get redirected to the developer console.
In the developer console, there will be a list of services in the left hand panel. Click the AuthN service to enable it.
In the modal, there will be a prompt to create a new Pangea API token or to extend an existing one. Choose Create a new token and click on Done.
In the left hand panel, click on OAuth Server, then navigate to the Scopes tab. We’ll create a new scope to represent one’s permission to authenticate with the MCP server.
To add a custom scope value, click the + Scope button on the right. In the Create Scope dialog, provide the new scope value details in the following fields:
Name: Define the scope value. Note this down for later. A good sample one could be “mcp”.
Display Name: Provide a recognizable name that will appear in the Display Name column in the scopes list.
Description: Explain what this scope value represents. For example, describe the permissions granted with this scope value.
Consent Required: Check this option to require explicit user approval for adding this scope value to the access token. This setting may remain unchecked for the purposes of this example.
Navigate back to the Clients tab, then click on the + OAuth Client button on the right to begin creating a new OAuth client.
Name: Assign a recognizable name to your client as it will appear in the list of clients in the OAuth Server settings. This name may be updated at any time.
Grant Type: must be Authorization Code.
Response Types: only Code is required.
Allowed Redirect URIs: enter http://localhost:8000/pangea/callback. Note that for a production MCP server, this should use the remote address of the server (e.g. https://mcp.example.org/pangea/callback) instead of a localhost address.
Allowed Scopes & Default Scopes: add the scope that was created earlier (e.g. “mcp”).
Note down the Client ID and Client Secret for later. Each of these will need to be set as environment variables
PANGEA_AUTHN_OAUTH_CLIENT_IDandPANGEA_AUTHN_OAUTH_CLIENT_SECRETrespectively.
Now let’s move on to the code. A typical definition of a FastMCP server might look like this:
from fastmcp import FastMCP
mcp = FastMCP(name="My MCP Server")
To add authentication, we’ll first need to implement an OAuthAuthorizationServerProvider:
import time
from secrets import token_hex, token_urlsafe
from typing import override
import httpx
from fastmcp.server.auth.auth import ClientRegistrationOptions, OAuthProvider, RevocationOptions
from mcp.server.auth.provider import (
AccessToken,
AuthorizationCode,
AuthorizationParams,
RefreshToken,
construct_redirect_uri,
)
from mcp.shared.auth import OAuthClientInformationFull, OAuthToken
from pydantic import AnyHttpUrl
from starlette.exceptions import HTTPException
MCP_SCOPE = "mcp"
class PangeaOAuthProvider(OAuthProvider):
def __init__(
self,
*,
mcp_issuer_url: str,
pangea_issuer_url: str,
client_id: str,
client_secret: str,
service_documentation_url: AnyHttpUrl | str | None = None,
client_registration_options: ClientRegistrationOptions | None = None,
revocation_options: RevocationOptions | None = None,
required_scopes: list[str] | None = None,
) -> None:
"""
Args:
mcp_issuer_url: URL of the MCP server.
pangea_issuer_url: Issuer URL of the Pangea AuthN project.
client_id: Pangea AuthN OAuth client ID.
client_secret: Pangea AuthN OAuth client secret.
"""
super().__init__(
issuer_url=mcp_issuer_url,
service_documentation_url=service_documentation_url,
client_registration_options=client_registration_options,
revocation_options=revocation_options,
required_scopes=required_scopes,
)
self.mcp_issuer_url = mcp_issuer_url
self.pangea_issuer_url = pangea_issuer_url
self.client_id = client_id
self.client_secret = client_secret
self.authorize_url = f"{self.pangea_issuer_url}/v2/oauth/authorize"
self.token_url = f"{self.pangea_issuer_url}/v2/oauth/token"
self.registration_endpoint = f"{self.pangea_issuer_url}/v2/oauth/clients/register"
self.auth_codes: dict[str, AuthorizationCode] = {}
self.tokens: dict[str, AccessToken] = {}
self.state_mapping: dict[str, dict[str, str]] = {}
# MCP token -> Pangea token.
self.token_mapping: dict[str, str] = {}
# In production, these should be stored in a database.
self.clients: dict[str, OAuthClientInformationFull] = {}
@override
async def get_client(self, client_id: str) -> OAuthClientInformationFull | None:
"""Retrieve an OAuth client by its ID."""
return self.clients.get(client_id)
@override
async def register_client(self, client_info: OAuthClientInformationFull) -> None:
"""Register a new OAuth client."""
self.clients[client_info.client_id] = client_info
@override
async def authorize(self, client: OAuthClientInformationFull, params: AuthorizationParams) -> str:
"""Generate an authorization URL for Pangea AuthN OAuth flow."""
state = params.state or token_urlsafe(32)
self.state_mapping[state] = {
"client_id": client.client_id,
"code_challenge": params.code_challenge,
"redirect_uri_provided_explicitly": str(params.redirect_uri_provided_explicitly),
"redirect_uri": str(params.redirect_uri),
}
external_params = {
"client_id": self.client_id,
"redirect_uri": f"{self.mcp_issuer_url}/pangea/callback",
"response_type": "code",
"state": state,
"scope": " ".join(params.scopes or []),
}
return construct_redirect_uri(self.authorize_url, **external_params)
@override
async def load_authorization_code(
self, client: OAuthClientInformationFull, authorization_code: str
) -> AuthorizationCode | None:
"""Retrieve an authorization code."""
code = self.auth_codes.get(authorization_code)
if code and code.client_id == client.client_id:
return code
return None
@override
async def exchange_authorization_code(
self, client: OAuthClientInformationFull, authorization_code: AuthorizationCode
) -> OAuthToken:
"""Exchange an authorization code for a token."""
if authorization_code.code not in self.auth_codes:
raise ValueError("Invalid authorization code")
mcp_token = f"mcp_{token_hex(32)}"
self.tokens[mcp_token] = AccessToken(
token=mcp_token,
client_id=client.client_id,
scopes=authorization_code.scopes,
expires_at=int(time.time()) + 3600,
)
pangea_token = next(
(token for token, data in self.tokens.items() if data.client_id == client.client_id),
None,
)
if pangea_token:
self.token_mapping[mcp_token] = pangea_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),
)
@override
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
if access_token.expires_at and access_token.expires_at < time.time():
del self.tokens[token]
return None
return access_token
@override
async def revoke_token(self, token: AccessToken | RefreshToken) -> None:
"""Revoke a token."""
if token.token in self.tokens:
del self.tokens[token.token]
@override
async def load_refresh_token(self, client: OAuthClientInformationFull, refresh_token: str) -> RefreshToken | None:
raise NotImplementedError()
@override
async def exchange_refresh_token(
self, client: OAuthClientInformationFull, refresh_token: RefreshToken, scopes: list[str]
) -> OAuthToken:
raise NotImplementedError()
async def handle_pangea_callback(self, code: str, state: str) -> str:
"""Handle Pangea AuthN OAuth callback."""
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"]
async with httpx.AsyncClient() as client:
response = await client.post(
self.token_url,
data={
"code": code,
"grant_type": "authorization_code",
"redirect_uri": f"{self.mcp_issuer_url}/pangea/callback",
},
headers={"Accept": "application/json"},
auth=httpx.BasicAuth(username=self.client_id, password=self.client_secret),
)
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"]))
pangea_token = data["access_token"]
# Create MCP authorization code
new_code = f"mcp_{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=[MCP_SCOPE],
code_challenge=code_challenge,
)
self.auth_codes[new_code] = auth_code
# Store Pangea token; the MCP token will be mapped to this later
self.tokens[pangea_token] = AccessToken(
token=pangea_token,
client_id=client_id,
scopes=[],
expires_at=None,
)
del self.state_mapping[state]
return construct_redirect_uri(redirect_uri, code=new_code, state=state)
Then create an instance of the provider and configure FastMCP to use it:
import os
from fastmcp import FastMCP
from fastmcp.server.auth.auth import ClientRegistrationOptions
MCP_SCOPE = "mcp"
# In production, this would be the remote URL of the MCP server.
mcp_issuer_url = "http://localhost:8000"
oauth_provider = PangeaOAuthProvider(
mcp_issuer_url=mcp_issuer_url,
pangea_issuer_url="https://pdn-ater32x5bh6wgxisqyjepu33couemzzz.login.aws.us.pangea.cloud",
client_id=os.getenv("PANGEA_AUTHN_OAUTH_CLIENT_ID", “”),
client_secret=os.getenv("PANGEA_AUTHN_OAUTH_CLIENT_SECRET", “”),
client_registration_options=ClientRegistrationOptions(
enabled=True, valid_scopes=[MCP_SCOPE], default_scopes=[MCP_SCOPE]
),
required_scopes=[MCP_SCOPE],
)
mcp = FastMCP(name="My MCP Server", auth=oauth_provider)
The pangea_issuer_url above should be replaced with the Hosted Login URL displayed on the AuthN Overview page.
Finally, set up the callback route that Pangea AuthN will redirect the user back to after they login:
from starlette.exceptions import HTTPException
from starlette.requests import Request
from starlette.responses import JSONResponse, RedirectResponse, Response
@mcp.custom_route("/pangea/callback", methods=["GET"])
async def pangea_callback_handler(request: Request) -> Response:
code = request.query_params.get("code")
state = request.query_params.get("state")
if not code or not state:
raise HTTPException(400, "Missing code or state parameter")
try:
redirect_uri = await oauth_provider.handle_pangea_callback(code, state)
return RedirectResponse(status_code=302, url=redirect_uri)
except HTTPException:
raise
except Exception:
return JSONResponse(
status_code=500,
content={
"error": "server_error",
"error_description": "Unexpected error",
},
)
With all of that in place, the FastMCP server should now be ready to authenticate users via Pangea AuthN. To learn more about Pangea AuthN’s many features—such as social login, single sign-on (SSO), and custom branding—check out the documentation.
