# Erstellen Sie einen benutzerdefinierten MCP-Server mit OAuth 2.1

### **Was Sie erstellen:**<br>

> 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](https://docs.en.theblockbrain.ai/for-admins/mcp-server).

***

### 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 erkennen*klicken, folgt Blockbrain [RFC 9728](https://datatracker.ietf.org/doc/html/rfc9728) + [RFC 8414](https://datatracker.ietf.org/doc/html/rfc8414) um Ihren Autorisierungsserver zu finden, und führt dann Authorization Code + PKCE aus:

```mermaid
sequenceDiagram
    participant BB as Blockbrain
    participant MCP as Ihr MCP-Server

    BB->>MCP: 1. GET /.well-known/oauth-protected-resource
    MCP-->>BB: { "authorization_servers": [...] }

    BB->>MCP: 2. GET /.well-known/oauth-authorization-server
    MCP-->>BB: { "authorization_endpoint": ..., "token_endpoint": ... }

    BB->>MCP: 3. Weiterleitung des Administrators → /authorize<br/>(response_type=code, code_challenge, ...)

    BB->>MCP: 4. POST /token  (code + code_verifier)
    MCP-->>BB: { "access_token": "...", "token_type": "Bearer" }

    BB->>MCP: 5. Nachfolgende /mcp-Aufrufe mit<br/>Authorization: Bearer <access_token>
```

***

### Das Python-Beispiel

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

#### `server.py`

```python
"""
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](https://docs.en.theblockbrain.ai/for-admins/mcp-server).

***

### 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.


---

# Agent Instructions: Querying This Documentation

If you need additional information that is not directly available in this page, you can query the documentation dynamically by asking a question.

Perform an HTTP GET request on the current page URL with the `ask` query parameter:

```
GET https://docs.blockbrain.ai/de/fur-entwickler/erstellen-sie-einen-benutzerdefinierten-mcp-server-mit-oauth-2.1.md?ask=<question>
```

The question should be specific, self-contained, and written in natural language.
The response will contain a direct answer to the question and relevant excerpts and sources from the documentation.

Use this mechanism when the answer is not explicitly present in the current page, you need clarification or additional context, or you want to retrieve related documentation sections.