Build a custom MCP server with OAuth 2.1

What you'll build:

A minimal, runnable Python MCP server that Blockbrain can connect to via OAuth 2.1 — so internal tools or data you expose can be safely consumed by a Blockbrain agent.

Audience: a developer at your side. Copy the four code blocks below into a folder, run five commands, point Blockbrain at the resulting URL, and you have a working authenticated integration.

Time required: ~20 minutes for a working local server; ~45 minutes including hardening.

See also: MCP Server — admin UI walkthrough.

---

Prerequisites

• Python 3.11+

• A public HTTPS URL for your local server while testing — easiest options: an HTTPS tunnelling tool such as cloudflared tunnel or ngrok

• Tenant Administrator access to your Blockbrain tenant

• Familiarity with OAuth 2.1 Authorization Code flow with PKCE (helpful but not required)

---

When to use which authentication mode

Blockbrain's MCP-server registration UI offers four authentication methods. Choose based on how your MCP server should know who is calling:

Method	Use when…	Trade-off
None	Internal/dev only and the server URL is not public	Anyone who reaches the URL can call the tools
API Key (Fixed Token)	Service-to-service. One shared secret. No per-user identity needed.	Cannot distinguish individual Blockbrain users on your side
OAuth 2.1 ← this guide	Production scenarios. Blockbrain's tenant admin authorizes the connection once via consent flow.	More moving parts, but the standard MCP-spec way
User Token (delegated)	You want per-user authorization on your MCP server (e.g. enforce that User A only sees their own data)	Your server must validate Blockbrain's forwarded JWT — see section 8

---

The OAuth 2.1 discovery flow Blockbrain uses

When you select OAuth 2.1 in the admin UI and click Discover OAuth, Blockbrain follows RFC 9728 + RFC 8414 to find your authorization server, then runs Authorization Code + PKCE:

---

The Python example

Three files in one folder. Copy each block verbatim.

server.py

"""
MCP Server Starter — OAuth Integration with Blockbrain
A minimal example demonstrating the full OAuth 2.1 + PKCE flow.
Run: python server.py
"""
import os
import secrets
import hashlib
import base64
from datetime import datetime, timedelta, timezone
from typing import Optional
from urllib.parse import urlencode

import httpx
import uvicorn
from dotenv import load_dotenv
from fastapi import FastAPI, HTTPException, Form, Request, Response
from fastapi.responses import JSONResponse, RedirectResponse
from jose import jwt
from mcp.server.fastmcp import FastMCP

load_dotenv()

# ---- Configuration ---------------------------------------------------------
SERVER_HOST          = os.getenv("SERVER_HOST", "http://localhost:8080")
JWKS_URL             = os.getenv("JWKS_URL", "https://auth.theblockbrain.ai/oauth/v2/keys")
AUDIENCE             = os.getenv("AUDIENCE", "your-api-audience")
OAUTH_CLIENT_ID      = os.getenv("OAUTH_CLIENT_ID", "demo-client-id")
OAUTH_CLIENT_SECRET  = os.getenv("OAUTH_CLIENT_SECRET", "demo-client-secret")
VALIDATE_USER_TOKEN  = os.getenv("VALIDATE_USER_TOKEN", "0") == "1"

# Scopes this server advertises in both .well-known documents and validates
# every /authorize request against.
SUPPORTED_SCOPES = {"mcp:read", "mcp:write"}

# ---- In-memory stores (DEMO ONLY — use Redis/DB in production) -------------
auth_codes: dict = {}
access_tokens: dict = {}

# ---- MCP server: tools + resources -----------------------------------------
mcp_server = FastMCP("blockbrain-oauth-example")

@mcp_server.tool()
def echo(message: str) -> str:
    """Echo back the provided message — verifies end-to-end connectivity."""
    return f"echo: {message}"

@mcp_server.resource("static://welcome")
def welcome() -> str:
    """A static welcome resource."""
    return "Hello from your custom MCP server!"

# ---- FastAPI app: OAuth endpoints + mounted MCP transport ------------------
app = FastAPI(title="MCP OAuth Example for Blockbrain")

# RFC 9728 — Protected Resource Metadata
@app.get("/.well-known/oauth-protected-resource")
async def oauth_protected_resource():
    return {
        "resource": SERVER_HOST,
        "authorization_servers": [SERVER_HOST],
        "scopes_supported": ["mcp:read", "mcp:write"],
        "bearer_methods_supported": ["header"],
    }

# RFC 8414 — Authorization Server Metadata
@app.get("/.well-known/oauth-authorization-server")
async def oauth_authorization_server():
    return {
        "issuer": SERVER_HOST,
        "authorization_endpoint": f"{SERVER_HOST}/authorize",
        "token_endpoint": f"{SERVER_HOST}/token",
        "response_types_supported": ["code"],
        "grant_types_supported": ["authorization_code"],
        "code_challenge_methods_supported": ["S256"],
        "scopes_supported": ["mcp:read", "mcp:write"],
        "token_endpoint_auth_methods_supported": ["client_secret_post"],
    }

# Authorization endpoint — issues an auth code bound to the PKCE challenge
@app.get("/authorize")
async def authorize(
    response_type: str,
    client_id: str,
    redirect_uri: str,
    code_challenge: str,
    code_challenge_method: str = "S256",
    scope: str = "mcp:read",
    state: Optional[str] = None,
):
    if response_type != "code":
        raise HTTPException(400, "unsupported response_type")
    if client_id != OAUTH_CLIENT_ID:
        raise HTTPException(400, "unknown client_id")
    if code_challenge_method != "S256":
        raise HTTPException(400, "code_challenge_method must be S256")

    # RFC 6749 §5.2 — reject any scope the server doesn't advertise.
    requested_scopes = set(scope.split())
    unknown = requested_scopes - SUPPORTED_SCOPES
    if unknown:
        raise HTTPException(400, f"invalid_scope: {sorted(unknown)}")

    code = secrets.token_urlsafe(32)
    auth_codes[code] = {
        "client_id":      client_id,
        "redirect_uri":   redirect_uri,
        "code_challenge": code_challenge,
        "scope":          " ".join(sorted(requested_scopes)),
        "expires_at":     datetime.now(timezone.utc) + timedelta(minutes=10),
    }
    params = {"code": code}
    if state:
        params["state"] = state
    return RedirectResponse(f"{redirect_uri}?{urlencode(params)}")

# Token endpoint — exchanges code + verifier for an access token
@app.post("/token")
async def token(
    grant_type:    str = Form(...),
    code:          str = Form(...),
    redirect_uri:  str = Form(...),
    client_id:     str = Form(...),
    client_secret: str = Form(...),
    code_verifier: str = Form(...),
):
    if grant_type != "authorization_code":
        raise HTTPException(400, "unsupported grant_type")
    if client_id != OAUTH_CLIENT_ID or client_secret != OAUTH_CLIENT_SECRET:
        raise HTTPException(401, "invalid client credentials")

    record = auth_codes.pop(code, None)
    if not record or record["expires_at"] < datetime.now(timezone.utc):
        raise HTTPException(400, "invalid or expired code")
    if record["redirect_uri"] != redirect_uri:
        raise HTTPException(400, "redirect_uri mismatch")

    # PKCE verification: base64url(SHA256(code_verifier)) == code_challenge
    expected = base64.urlsafe_b64encode(
        hashlib.sha256(code_verifier.encode()).digest()
    ).rstrip(b"=").decode()
    if expected != record["code_challenge"]:
        raise HTTPException(400, "invalid code_verifier")

    access_token = secrets.token_urlsafe(32)
    access_tokens[access_token] = {
        "client_id":  client_id,
        "scope":      record["scope"],
        "expires_at": datetime.now(timezone.utc) + timedelta(hours=1),
    }
    return {
        "access_token": access_token,
        "token_type":   "Bearer",
        "expires_in":   3600,
        "scope":        record["scope"],
    }

# ---- Optional: validate Blockbrain's forwarded user JWT (delegated mode) ---
async def validate_user_jwt(authorization_header: str) -> dict:
    token_str = authorization_header.replace("Bearer ", "", 1)
    if not token_str:
        raise HTTPException(401, "missing bearer token")
    async with httpx.AsyncClient() as client:
        jwks = (await client.get(JWKS_URL)).json()["keys"]
    header = jwt.get_unverified_header(token_str)
    key = next((k for k in jwks if k["kid"] == header["kid"]), None)
    if not key:
        raise HTTPException(401, "signing key not found in JWKs")
    return jwt.decode(token_str, key, algorithms=["RS256"], audience=AUDIENCE)

# ---- Bearer-token middleware for /mcp --------------------------------------
# RFC 6750: every MCP transport call must carry a valid access token.
# This runs BEFORE the request reaches the mounted MCP app.
@app.middleware("http")
async def require_bearer_for_mcp(request: Request, call_next):
    if request.url.path.startswith("/mcp"):
        auth = request.headers.get("authorization", "")
        if not auth.lower().startswith("bearer "):
            return JSONResponse(
                {"error": "invalid_token", "error_description": "missing bearer token"},
                status_code=401,
                headers={"WWW-Authenticate": 'Bearer realm="mcp"'},
            )
        token_str = auth.split(" ", 1)[1].strip()
        record = access_tokens.get(token_str)
        if not record or record["expires_at"] < datetime.now(timezone.utc):
            return JSONResponse(
                {"error": "invalid_token", "error_description": "expired or unknown token"},
                status_code=401,
                headers={"WWW-Authenticate": 'Bearer error="invalid_token"'},
            )
        # Optionally enforce per-call scope here, e.g. require "mcp:write"
        # for state-changing tools by inspecting record["scope"].
    return await call_next(request)

# ---- Mount MCP server (streamable HTTP transport) at /mcp ------------------
app.mount("/mcp", mcp_server.streamable_http_app())

if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8080)

requirements.txt

mcp>=1.2.0
fastapi>=0.110.0
uvicorn[standard]>=0.27.0
python-jose[cryptography]>=3.3.0
httpx>=0.26.0
python-dotenv>=1.0.0

.env.example

Replace the demo client credentials with strong random values before deploying anywhere reachable from the public internet.

# Public URL where your server is reachable from Blockbrain.
# When tunnelling locally with cloudflared/ngrok, set this to the tunnel URL.
SERVER_HOST=http://localhost:8080

# Blockbrain's JWKs endpoint — only used when VALIDATE_USER_TOKEN=1
JWKS_URL=https://auth.theblockbrain.ai/oauth/v2/keys
AUDIENCE=your-api-audience

# Static OAuth client credentials. Generate strong random values for production.
OAUTH_CLIENT_ID=demo-client-id
OAUTH_CLIENT_SECRET=demo-client-secret

# Set to 1 to additionally validate Blockbrain's forwarded user JWT
# on every /mcp request (delegated-access mode). Leave 0 for OAuth-only.
VALIDATE_USER_TOKEN=0

---

Run it locally

python -m venv .venv
source .venv/bin/activate          # Windows: .venv\Scripts\activate
pip install -r requirements.txt
cp .env.example .env               # then edit .env
python server.py                   # listens on :8080

---

Verify the endpoints with curl

# Discovery — RFC 9728
curl -s http://localhost:8080/.well-known/oauth-protected-resource | jq
# → { "resource": "...", "authorization_servers": ["..."], ... }

# Discovery — RFC 8414
curl -s http://localhost:8080/.well-known/oauth-authorization-server | jq
# → { "authorization_endpoint": "...", "token_endpoint": "...",
#     "code_challenge_methods_supported": ["S256"], ... }

# MCP transport handshake
npx @modelcontextprotocol/inspector http://localhost:8080/mcp
# → lists tool "echo" and resource "static://welcome"

---

Register your MCP server in Blockbrain

1. Expose your local server publicly: cloudflared tunnel --url http://localhost:8080 (or ngrok http 8080). Note the public HTTPS URL.

2. Set SERVER_HOST in .env to that public URL and restart the server.

3. Open Blockbrain → Admin → Agents → MCP Servers → + Add MCP Server.

4. Fill in:

   • Server Name: e.g. my-mcp-demo

   • Server URL: https://<your-tunnel>/mcp

   • Transport: HTTP

   • Authentication: OAuth 2.1

5. Click Discover OAuth. Blockbrain reads your two .well-known endpoints and pre-fills the OAuth config.

6. Click Configure, complete the consent flow, and confirm the access token comes back.

7. Save. Assign the integration to a test agent. In a chat, ask the agent to call echo "hello" — you should see echo: hello come back.

Full admin-UI walkthrough with screenshots: MCP Server — for admins.

---

Optional — validate Blockbrain's forwarded user JWT

If you also want per-user authorization (delegated-access mode), set VALIDATE_USER_TOKEN=1. The validate_user_jwt helper in server.py checks every incoming bearer token against Blockbrain's public JWKs at https://auth.theblockbrain.ai/oauth/v2/keys, verifies signature, expiration, and audience, and returns the claims (including external_user_id, urn:zitadel:iam:org:id, etc.).

---

Headers Blockbrain sends with every request

Your MCP server can read these to identify the calling user, tenant, and thread context:

Header	Description	Example
X-User-ID	End-user identifier	user_12345
X-External-User-ID	Your system's user identifier (if activated)	ext_user_abc
X-Tenant-ID	Tenant identifier	tenant_xyz
X-Agent-ID	Agent making the request	agent_007
X-Data-Room-ID	Associated data room	room_456
X-Thread-ID	Conversation thread	thread_789
Authorization	Bearer token (OAuth access token, or forwarded user JWT)	Bearer ...

---

Troubleshooting

Symptom	Likely cause	Fix
"Discover OAuth" returns 404	.well-known paths not exposed at the public URL	Confirm the tunnel forwards root paths; re-curl both well-known endpoints
Redirect-URI mismatch on consent	IdP does not support Dynamic Client Registration (e.g. Microsoft Entra)	Use the static OAUTH_CLIENT_ID in this example rather than a dynamically registered one
Token returned but tools list is empty	Scopes not preserved during configure	Make sure your authorization-server metadata includes scopes_supported

---

Production hardening checklist

Before shipping to a production environment, replace or add:

• The in-memory auth_codes / access_tokens dicts → a persistent store (e.g. Redis), so tokens survive restarts and can be shared across replicas.

• Refresh-token rotation. The example issues access tokens only.

• Structured logging and an audit trail for every /authorize and /token call.

• Move OAUTH_CLIENT_SECRET to a secrets manager — never check it into source control.

• Terminate HTTPS in front of the server (reverse proxy, load balancer, or your platform's ingress).

• Rate limiting on /token and /authorize.

• If multi-tenant, scope OAUTH_CLIENT_ID per tenant rather than reusing one shared value.

---

Next steps

• SSE transport — Blockbrain also supports SSE (https://your-server/sse). To switch, replace the app.mount("/mcp", ...) line with the SSE app from mcp.server.sse.

• Real tools — replace the echo tool with calls into your domain (database queries, internal APIs, etc.).

• JavaScript / TypeScript example — a Node/Express equivalent of this guide using @modelcontextprotocol/sdk is being prepared as a follow-up.