Understanding Authorization in MCP

pip install "mcp>=2.2" "httpx2>=2.5" "pyjwt[crypto]>=2.14"

import asyncio

import httpx2
import jwt

from mcp.server import MCPServer
from mcp.server.auth.provider import AccessToken, TokenVerifier
from mcp.server.auth.settings import AuthSettings

# The authorization server's issuer identifier, and where it publishes metadata
ISSUER = "https://auth.example.com"
METADATA_URL = "https://auth.example.com/.well-known/oauth-authorization-server"

# The URL clients connect to, which is also the audience of their tokens
MCP_SERVER_URL = "http://localhost:3000/mcp"


class JWTTokenVerifier(TokenVerifier):
    def __init__(self) -> None:
        metadata = httpx2.get(METADATA_URL).raise_for_status().json()
        if metadata["issuer"] != ISSUER:
            raise RuntimeError(f"Unexpected issuer: {metadata['issuer']}")
        self.jwks_client = jwt.PyJWKClient(metadata["jwks_uri"], timeout=5)

    async def verify_token(self, token: str) -> AccessToken | None:
        try:
            # PyJWKClient blocks while it fetches keys, so keep it off the event loop
            signing_key = await asyncio.to_thread(
                self.jwks_client.get_signing_key_from_jwt, token
            )
            claims = jwt.decode(
                token,
                signing_key,
                algorithms=["RS256"],
                issuer=ISSUER,
                audience=MCP_SERVER_URL,
                leeway=30,
                options={"require": ["exp"]},
            )
        except jwt.PyJWKClientConnectionError:
            # An unreachable JWKS endpoint is an outage, not an invalid token
            raise
        except jwt.PyJWTError:
            return None

        scope = claims.get("scope")

        return AccessToken(
            token=token,
            client_id=claims.get("client_id") or claims.get("azp") or "",
            scopes=scope.split() if isinstance(scope, str) else [],
            expires_at=int(claims["exp"]),
            # jwt.decode has checked that the token's audience is this server
            resource=MCP_SERVER_URL,
            subject=claims.get("sub"),
            claims=claims,
        )


mcp = MCPServer(
    "Math",
    token_verifier=JWTTokenVerifier(),
    # model_validate keeps ISSUER as it is. AnyHttpUrl(ISSUER) would add a trailing
    # slash, and the issuer is compared character by character.
    auth=AuthSettings.model_validate(
        {
            "issuer_url": ISSUER,
            "resource_server_url": MCP_SERVER_URL,
            "required_scopes": ["mcp:tools"],
            "validate_token_resource": True,
        }
    ),
)


@mcp.tool()
def add(a: float, b: float) -> float:
    """Add two numbers together."""
    return a + b


if __name__ == "__main__":
    mcp.run(transport="streamable-http", port=3000)

python server.py

<Project Sdk="Microsoft.NET.Sdk.Web">

  <PropertyGroup>
    <TargetFramework>net10.0</TargetFramework>
    <Nullable>enable</Nullable>
    <ImplicitUsings>enable</ImplicitUsings>
  </PropertyGroup>

  <ItemGroup>
    <PackageReference Include="Microsoft.AspNetCore.Authentication.JwtBearer" Version="10.0.14" />
    <PackageReference Include="ModelContextProtocol.AspNetCore" Version="2.4.0" />
  </ItemGroup>

</Project>

using System.ComponentModel;
using Microsoft.AspNetCore.Authentication.JwtBearer;
using Microsoft.IdentityModel.Tokens;
using ModelContextProtocol.Server;

// The authorization server's issuer identifier, and where it publishes metadata
var issuer = "https://auth.example.com";
var metadataUrl = "https://auth.example.com/.well-known/oauth-authorization-server";

// The URL clients connect to, which is also the audience of their tokens
var mcpServerUrl = "http://localhost:3000/mcp";

var serverOrigin = new Uri(mcpServerUrl).GetLeftPart(UriPartial.Authority);
var mcpPath = new Uri(mcpServerUrl).AbsolutePath.TrimEnd('/');
var resourceMetadataUrl = $"{serverOrigin}/.well-known/oauth-protected-resource{mcpPath}";

// Browser origins and host names that the server doesn't reject
string[] allowedOrigins = [serverOrigin];
var allowedHosts = "localhost;127.0.0.1;[::1]";

var builder = WebApplication.CreateBuilder(args);
builder.Configuration["AllowedHosts"] = allowedHosts;

builder.Services.AddAuthentication(JwtBearerDefaults.AuthenticationScheme)
.AddJwtBearer(options =>
{
    options.MetadataAddress = metadataUrl;
    options.BackchannelTimeout = TimeSpan.FromSeconds(5);
    options.RefreshInterval = TimeSpan.FromSeconds(30);
    options.MapInboundClaims = false;
    options.TokenValidationParameters = new TokenValidationParameters
    {
        // ValidIssuer alone would also accept the issuer named in the metadata
        IssuerValidatorUsingConfiguration = (tokenIssuer, _, _, configuration) =>
            tokenIssuer == issuer && configuration?.Issuer == issuer
                ? tokenIssuer
                : throw new SecurityTokenInvalidIssuerException("Unexpected issuer"),
        ValidAudience = mcpServerUrl,
        IgnoreTrailingSlashWhenValidatingAudience = false,
        ClockSkew = TimeSpan.FromSeconds(30),
        NameClaimType = "sub",
        ValidAlgorithms = [SecurityAlgorithms.RsaSha256],
    };
    options.Events = new JwtBearerEvents
    {
        // Point clients to the PRM document and name the scope they need
        OnChallenge = context =>
        {
            context.HandleResponse();
            var error = context.AuthenticateFailure is null ? "" : "error=\"invalid_token\", ";
            context.Response.StatusCode = StatusCodes.Status401Unauthorized;
            context.Response.Headers.WWWAuthenticate =
                $"Bearer {error}scope=\"mcp:tools\", resource_metadata=\"{resourceMetadataUrl}\"";
            return Task.CompletedTask;
        },
        // If the signing keys were never downloaded, this throws and the client gets a 500
        OnAuthenticationFailed = context => context.Options.ConfigurationManager!
            .GetConfigurationAsync(context.HttpContext.RequestAborted),
        // Tell clients which scope they need when a valid token lacks it
        OnForbidden = context =>
        {
            context.Response.Headers.WWWAuthenticate =
                $"Bearer error=\"insufficient_scope\", scope=\"mcp:tools\", resource_metadata=\"{resourceMetadataUrl}\"";
            return Task.CompletedTask;
        },
    };
})
.AddMcp(options =>
{
    options.ResourceMetadata = new()
    {
        Resource = mcpServerUrl,
        AuthorizationServers = { issuer },
        ScopesSupported = ["mcp:tools"],
    };
});

builder.Services.AddAuthorizationBuilder()
    .AddPolicy("mcp:tools", policy => policy
        .RequireAuthenticatedUser()
        .RequireAssertion(context => context.User.FindAll("scope")
            .SelectMany(claim => claim.Value.Split(' '))
            .Contains("mcp:tools")));

builder.Services.AddMcpServer()
    .WithHttpTransport()
    .WithTools<MathTools>();

var app = builder.Build();

// Reject browser requests from other origins, which protects against DNS rebinding
app.Use(async (context, next) =>
{
    var origin = context.Request.Headers.Origin.ToString();
    if (origin.Length > 0 && context.Request.Path.StartsWithSegments(mcpPath)
        && !allowedOrigins.Contains(origin))
    {
        context.Response.StatusCode = StatusCodes.Status403Forbidden;
        return;
    }
    await next();
});

app.UseAuthentication();
app.UseAuthorization();

app.MapMcp(mcpPath).RequireAuthorization("mcp:tools");

app.Run("http://localhost:3000");

[McpServerToolType]
public sealed class MathTools
{
    [McpServerTool, Description("Add two numbers together.")]
    public static double Add(double a, double b) => a + b;
}

dotnet run