Einen benutzerdefinierten MCP-Server mit OAuth 2.1 erstellen

Was Sie erstellen:

Ein minimaler, lauffähiger Python-MCP-Server, mit dem sich Blockbrain über OAuth 2.1 verbinden kann — sodass interne Tools oder Daten, die Sie bereitstellen, sicher von einem Blockbrain-Agenten genutzt werden können.

Zielgruppe: ein Entwickler an Ihrer Seite. Kopieren Sie die vier Codeblöcke unten in einen Ordner, führen Sie fünf Befehle aus, zeigen Sie Blockbrain auf die resultierende URL, und Sie haben eine funktionierende authentifizierte Integration.

Benötigte Zeit: ~20 Minuten für einen funktionierenden lokalen Server; ~45 Minuten inklusive Härtung.

Siehe auch: MCP-Server — Rundgang durch die Admin-Oberfläche.

---

Voraussetzungen

• Python 3.11+

• Eine öffentliche HTTPS-URL für Ihren lokalen Server während des Tests — die einfachsten Optionen: ein HTTPS-Tunneling-Tool wie cloudflared-Tunnel oder ngrok

• Tenant-Administrator Zugriff auf Ihren Blockbrain-Tenant

• Vertrautheit mit dem OAuth-2.1-Authorization-Code-Flow mit PKCE (hilfreich, aber nicht erforderlich)

---

Wann welcher Authentifizierungsmodus verwendet wird

Die MCP-Server-Registrierungsoberfläche von Blockbrain bietet vier Authentifizierungsmethoden. Wählen Sie je nachdem, wie Ihr MCP-Server erkennen soll, wer aufruft:

Methode	Verwenden, wenn…	Kompromiss
Keine	Nur intern/zu Entwicklungszwecken und die Server-URL nicht öffentlich ist	Jeder, der die URL erreicht, kann die Tools aufrufen
API-Schlüssel (Fester Token)	Service-zu-Service. Ein gemeinsames Secret. Keine Identität pro Benutzer erforderlich.	Kann auf Ihrer Seite einzelne Blockbrain-Benutzer nicht unterscheiden
OAuth 2.1 ← dieser Leitfaden	Produktionsszenarien. Der Tenant-Administrator von Blockbrain autorisiert die Verbindung einmalig über einen Consent-Flow.	Mehr bewegliche Teile, aber der Standardweg gemäß MCP-Spezifikation
Benutzertoken (delegiert)	Sie möchten eine Autorisierung pro Benutzer auf Ihrem MCP-Server (z. B. erzwingen, dass Benutzer A nur seine eigenen Daten sieht)	Ihr Server muss das weitergeleitete JWT von Blockbrain validieren — siehe Abschnitt 8

---

Der OAuth-2.1-Discovery-Flow, den Blockbrain verwendet

Wenn Sie im Admin-UI OAuth 2.1 auswählen und auf OAuth erkennenklicken, folgt Blockbrain RFC 9728 + RFC 8414 um Ihren Autorisierungsserver zu finden, und führt dann Authorization Code + PKCE aus:

---

Das Python-Beispiel

Drei Dateien in einem Ordner. Kopieren Sie jeden Block unverändert.

server.py

"""
MCP-Server-Startvorlage — OAuth-Integration mit Blockbrain
Ein minimales Beispiel, das den vollständigen OAuth-2.1- + PKCE-Flow demonstriert.
Ausführen: 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()

# ---- Konfiguration ---------------------------------------------------------
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, die dieser Server in beiden .well-known-Dokumenten ankündigt und gegen die
# jede /authorize-Anfrage geprüft wird.
SUPPORTED_SCOPES = {"mcp:read", "mcp:write"}

# ---- In-Memory-Speicher (NUR DEMO — in der Produktion Redis/DB verwenden) -------------
auth_codes: dict = {}
access_tokens: dict = {}

# ---- MCP-Server: Tools + Ressourcen -----------------------------------------
mcp_server = FastMCP("blockbrain-oauth-example")

@mcp_server.tool()
def echo(message: str) -> str:
    """Die übergebene Nachricht zurückgeben — prüft die End-to-End-Konnektivität."""
    return f"echo: {message}"

@mcp_server.resource("static://welcome")
def welcome() -> str:
    """Eine statische Willkommensressource."""
    return "Hallo von Ihrem benutzerdefinierten MCP-Server!"

# ---- FastAPI-App: OAuth-Endpunkte + eingebundener MCP-Transport ------------------
app = FastAPI(title="MCP-OAuth-Beispiel für Blockbrain")

# RFC 9728 — Metadaten der geschützten Ressource
@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 — Metadaten des Autorisierungsservers
@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"],
    }

# Autorisierungsendpunkt — stellt einen Auth-Code aus, der an den PKCE-Challenge gebunden ist
@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 — jeden Scope ablehnen, den der Server nicht ankündigt.
    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-Endpunkt — tauscht Code + Verifier gegen ein Zugriffstoken
@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-Prüfung: 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: weitergeleitetes Benutzer-JWT von Blockbrain validieren (delegierter Modus) ---
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)

# ---- Middleware für Bearer-Token für /mcp --------------------------------------
# RFC 6750: Jeder MCP-Transportaufruf muss ein gültiges Zugriffstoken mitführen.
# Dies läuft, BEVOR die Anfrage die eingebundene MCP-App erreicht.
@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"'},
            )
        # Optional hier den Scope pro Aufruf erzwingen, z. B. "mcp:write" verlangen
        # für zustandsändernde Tools, indem record["scope"] geprüft wird.
    return await call_next(request)

# ---- MCP-Server (Streamable-HTTP-Transport) unter /mcp einbinden ------------------
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

Ersetzen Sie die Demo-Client-Anmeldedaten vor dem Einsatz irgendwo, das vom öffentlichen Internet aus erreichbar ist, durch starke Zufallswerte.

# Öffentliche URL, unter der Ihr Server von Blockbrain aus erreichbar ist.
# Beim lokalen Tunneln mit cloudflared/ngrok hier die Tunnel-URL eintragen.
SERVER_HOST=http://localhost:8080

# Blockbrains JWKs-Endpunkt — wird nur verwendet, wenn VALIDATE_USER_TOKEN=1
JWKS_URL=https://auth.theblockbrain.ai/oauth/v2/keys
AUDIENCE=your-api-audience

# Statische OAuth-Client-Anmeldedaten. Für die Produktion starke Zufallswerte erzeugen.
OAUTH_CLIENT_ID=demo-client-id
OAUTH_CLIENT_SECRET=demo-client-secret

# Auf 1 setzen, um zusätzlich das weitergeleitete Benutzer-JWT von Blockbrain zu validieren
# bei jeder /mcp-Anfrage (delegierter Zugriffsmodus). Für reines OAuth 0 lassen.
VALIDATE_USER_TOKEN=0

---

Lokal ausführen

python -m venv .venv
source .venv/bin/activate          # Windows: .venv\Scripts\activate
pip install -r requirements.txt
cp .env.example .env               # dann .env bearbeiten
python server.py                   # lauscht auf :8080

---

Die Endpunkte mit curl überprüfen

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

# Erkennung — 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
# → listet das Tool "echo" und die Ressource "static://welcome" auf

---

Registrieren Sie Ihren MCP-Server in Blockbrain

1. Machen Sie Ihren lokalen Server öffentlich erreichbar: cloudflared tunnel --url http://localhost:8080 (oder ngrok http 8080). Notieren Sie die öffentliche HTTPS-URL.

2. Setzen Sie SERVER_HOST in .env auf diese öffentliche URL und starten Sie den Server neu.

3. Öffnen Sie Blockbrain → Admin → Agents → MCP-Server → + MCP-Server hinzufügen.

4. Füllen Sie aus:

   • Servername: z. B. my-mcp-demo

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

   • Transport: HTTP

   • Authentifizierung: OAuth 2.1

5. Klicken Sie auf OAuth erkennen. Blockbrain liest Ihre beiden .well-known -Endpunkte und füllt die OAuth-Konfiguration vor.

6. Klicken Sie auf Konfigurieren Sie, schließen Sie den Consent-Flow ab und bestätigen Sie, dass das Zugriffstoken zurückkommt.

7. Speichern. Weisen Sie die Integration einem Test-Agenten zu. Bitten Sie den Agenten im Chat, echo "hello" aufzurufen — Sie sollten echo: hello zurückbekommen.

Vollständiger Rundgang durch die Admin-Oberfläche mit Screenshots: MCP-Server — für Administratoren.

---

Optional — weitergeleitetes Benutzer-JWT von Blockbrain validieren

Wenn Sie zusätzlich eine Autorisierung pro Benutzer möchten (delegierter Zugriffsmodus), setzen Sie VALIDATE_USER_TOKEN=1. Die validate_user_jwt Hilfe-Funktion in server.py prüft jedes eingehende Bearer-Token gegen Blockbrains öffentliche JWKs unter https://auth.theblockbrain.ai/oauth/v2/keys, verifiziert Signatur, Ablauf und Audience und gibt die Claims zurück (einschließlich external_user_id, urn:zitadel:iam:org:id, usw.).

---

Header, die Blockbrain mit jeder Anfrage sendet

Ihr MCP-Server kann diese lesen, um den aufrufenden Benutzer, den Tenant und den Thread-Kontext zu identifizieren:

Header	Beschreibung	Beispiel
X-User-ID	Endbenutzer-Identifikator	user_12345
X-External-User-ID	Benutzer-Identifikator Ihres Systems (falls aktiviert)	ext_user_abc
X-Tenant-ID	Tenant-Identifikator	tenant_xyz
X-Agent-ID	Agent, der die Anfrage stellt	agent_007
X-Data-Room-ID	Zugeordneter Data Room	room_456
X-Thread-ID	Unterhaltungs-Thread	thread_789
Authorization	Bearer-Token (OAuth-Zugriffstoken oder weitergeleitetes Benutzer-JWT)	Bearer ...

---

Fehlerbehebung

Symptom	Wahrscheinliche Ursache	Behebung
„OAuth erkennen“ gibt 404 zurück	.well-known Pfade sind unter der öffentlichen URL nicht freigegeben	Bestätigen Sie, dass der Tunnel Root-Pfade weiterleitet; rufen Sie beide well-known-Endpunkte erneut mit curl auf
Redirect-URI-Fehler beim Consent	Der IdP unterstützt Dynamic Client Registration nicht (z. B. Microsoft Entra)	Verwenden Sie das statische OAUTH_CLIENT_ID in diesem Beispiel statt eines dynamisch registrierten
Token zurückgegeben, aber die Tool-Liste ist leer	Scopes wurden bei der Konfiguration nicht beibehalten	Stellen Sie sicher, dass Ihre Metadaten des Autorisierungsservers scopes_supported

---

Checkliste zur Härtung für die Produktion

Bevor Sie in einer Produktionsumgebung ausliefern, ersetzen oder ergänzen Sie:

• Die In-Memory- auth_codes / access_tokens -Dictionaries → einen persistenten Speicher (z. B. Redis), damit Tokens Neustarts überstehen und zwischen Replikas geteilt werden können.

• Refresh-Token-Rotation. Das Beispiel stellt nur Access Tokens aus.

• Strukturiertes Logging und einen Audit-Trail für jeden /authorize und /token Aufruf.

• Verschieben Sie OAUTH_CLIENT_SECRET in einen Secrets Manager — niemals in die Versionsverwaltung einchecken.

• Beenden Sie HTTPS vor dem Server (Reverse Proxy, Load Balancer oder das Ingress Ihrer Plattform).

• Ratenbegrenzung für /token und /authorize.

• Wenn Multi-Tenant, dann Scope OAUTH_CLIENT_ID pro Tenant statt einen gemeinsamen Wert wiederzuverwenden.

---

Nächste Schritte

• SSE-Transport — Blockbrain unterstützt ebenfalls SSE (https://your-server/sse). Um zu wechseln, ersetzen Sie die app.mount("/mcp", ... ) Zeile durch die SSE-App aus mcp.server.sse.

• Echte Tools — ersetzen Sie das echo Tool durch Aufrufe in Ihrer Domäne (Datenbankabfragen, interne APIs usw.).

• JavaScript-/TypeScript-Beispiel — eine Node/Express-Entsprechung dieses Leitfadens mit @modelcontextprotocol/sdk wird als Folgebeitrag vorbereitet.