yourco-mcp — AI Registry 向け MCP SDK

ツールのメタデータが AI Registry に存在し、実行時にホットリロードされる MCP サーバーを構築します。あなたはハンドラー関数を書くだけです。それ以外のすべて — 説明、スキーマ、オーディエンス、ツールごとのスコープ、公開設定 — はレジストリ 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 は server.run() 用に uvicorn を同梱します。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 はあなたのパスワード、キー、トークン形式を一切目にしません。実装するのは正確に1つのメソッドです: ヘッダー入力、ユーザー出力。

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）

オーディエンス、非表示パラメータ、固定値

管理者は、オーディエンスごとに1つのツールを異なる形で公開できます（例: external と 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. 管理者がレジストリに保存すると → 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 個のレプリカを実行できます。スティッキーセッションは不要です。

新規プロダクトチーム向けチェックリスト

1. レジストリ管理者にプロダクトのオンボーディングを依頼し、API キーを受け取ってください。

2. （上記の）pip install を実行し、デプロイ環境の環境変数に REGISTRY_API_KEY を設定します。

3. プロダクトが所有するツールのハンドラーを書きます（名前はレジストリと一致させる必要があります）。

4. 既存の認証を1つの AuthProvider.authenticate メソッドに組み込みます。

5. どのトークンに audience:* スコープを持たせるかを決めます（内部エージェントなど）。

6. server.run() — curl localhost:8080/healthz と MCP の tools/list で確認します。レジストリ UI で説明を編集し、ライブで変わる様子を確認します。