# Einen benutzerdefinierten MCP-Server mit OAuth 2.1 erstellen

## **Was Sie erstellen werden:**

> Ein minimalistischer, ausführbarer Python-MCP-Server, mit dem sich Blockbrain über OAuth 2.1 verbinden kann — sodass interne Tools oder Daten, die Sie freigeben, 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 Testens — 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 Registrierungsoberfläche für MCP-Server von Blockbrain bietet vier Authentifizierungsmethoden. Wählen Sie je nachdem, wie Ihr MCP-Server erkennen soll, wer anfragt:

| 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-Key** (Fester Token)         | Service-zu-Service. Ein gemeinsames Geheimnis. Keine Identität pro Benutzer erforderlich.                                              | Kann einzelne Blockbrain-Benutzer auf Ihrer Seite nicht unterscheiden                 |
| **OAuth 2.1** ← *dieser Leitfaden* | Produktionsszenarien. Der Tenant-Admin 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 Benutzerautorisierung pro Nutzer auf Ihrem MCP-Server (z. B. erzwingen, dass Benutzer A nur ihre 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 **OAuth 2.1** in der Admin-Oberfläche 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. Admin umleiten → /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. Weitere /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
from typing import Optional
from urllib.parse import urlencode

import httpx
import uvicorn
from dotenv import load_dotenv
from fastapi import FastAPI, HTTPException, Form
from fastapi.responses import 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"

# ---- In-Memory-Speicher (NUR DEMO — in 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 — überprü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 geschützter Ressourcen
@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 an die PKCE-Herausforderung gebundenen Auth-Code aus
@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")

    code = secrets.token_urlsafe(32)
    auth_codes[code] = {
        "client_id":      client_id,
        "redirect_uri":   redirect_uri,
        "code_challenge": code_challenge,
        "scope":          scope,
        "expires_at":     datetime.utcnow() + 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 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.utcnow():
        raise HTTPException(400, "invalid or expired code")
    if record["redirect_uri"] != redirect_uri:
        raise HTTPException(400, "redirect_uri mismatch")

    # PKCE-Überprü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,
        "expires_at": datetime.utcnow() + 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)

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

> **Ersetzen Sie die Demo-Client-Anmeldedaten vor der Bereitstellung an einem Ort, der aus dem öffentlichen Internet erreichbar ist, durch starke zufällige Werte.**

```
# Öffentliche URL, unter der Ihr Server von Blockbrain aus erreichbar ist.
# Beim lokalen Tunneling mit cloudflared/ngrok diesen Wert auf die Tunnel-URL setzen.
SERVER_HOST=http://localhost:8080

# JWKs-Endpunkt von Blockbrain — 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 zufällige Werte 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 belassen.
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 prüfen

```
# 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
# → 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 sich 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://<Ihr-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**, schließen Sie den Consent-Flow ab und bestätigen Sie, dass das Access Token zurückkommt.
7. Speichern. Weisen Sie die Integration einem Test-Agenten zu. Bitten Sie den Agenten in einem Chat, `echo "hello"` aufzurufen — Sie sollten sehen, dass `echo: hello` zurückkommt.

Vollständiger Rundgang durch die Admin-Oberfläche mit Screenshots: [MCP-Server — für Admins](https://docs.en.theblockbrain.ai/for-admins/mcp-server).

***

### Optional — das weitergeleitete Benutzer-JWT von Blockbrain validieren

Wenn Sie zusätzlich eine Benutzerautorisierung pro Nutzer wünschen (delegierter Zugriffsmodus), setzen Sie `VALIDATE_USER_TOKEN=1`. Die `validate_user_jwt` Hilfsfunktion in `server.py` prüft jedes eingehende Bearer-Token gegen die öffentlichen JWKs von Blockbrain unter `https://auth.theblockbrain.ai/oauth/v2/keys`, verifiziert Signatur, Ablaufzeit und Zielgruppe 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 anfragenden Benutzer, Tenant und Thread-Kontext zu identifizieren:

| Header               | Beschreibung                                                         | Beispiel       |
| -------------------- | -------------------------------------------------------------------- | -------------- |
| `X-User-ID`          | Endbenutzer-Identifikator                                            | `user_12345`   |
| `X-External-User-ID` | Benutzerkennzeichen 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`        | Konversations-Thread                                                 | `thread_789`   |
| `Authorization`      | Bearer-Token (OAuth-Access-Token oder weitergeleitetes Benutzer-JWT) | `Bearer ...`   |

***

### Fehlerbehebung

| Symptom                                           | Wahrscheinliche Ursache                                                      | Behebung                                                                                                               |
| ------------------------------------------------- | ---------------------------------------------------------------------------- | ---------------------------------------------------------------------------------------------------------------------- |
| „OAuth erkennen“ liefert 404                      | `.well-known` Pfade werden unter der öffentlichen URL nicht bereitgestellt   | Bestätigen Sie, dass der Tunnel die Root-Pfade weiterleitet; rufen Sie beide .well-known-Endpunkte erneut mit curl auf |
| Redirect-URI passt beim Consent nicht             | Der IdP unterstützt kein Dynamic Client Registration (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 während der Konfiguration nicht beibehalten                    | Stellen Sie sicher, dass Ihre Metadaten des Autorisierungsservers `scopes_supported`                                   |

***

### Checkliste für Produktionshärtung

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

* Die In-Memory- `auth_codes` / `access_tokens` Dictionaries → ein persistenter Speicher (z. B. Redis), damit Token Neustarts überstehen und über Replikate hinweg gemeinsam genutzt werden können.
* Refresh-Token-Rotation. Das Beispiel stellt nur Access Tokens aus.
* Strukturiertes Logging und ein Audit-Log für jeden `/authorize` und `/token` Aufruf.
* Verschieben Sie `OAUTH_CLIENT_SECRET` in einen Secrets Manager — niemals in die Quellcodeverwaltung einchecken.
* Beenden Sie HTTPS vor dem Server (Reverse Proxy, Load Balancer oder der Ingress Ihrer Plattform).
* Rate Limiting auf `/token` und `/authorize`.
* Bei Multi-Tenancy den Bereich `OAUTH_CLIENT_ID` pro Tenant festlegen, statt einen gemeinsam genutzten Wert wiederzuverwenden.

***

### Nächste Schritte

* **SSE-Transport** — Blockbrain unterstützt auch SSE (`https://your-server/sse`). Zum Umschalten ersetzen Sie die Zeile `app.mount("/mcp", ... )` durch die SSE-App aus `mcp.server.sse`.
* **Echte Tools** — ersetzen Sie das `echo` Tool durch Aufrufe in Ihre Domäne (Datenbankabfragen, interne APIs usw.).
* **JavaScript-/TypeScript-Beispiel** — ein Node/Express-Äquivalent zu diesem Leitfaden mit `@modelcontextprotocol/sdk` wird als Nachfolgebeitrag 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.en.theblockbrain.ai/de/fur-entwickler/einen-benutzerdefinierten-mcp-server-mit-oauth-2.1-erstellen.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.