[pydantic-ai] FastMCPToolset 추가 — FastMCP 기반 MCP 클라이언트 통합

PR 링크: pydantic/pydantic-ai#2784 상태: Merged | 변경: +1205 / -15

들어가며

MCP(Model Context Protocol)는 LLM이 외부 도구를 호출하는 표준 프로토콜입니다. 기존 Pydantic AI는 MCP SDK 기반의 MCPServer 클라이언트만 제공했는데, FastMCP 프레임워크는 Tool Transformation, OAuth, 다양한 트랜스포트 등 추가 기능을 제공합니다. 이 PR은 FastMCP Client를 활용한 FastMCPToolset을 새로 도입합니다.

핵심 코드 분석

1. 유연한 입력 타입

@dataclass(init=False)
class FastMCPToolset(AbstractToolset[AgentDepsT]):
    def __init__(
        self,
        client: Client[Any] | ClientTransport | FastMCP | FastMCP1Server 
               | AnyUrl | Path | MCPConfig | dict[str, Any] | str,
        *,
        max_retries: int = 1,
        tool_error_behavior: Literal['model_retry', 'error'] = 'model_retry',
    ) -> None:
        if isinstance(client, Client):
            self.client = client
        else:
            self.client = Client[Any](transport=client)

FastMCP Server 인스턴스, URL, 파일 경로, JSON 설정 등 다양한 입력을 받아 자동으로 Client를 생성합니다. 이로써 사용자는 FastMCPToolset('http://localhost:8000/mcp') 한 줄로 연결할 수 있습니다.

2. 참조 카운팅 기반 수명 관리

async def __aenter__(self) -> Self:
    async with self._enter_lock:
        if self._running_count == 0:
            self._exit_stack = AsyncExitStack()
            await self._exit_stack.enter_async_context(self.client)
        self._running_count += 1
    return self

async def __aexit__(self, *args: Any) -> bool | None:
    async with self._enter_lock:
        self._running_count -= 1
        if self._running_count == 0 and self._exit_stack:
            await self._exit_stack.aclose()
            self._exit_stack = None
    return None

여러 agent가 동일한 FastMCPToolset을 공유할 때, 참조 카운팅으로 마지막 사용자가 나갈 때만 연결을 종료합니다. asyncio.Lock으로 동시 진입/퇴장을 보호합니다.

3. tool_error_behavior 설정

tool_error_behavior: Literal['model_retry', 'error'] = 'model_retry'

MCP tool 실행 에러 시 'model_retry'(LLM에게 재시도 요청)와 'error'(즉시 에러 전파) 중 선택할 수 있습니다. 기본값은 'model_retry'로, LLM이 입력을 수정하여 재시도할 기회를 줍니다.

왜 이게 좋은가

• 다양한 연결 방식: URL, stdio, Python/Node.js 스크립트, JSON 설정 등 거의 모든 MCP 서버 연결 방식을 지원합니다.
• 로컬 서버 직접 연결: 동일 프로세스의 FastMCP Server에 네트워크 없이 직접 연결하여 지연 시간을 제거합니다.
• 기존 MCP 생태계 호환: FastMCP로 만들지 않은 서버에도 연결 가능합니다.

정리

• 다형적 입력은 사용자 편의를 극대화한다: 여러 타입을 받아 내부에서 통일된 객체로 변환하는 패턴은 DX를 크게 개선합니다.
• 공유 리소스는 참조 카운팅으로 관리하라: 여러 소비자가 공유하는 네트워크 연결은 마지막 소비자가 정리해야 합니다.

참고 자료

• pydantic/pydantic-ai#2784 — PR 전체 diff
• FastMCP 공식 문서 — FastMCP 프레임워크
• MCP 프로토콜 — Model Context Protocol 사양

⚠️ 알림: 이 분석은 AI가 실제 코드 diff를 기반으로 작성했습니다.