MCP Authentication in Python: API Keys, JWT and OAuth

# Trust access tokens from this issuer, minted for this server's audience, and nothing else.
tokens = JwksAuth.from_discovery(issuer="http://127.0.0.1:9431", audience="api://billing")

server = MCPServer("billing", require_auth=True)
server.add_middleware(AuthMiddleware(tokens))

pip install "promptise>=1.3.3"
export OPENAI_API_KEY="sk-..."

from promptise.mcp.server import HasScope, MCPServer, RequestContext, ToolError

# A stand-in for your billing system. Amounts are in EUR.
INVOICES = {
    "INV-7": {"customer": "acme", "amount": 1200.0, "credited": 0.0},
    "INV-8": {"customer": "globex", "amount": 310.0, "credited": 0.0},
}


def add_billing_tools(server: MCPServer) -> None:
    """The same four tools, whichever way the server authenticates."""

    @server.tool(read_only_hint=True)
    async def whoami(ctx: RequestContext) -> dict:
        """Show the identity this server verified for the current call."""
        client = ctx.client
        return {
            "client_id": client.client_id,
            "roles": sorted(client.roles),
            "scopes": sorted(client.scopes),
        }

    @server.tool(read_only_hint=True)
    async def list_invoices() -> list[dict]:
        """List invoices with their amount and what has been credited."""
        return [{"invoice_id": iid, **inv} for iid, inv in INVOICES.items()]

    @server.tool(roles=["finance"], destructive_hint=True)
    async def issue_credit(invoice_id: str, amount: float, reason: str) -> dict:
        """Credit part of an invoice back to the customer. Finance only."""
        invoice = INVOICES.get(invoice_id)
        if invoice is None:
            raise ToolError(f"There is no invoice {invoice_id}.")
        invoice["credited"] += amount
        return {"invoice_id": invoice_id, "credited": invoice["credited"], "reason": reason}

    @server.tool(read_only_hint=True, guards=[HasScope("ledger:export")])
    async def export_ledger(month: str) -> dict:
        """Export one month's ledger as CSV. Needs the ledger:export scope."""
        rows = [f"{iid},{inv['amount']},{inv['credited']}" for iid, inv in INVOICES.items()]
        return {"month": month, "csv": "\n".join(["invoice_id,amount,credited", *rows])}

import json

from promptise.mcp.client import MCPClient

CALLS = [
    ("whoami", {}),
    ("issue_credit", {"invoice_id": "INV-7", "amount": 50, "reason": "late delivery"}),
    ("export_ledger", {"month": "2026-09"}),
]


def summary(result) -> str:
    """One line per call: the result, or the error code and message."""
    text = result.content[0].text
    if result.isError:
        error = json.loads(text)["error"]
        return f"{error['code']}: {error['message']}"
    return " ".join(text.split())[:100]


async def probe(url: str, label: str, **credentials) -> None:
    """Connect with the given credentials and try each tool once."""
    print(f"== {label}")
    async with MCPClient(url=url, **credentials) as client:
        for name, args in CALLS:
            print(f"{name:<14} {summary(await client.call_tool(name, args))}")

import os

from billing_tools import add_billing_tools
from promptise.mcp.server import APIKeyAuth, AuthMiddleware, MCPServer

# Each key names a client and its roles. Keep the keys themselves in a secret store.
keys = APIKeyAuth(
    keys={
        os.environ["BILLING_AGENT_KEY"]: {"client_id": "billing-agent", "roles": ["finance"]},
        os.environ["REPORTING_KEY"]: {"client_id": "reporting-job", "roles": ["reporting"]},
    }
)

server = MCPServer("billing", require_auth=True)
server.add_middleware(AuthMiddleware(keys))
add_billing_tools(server)

if __name__ == "__main__":
    server.run(transport="http", host="127.0.0.1", port=9432)

import asyncio
import os

from probe import probe

URL = "http://127.0.0.1:9432/mcp"


async def main():
    await probe(URL, "billing-agent (API key)", api_key=os.environ["BILLING_AGENT_KEY"])
    await probe(URL, "reporting-job (API key)", api_key=os.environ["REPORTING_KEY"])


asyncio.run(main())

export BILLING_AGENT_KEY="sk-billing-agent-local-only" REPORTING_KEY="sk-reporting-local-only"
python server_apikey.py      # in one terminal
python client_apikey.py      # in another

import os

from billing_tools import add_billing_tools
from promptise.mcp.server import AuthMiddleware, JWTAuth, MCPServer

jwt_auth = JWTAuth(secret=os.environ["BILLING_JWT_SECRET"], audience="api://billing")

server = MCPServer("billing", require_auth=True)
server.add_middleware(AuthMiddleware(jwt_auth))

# A built-in OAuth-style token endpoint at /auth/token, for development.
server.enable_token_endpoint(
    jwt_auth=jwt_auth,
    clients={
        "billing-agent": {
            "secret": os.environ["BILLING_AGENT_SECRET"],
            "roles": ["finance"],
            "claims": {"scope": "invoices:read ledger:export"},
            "expires_in": 900,
        },
        "reporting-job": {
            "secret": os.environ["REPORTING_SECRET"],
            "roles": ["reporting"],
            "claims": {"scope": "invoices:read"},
            "expires_in": 900,
        },
    },
)
add_billing_tools(server)

if __name__ == "__main__":
    server.run(transport="http", host="127.0.0.1", port=9433)

import asyncio
import os

import jwt

from promptise.mcp.client import MCPClient
from probe import probe

SERVER = "http://127.0.0.1:9433"


async def main():
    token = await MCPClient.fetch_token(
        f"{SERVER}/auth/token",
        client_id="billing-agent",
        client_secret=os.environ["BILLING_AGENT_SECRET"],
    )
    claims = jwt.decode(token, options={"verify_signature": False})  # for display only
    print("claims:", {k: claims[k] for k in ("sub", "aud", "roles", "scope")})
    await probe(f"{SERVER}/mcp", "billing-agent (JWT)", bearer_token=token)


asyncio.run(main())

export BILLING_JWT_SECRET="local-only-jwt-secret-0f3a9c"
export BILLING_AGENT_SECRET="agent-secret-local-only" REPORTING_SECRET="reporting-secret-local-only"
python server_jwt.py         # in one terminal
python client_jwt.py         # in another

"""The built-in token endpoint, called the way RFC 6749 describes."""

import os

import httpx

URL = "http://127.0.0.1:9433/auth/token"
CLIENT = ("reporting-job", os.environ["REPORTING_SECRET"])

# Form-encoded, with the client secret in HTTP Basic auth.
grant = httpx.post(URL, data={"grant_type": "client_credentials"}, auth=CLIENT)
print("client_credentials:", grant.status_code, sorted(grant.json()))
print("  Content-Type:", grant.headers["content-type"], " Cache-Control:", grant.headers.get("cache-control"))

refresh = {"grant_type": "refresh_token", "refresh_token": grant.json()["refresh_token"]}
print("refresh:", httpx.post(URL, data=refresh).status_code)
again = httpx.post(URL, data=refresh).json()
print("same refresh token again:", again["error"], "-", again["error_description"])

no_grant = httpx.post(URL, data={"scope": "invoices:read"}, auth=CLIENT).json()
print("form without grant_type:", no_grant["error"], "-", no_grant["error_description"])

"""A tiny OAuth 2.0 authorization server for local development. Never run it in production.

It does what Entra ID, Okta, Auth0 or Keycloak do for machine clients: checks a
client's ID and secret (the client_credentials grant), signs a short-lived RS256
access token, and publishes its discovery document and public keys.
"""

import base64
import hmac
import json
import time
import uuid
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from urllib.parse import parse_qs

import jwt
from cryptography.hazmat.primitives.asymmetric import rsa

ISSUER = "http://127.0.0.1:9431"
DEFAULT_AUDIENCE = "api://billing"
TOKEN_TTL = 600  # seconds

# Registered machine clients: their secret, app roles and the scopes they may request.
CLIENTS = {
    "billing-agent": {
        "secret": "agent-secret-for-local-dev",
        "roles": ["finance"],
        "scopes": ["invoices:read", "ledger:export"],
    },
    "reporting-job": {
        "secret": "reporting-secret-for-local-dev",
        "roles": ["reporting"],
        "scopes": ["invoices:read"],
    },
}
# People who sign in to your app. A real IdP issues their tokens after a login.
USERS = {"dana": {"roles": ["support"], "scopes": ["invoices:read"]}}

KEY = rsa.generate_private_key(public_exponent=65537, key_size=2048)
KID = uuid.uuid4().hex[:8]
JWK = {
    **json.loads(jwt.algorithms.RSAAlgorithm.to_jwk(KEY.public_key())),
    "kid": KID,
    "use": "sig",
    "alg": "RS256",
}


def sign(subject: str, audience: str, roles: list[str], scopes: list[str]) -> str:
    now = int(time.time())
    claims = {
        "iss": ISSUER,
        "sub": subject,
        "aud": audience,
        "iat": now,
        "exp": now + TOKEN_TTL,
        "jti": uuid.uuid4().hex,
        "roles": roles,
        "scope": " ".join(scopes),
    }
    return jwt.encode(claims, KEY, algorithm="RS256", headers={"kid": KID})


class Handler(BaseHTTPRequestHandler):
    def do_GET(self):
        if self.path == "/.well-known/openid-configuration":
            self.reply(200, {
                "issuer": ISSUER,
                "token_endpoint": f"{ISSUER}/oauth/token",
                "jwks_uri": f"{ISSUER}/jwks.json",
                "grant_types_supported": ["client_credentials"],
            })
        elif self.path == "/jwks.json":
            self.reply(200, {"keys": [JWK]})
        elif self.path.startswith("/dev/login?user="):
            # Stands in for your app's sign-in (the authorization code flow).
            user = self.path.split("=", 1)[1]
            if user not in USERS:
                return self.reply(404, {"error": "unknown user"})
            u = USERS[user]
            self.reply(200, {"access_token": sign(user, DEFAULT_AUDIENCE, u["roles"], u["scopes"])})
        else:
            self.reply(404, {"error": "not_found"})

    def do_POST(self):
        if self.path != "/oauth/token":
            return self.reply(404, {"error": "not_found"})
        content_type = self.headers.get("content-type", "").split(";")[0]
        if content_type != "application/x-www-form-urlencoded":
            hint = "Use application/x-www-form-urlencoded (RFC 6749)."
            return self.reply(400, {"error": "invalid_request", "error_description": hint})
        body = self.rfile.read(int(self.headers["content-length"])).decode()
        form = {key: values[0] for key, values in parse_qs(body).items()}
        client_id, secret = form.get("client_id"), form.get("client_secret")
        basic = self.headers.get("authorization", "")
        if basic.startswith("Basic "):
            client_id, secret = base64.b64decode(basic[6:]).decode().split(":", 1)
        client = CLIENTS.get(client_id or "")
        if client is None or not hmac.compare_digest(client["secret"], secret or ""):
            return self.reply(401, {"error": "invalid_client"})
        if form.get("grant_type") != "client_credentials":
            return self.reply(400, {"error": "unsupported_grant_type"})
        asked = form.get("scope", "").split() or client["scopes"]
        if not set(asked) <= set(client["scopes"]):
            return self.reply(400, {"error": "invalid_scope"})
        audience = form.get("resource") or DEFAULT_AUDIENCE  # RFC 8707 resource indicator
        token = sign(client_id, audience, client["roles"], asked)
        print(f"[oauth] {client_id}: token for {audience}, scope '{' '.join(asked)}'", flush=True)
        grant = {"access_token": token, "token_type": "Bearer", "expires_in": TOKEN_TTL}
        self.reply(200, grant | {"scope": " ".join(asked)})

    def reply(self, status, body):
        data = json.dumps(body).encode()
        self.send_response(status)
        self.send_header("content-type", "application/json")
        self.send_header("content-length", str(len(data)))
        self.end_headers()
        self.wfile.write(data)

    def log_message(self, *args):
        pass


if __name__ == "__main__":
    print(f"[oauth] issuer {ISSUER}, key {KID}", flush=True)
    ThreadingHTTPServer(("127.0.0.1", 9431), Handler).serve_forever()

from billing_tools import add_billing_tools
from promptise.mcp.server import AuthMiddleware, JwksAuth, MCPServer

# Trust access tokens from this issuer, minted for this server's audience, and nothing else.
tokens = JwksAuth.from_discovery(issuer="http://127.0.0.1:9431", audience="api://billing")

server = MCPServer("billing", require_auth=True)
server.add_middleware(AuthMiddleware(tokens))
add_billing_tools(server)

if __name__ == "__main__":
    server.run(transport="http", host="127.0.0.1", port=9434)

import time

import httpx

TOKEN_URL = "http://127.0.0.1:9431/oauth/token"


def client_credentials(
    client_id: str, secret: str, *, scope: str = "", resource: str = "api://billing"
) -> dict:
    """The OAuth 2.0 client_credentials grant: form-encoded, client secret in HTTP Basic auth."""
    form = {"grant_type": "client_credentials", "resource": resource}
    if scope:
        form["scope"] = scope
    response = httpx.post(TOKEN_URL, data=form, auth=(client_id, secret))
    response.raise_for_status()
    return response.json()


class ClientCredentialsToken:
    """Hands MCPClient a cached access token, and fetches a new one before it expires."""

    def __init__(self, client_id: str, secret: str, *, scope: str = "", margin: float = 60) -> None:
        self._args = (client_id, secret)
        self._scope = scope
        self._margin = margin
        self._token, self._expires_at = None, 0.0

    def __call__(self, force_refresh: bool = False) -> str:
        if force_refresh or self._token is None or time.time() > self._expires_at - self._margin:
            grant = client_credentials(*self._args, scope=self._scope)
            self._token, self._expires_at = grant["access_token"], time.time() + grant["expires_in"]
        return self._token

import asyncio

from client_credentials import ClientCredentialsToken
from probe import probe

URL = "http://127.0.0.1:9434/mcp"


async def main():
    agent = ClientCredentialsToken("billing-agent", "agent-secret-for-local-dev")
    reporting = ClientCredentialsToken("reporting-job", "reporting-secret-for-local-dev")
    await probe(URL, "billing-agent (OAuth)", bearer_token_provider=agent)
    await probe(URL, "reporting-job (OAuth)", bearer_token_provider=reporting)


asyncio.run(main())

python local_oauth.py        # terminal 1: the identity provider on port 9431
python server_oauth.py       # terminal 2: the MCP server on port 9434
python client_oauth.py       # terminal 3

import asyncio

import jwt

from promptise.mcp.client import MCPClient


async def main():
    token = await MCPClient.fetch_token(
        "http://127.0.0.1:9431/oauth/token",
        client_id="billing-agent",
        client_secret="agent-secret-for-local-dev",
    )
    claims = jwt.decode(token, options={"verify_signature": False})  # for display only
    print({k: claims[k] for k in ("iss", "sub", "aud", "roles", "scope")})


asyncio.run(main())

"""What each kind of failure looks like on the wire, as raw HTTP."""

import json

import httpx
import jwt
from cryptography.hazmat.primitives.asymmetric import rsa

from client_credentials import client_credentials

MCP = "http://127.0.0.1:9434/mcp"
HEADERS = {"accept": "application/json, text/event-stream", "content-type": "application/json"}
VERSION = "2025-06-18"
INITIALIZE = {
    "jsonrpc": "2.0", "id": 1, "method": "initialize",
    "params": {"protocolVersion": VERSION, "capabilities": {}, "clientInfo": {"name": "probe", "version": "1"}},
}


def bearer(token):
    return {"authorization": f"Bearer {token}"}


def show(label, response):
    challenge = response.headers.get("www-authenticate", "(none)")
    print(f"{label:<26} HTTP {response.status_code}  WWW-Authenticate: {challenge}  {response.text[:58]}")


def forged():
    """Claims to be billing-agent, but is signed with an attacker's key."""
    kid = httpx.get("http://127.0.0.1:9431/jwks.json").json()["keys"][0]["kid"]  # key ids are public
    claims = {"iss": "http://127.0.0.1:9431", "sub": "billing-agent", "aud": "api://billing", "exp": 4102444800}
    attacker_key = rsa.generate_private_key(public_exponent=65537, key_size=2048)
    return jwt.encode(claims, attacker_key, "RS256", headers={"kid": kid})


def call_tool(token, name, arguments):
    """An MCP session by hand: initialize, initialized, then one tools/call."""
    headers = HEADERS | bearer(token)
    opened = httpx.post(MCP, json=INITIALIZE, headers=headers)
    headers |= {"mcp-session-id": opened.headers["mcp-session-id"], "mcp-protocol-version": VERSION}
    httpx.post(MCP, json={"jsonrpc": "2.0", "method": "notifications/initialized"}, headers=headers)
    call = {"jsonrpc": "2.0", "id": 2, "method": "tools/call", "params": {"name": name, "arguments": arguments}}
    response = httpx.post(MCP, json=call, headers=headers)
    data = [line[5:] for line in response.text.splitlines() if line.startswith("data:")]
    result = json.loads(data[-1])["result"]
    code = json.loads(result["content"][0]["text"])["error"]["code"] if result["isError"] else "-"
    print(f"{'tools/call ' + name:<26} HTTP {response.status_code}  isError: {result['isError']}  error: {code}")


secret = "reporting-secret-for-local-dev"
reporting = client_credentials("reporting-job", secret)["access_token"]
payroll = client_credentials("reporting-job", secret, resource="api://payroll")["access_token"]

show("no credential", httpx.post(MCP, json=INITIALIZE, headers=HEADERS))
show("malformed token", httpx.post(MCP, json=INITIALIZE, headers=HEADERS | bearer("not-a-jwt")))
show("forged signature", httpx.post(MCP, json=INITIALIZE, headers=HEADERS | bearer(forged())))
show("token for api://payroll", httpx.post(MCP, json=INITIALIZE, headers=HEADERS | bearer(payroll)))
show("resource metadata (RFC 9728)", httpx.get("http://127.0.0.1:9434/.well-known/oauth-protected-resource"))
call_tool(reporting, "list_invoices", {})
call_tool(reporting, "issue_credit", {"invoice_id": "INV-7", "amount": 50, "reason": "late delivery"})

"""A 401 doesn't say why. Run the same token through the server in-process to find out."""

import asyncio
import json

from client_credentials import client_credentials
from promptise.mcp.server import TestClient
from server_oauth import server

secret = "reporting-secret-for-local-dev"
token = client_credentials("reporting-job", secret, resource="api://payroll")["access_token"]


async def main():
    client = TestClient(server, meta={"authorization": f"Bearer {token}"})
    result = await client.call_tool("whoami", {})
    print(json.loads(result[0].text)["error"]["message"])


asyncio.run(main())

"""The same agent, three ways to authenticate. Calls whoami directly: no model call."""

import asyncio
import os

from billing_identity import identity
from promptise import build_agent
from promptise.config import HTTPServerSpec
from promptise.mcp.client import MCPClient


async def whoami(spec: HTTPServerSpec, **agent_options) -> str:
    agent = await build_agent(model="openai:gpt-5-mini", servers={"billing": spec}, **agent_options)
    try:
        tool = next(tool for tool in agent.tools if tool.name == "whoami")
        return await tool.ainvoke({})
    finally:
        await agent.shutdown()


async def main():
    api_key = HTTPServerSpec(url="http://127.0.0.1:9432/mcp", api_key=os.environ["BILLING_AGENT_KEY"])
    print("api_key=           ", await whoami(api_key))
    print("api_key= + identity", await whoami(api_key, identity=identity))  # the key still wins

    token = await MCPClient.fetch_token(
        "http://127.0.0.1:9433/auth/token",
        client_id="billing-agent",
        client_secret=os.environ["BILLING_AGENT_SECRET"],
    )
    print("bearer_token=      ", await whoami(HTTPServerSpec(url="http://127.0.0.1:9433/mcp", bearer_token=token)))

    oauth = HTTPServerSpec(url="http://127.0.0.1:9434/mcp", audience="api://billing")
    print("identity=          ", await whoami(oauth, identity=identity))


asyncio.run(main())

import httpx

from client_credentials import client_credentials
from promptise import CallerContext
from promptise.identity import AgentIdentity, CallableTokenProvider


def agent_token(audience: str | None) -> str:
    """The agent's own credential: a client_credentials token for the server's audience."""
    grant = client_credentials("billing-agent", "agent-secret-for-local-dev", resource=audience or "api://billing")
    return grant["access_token"]


identity = AgentIdentity(
    "billing-agent",
    credential=CallableTokenProvider(token_fn=agent_token, provider_label="client-credentials"),
)


def signed_in_user(name: str) -> CallerContext:
    """Stands in for your app's login: the user's own access token from the same IdP."""
    login = httpx.get("http://127.0.0.1:9431/dev/login", params={"user": name}).json()
    return CallerContext(user_id=f"{name}@shop.example", bearer_token=login["access_token"])

from langchain_core.callbacks import BaseCallbackHandler


class Usage(BaseCallbackHandler):
    """Count model calls and tokens for one run."""

    def __init__(self):
        self.calls = self.input_tokens = self.output_tokens = 0

    def on_llm_end(self, response, **kwargs):
        self.calls += 1
        usage = response.generations[0][0].message.usage_metadata or {}
        self.input_tokens += usage.get("input_tokens", 0)
        self.output_tokens += usage.get("output_tokens", 0)

    def __str__(self):
        tokens = f"{self.input_tokens:,} input + {self.output_tokens:,} output tokens"
        return f"{self.calls} model calls, {tokens}"

import asyncio
import sys

from billing_identity import identity, signed_in_user
from promptise import build_agent
from promptise.config import HTTPServerSpec
from usage import Usage

REQUEST = "Customer acme had a late delivery. Issue a 50 EUR credit on invoice INV-7."


async def main(forward: bool) -> None:
    agent = await build_agent(
        model="openai:gpt-5-mini",
        servers={
            "billing": HTTPServerSpec(
                url="http://127.0.0.1:9434/mcp",
                audience="api://billing",
                forward_caller_token=forward,
            ),
        },
        identity=identity,
        instructions="You are a billing assistant. Use the billing tools. If a tool refuses, say why.",
        trace_tools=True,
    )
    usage = Usage()
    try:
        result = await agent.ainvoke(
            {"messages": [{"role": "user", "content": REQUEST}]},
            caller=signed_in_user("dana"),
            config={"callbacks": [usage]},
        )
        print(">>>", result["messages"][-1].content)
        print("usage:", usage)
    finally:
        await agent.shutdown()


if __name__ == "__main__":
    asyncio.run(main(forward="--no-forward" not in sys.argv))

python billing_agent.py                # forwards Dana's token, the default here
python billing_agent.py --no-forward   # sends the agent's own token

"""Which credential a server gets when the agent acts for a signed-in user."""

import asyncio
import os

from billing_identity import identity, signed_in_user
from promptise import build_agent
from promptise.config import HTTPServerSpec

API_KEY_SERVER = "http://127.0.0.1:9432/mcp"
SPECS = {
    "OAuth server, default": HTTPServerSpec(url="http://127.0.0.1:9434/mcp", audience="api://billing"),
    "API-key server, default": HTTPServerSpec(url=API_KEY_SERVER, api_key=os.environ["BILLING_AGENT_KEY"]),
    "API-key server, forward=True": HTTPServerSpec(
        url=API_KEY_SERVER, api_key=os.environ["BILLING_AGENT_KEY"], forward_caller_token=True
    ),
}


async def main():
    for label, spec in SPECS.items():
        agent = await build_agent(
            model="openai:gpt-5-mini",
            servers={"billing": spec},
            identity=identity,
            instructions="Call the whoami tool once, then reply in one short sentence.",
        )
        try:
            result = await agent.ainvoke(
                {"messages": [{"role": "user", "content": "Who does the billing server think I am?"}]},
                caller=signed_in_user("dana"),
            )
            seen = [m.content for m in result["messages"] if m.type == "tool"]
            print(f"{label:<29} forwards_caller_token={spec.forwards_caller_token!s:<5}  whoami -> {seen[-1][:110]}")
        finally:
            await agent.shutdown()


asyncio.run(main())