yourco-mcp — MCP SDK для AI Registry

Создайте MCP-сервер, чьи метаданные инструментов хранятся в AI Registry и перезагружаются на лету в рантайме. Вы пишете функции-обработчики; всё остальное — описания, схемы, аудитории, области видимости для каждого инструмента, доступность — управляется администраторами в UI реестра и доходит до вашего запущенного сервера за миллисекунды, без передеплоя.

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

Ваш сервер никогда не блокируется из-за реестра: весь MCP-трафик обслуживается из памяти, и если реестр недоступен, ваш сервер продолжает работать (см. «Отказоустойчивость»).

Установка

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

Зафиксируйте тег в продакшене (...mcp-sdk.git@v0.1.0). Дополнительные пакеты: server включает uvicorn для server.run(); redis включает Redis pub/sub (рекомендуется в проде — без него SDK автоматически переключается на SSE-поток реестра).

Related MCP server: dynamic-mcp

Быстрый старт — вся интеграция

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

Обратите внимание, чего нет: никаких описаний, JSON-схем, конфигурации Redis, шаблонного кода аутентификации. Реестру принадлежат метаданные; вашему коду принадлежит поведение. Если реестр перечисляет инструмент, для которого у вас нет обработчика, он исключается из tools/list с предупреждением (безопасное поведение при сбое, никогда не приводит к падению).

Аутентификация — идентификация за вами, применение — за SDK

Каждый продукт сам обрабатывает аутентификацию для своих инструментов. SDK никогда не видит ваши пароли, ключи или форматы токенов — вы реализуете ровно один метод: заголовки на входе, пользователь на выходе.

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

Подойдёт и обычный async def fn(headers) -> AuthUser | None.

Ответственность фреймворка заканчивается на интерфейсе. То, что происходит внутри authenticate, — это ваша бизнес-логика и только ваша: Firebase, Auth0, Keycloak, собственный JWT-эмитент, таблица сессий, mTLS, LDAP, что угодно. SDK никогда не импортирует, не включает в свою поставку и не отдаёт предпочтение ни одной системе идентификации; он лишь потребляет возвращаемый вами AuthUser. Пример ниже использует Firebase исключительно для иллюстрации:

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)

Затем требуйте роли для каждого инструмента, прямо в декораторе:

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

Объявленные в коде scopes применяются в объединении с заданными в реестре required_scopes — любая сторона может ужесточить требования к инструменту, но ни одна не может их ослабить. Встроенные провайдеры: ApiKeyAuthProvider({key: {...}}), StaticTokenProvider({token: {...}}) и NoAuth() — явный отказ от аутентификации для по-настоящему открытых серверов (ничто никогда не открыто случайно).

Контракт в одной строке: обнаружение всегда публично; всё, что связано с выполнением, — ваш подключаемый выбор.

• tools/list (а также initialize/ping) никогда не требует аутентификации — это инвариант, а не значение по умолчанию. Шлюзы и каталоги (например, Bifrost) могут перечислять инструменты любого продукта без каких-либо учётных данных. Анонимные вызывающие видят представление аудитории по умолчанию.

• Аутентификация подключаема: ваш AuthProvider — или NoAuth() для полностью открытого сервера (осознанный выбор, а не случайность).

• Авторизация подключаема: ваши области видимости, выпущенные вашей системой аутентификации, проверяются по заданным в реестре required_scopes для каждого инструмента, плюс ваш хук @server.authorize для всего, что области видимости не могут выразить.

• Аутентификация выполнения для каждого инструмента — ваш выбор:

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

Правило безопасности: если администратор привязывает required_scopes к инструменту в реестре, аутентификация снова обязательна, даже если код помечает его публичным — ужесточение в рантайме всегда побеждает; отказ со стороны кода никогда не может его переопределить.

Как только у вас появляется верификатор, SDK обеспечивает соблюдение — вам не нужно писать ничего из этого:

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

Соглашения об областях видимости в компании (согласуйте один раз, в масштабах всей организации):

• audience:<key> — предоставляет аудиторию (например, audience:internal для внутренних агентов)

• <domain>:<action> — требования для каждого инструмента, которые администраторы задают в реестре (например, payments:write, invoices:read)

Аудитории, скрытые параметры, фиксированные значения

Администраторы могут по-разному показывать один инструмент для каждой аудитории (например, external vs internal): разные описания, дополнительные параметры только для внутренних или параметры, которые скрыты от аудитории, а вместо этого вашему обработчику отправляется фиксированное значение — вызывающие никогда не могут его увидеть или переопределить. Ваш обработчик просто объявляет параметр со значением по умолчанию; SDK проверяет аргументы по схеме аудитории вызывающего, отбрасывает неизвестные аргументы и подставляет фиксированные значения до выполнения вашего кода.

@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 даёт вам ctx.user (объект AuthUser), ctx.audience и ctx.tool.

Живые обновления — как сохранение в реестре доходит до вашего сервера

1. Администратор сохраняет изменения в реестре → одна транзакция увеличивает номер последовательности продукта и публикует событие, несущее уже сформированные представления.

2. Ваш сервер (подписанный с момента запуска — на Redis, если он настроен для вашего продукта, в противном случае на SSE-поток реестра; манифест сообщает SDK, какой вариант использовать) получает его.

3. Проверка последовательности: следующий по порядку → применяется как атомарная замена манифеста; устаревший → игнорируется; пропуск → полная повторная загрузка и сверка. Сходимость гарантирована.

4. Следующий вызов tools/list/tools/call отдаёт новые метаданные. Типичная задержка: единицы миллисекунд (Redis) до нескольких сотен миллисекунд (SSE).

Отказоустойчивость

• Реестр недоступен → ваш сервер продолжает обслуживать запросы из памяти, включая последнее применённое обновление. Реестр — это плоскость управления, а не зависимость в рантайме.

• Холодный старт при недоступном реестре → обслуживание из последнего известного рабочего снимка, который SDK ведёт автоматически (кэш в ~/.cache/yourco-mcp/; переопределить расположение можно через YOURCO_MCP_CACHE_DIR, если это нужно вашей среде выполнения).

• Сбой Pub/sub → автоматический переход на дешёвый условный опрос (ETag/304) с экспоненциальной паузой, при этом непрерывно повторяются попытки подписки — обновления продолжают поступать, просто на несколько секунд медленнее.

• Некорректное или повреждённое обновление → логируется, игнорируется, повторно синхронизируется; рабочий манифест никогда не заменяется сломанным.

• HTTP без состояния → запускайте N реплик за любым балансировщиком; никаких липких сессий.