--- title: "도구 연동과 MCP 규격: Tool Calling과 스키마 검증" slug: "ai-agent-engineering-03-tools-mcp" canonicalUrl: "https://moonshotnotes.com/posts/ai-agent-engineering-03-tools-mcp/" sourceUrl: "https://moonshotnotes.com/posts/ai-agent-engineering-03-tools-mcp/" markdownUrl: "https://moonshotnotes.com/agent/posts/ai-agent-engineering-03-tools-mcp.md" language: "ko" category: "AI Agent" updatedAt: "2026-07-25" agentTokenEstimate: 7495 --- # 도구 연동과 MCP 규격: Tool Calling과 스키마 검증 LLM Tool Calling 메커니즘부터 Model Context Protocol(MCP) 표준 규격, Pydantic v2 기반 스키마 검증, 방어적 인자 검증 체크리스트까지 엔터프라이즈 에이전트 도구 연동의 전 과정을 다룹니다. ## Agent metadata - Source: https://moonshotnotes.com/posts/ai-agent-engineering-03-tools-mcp/ - Markdown: https://moonshotnotes.com/agent/posts/ai-agent-engineering-03-tools-mcp.md - Language: ko - Category: AI Agent - Tags: AI Agent, MCP, Model Context Protocol, Tool Calling, Function Calling, Pydantic v2, Schema Validation, Agent Engineering - Updated: 2026-07-25 - Estimated tokens: 7495 > 본 포스트는 'AI 에이전트 엔지니어링(AI Agent Engineering)' 시리즈의 3편입니다. LLM Tool Calling의 5단계 생명주기, Model Context Protocol(MCP) 규격, 그리고 Pydantic v2 기반 방어적 스키마 검증 기법을 심층 파헤칩니다. --- ## 1. 도구 정의와 LLM Tool Calling 메커니즘 LLM(대화형 언어 모델)이 격리된 상자(Sandbox)를 벗어나 외부 실세계를 조작하기 위해서는 도구(Tool / Function)를 호출할 수 있어야 합니다. 초기 프롬프트 엔지니어링 기법에서는 ReAct(Reasoning + Acting) 패턴처럼 일반 텍스트 출력 속에 `Action: search_database("query")` 형태의 문자열 파싱을 사용했으나, 이는 괄호 누락이나 인자 타입 오염 등 심각한 구문 오류(Syntax Error)를 야기했습니다. 현대의 프로덕션 에이전트 런타임은 LLM 공급업체(OpenAI, Anthropic, Google 등)가 제공하는 네이티브 Tool Calling (Function Calling) 기능에 의존합니다. ### 1.1 ReAct 텍스트 파싱 vs 네이티브 Tool Calling 비교 | 비교 항목 | ReAct 텍스트 파싱 (Prompt Parsing) | 네이티브 Tool Calling (Native API) | | :--- | :--- | :--- | | 구문 파싱 방식 | LLM 텍스트 출력에서 Regex/문자열 파싱 | API 제공 JSON/Protobuf Structured Payload | | 타입 안전성 | 낮음 (문자열 오염, 괄호 누락 빈번) | 높음 (JSON Schema 기반 문법 제어 생성) | | 토큰 효율성 | 중간 (ReAct 템플릿 프롬프트 오버헤드) | 최적화 (API 레벨 `tools` 매개변수 주입) | | 복수 도구 호출 | 구현 복잡 (병렬 호출 처리 어려움) | 기본 지원 (`tool_calls` 배열로 병렬 수행) | | 오류 감지 속도 | 파싱 시점 (런타임 Regex Fail) | API 응답 시점 (구문 검증 완료된 JSON) | ### 1.2 네이티브 Tool Calling 5단계 생명주기 (Lifecycle Architecture): ```mermaid graph TD subgraph "Stage 1: Tool Schema Registration" A1["Pydantic / Python Handler"] --> A2["JSON Schema Generator"] A2 --> A3["Tools Array (JSON Schemas)"] end subgraph "Stage 2: API Payload Transmission" A3 --> B1["API Request Payload Inject"] B2["User Goal & History"] --> B1 end subgraph "Stage 3: LLM Reasoning & Constrained Decoding" B1 --> C1["LLM Provider Engine"] C1 --> C2{"Tool Execution Needed?"} C2 -- "No" --> C3["Standard Content Text Output"] C2 -- "Yes" --> C4["Grammar-Guided Constrained Sampler"] C4 --> C5["Native tool_calls JSON Output"] end subgraph "Stage 4: Runtime Validation & Dispatch" C5 --> D1["Agent Runtime Parser"] D1 --> D2{"Pydantic Schema Validation"} D2 -- "Fail" --> D3["Diagnostic Error Generation"] D2 -- "Pass" --> D4["Tool Handler Execution"] end subgraph "Stage 5: Context Feedback Loop" D3 --> E1["Inject Role: tool (Error Feedback)"] D4 --> E2["Inject Role: tool (Success Result)"] E1 --> B1 E2 --> B1 end ``` ### LLM Tool Calling의 5단계 세부 동작 1. 도구 정의 전달 (Tool Schema Injection): 호스트 애플리케이션은 사용 가능한 도구 목록을 JSON Schema 래퍼로 변환하여 LLM 요청 파라미터(`tools`)로 주입합니다. 2. 페이로드 전송 (Payload Transmission): 사용자 입력, 대화 히스토리, 도구 스키마를 묶어 LLM API로 전송합니다. 3. 문법 제어 샘플링 (Constrained Decoding & Tool Call Generation): 모델은 프롬프트 컨텍스트와 사용자의 목표를 분석하여 정형화된 JSON 형태의 `tool_calls` 객체(`id`, `name`, `arguments`)를 생성합니다. 4. 호스트 실행 및 검증 (Host Validation & Execution): 에이전트 런타임은 `tool_calls` 배열을 수신하여 해당 도구의 핸들러(Handler)를 찾고, 인자값을 검증한 뒤 실행합니다. 5. 결과 반환 및 대화 환류 (Context Feedback Loop): 실행 결과(성공 텍스트 또는 JSON 에러)를 `role: "tool"` 메시지로 변환하여 컨텍스트 히스토리에 추가하고 모델에게 재추론을 요청합니다. > 컨텍스트 오버헤드 주의: 수십 개의 도구 스키마를 무분별하게 `tools` 배열에 주입하면 매 요청마다 수천 토큰 이상의 프롬프트 비용이 발생하며, 도구 간 이름/설명 충돌로 인해 모델의 도구 선택 정확도가 저하됩니다. > 스키마 최적화 노하우: 도구 설정을 작성할 때 `description`을 명확하고 구체적으로 기술해야 합니다. 모델은 함수 이름보다 `description`을 우선시하여 도구를 선택합니다. --- ## 2. 도구 실행 제어 루프 및 시퀀스 다이어그램 도구 실행은 단발성 호출로 끝나지 않습니다. 도구 실행 결과가 예외를 반환하거나 예상과 다른 데이터가 반환될 경우, LLM이 오류 메시지를 관찰(Observe)하여 스스로 인자값을 수정하거나 대체 도구를 호출하는 자율 복구 루프(Self-Correction Loop)가 동작해야 합니다. ### 2.1 에이전트 도구 실행 전과정 시퀀스 다이어그램: ```mermaid flowchart TD User["User"] --> |1. 과업 요청 (Goal)| Agent["Agent"] Agent --> |2. 활성 도구 스키마 목록 조회 (JSON Schema)| Reg["Registry"] Reg --> |3. Top-K 선택된 Tool Schemas| Agent Agent --> |4. Prompt + Conversation History + Tools| LLM["LLM"] LLM --> |5. Response (with tool_calls)| Agent Agent --> |6. 핸들러 검색 & Pydantic 검증| Reg Reg --> |7. MCP stdio/SSE tools/call 요청| MCP["MCP"] MCP --> |8. 실행 결과 (JSON / Text Result)| Reg Reg --> |9. Standardized ToolResult| Agent Agent --> |10. 대화 히스토리에 role: tool 결과 추가| Agent Agent --> |11. 구문/타입 에러 진단 텍스트 생성| Agent Agent --> |12. Diagnostic Error 추가 (Self-Correction 유도)| Agent Agent --> |13. Final Result Context| LLM LLM --> |14. 최종 텍스트 답변 (Final Answer)| Agent Agent --> |15. 과업 완수 결과 반환| EndState["과업 완수 결과 반환"] ``` ### 2.2 자율 복구 루프 (Self-Correction Loop) 메커니즘: ```mermaid stateDiagram-v2 [*] --> LLM_Inference: 사용자 요청 전송 LLM_Inference --> ToolCall_Generated: tool_calls 제너레이션 LLM_Inference --> Final_Answer: 일반 텍스트 응답 (종료) ToolCall_Generated --> Schema_Validation: JSON 파싱 및 Pydantic 검증 state Schema_Validation { [*] --> Syntax_Check Syntax_Check --> Type_Check: Valid JSON Syntax Syntax_Check --> Diagnostic_Syntax_Error: JSONDecodeError Type_Check --> Business_Rule_Check: Valid Types Type_Check --> Diagnostic_Type_Error: ValidationError Business_Rule_Check --> Validated: All Rules Pass Business_Rule_Check --> Diagnostic_Rule_Error: Business Rule Fail } Diagnostic_Syntax_Error --> LLM_Self_Correction: 진단 힌트 포함 (role: tool) Diagnostic_Type_Error --> LLM_Self_Correction: 필드 타입 오류 힌트 포함 (role: tool) Diagnostic_Rule_Error --> LLM_Self_Correction: 비즈니스 제약 실패 힌트 포함 (role: tool) LLM_Self_Correction --> Retry_Check: 재시도 횟수 확인 Retry_Check --> LLM_Inference: 인자 수정 후 재추론 Retry_Check --> Max_Retry_Exceeded: Threshold 초과 Validated --> Tool_Execution: External API / Handler 실행 state Tool_Execution { [*] --> Executing Executing --> Execution_Success: Success 200 OK Executing --> Execution_Failure: Exception / HTTP 5xx } Execution_Failure --> LLM_Self_Correction: 실행 예외 메시지 피드백 Execution_Success --> LLM_Inference: 성공 결과 피드백 (role: tool) Max_Retry_Exceeded --> Final_Answer: 에러 안내 및 안전한 폴백 출력 ``` ### 2.3 에이전트 도구 실패 처리 전략 비교 | 처리 전략 | 동작 메커니즘 | 장점 | 단점 / 고려사항 | | :--- | :--- | :--- | :--- | | Self-Correction (자율 복구) | LLM에 진단 에러 반환하여 인자 수정 유도 | 모델의 자율적 수정 능력 활용 | 토큰 추가 소비, 재시도 횟수 제한 필수 | | Tool Fallback (대체 도구) | 특정 도구 실패 시 동종 대체 도구로 전환 | 서비스 연속성 보장 | 대체 도구 등록 및 라우팅 로직 필요 | | Human-in-the-Loop (인간 개입) | 치명적 에러 시 운영자 승인/수정 요청 | 안전성 및 정확도 극대화 | 실시간 자동화 지연 발생 | | Graceful Degradation (우아한 강등) | 실패 시 기본값/부족한 정보로 대치 응답 | 전체 시스템 멈춤 방지 | 정보 불완전성 가능성 | > 무한 루프 방지 (Max Retries Bound): 자율 복구 루프는 반드시 `max_retries` (예: 3회) 제한을 두어야 합니다. 잘못된 스키마로 인해 모델이 동일 오류를 반복 생성하는 경우, 토큰 폭발과 무한 대기 현상을 차단해야 합니다. --- ## 3. Model Context Protocol (MCP) 규격과 아키텍처 에이전트 생태계가 확장됨에 따라 호스트 애플리케이션(Claude Desktop, Cursor, Custom Agent Framework)과 개별 도구(PostgreSQL, GitHub, Slack, Local Filesystem) 간의 표준화된 통신 인터페이스가 필요해졌습니다. ### 3.1 MCP 3대 핵심 원시 요소 (Primitives) 비교 MCP는 JSON-RPC 2.0 프로토콜 기반 위에서 세 가지 원시 요소를 제공합니다. | Primitive | 주요 JSON-RPC 메서드 | 접근 방식 | Side-Effect 여부 | 주요 용도 및 엔터프라이즈 활용 예시 | | :--- | :--- | :--- | :--- | :--- | | Tools | `tools/list`, `tools/call` | Executable | 있음 (상태 변경 가능) | DB Write, API 호출, 이메일 발송, 파일 수정 | | Resources | `resources/list`, `resources/read` | Read-Only | 없음 (안전함) | 파일 읽기, DB Select 조회, 로그 스트림 수집 | | Prompts | `prompts/list`, `prompts/get` | Template | 없음 (안전함) | 서버 측 중앙 관리 표준 프롬프트, 템플릿 주입 | ### 3.2 MCP Stdio vs SSE Transport 레이어 아키텍처: ```mermaid graph TD subgraph "MCP Client Host Runtime" HostApp["Host Application (Claude / Cursor / Custom Agent)"] ClientMgr["MCP Client Manager"] HostApp --> ClientMgr end subgraph "Transport Layer Breakdown" direction LR subgraph "Stdio Transport (Local IPC)" StdioClient["Stdio Transport Client"] StdinChannel["Child Process stdin (NDJSON)"] StdoutChannel["Child Process stdout (NDJSON)"] StdioClient --> StdinChannel StdoutChannel --> StdioClient end subgraph "SSE Transport (Remote Network)" SSEClient["SSE Transport Client"] HTTPPost["HTTP POST Channel (/messages)"] SSERequest["SSE Event Stream Channel (/sse)"] SSEClient --> HTTPPost SSERequest --> SSEClient end end ClientMgr --> StdioClient ClientMgr --> SSEClient subgraph "MCP Server Layer" StdioServer["Local MCP Server Process"] SSEServer["Remote MCP HTTP/SSE Server"] StdinChannel --> StdioServer StdioServer --> StdoutChannel HTTPPost --> SSEServer SSEServer --> SSERequest end subgraph "Server Target Execution" StdioServer --> LocalDB[("Local File System / CLI Tools")] SSEServer --> CloudAPI["Cloud Enterprise APIs / PostgreSQL"] end ``` ### 3.3 MCP Stdio vs SSE 전송 레이어 상세 비교 | 전송 레이어 (Transport) | 통신 방식 | 인증 & 보안 | 레이턴시 | 주요 유즈케이스 | | :--- | :--- | :--- | :--- | :--- | | Stdio Transport | 로컬 하위 프로세스 `stdin`/`stdout` | OS 프로세스 격리 및 OS 파일 권한 | 극히 낮음 (Zero Network) | 개발자 로컬 도구, CLI 도구, Claude Desktop | | SSE Transport | HTTP POST + Server-Sent Events | Bearer Token, OAuth2, TLS | 네트워크 RTT 발생 | 원격 SaaS, 분산 서버, 멀티테넌트 MCP | > JSON-RPC 2.0 표준 규격: MCP는 모든 메시지 교환 시 `jsonrpc: "2.0"`, `method`, `params`, `id` 필드를 갖는 JSON-RPC 규격을 따릅니다. 이를 통해 클라이언트와 서버의 구현 언어가 달라도 완벽한 호환성을 제공합니다. > 로컬 개발과 원격 분산 구동: 개발 단계에서는 오버헤드가 없는 Stdio 트랜스포트로 구축한 후, 프로덕션 배포 시 SSE/HTTP 트랜스포트로 전환하여 마이크로서비스로 확장하는 패턴이 권장됩니다. --- ## 4. MCP 파이썬 실전 구현 (Server & Client) 이 절에서는 현대적인 파이썬 백엔드 환경에서 MCP 서버를 구축하고, 에이전트 클라이언트가 이와 연결하여 도구를 동적으로 가져와 실행하는 프로덕션 레벨 코드 구조를 구축합니다. ### 4.1 MCP Handshake 및 메시지 교환 통신 시퀀스: ```mermaid flowchart TD Client["Client"] --> |1. connect() spawn process / open stream| Transport["Transport"] Client --> |2. JSON-RPC request: initialize (protocolVersion, capabilities)| Server["Server"] Server --> |3. JSON-RPC response: initialize (serverInfo, capabilities)| Client Client --> |4. JSON-RPC notification: initialized| Server Client --> |5. JSON-RPC request: tools/list| Server Server --> |6. JSON-RPC response: tools/list (Tool Schemas)| Client Client --> |7. MCP Schema to OpenAI / Anthropic Schema Transform| Client Client --> |8. JSON-RPC request: tools/call (name='get_user_profile', arguments={...})| Server Server --> |9. Pydantic Validation & Handler Execution| Server Server --> |10. JSON-RPC response: tools/call (content: [{type: 'text', text: '...'}])| Client ``` ### 4.2 MCP 도구 서버 구현 (`mcp_server.py`) ```python """ MCP Server Implementation using FastMCP / Async patterns. 이 서버는 엔터프라이즈 데이터베이스 조회 및 사용자 권한 변경 도구를 제공합니다. """ import asyncio import logging from typing import Dict, Any from pydantic import BaseModel, Field from mcp.server.fastmcp import FastMCP logging.basicConfig(level=logging.INFO) logger = logging.getLogger("EnterpriseMCPServer") # FastMCP 서버 인스턴스 초기화 mcp = FastMCP("Enterprise-Data-MCP-Server") # ========================================== # Pydantic Input Schemas # ========================================== class UserQueryInput(BaseModel): user_id: str = Field(description="조회할 사용자의 고유 UUID 키", json_schema_extra={"example": "usr_102938"}) include_deleted: bool = Field(default=False, description="탈퇴된 사용자 계정 포함 여부") class UserRoleUpdateInput(BaseModel): user_id: str = Field(description="권한을 변경할 사용자의 ID") target_role: str = Field(description="부여할 새로운 역할 (admin, operator, viewer)") reason: str = Field(min_length=10, description="권한 변경 사유 (최소 10자 이상 필수)") # ========================================== # MCP Tool Registration # ========================================== @mcp.tool( name="get_user_profile", description="사용자 ID를 기반으로 상세 프로필 데이터와 최근 활동 내역을 데이터베이스에서 조회합니다." ) async def get_user_profile(user_id: str, include_deleted: bool = False) -> Dict[str, Any]: # 방어적 검증 (서버 내부 이차 검증) input_data = UserQueryInput(user_id=user_id, include_deleted=include_deleted) logger.info(f"User Query Executed for ID: {input_data.user_id}") # 가상 DB 조회 시뮬레이션 if input_data.user_id == "usr_404": return {"status": "error", "message": f"User '{input_data.user_id}' not found."} return { "status": "success", "data": { "user_id": input_data.user_id, "name": "Jane Doe", "email": "user_account_id", "role": "operator", "is_active": True, } } @mcp.tool( name="update_user_role", description="사용자의 시스템 접근 권한(Role)을 변경합니다. 고위험 도구이므로 사유 작성이 필수적입니다." ) async def update_user_role(user_id: str, target_role: str, reason: str) -> Dict[str, Any]: # Pydantic 스키마 기반 데이터 정제 try: payload = UserRoleUpdateInput(user_id=user_id, target_role=target_role, reason=reason) except Exception as e: return {"status": "error", "message": f"Schema Validation Failed: {str(e)}"} if payload.target_role not in ["admin", "operator", "viewer"]: return {"status": "error", "message": f"Invalid role '{payload.target_role}'. Must be one of [admin, operator, viewer]."} logger.warning(f"SECURITY EVENT: User {payload.user_id} role changed to {payload.target_role} (Reason: {payload.reason})") return { "status": "success", "result": { "user_id": payload.user_id, "updated_role": payload.target_role, "audit_logged": True } } if __name__ == "__main__": # Stdio 모드로 MCP 서버 구동 mcp.run(transport="stdio") ``` ### 4.3 MCP 에이전트 클라이언트 연동 (`mcp_client.py`) ```python """ MCP Client Manager that connects to stdio MCP Server and exposes JSON-RPC tools to LLM. """ import sys import asyncio from typing import List, Dict, Any, Optional from contextlib import AsyncExitStack from mcp import ClientSession, StdioServerParameters from mcp.client.stdio import stdio_client class AgentMCPClientManager: """MCP 서버들과의 라이프사이클을 관리하고 도구 스키마를 LLM 표준 규격으로 변환하는 클라이언트 클래스""" def __init__(self, server_script_path: str): self.server_script_path = server_script_path self.session: Optional[ClientSession] = None self._exit_stack = AsyncExitStack() async def connect(self): """Python 하위 프로세스로 MCP 서버를 시작하고 stdio 채널을 개설합니다.""" server_params = StdioServerParameters( command=sys.executable, args=[self.server_script_path], env=None ) # stdio 트랜스포트 및 세션을 AsyncExitStack으로 수명주기 안전 관리 read_stream, write_stream = await self._exit_stack.enter_async_context( stdio_client(server_params) ) self.session = await self._exit_stack.enter_async_context( ClientSession(read_stream, write_stream) ) await self.session.initialize() print("[MCP Client] Connection successfully established to MCP Server.") async def get_llm_tool_definitions(self) -> List[Dict[str, Any]]: """MCP 서버의 tools/list를 호출하여 LLM(OpenAI/Anthropic) 규격의 JSON Schema 목록으로 변환합니다.""" if not self.session: raise RuntimeError("MCP Session is not initialized.") mcp_tools_response = await self.session.list_tools() llm_tools = [] for tool in mcp_tools_response.tools: # MCP Tool Schema to OpenAI Function Calling Schema 변환 input_schema = getattr(tool, "inputSchema", getattr(tool, "input_schema", {})) llm_tools.append({ "type": "function", "function": { "name": tool.name, "description": tool.description or "", "parameters": input_schema } }) return llm_tools async def call_tool(self, tool_name: str, arguments: Dict[str, Any]) -> Dict[str, Any]: """MCP 서버로 tools/call 메시지를 전송하고 결과를 수신합니다.""" if not self.session: raise RuntimeError("MCP Session is not initialized.") result = await self.session.call_tool(tool_name, arguments) # MCP TextContent / ImageContent 응답 파싱 output_chunks = [] for content in result.content: if getattr(content, "type", None) == "text": output_chunks.append(content.text) is_error = getattr(result, "isError", getattr(result, "is_error", False)) return { "is_error": bool(is_error), "raw_output": "\n".join(output_chunks) } async def close(self): """MCP 클라이언트 자원을 안전하게 해제합니다.""" await self._exit_stack.aclose() self.session = None print("[MCP Client] Connection safely closed.") ``` > 보안 위협 주의 (Command & Tool Injection): Stdio MCP 서버가 하위 프로세스로 구동될 때, 클라이언트 입력이 검증 없이 `os.system`이나 Shell 명령으로 실행되면 시스템 전체 권한이 탈취될 수 있습니다. 반드시 파라미터를 Sanitization하고 Pydantic 검증을 수행하세요. > `AsyncExitStack` 자원 관리: MCP 클라이언트 구현 시 `AsyncExitStack`을 활용하면 프로세스 예외 종료나 네트워크 단절 시에도 하위 프로세스의 `stdin`/`stdout` 스트림 및 세션을 안전하게 정리할 수 있습니다. --- ## 5. Tool Registry 패턴과 동적 도구 선택 (Dynamic Tool Selection) 엔터프라이즈 환경에서는 에이전트에 등록된 도구 수가 수십 개 이상으로 늘어납니다. 모든 도구의 스키마를 매 프롬프트마다 주입하면 세 가지 치명적 문제가 발생합니다. 1. 컨텍스트 비용 폭발 (Token Cost Bloat): 도구 정의만으로 수만 토큰 소비. 2. 도구 선택 환각 (Tool Confusion): 무관한 도구들이 유사한 매개변수를 가질 때 모델이 엉뚱한 도구를 호출. 3. 추론 속도 지연 (Latency Penalty): 프롬프트 길이 증가에 따른 TTFT(Time to First Token) 지연. 이를 해결하기 위해 Tool Registry 패턴과 시맨틱 동적 도구 선택(Semantic Tool Selection) 기법을 도입해야 합니다. ### 5.1 정적 도구 주입 vs 동적 시맨틱 레지스트리 비교 | 평가 항목 | 정적 도구 주입 (Static Tool Injection) | 동적 시맨틱 레지스트리 (Dynamic Tool Registry) | | :--- | :--- | :--- | | 도구 수용 한계 | 10개 이하 제한 | 수백 - 수천 개 확장 가능 | | 토큰 소모 비용 | 매우 높음 (모든 도구 스키마 매번 주입) | 극소화 (질의 관련 Top-K 도구만 선택) | | 도구 선택 정확도 | 저하 (유사한 도구 간 충돌 발생) | 높음 (시맨틱 유사도 높은 도구 집중) | | 초기 응답 속도 (TTFT) | 지연 (긴 프롬프트 헤더) | 빠름 (최적화된 최소 프롬프트) | | 구현 난이도 | 단순 (배열에 포함) | 중간 (임베딩 및 레지스트리 검색 필요) | ### 5.2 동적 도구 선택 코사인 유사도 필터링 흐름: ```mermaid graph TD subgraph "Phase 1: Tool Registry Indexing (Offline / Startup)" T1["Tool 1: get_user_profile"] --> E1["Generate Text Embedding"] T2["Tool 2: update_user_role"] --> E2["Generate Text Embedding"] TN["Tool N: 100+ Enterprise Tools"] --> EN["Generate Text Embedding"] E1 --> VDB[("Vector Registry Index")] E2 --> VDB EN --> VDB end subgraph "Phase 2: Runtime Semantic Filtering" UQ["User Input Goal: 'Jane의 프로필과 권한 확인해줘'"] --> QE["Query Text Embedder"] QE --> QV["Query Vector (V_q)"] QV --> SimEngine["Cosine Similarity Calculator"] VDB --> SimEngine SimEngine --> ScoreList["Similarity Score Matrix"] ScoreList --> Ranker["Top-K Ranking Filter (K=3)"] end subgraph "Phase 3: Payload Building & Prompt Injection" Ranker --> SelectedTools["Selected Schemas (Top-3)"] SelectedTools --> PromptBuilder["Construct Final LLM Request"] PromptBuilder --> LLMCall["LLM Inference Call"] end ``` ### 5.3 시맨틱 동적 도구 레지스트리 구현 (`tool_registry.py`) ```python """ Dynamic Tool Registry with Semantic Tool Filtering. """ import math from typing import List, Dict, Any, Callable, Awaitable, Optional from pydantic import BaseModel, Field, ConfigDict class RegisteredTool(BaseModel): model_config = ConfigDict(arbitrary_types_allowed=True) name: str description: str parameters_schema: Dict[str, Any] handler: Callable[..., Awaitable[Dict[str, Any]]] = Field(exclude=True) tags: List[str] = Field(default_factory=list) class DynamicToolRegistry: """도구를 체계적으로 등록하고, 사용자 질의에 맞춰 동적으로 Top-K 도구를 주입하는 레지스트리""" def __init__(self): self._tools: Dict[str, RegisteredTool] = {} def register_tool( self, name: str, description: str, parameters_schema: Dict[str, Any], handler: Callable[..., Awaitable[Dict[str, Any]]], tags: Optional[List[str]] = None ): tool = RegisteredTool( name=name, description=description, parameters_schema=parameters_schema, handler=handler, tags=tags or [] ) self._tools[name] = tool def _dummy_embedding(self, text: str) -> List[float]: """간이 임베딩 벡터 생성기 (실제 프로덕션에서는 OpenAI text-embedding-3 등 사용)""" vocab = ["user", "profile", "role", "update", "delete", "database", "search", "order", "email"] tokens = text.lower().replace("_", " ").split() vector = [1.0 if word in tokens else 0.0 for word in vocab] norm = math.sqrt(sum(v * v for v in vector)) or 1.0 return [v / norm for v in vector] def _cosine_similarity(self, vec1: List[float], vec2: List[float]) -> float: return sum(a * b for a, b in zip(vec1, vec2)) def select_tools(self, query: str, top_k: int = 3) -> List[Dict[str, Any]]: """사용자의 입력 질의와의 유사도를 계산하여 가장 연관성이 높은 Top-K 도구만 필터링합니다.""" if len(self._tools) <= top_k: return [ { "type": "function", "function": { "name": t.name, "description": t.description, "parameters": t.parameters_schema } } for t in self._tools.values() ] query_vec = self._dummy_embedding(query) scored_tools = [] for tool in self._tools.values(): tool_text = f"{tool.name} {tool.description} {' '.join(tool.tags)}" tool_vec = self._dummy_embedding(tool_text) score = self._cosine_similarity(query_vec, tool_vec) scored_tools.append((score, tool)) # 코사인 유사도 기준 내림차순 정렬 scored_tools.sort(key=lambda x: x[0], reverse=True) selected = scored_tools[:top_k] return [ { "type": "function", "function": { "name": t.name, "description": t.description, "parameters": t.parameters_schema } } for _, t in selected ] async def execute_tool(self, name: str, arguments: Dict[str, Any]) -> Dict[str, Any]: if name not in self._tools: raise KeyError(f"Tool '{name}' is not registered in the Tool Registry.") return await self._tools[name].handler(**arguments) ``` > K값 설정의 딜레마 (Top-K Balancing): `top_k`를 너무 작게(예: 2) 설정하면 복잡한 다단계 과업 시 필요한 도구가 누락될 수 있고, 너무 크게(예: 20 이상) 설정하면 동적 필터링의 토큰 절감 효과가 반감됩니다. 일반적으로 3–5개가 가장 이상적입니다. > 하이브리드 필터링 (Tag + Semantic Search): 태그 기반 1차 하드 필터링(예: 카테고리가 `database`인 도구만 선별)과 임베딩 기반 2차 시맨틱 필터링을 결합하면 선택 정확도를 크게 높일 수 있습니다. --- ## 6. Pydantic v2 기반 엄격한 스키마 검증 및 자율 복구 (Self-Correction) LLM이 생성한 `tool_calls.arguments`는 단순 문자열(JSON String)입니다. 모델은 종종 필수 필드를 누락하거나, 정수(int) 자리에 문자열(str)을 넣거나, JSON 닫는 괄호를 누락하는 오류를 유발합니다. 따라서 호스트 에이전트 런타임에는 Pydantic v2 기반의 검증 파이프라인(Validation Pipeline)이 필수적입니다. ### 6.1 Pydantic v2 방어적 검증 파이프라인 흐름: ```mermaid graph TD RawJSON["LLM Raw arguments String"] --> Step1{"1. JSON Syntax Parse"} Step1 -- "JSONDecodeError" --> Err1["Generate Syntax Fix Diagnostic Prompt"] Step1 -- "Valid JSON Dict" --> Step2{"2. Type & Schema Validation"} Step2 -- "ValidationError (Type/Missing)" --> Err2["Generate Schema Fix Diagnostic Prompt"] Step2 -- "Valid DTO" --> Step3{"3. Business Rule Validation"} Step3 -- "field_validator Error" --> Err3["Generate Business Restriction Diagnostic Prompt"] Step3 -- "Pass All Checks" --> Exec["Safe Tool Handler Execution"] Err1 --> Feedback["Feedback Loop (role: 'tool')"] Err2 --> Feedback Err3 --> Feedback Feedback --> ReInference["LLM Re-Inference Attempt"] ``` ### 6.2 Pydantic 도구 검증 오류 분류 및 자율 복구 피드백 맵 | 에러 유형 | 원인 예시 | Pydantic 처리 및 진단 피드백 힌트 | LLM 자율 복구 동작 | | :--- | :--- | :--- | :--- | | JSON Syntax Error | `{"user_id": "usr_100"` (괄호 닫기 누락) | `[Syntax Error] 올바른 JSON 형식이 아닙니다.` | JSON 문법 구문 수정 재생성 | | Type Mismatch | `"limit": "ten"` (숫자 자리에 문자열) | `Field 'limit': Input should be a valid integer` | 정수형 데이터로 값 변경 | | Missing Required Field | `"reason"` 누락 | `Field 'reason': Field required` | 누락된 필수 필드 추가 | | Business Constraint | `"table_name": "passwords"` | `허용되지 않은 테이블 접근 시도입니다.` | 허용된 테이블 목록 중 재선택 | ### 6.3 방어적 스키마 파서와 자율 복구 루프 (`schema_validator.py`) ```python """ Pydantic v2 Defensive Argument Parsing & LLM Diagnostic Error Feedback Loop. """ import json from typing import Type, TypeVar, Tuple, Optional, Any, Dict from pydantic import BaseModel, ValidationError, Field, field_validator T = TypeVar("T", bound=BaseModel) class DatabaseQueryArgs(BaseModel): table_name: str = Field(description="조회 대상 테이블 이름") limit: int = Field(default=10, ge=1, le=100, description="반환 레코드 수") filters: Dict[str, Any] = Field(default_factory=dict, description="검색 필터 조건") @field_validator("table_name") @classmethod def validate_table_name(cls, v: str) -> str: allowed_tables = ["users", "orders", "audit_logs", "products"] clean_v = v.strip().lower() if clean_v not in allowed_tables: raise ValueError(f"허용되지 않은 테이블 '{v}' 접근 시도입니다. 허용 테이블: {allowed_tables}") return clean_v class SafeToolCallExecutor: """LLM이 생성한 JSON 문자열 인자를 안전하게 파싱하고 Pydantic v2로 검증하는 실행기""" @staticmethod def parse_and_validate( raw_arguments_json: str, schema_cls: Type[T] ) -> Tuple[Optional[T], Optional[str]]: """ JSON 문자열을 파싱하고 Pydantic 스키마 검증을 진행합니다. 성공 시 (ValidatedModel, None) 반환. 실패 시 (None, DiagnosticErrorMessage) 반환하여 LLM 재추론 힌트로 사용. """ # 1. JSON Syntax Validation try: parsed_dict = json.loads(raw_arguments_json) except json.JSONDecodeError as e: error_hint = ( f"[Tool Argument Syntax Error] 입력된 인자가 올바른 JSON 형식이 아닙니다.\n" f"Error Details: {str(e)}\n" f"Raw Input: {raw_arguments_json}\n" f"조치 사항: 올바른 JSON 문법 형식으로 인자값을 재작성하세요." ) return None, error_hint # 2. Pydantic v2 Schema & Business Rule Validation try: validated_model = schema_cls.model_validate(parsed_dict) return validated_model, None except ValidationError as ve: # Pydantic 진단 에러 세부 정보 구조화 formatted_errors = [] for err in ve.errors(): loc = " -> ".join(str(p) for p in err["loc"]) msg = err["msg"] formatted_errors.append(f"- Field '{loc}': {msg}") error_hint = ( f"[Tool Argument Schema Validation Failed]\n" f"도구 실행 인자 검증에 실패했습니다. 아래 오류 사유를 확인하고 올바른 인자값으로 재호출하세요:\n" + "\n".join(formatted_errors) + f"\n요청된 스키마 구조: {json.dumps(schema_cls.model_json_schema(), indent=2, ensure_ascii=False)}" ) return None, error_hint ``` > 진단 메시지의 중요성: LLM에게 단순히 "검증 실패"라고만 전달하면 동일한 실수를 반복합니다. 실패한 필드의 exact path, 원인, 그리고 허용 가능한 값 스키마(`model_json_schema()`)를 피드백에 명확히 작성해야 LLM이 한 번만에 자율 복구에 성공합니다. --- ## 7. 방어적 도구 연동 엔지니어링 체크리스트 프로덕션 환경에 도구를 노출할 때 발생할 수 있는 보안 문제와 서비스 장애를 방지하기 위해 작성된 엔지니어링 검증 체크리스트입니다. ### 7.1 엔터프라이즈 도구 연동 보안 & 신뢰성 매트릭스 | 검증 영역 | 잠재적 위협 / 위협 시나리오 | 방어 가드레일 (Guardrail) | 엔지니어링 구현 방법 | | :--- | :--- | :--- | :--- | | 인자 인젝션 | Path Traversal (`../etc/passwd`), Command Injection | Strict Path Isolation & Param Extraction | `os.path.abspath` 검증, `shell=True` 금지 | | 타임아웃 & 브레이커 | 무응답 API로 인한 에이전트 스레드 영구 블로킹 | Hard Timeout & Circuit Breaker | `asyncio.wait_for(timeout=15.0)` 강제 | | 권한 격리 (HITL) | 무단 데이터 삭제, 고위험 결제/메일 발송 | Human-in-the-Loop Approval Interruption | Read-Only vs Mutating 도구 분류 및 승인 대기 | | 중복 실행 (멱등성) | 재시도 루프로 인한 중복 결제/중복 DB 레코드 생성 | Idempotency Key Validation | Header 기반 `X-Idempotency-Key` 추적 | | 에러 마스킹 | DB 접속 비번, Internal IP, API Key 유출 | Error Sanitization Layer | 정규식 마스킹 및 비즈니스 힌트 변환 | ### 7.2 Human-in-the-Loop & 권한 경계 메커니즘: ```mermaid graph TD ToolCall["LLM Tool Call Request"] --> RiskCheck{"Tool Risk Level Analysis"} RiskCheck -- "Safe / Read-Only (e.g., get_user)" --> DirectExec["Direct System Execution"] RiskCheck -- "Mutating / High-Risk (e.g., delete_db)" --> HITL["Interrupt Agent Loop & Request HITL Approval"] HITL --> UserChoice{"Human Operator Action"} UserChoice -- "Approve" --> DirectExec UserChoice -- "Reject" --> RejectFeedback["Return Rejection Error to LLM Context"] DirectExec --> ContextReturn["Return Execution Result (role: 'tool')"] RejectFeedback --> ContextReturn ``` ### 7.3 방어적 도구 연동 체크리스트 인자 인젝션 방어 (Argument Injection Prevention) - [ ] Path Traversal 방지: 파일 읽기/쓰기 도구에서 `../` 경로 조작을 차단하기 위해 `os.path.abspath` 및 샌드박스 루틴 기본 적용 여부. - [ ] Command & SQL Injection 방지: Shell 명령을 실행하는 도구에서 `shell=True` 사용 금지 및 매개변수화된 쿼리(Parameterized Queries) 강제 적용 여부. 비동기 타임아웃 및 서킷 브레이커 (Timeout & Resilience) - [ ] 모든 도구 핸들러 호출부에 하드 타임아웃 적용 여부 (`asyncio.wait_for(..., timeout=15.0)`). - [ ] 외부 3rd-party API 호출 도구 실패 시 타임아웃 지연으로 인한 에이전트 루프 전체 블로킹 방지 여부. 인간 개입 승인 (Human-in-the-Loop & Permission Boundaries) - [ ] Read-Only vs Mutating 구분: 시스템 상태를 변경하거나 파괴적인 작업(DB 삭제, 외부 이메일 전송, 결제 등) 수행 시 HITL 승인 단계 강제 배치 여부. 멱등성 보장 (Idempotency) - [ ] 네트워크 재시도 또는 LLM self-correction 재시도 시 동일한 요청이 중복 실행되지 않도록 `idempotency_key` 또는 유일 식별자 처리 여부. 진단 메시지 vs 보안 마스킹 (Error Masking Policy) - [ ] LLM에게 반환되는 도구 에러 텍스트에 내부 데이터베이스 접속 정보, API Key, Stack trace 등 민감 정보가 유출되지 않도록 마스킹 처리 되었는가? - [ ] 동시에 LLM이 오류 원인을 파악하고 자율 복구할 수 있는 충분한 비즈니스 힌트가 포함되었는가? > 고위험 도구 자동화 금지: DB 드롭(Drop Table), 결제 승인, 외부 일괄 메일 발송 등 원복 불가능한 작업은 절대로 LLM 자율 실행에 맡기지 말고, 반드시 HITL 승인 레이어를 통해 최종 인간 검증을 거치도록 설계하십시오. --- ## 8. 요약 및 다음 포스트 예고 ### 8.1 3편 핵심 요약 매트릭스 | 영역 | 핵심 개념 | 핵심 테이크어웨이 (Takeaway) | | :--- | :--- | :--- | | Tool Calling | Native Function Calling | JSON Schema 문법 제어 샘플링으로 타입 안전성 및 병렬 호출 보장 | | MCP | Tools, Resources, Prompts Primitives | Stdio(로컬 IPC) 및 SSE(원격 HTTP) 전송 레이어로 M x N 인터페이스 통합 | | Tool Registry | Dynamic Semantic Selection | 코사인 유사도 벡터 필터링으로 Top-K 도구만 선택하여 프롬프트 토큰 절감 | | Schema Guard | Pydantic v2 Defensive Pipeline | 엄격한 타입/비즈니스 검증 및 구조화된 진단 피드백으로 자율 복구(Self-Correction) 구현 | 다음 회차 `04편: 오케스트레이션과 멀티 에이전트`에서는 단일 에이전트의 한계를 넘어 복수의 전문 에이전트들이 역할을 분담하고 그래프 상태(Graph State) 기반으로 협업하는 멀티 에이전트 오케스트레이션 패턴을 심층 분석합니다.