yourco-mcp — AI 레지스트리용 MCP SDK

도구 메타데이터가 AI 레지스트리에 저장되고 런타임에 핫 리로드되는 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는 여러분의 비밀번호, 키, 토큰 형식을 절대 보지 않습니다. 정확히 하나의 메서드만 구현하면 됩니다: 헤더 입력, 사용자 출력.

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와 합집합(union) 으로 적용됩니다 — 어느 쪽이든 도구를 더 엄격하게 할 수 있지만, 어느 쪽도 다른 쪽을 느슨하게 할 수는 없습니다. 기본 내장: ApiKeyAuthProvider({key: {...}}), StaticTokenProvider({token: {...}}), 그리고 NoAuth() — 완전히 공개된 서버를 위한 명시적 옵트아웃입니다(어떤 것도 우연히 공개되지 않습니다).

계약을 한 줄로 정리하면: 탐색(discovery)은 항상 공개됩니다. 실행에 관한 모든 것은 여러분 제품의 플러그형 선택입니다.

• 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가 구성되어 있으면 Redis, 아니면 레지스트리의 SSE 스트림; 매니페스트가 SDK에 어떤 것을 사용할지 알려줍니다)가 이를 수신합니다.

3. 시퀀스 확인: 순서상 다음이면 → 원자적 매니페스트 교체로 적용; 오래된 것이면 → 무시; 공백이 있으면 → 전체 재조회 및 조정. 수렴이 보장됩니다.

4. 다음 tools/list/tools/call부터 새 메타데이터를 제공합니다. 일반적인 지연 시간: 수 밀리초(Redis)에서 수백 ms(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. 기존 인증을 하나의 AuthProvider.authenticate 메서드에 연결하세요.

5. 어떤 토큰이 audience:* 스코프를 지닐지 결정하세요(내부 에이전트 등).

6. server.run() — curl localhost:8080/healthz와 MCP tools/list로 검증하세요. 레지스트리 UI에서 설명을 편집하고 실시간으로 변경되는 것을 확인하세요.