AI 에이전트 최소 권한 토큰 구현 가이드: MCP OAuth 2.1 + RFC 8707

토큰 요청 (resource 파라미터 포함):
 
POST /token HTTP/1.1
Host: auth.example.com
Content-Type: application/x-www-form-urlencoded
 
grant_type=authorization_code
&code=SplxlOBeZQQYbYS6WxSbIA
&redirect_uri=https://agent.example.com/callback
&resource=https://mcp-server.example.com        ← RFC 8707 핵심
&code_verifier=dBjftJeZ4CVP-mB92K27uhbUJU1p1r_wW1gFWFOEjXk

{
  "iss": "https://auth.example.com",
  "sub": "agent-user-123",
  "aud": "https://mcp-server.example.com",
  "scope": "tool:files:read",
  "exp": 1713100800,
  "iat": 1713100500,
  "jti": "a1b2c3d4-e5f6-..."
}

scope=tool:files:read       → 파일 읽기, TTL 5분
scope=tool:files:write      → 파일 쓰기, TTL 3분
scope=tool:code:execute     → 코드 실행, TTL 2분 (고위험)
scope=tool:db:write:prod    → 프로덕션 DB 쓰기, TTL 1분 + 인간 승인 필요

[사용자] → Orchestrator Agent
                 ↓ RFC 8693 Token Exchange
          Sub-Agent A (act 클레임에 상위 에이전트 정보 포함)
                 ↓
          MCP Server (aud=서버 URI, TTL=30초)

POST /token HTTP/1.1
Host: auth.example.com
Content-Type: application/x-www-form-urlencoded
 
grant_type=urn:ietf:params:oauth:grant-type:token-exchange
&subject_token=<orchestrator_access_token>
&subject_token_type=urn:ietf:params:oauth:token-type:access_token
&requested_token_type=urn:ietf:params:oauth:token-type:access_token
&resource=https://mcp-subagent.example.com
&scope=tool:search:read
&actor_token=<subagent_client_assertion>
&actor_token_type=urn:ietf:params:oauth:token-type:jwt

{
  "sub": "user-456",
  "aud": "https://mcp-subagent.example.com",
  "scope": "tool:search:read",
  "exp": 1713100830,
  "act": {
    "sub": "orchestrator-agent",
    "act": {
      "sub": "user-456"
    }
  }
}

# requirements: mcp[auth], python-jose, fastapi, httpx
from fastapi import FastAPI, Depends, HTTPException, status
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
from jose import jwt, JWTError
import httpx
 
app = FastAPI()
security = HTTPBearer()
 
MCP_SERVER_URI = "https://mcp-server.example.com"
AUTH_SERVER_JWKS_URI = "https://auth.example.com/.well-known/jwks.json"
 
async def get_jwks():
    """
    Authorization Server의 공개키를 가져와 토큰 서명을 검증합니다.
 
    주의: 이 구현은 설명 목적으로 캐싱 로직을 생략했습니다.
    프로덕션에서는 cachetools, httpx_auth 등을 이용한 캐싱 처리가 반드시
    필요합니다. 캐싱 없이 사용하면 매 요청마다 외부 호출이 발생해 레이턴시가
    증가하고, Authorization Server 장애 시 서비스가 함께 중단될 수 있습니다.
    """
    async with httpx.AsyncClient() as client:
        response = await client.get(AUTH_SERVER_JWKS_URI)
        return response.json()
 
async def verify_ephemeral_token(
    credentials: HTTPAuthorizationCredentials = Depends(security)
) -> dict:
    """
    RFC 8707 준수 검증:
    1. 서명 유효성
    2. 만료 여부 (exp)
    3. audience 클레임이 이 서버 URI와 일치하는지
    4. 필요한 scope 포함 여부
    """
    token = credentials.credentials
    jwks = await get_jwks()
 
    try:
        payload = jwt.decode(
            token,
            jwks,
            algorithms=["RS256"],
            audience=MCP_SERVER_URI,   # ← RFC 8707 핵심: aud 검증
            options={"verify_exp": True}
        )
    except JWTError as e:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail=f"토큰 검증 실패: {str(e)}",
            headers={"WWW-Authenticate": "Bearer"}
        )
 
    return payload
 
@app.get("/tools/files/read")
async def read_file(
    path: str,
    token_payload: dict = Depends(verify_ephemeral_token)
):
    """tool:files:read 스코프를 가진 단기 토큰만 허용합니다."""
    scopes = token_payload.get("scope", "").split()
 
    if "tool:files:read" not in scopes:
        raise HTTPException(
            status_code=status.HTTP_403_FORBIDDEN,
            detail="이 도구에 필요한 스코프가 없습니다: tool:files:read"
        )
 
    # TTL이 매우 짧으므로 여기까지 도달한 토큰은 유효한 것으로 신뢰
    # TODO: 실제 파일 읽기 로직 구현 필요
    return {"content": f"파일 읽기: {path}", "ttl_remaining": "검증됨"}

import { Client } from "@modelcontextprotocol/sdk/client/index.js";
import { StdioClientTransport } from "@modelcontextprotocol/sdk/client/stdio.js";
 
interface TokenRequest {
  resource: string;  // RFC 8707: 대상 서버 URI
  scope: string;     // 도구별 최소 권한
}
 
/**
 * client_credentials 그랜트로 단기 토큰을 요청합니다.
 *
 * 참고: PKCE(code_verifier/code_challenge)는 authorization_code 플로우에서만
 * 사용합니다. client_credentials 그랜트에는 적용되지 않습니다(RFC 7636).
 * 사용자 인가가 필요한 플로우라면 별도로 PKCE를 적용한 authorization_code
 * 플로우를 사용하세요.
 */
async function requestEphemeralToken(req: TokenRequest): Promise<string> {
  const response = await fetch("https://auth.example.com/token", {
    method: "POST",
    headers: { "Content-Type": "application/x-www-form-urlencoded" },
    body: new URLSearchParams({
      grant_type: "client_credentials",
      resource: req.resource,  // ← RFC 8707
      scope: req.scope,
      client_assertion_type:
        "urn:ietf:params:oauth:client-assertion-type:jwt-bearer",
      client_assertion: await generateClientAssertion(),
    }),
  });
 
  const { access_token } = await response.json();
  return access_token;
}
 
/**
 * authorization_code 플로우에서 PKCE를 사용할 경우,
 * code_verifier는 RFC 7636 §4.1에 따라 43-128자의 Base64URL 문자열이어야 합니다.
 * crypto.randomUUID()는 엔트로피가 충분하지 않으므로 아래 방식을 사용하세요.
 */
function generateCodeVerifier(): string {
  const array = new Uint8Array(32);
  crypto.getRandomValues(array);
  return btoa(String.fromCharCode(...array))
    .replace(/\+/g, "-")
    .replace(/\//g, "_")
    .replace(/=/g, "");
}
 
async function callToolWithEphemeralToken(
  toolName: string,
  toolArgs: Record<string, unknown>
) {
  const toolScopeMap: Record<string, { scope: string; ttl: number }> = {
    "read_file":    { scope: "tool:files:read",    ttl: 300 },
    "write_file":   { scope: "tool:files:write",   ttl: 180 },
    "execute_code": { scope: "tool:code:execute",  ttl: 120 },
  };
 
  const toolConfig = toolScopeMap[toolName];
  if (!toolConfig) throw new Error(`알 수 없는 도구: ${toolName}`);
 
  // 1. 도구 호출 직전 단기 토큰 요청
  const token = await requestEphemeralToken({
    resource: "https://mcp-server.example.com",
    scope: toolConfig.scope,
  });
 
  // 2. 토큰을 헤더에 담아 MCP 도구 호출
  // 참고: 이 예시는 도구 호출마다 새 연결을 생성합니다.
  //        프로덕션에서는 연결을 재사용하고 토큰만 갱신하는 방식이 일반적입니다.
  const transport = new StdioClientTransport({
    command: "mcp-server",
    env: { MCP_AUTH_TOKEN: token },
  });
 
  const client = new Client({ name: "agent", version: "1.0.0" }, {});
  await client.connect(transport);
 
  const result = await client.callTool({ name: toolName, arguments: toolArgs });
  await client.close();
 
  return result;
}

import httpx
 
# 취약한 방식: 장기 API 키 사용
headers_unsafe = {
    "Authorization": f"Bearer {LONG_LIVED_API_KEY}",
    # 만료: 없음, 범위: 전체, 대상: 모든 서비스
}
 
# 안전한 방식: Ephemeral Scoped Token + RFC 8707
async def publish_post_safely(post_content: str) -> dict:
    """
    scope=create:post, TTL=60초로 발급된 토큰만 사용.
    토큰이 탈취되더라도:
    - 60초 후 자동 만료
    - dev.to 서버 외 재사용 불가 (aud 클레임)
    - 새 글 생성 외 다른 작업 불가 (scope 제한)
 
    request_ephemeral_token()은 클라이언트 측 토큰 요청 함수입니다.
    시그니처:
        async def request_ephemeral_token(
            resource: str,   # RFC 8707 대상 서버 URI
            scope: str,      # 최소 권한 스코프
            ttl: int         # 토큰 수명 (초)
        ) -> str: ...
    예시 2의 requestEphemeralToken()과 동일한 역할을 Python으로 구현한 것입니다.
    """
    token = await request_ephemeral_token(
        resource="https://dev.to/api",
        scope="create:post",
        ttl=60  # 60초 TTL
    )
 
    async with httpx.AsyncClient() as client:
        response = await client.post(
            "https://dev.to/api/articles",
            headers={"Authorization": f"Bearer {token}"},
            json={"article": {"title": "...", "body_markdown": post_content}}
        )
 
    return response.json()