[pydantic-ai] FastMCPToolset Temporal 통합 — MCP 툴셋 공통 추상화

PR 링크: pydantic/pydantic-ai#3413 상태: Merged | 변경: +1454 / -147

들어가며

Pydantic AI는 MCP(Model Context Protocol) 서버와 통신하는 두 가지 방식을 지원합니다: 표준 MCPServer와 FastMCP 기반의 FastMCPToolset. Temporal 워크플로우 환경에서는 모든 I/O 작업이 Activity로 래핑되어야 하는데, 기존에는 TemporalMCPServer만 존재했고 FastMCPToolset은 Temporal과 통합되지 않았습니다. 이 PR은 두 구현의 공통 로직을 TemporalMCPToolset 추상 클래스로 추출하고, TemporalFastMCPToolset을 새로 만들어 FastMCP도 Temporal에서 동작하도록 합니다.

핵심 코드 분석

1. 공통 추상 클래스 추출

Before: TemporalMCPServer가 모든 Temporal 통합 로직을 직접 포함

After (_mcp.py):

class TemporalMCPToolset(TemporalWrapperToolset[AgentDepsT], ABC):
    @abstractmethod
    def tool_for_tool_def(self, tool_def: ToolDefinition) -> ToolsetTool[AgentDepsT]:
        raise NotImplementedError

    async def get_tools(self, ctx: RunContext[AgentDepsT]) -> dict[str, ToolsetTool[AgentDepsT]]:
        if not workflow.in_workflow():
            return await super().get_tools(ctx)
        # Temporal activity로 위임
        tool_defs = await workflow.execute_activity(
            activity=self.get_tools_activity,
            args=[_GetToolsParams(serialized_run_context=serialized_run_context), ctx.deps],
            **self.activity_config,
        )
        return {name: self.tool_for_tool_def(tool_def) for name, tool_def in tool_defs.items()}

get_tools()와 call_tool() 로직을 추상 클래스로 올리고, tool_for_tool_def()만 하위 클래스에서 구현하도록 합니다.

2. TemporalMCPServer 단순화

Before: 147줄의 전체 Temporal 통합 로직

After (_mcp_server.py):

class TemporalMCPServer(TemporalMCPToolset[AgentDepsT]):
    def tool_for_tool_def(self, tool_def: ToolDefinition) -> ToolsetTool[AgentDepsT]:
        assert isinstance(self.wrapped, MCPServer)
        return self.wrapped.tool_for_tool_def(tool_def)

공통 로직이 상위 클래스로 이동하여, 기존 클래스는 tool_for_tool_def() 구현만 남습니다.

3. TemporalFastMCPToolset 추가

class TemporalFastMCPToolset(TemporalMCPToolset[AgentDepsT]):
    def tool_for_tool_def(self, tool_def: ToolDefinition) -> ToolsetTool[AgentDepsT]:
        assert isinstance(self.wrapped, FastMCPToolset)
        return self.wrapped.tool_for_tool_def(tool_def)

동일한 패턴으로 FastMCP 버전을 구현합니다. beartype 의존성도 Temporal worker의 sandbox 허용 목록에 추가했습니다.

왜 이게 좋은가

• 중복 제거: 147줄의 코드가 공통 추상 클래스로 합쳐져, 두 구현 모두 10줄 미만의 하위 클래스가 됩니다.
• Template Method 패턴: tool_for_tool_def()라는 하나의 추상 메서드만 구현하면 새로운 MCP 툴셋을 Temporal에 통합할 수 있습니다.
• 워크플로우 외 동작: if not workflow.in_workflow() 분기로 Temporal 외부에서도 정상 동작합니다.

정리

• 공통 로직은 추상 클래스로 추출하라: 유사한 구현이 2개 이상이면 Template Method 패턴을 고려하십시오.
• 분산 환경의 직렬화 제약을 고려하라: ToolsetTool에 SchemaValidator가 포함되어 직렬화가 불가하므로, ToolDefinition만 전달하고 Activity 밖에서 재구성합니다.

참고 자료

• pydantic/pydantic-ai#3413 — PR 전체 diff
• Temporal Activity 문서 — Activity 개념 설명

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