MCP(Model Context Protocol) 멀티에이전트 위임 패턴 심층 분석 — OAuth Token Exchange(RFC 8693)와 단방향 권한 축소로 에이전트 체인 보안 설계하기

{
  "sub": "user-alice-001",       // 원본 사용자 (변경되지 않음)
  "act": {
    "sub": "orchestrator-agent-A",  // 1차 위임: 오케스트레이터
    "act": {
      "sub": "subagent-A1",         // 2차 위임: 서브에이전트
      "client_id": "mcp-tool-caller"
    }
  },
  "scope": "read:documents",    // 실제 부여된 권한
  "aud": "mcp-server-docs",     // 이 토큰을 수신할 MCP 서버
  "exp": 1775000000             // 만료 시각 (Unix timestamp)
}

// RFC 8693 Token Exchange 요청 예시 (TypeScript)
async function exchangeTokenForAgent(
  userToken: string,
  agentClientId: string,
  requestedScopes: string[]
): 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: "urn:ietf:params:oauth:grant-type:token-exchange",
      subject_token: userToken,
      subject_token_type: "urn:ietf:params:oauth:token-type:access_token",
      requested_token_type: "urn:ietf:params:oauth:token-type:access_token",
      audience: "mcp-server-docs",
      scope: requestedScopes.join(" "),
      client_id: agentClientId,
    }),
  });
 
  if (!response.ok) {
    const errorBody = await response.text();
    throw new Error(
      `Token exchange failed (${response.status}): ${errorBody}`
    );
  }
 
  const body = await response.json();
 
  if (!body.access_token) {
    throw new Error("Token exchange succeeded but no access_token returned");
  }
 
  return body.access_token; // act 클레임이 포함된 위임 토큰
}

# Biscuit 토큰으로 단방향 권한 축소 구현 (Python)
# 실제 biscuit-python 패키지 API는 아래와 다를 수 있습니다.
# 블록 추가 시 키페어 처리 방식이 다르므로 공식 문서를 참고하세요.
# pip install biscuit-python
from biscuit_auth import BiscuitBuilder, BlockBuilder, KeyPair
 
root_keypair = KeyPair()  # 실제 환경에서는 환경 변수나 HSM에서 로드합니다
 
# 오케스트레이터용 루트 토큰 발급
root_token = (
    BiscuitBuilder()
    .fact('right("docs", "read")')
    .fact('right("logs", "write")')
    .fact("budget(100)")
    .build(root_keypair.private_key)
)
 
# 서브에이전트 A에게 위임 — 권한을 read only, 예산 40으로 축소
attenuated_token = root_token.append_block(
    BlockBuilder()
    .check('check if right($resource, "read")')   # write 제거
    .check("check if budget($b), $b <= 40")        # 예산 40으로 제한
)
 
# 서브에이전트 A-1에게 재위임 — docs만 허용, 예산 10으로 추가 축소
further_attenuated = attenuated_token.append_block(
    BlockBuilder()
    .check('check if right("docs", "read")')       # logs 제거
    .check("check if budget($b), $b <= 10")
)

사용자(Alice)         IdP              에이전트          MCP 서버
    │                  │                   │                │
    │── 로그인 ────────>│                   │                │
    │<─ id_token ───────│                   │                │
    │── id_token 전달 ──────────────────────>│                │
    │                  │<── Token Exchange ─│                │
    │                  │    (RFC 8693 OBO)  │                │
    │                  │── act 토큰 ────────>│                │
    │                  │                   │── API 호출 ────>│
    │                  │                   │   (sub=Alice,   │
    │                  │                   │    act=AgentA)  │
    │                  │                   │<── 응답 ────────│
    │                  │                   │                │── 감사 로그
    │                  │                   │                    (sub+act 기록)

import { DelegationChain, DelegationToken, Budget } from "@newtro/delegateos";
 
// orchestratorPrivateKey는 환경 변수나 HSM(Hardware Security Module)에서
// 로드해야 합니다. 코드에 하드코딩하면 안 됩니다.
// 예: const orchestratorPrivateKey = loadKeyFromEnv("ORCHESTRATOR_PRIVATE_KEY");
 
// 1. 오케스트레이터 루트 토큰 발급
const orchestratorChain = await DelegationChain.create({
  subject: "user-alice-001",
  actor: "orchestrator-agent",
  permissions: ["read", "write"],
  budget: new Budget({ tokens: 100, apiCalls: 500 }),
  expiresIn: "1h",
  signingKey: orchestratorPrivateKey,
});
 
// 2. 서브에이전트 A에게 위임 (read only, 예산 40)
const subAgentAToken: DelegationToken = await orchestratorChain.delegate({
  to: "subagent-a",
  permissions: ["read"],                     // write 제거
  budget: new Budget({ tokens: 40, apiCalls: 200 }),
  attestation: {
    taskId: "task-summarize-docs",
    reason: "문서 요약 작업",
    timestamp: Date.now(),
  },
});
 
// 3. 서브에이전트 A-1에게 재위임 (read:docs only, 예산 10)
const subAgentA1Token: DelegationToken = await subAgentAToken.delegate({
  to: "subagent-a1",
  permissions: ["read:docs"],               // 리소스 범위 추가 축소
  budget: new Budget({ tokens: 10, apiCalls: 50 }),
  attestation: {
    taskId: "task-summarize-docs-section-1",
    reason: "1장 요약",
    timestamp: Date.now(),
  },
});
 
// 4. 서브에이전트 B에게 위임 (write:logs only, 예산 60)
const subAgentBToken: DelegationToken = await orchestratorChain.delegate({
  to: "subagent-b",
  permissions: ["write:logs"],
  budget: new Budget({ tokens: 60, apiCalls: 300 }),
  attestation: {
    taskId: "task-write-audit-log",
    reason: "감사 로그 기록",
    timestamp: Date.now(),
  },
});
 
// 5. MCP 서버 호출 시 토큰 전달
const mcpResponse = await mcpClient.callTool(
  "read_document",
  { documentId: "doc-001" },
  {
    headers: {
      Authorization: `Bearer ${subAgentA1Token.serialize()}`,
      "X-Delegation-Chain": subAgentA1Token.getChainProof(),
    },
  }
);

from agent_governance_toolkit import AgentRuntime, ExecutionRing, Policy
 
# 정책 정의: Ring 별 허용 작업
policy = Policy.from_yaml("""
rings:
  ring0:  # 오케스트레이터 (최고 신뢰)
    allowed_actions: ["*"]
    budget_limit: 10000
  ring1:  # 내부 서브에이전트
    allowed_actions: ["read:*", "write:logs", "call:mcp_tools"]
    budget_limit: 1000
    max_delegation_depth: 2
  ring2:  # 외부 API 호출 에이전트
    allowed_actions: ["read:public", "call:external_api"]
    budget_limit: 100
    max_delegation_depth: 0  # 재위임 금지
  ring3:  # 미검증 외부 에이전트
    allowed_actions: []       # 모든 작업 차단
    require_human_approval: true
""")
 
runtime = AgentRuntime(policy=policy)
 
# 에이전트 등록 및 링 배정
orchestrator = runtime.register_agent(
    agent_id="orchestrator-001",
    ring=ExecutionRing.RING0,
    identity_token=orchestrator_jwt,
)
 
subagent_a = runtime.register_agent(
    agent_id="subagent-a",
    ring=ExecutionRing.RING1,
    identity_token=subagent_a_jwt,
)
 
# 위임 실행 — Ring 정책 자동 적용
async def run_task():
    async with runtime.delegation_context(
        delegator=orchestrator,
        delegatee=subagent_a,
        task_scope=["read:documents"],
    ) as ctx:
        result = await subagent_a.call_mcp_tool(
            "read_document",
            args={"id": "doc-001"},
            delegation_context=ctx,
            # Ring 정책 엔진이 모든 호출을 실시간으로 검사합니다
        )
        # 감사 로그 자동 생성 (EU AI Act, HIPAA, SOC 2 매핑 포함)
        return result

import { TokenVault } from "@auth0/agent-toolkit";
 
const vault = new TokenVault({
  domain: "your-tenant.auth0.com",
  clientId: process.env.AUTH0_CLIENT_ID,
});
 
// 에이전트가 Google Drive 접근이 필요할 때
async function getGoogleDriveAccess(
  userToken: string,
  agentId: string
): Promise<string> {
  // Token Vault가 내부적으로 RFC 8693 Token Exchange 수행
  const { access_token } = await vault.exchangeForService({
    subjectToken: userToken,
    targetService: "google-drive",
    actor: agentId,
    scopes: ["https://www.googleapis.com/auth/drive.readonly"],
    maxDuration: 3600, // 원본 사용자 세션보다 길어질 수 없습니다
  });
 
  // 원본 Google OAuth 토큰은 Vault에만 존재, 에이전트에 노출되지 않습니다
  return access_token; // 단기 위임 토큰만 반환
}
 
// MCP 도구 구현
const googleDriveTool = {
  name: "list_drive_files",
  handler: async (args: { folderId: string }, context: MCPContext) => {
    const driveToken = await getGoogleDriveAccess(
      context.userToken,
      context.agentId
    );
    // driveToken은 1시간짜리 단기 토큰으로, 유출되더라도 피해가 최소화됩니다
    return await googleDriveClient.listFiles(args.folderId, driveToken);
  },
};