yourco-mcp — SDK de MCP para el AI Registry

Construye un servidor MCP cuyos metadatos de herramientas viven en el AI Registry y se recargan en caliente en tiempo de ejecución. Tú escribes las funciones handler; todo lo demás — descripciones, esquemas, audiencias, scopes por herramienta, exposición — lo gestionan los administradores en la UI del registry y llega a tu servidor en ejecución en milisegundos, sin volver a desplegar.

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

Tu servidor nunca se bloquea esperando al registry: todo el tráfico MCP se sirve desde memoria, y si el registry está caído, tu servidor sigue funcionando (consulta Resiliencia).

Instalación

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

Fija una tag en producción (...mcp-sdk.git@v0.1.0). Extras: server incluye uvicorn para server.run(); redis habilita Redis pub/sub (recomendado en prod — sin él, el SDK recurre automáticamente al stream SSE del registry).

Related MCP server: mcp-toolkit-hub

Inicio rápido — la integración completa

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

Observa lo que falta: no hay descripciones, ni esquemas JSON, ni configuración de Redis, ni código repetitivo de autenticación. El registry es dueño de los metadatos; tu código es dueño del comportamiento. Si el registry lista una herramienta para la que no tienes handler, se excluye de tools/list con un aviso (a prueba de fallos: nunca falla ni crashea).

Autenticación: tú eres dueño de la identidad, el SDK es dueño de hacerla cumplir

Cada producto gestiona la autenticación de sus propias herramientas. El SDK nunca ve tus contraseñas, claves ni formatos de token: implementas exactamente un método: cabeceras de entrada, usuario de salida.

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())

También funciona un simple async def fn(headers) -> AuthUser | None.

La responsabilidad del framework termina en la interfaz. Lo que ocurre dentro de authenticate es tu lógica de negocio y solo tuya — Firebase, Auth0, Keycloak, tu propio emisor de JWT, una tabla de sesiones, mTLS, LDAP, lo que sea. El SDK nunca importa, empaqueta ni favorece ningún sistema de identidad; solo consume el AuthUser que tú devuelves. El siguiente ejemplo usa Firebase únicamente como ilustración:

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)

Luego exige roles por herramienta, directamente en el decorador:

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

Los scopes declarados en código se aplican en unión con los required_scopes establecidos en el registry — cualquiera de las dos partes puede endurecer una herramienta, pero ninguna puede flexibilizar a la otra. Integrados: ApiKeyAuthProvider({key: {...}}), StaticTokenProvider({token: {...}}), y NoAuth() — la opción de exclusión explícita para servidores genuinamente abiertos (nada se abre jamás por accidente).

El contrato en una línea: el descubrimiento siempre es público; todo lo relativo a la ejecución es una opción conectable de tu producto.

• tools/list (e initialize/ping) nunca requieren autenticación — es un invariante, no un valor predeterminado. Las pasarelas y catálogos (p. ej., Bifrost) pueden enumerar las herramientas de todos los productos sin credenciales. Quienes llaman de forma anónima ven la vista de la audiencia predeterminada.

• La autenticación es conectable: tu AuthProvider — o NoAuth() para un servidor completamente abierto (una elección explícita, nunca un accidente).

• La autorización es conectable: tus scopes, emitidos por tu sistema de autenticación, se comprueban contra los required_scopes del registry por herramienta, más tu hook @server.authorize para cualquier cosa que los scopes no puedan expresar.

• La autenticación de ejecución por herramienta es tu elección:

@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): ...

Regla de seguridad: si un administrador asigna required_scopes a una herramienta en el registry, la autenticación vuelve a ser obligatoria aunque el código la marque como pública — el endurecimiento en tiempo de ejecución siempre gana; la exclusión desde el código nunca puede anularlo.

Una vez que tu verificador existe, el SDK se encarga de hacerlo cumplir — tú no escribes nada de esto:

@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)

Convenciones de scopes de la empresa (alínealas una vez, en toda la organización):

• audience:<key> — otorga una audiencia (p. ej., audience:internal para agentes internos)

• <domain>:<action> — requisitos por herramienta que los administradores definen en el registry (p. ej., payments:write, invoices:read)

Audiencias, parámetros ocultos, valores fijos

Los administradores pueden exponer una misma herramienta de forma diferente según la audiencia (p. ej., external vs internal): descripciones distintas, parámetros adicionales solo internos, o parámetros que están ocultos para una audiencia y que en su lugar se envían a tu handler con un valor fijo — quienes llaman nunca pueden verlo ni modificarlo. Tu handler solo declara el parámetro con un valor predeterminado; el SDK valida los argumentos contra el esquema de audiencia de quien llama, elimina los argumentos desconocidos e inyecta los valores fijos antes de que tu código se ejecute.

@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 te da ctx.user (el AuthUser), ctx.audience y ctx.tool.

Actualizaciones en vivo — cómo un guardado en el registry llega a tu servidor

1. Un administrador guarda en el registry → una transacción incrementa el número de secuencia del producto y publica un evento con las vistas ya resueltas.

2. Tu servidor (suscrito desde el arranque — Redis si tu producto lo tiene configurado, el stream SSE del registry en caso contrario; el manifiesto le indica al SDK cuál) lo recibe.

3. Comprobación de secuencia: el siguiente en orden → se aplica como un intercambio atómico de manifiesto; obsoleto → se ignora; una laguna → reobtención completa y reconciliación. La convergencia está garantizada.

4. El siguiente tools/list/tools/call sirve los nuevos metadatos. Latencia típica: milisegundos de un solo dígito (Redis) a unos pocos cientos de ms (SSE).

Resiliencia

• Registry caído → tu servidor sigue sirviendo desde memoria, incluida la última actualización aplicada. El registry es un plano de control, nunca una dependencia en tiempo de ejecución.

• Arranque en frío con el registry caído → se sirve desde una instantánea válida conocida (last-known-good) que el SDK mantiene automáticamente (almacenada en caché en ~/.cache/yourco-mcp/; sobrescribe la ubicación con YOURCO_MCP_CACHE_DIR si tu runtime lo necesita).

• Caída de pub/sub → respaldo automático a sondeo condicional económico (ETag/304) con backoff exponencial, mientras sigue intentando volver a suscribirse — las actualizaciones siguen fluyendo, solo unos segundos más lentas.

• Actualización errónea o malformada → se registra en el log, se ignora y se vuelve a sincronizar; un buen manifiesto nunca se sustituye por uno roto.

• HTTP sin estado → ejecuta N réplicas detrás de cualquier balanceador de carga; sin sesiones persistentes.

Lista de verificación para un producto nuevo

1. Pide a un administrador del registry que dé de alta tu producto y te entregue una API key.

2. pip install (arriba), define REGISTRY_API_KEY en el entorno de tu despliegue.

3. Escribe handlers para las herramientas que tu producto posee (los nombres deben coincidir con el registry).

4. Integra tu autenticación existente en un único método AuthProvider.authenticate.

5. Decide cuáles de tus tokens llevan scopes audience:* (agentes internos, etc.).

6. server.run() — verifica con curl localhost:8080/healthz y un tools/list de MCP. Edita una descripción en la UI del registry y observa cómo cambia en vivo.