yourco-mcp — MCP-SDK für das AI Registry

Erstellen Sie einen MCP-Server, dessen Tool-Metadaten im AI Registry leben und zur Laufzeit heiß neu geladen werden. Sie schreiben Handler-Funktionen; alles andere — Beschreibungen, Schemas, Zielgruppen, Scope-Einstellungen pro Tool, Sichtbarkeit — wird von Admins in der Registry-Oberfläche verwaltet und erreicht Ihren laufenden Server in Millisekunden, ohne Redeploy.

Registry (control plane)          Your server (data plane, this SDK)
  admins edit metadata   ──push──▶  in-memory manifest ──▶ answers MCP calls
  UI / RBAC / versions              your handlers      ──▶ your business logic

Ihr Server wird nie durch das Registry blockiert: Der gesamte MCP-Datenverkehr wird aus dem Speicher bedient, und wenn das Registry ausfällt, läuft Ihr Server weiter (siehe Resilienz).

Installation

pip install "yourco-mcp[server,redis] @ git+https://github.com/amitmohapatra/mcp-sdk.git"

Pinnen Sie in Produktion einen Tag an (...mcp-sdk.git@v0.1.0). Extras: server bündelt uvicorn für server.run(); redis aktiviert Redis-Pub/Sub (in Produktion empfohlen — ohne dieses Extra fällt das SDK automatisch auf den SSE-Stream des Registry zurück).

Related MCP server: dynamic-mcp

Schnellstart — die gesamte Integration

import os
from yourco_mcp import ProductServer

server = ProductServer(
    registry_url="https://registry.yourco.com",
    product_key="billing",                    # your product's key in the registry
    api_key=os.environ["REGISTRY_API_KEY"],   # issued in the UI: Manage -> SDK API keys
)

@server.tool("get_invoice")                   # bound by NAME — metadata comes from the registry
async def get_invoice(ctx, invoice_id: str, max_results: int = 100):
    return {"invoice_id": invoice_id, "max_results": max_results}

if __name__ == "__main__":
    server.run(port=8080)                     # stateless MCP over HTTP at POST /mcp

Beachten Sie, was fehlt: keine Beschreibungen, keine JSON-Schemas, keine Redis-Konfiguration, kein Auth-Boilerplate. Das Registry besitzt die Metadaten; Ihr Code besitzt das Verhalten. Wenn das Registry ein Tool auflistet, für das Sie keinen Handler haben, wird es mit einer Warnung aus tools/list ausgeschlossen (ausfallsicher, niemals fail-crash).

Authentifizierung — Sie besitzen die Identität, das SDK besitzt die Durchsetzung

Jedes Produkt übernimmt die Authentifizierung für seine eigenen Tools. Das SDK sieht niemals Ihre Passwörter, Schlüssel oder Token-Formate — Sie implementieren genau eine Methode: Header hinein, Benutzer heraus.

from yourco_mcp import ProductServer, AuthProvider, AuthUser

class MyProductAuth(AuthProvider):
    async def authenticate(self, headers) -> AuthUser | None:
        token = headers.get("authorization", "").removeprefix("Bearer ")
        claims = my_jwt_verify(token)          # YOUR auth: your JWT lib, your OAuth
        if not claims:                         # introspection, your session store
            return None
        return AuthUser(id=claims["sub"], scopes=claims.get("scopes", []))

server = ProductServer(..., auth=MyProductAuth())

Auch ein einfaches async def fn(headers) -> AuthUser | None funktioniert.

Die Verantwortung des Frameworks endet an der Schnittstelle. Was in authenticate passiert, ist Ihre Geschäftslogik und nur Ihre — Firebase, Auth0, Keycloak, Ihr eigener JWT-Aussteller, eine Session-Tabelle, mTLS, LDAP, was auch immer. Das SDK importiert, bündelt oder bevorzugt nie ein Identitätssystem; es konsumiert nur das von Ihnen zurückgegebene AuthUser. Das folgende Beispiel verwendet Firebase rein zur Veranschaulichung:

import asyncio
import firebase_admin
from firebase_admin import auth as fb_auth
from yourco_mcp import AuthProvider, AuthUser

firebase_admin.initialize_app()                      # your service account creds

class FirebaseAuth(AuthProvider):
    async def authenticate(self, headers) -> AuthUser | None:
        token = headers.get("authorization", "").removeprefix("Bearer ").strip()
        try:                                          # verify_id_token is blocking:
            decoded = await asyncio.to_thread(fb_auth.verify_id_token, token)
        except Exception:
            return None
        roles = await my_db.fetch_roles(decoded["uid"])       # YOUR roles table
        return AuthUser(id=decoded["uid"],
                        scopes=[f"role:{r}" for r in roles],   # roles become scopes
                        claims=decoded)

Verlangen Sie dann Rollen pro Tool, direkt am Dekorator:

@server.tool("refund_payment", scopes=["role:finance-admin"])
async def refund_payment(ctx, payment_id: str, amount: float): ...

Im Code deklarierte scopes greifen in Union mit den im Registry gesetzten required_scopes — jede Seite kann ein Tool verschärfen, keine kann die andere lockern. Eingebaut: ApiKeyAuthProvider({key: {...}}), StaticTokenProvider({token: {...}}) und NoAuth() — der explizite Opt-out für wirklich offene Server (nichts ist je versehentlich offen).

Der Vertrag in einem Satz: Entdeckung ist immer öffentlich; alles, was die Ausführung betrifft, ist eine steckbare Entscheidung Ihres Produkts.

• tools/list (sowie initialize/ping) erfordert nie Authentifizierung — eine Invariante, kein Standard. Gateways und Kataloge (z. B. Bifrost) können die Tools jedes Produkts ohne jegliche Zugangsdaten auflisten. Anonyme Aufrufer sehen die Ansicht der Standard-Zielgruppe.

• Authentifizierung ist steckbar: Ihr AuthProvider — oder NoAuth() für einen vollständig offenen Server (eine explizite Entscheidung, niemals ein Versehen).

• Autorisierung ist steckbar: Ihre Scopes, von Ihrem Auth-System ausgestellt, werden gegen die pro Tool im Registry gesetzten required_scopes geprüft, plus Ihr @server.authorize-Hook für alles, was Scopes nicht ausdrücken können.

• Die Ausführungs-Authentifizierung pro Tool ist Ihre Entscheidung:

@server.tool("ping", public=True)          # executes without auth
async def ping(ctx): ...

@server.tool("refund_payment")             # gated (the default)
async def refund(ctx, payment_id: str): ...

Sicherheitsregel: Wenn ein Admin im Registry required_scopes an ein Tool hängt, ist Authentifizierung auch dann wieder erforderlich, wenn der Code das Tool als öffentlich markiert — Laufzeitverschärfung gewinnt immer; der Code-seitige Opt-out kann das nie überschreiben.

Sobald Ihr Verifier existiert, erzwingt das SDK — Sie müssen nichts davon selbst schreiben:

@server.authorize
async def gate(user, tool, args) -> bool:
    return not (tool == "refund_payment" and args["amount"] > 10_000
                and "payments:admin" not in user.scopes)

Unternehmensweite Scope-Konventionen (einmal vereinbart, organisationsweit):

• audience:<key> — gewährt eine Zielgruppe (z. B. audience:internal für interne Agenten)

• <domain>:<action> — pro Tool Anforderungen, die Admins im Registry festlegen (z. B. payments:write, invoices:read)

Zielgruppen, versteckte Parameter, feste Werte

Admins können ein Tool pro Zielgruppe unterschiedlich bereitstellen (z. B. external vs. internal): verschiedene Beschreibungen, zusätzliche nur-interne Parameter oder Parameter, die für eine Zielgruppe versteckt sind und stattdessen mit einem festen Wert an Ihren Handler gesendet werden — Aufrufer können sie nie sehen oder überschreiben. Ihr Handler deklariert den Parameter einfach mit einem Standardwert; das SDK validiert Argumente gegen das Zielgruppen-Schema des Aufrufers, entfernt unbekannte Argumente und injiziert feste Werte, bevor Ihr Code läuft.

@server.tool("charge_card")
async def charge_card(ctx, card_id: str, amount: float, currency: str = "USD"):
    # external callers can't even see `currency` — the SDK always passes the
    # admin-fixed value; internal callers control it. ctx.audience tells you which.
    ...

ctx gibt Ihnen ctx.user (das AuthUser-Objekt), ctx.audience und ctx.tool.

Live-Updates — wie ein Registry-Speichervorgang Ihren Server erreicht

1. Admin speichert im Registry → eine Transaktion erhöht die Sequenznummer des Produkts und veröffentlicht ein Ereignis mit den bereits aufgelösten Ansichten.

2. Ihr Server (seit dem Start abonniert — Redis, falls Ihr Produkt es konfiguriert hat, sonst der SSE-Stream des Registry; das Manifest teilt dem SDK mit, welcher) empfängt es.

3. Sequenzprüfung: in der richtigen Reihenfolge → als atomarer Manifestwechsel angewendet; veraltet → ignoriert; Lücke → vollständiger Neuabruf und Abgleich. Konvergenz ist garantiert.

4. Das nächste tools/list/tools/call bedient die neuen Metadaten. Typische Latenz: einstellige Millisekunden (Redis) bis einige hundert ms (SSE).

Resilienz

• Registry ausgefallen → Ihr Server bedient weiterhin aus dem Speicher, einschließlich des zuletzt angewendeten Updates. Das Registry ist eine Kontrollebene, niemals eine Laufzeitabhängigkeit.

• Kaltstart bei ausgefallenem Registry → wird aus einem letzten bekannten guten Snapshot bedient, den das SDK automatisch pflegt (zwischengespeichert unter ~/.cache/yourco-mcp/; überschreiben Sie den Speicherort mit YOURCO_MCP_CACHE_DIR, falls Ihre Laufzeit das benötigt).

• Pub/Sub-Ausfall → automatischer Fallback auf kostengünstiges bedingtes Polling (ETag/304) mit exponentiellem Backoff, während kontinuierlich versucht wird, sich erneut zu abonnieren — Updates fließen weiter, nur Sekunden langsamer.

• Schlechtes/fehlerhaftes Update → protokolliert, ignoriert, neu synchronisiert; ein gutes Manifest wird niemals durch ein defektes ersetzt.

• Zustandsloses HTTP → betreiben Sie N Replikate hinter einem beliebigen Load Balancer; keine Sticky-Sessions.

Checkliste für ein neues Produkt

1. Bitten Sie einen Registry-Admin, Ihr Produkt einzubinden und Ihnen einen API-Schlüssel auszuhändigen.

2. pip install (siehe oben), setzen Sie REGISTRY_API_KEY in Ihrer Deployment-Umgebung.

3. Schreiben Sie Handler für die Tools, die Ihr Produkt besitzt (Namen müssen mit dem Registry übereinstimmen).

4. Binden Sie Ihre bestehende Authentifizierung in eine AuthProvider.authenticate-Methode ein.

5. Entscheiden Sie, welche Ihrer Token audience:*-Scopes tragen (interne Agenten usw.).

6. server.run() — verifizieren Sie mit curl localhost:8080/healthz und einem MCP-tools/list. Bearbeiten Sie eine Beschreibung in der Registry-Oberfläche und beobachten Sie, wie sie sich live ändert.