Tool Integration and the MCP Standard: Tool Calling and Schema Validation

This post delves deeply into LLM Native Tool Calling mechanisms, Model Context Protocol (MCP) standard specifications, and Pydantic v2-based defensive schema validation techniques.

---

1. Tool Definition and LLM Tool Calling Mechanism

For an LLM (Large Language Model) to escape an isolated sandbox and manipulate the external real world, it must be capable of calling tools (Tool / Function). Early prompt engineering techniques relied on string parsing from general text outputs, such as Action: search_database("query") in ReAct (Reasoning + Acting) patterns. However, this often caused severe syntax errors, missing parentheses, or argument type corruption.

Modern production agent runtimes depend on native Tool Calling (Function Calling) features provided by LLM providers (OpenAI, Anthropic, Google, etc.).

1.1 ReAct Text Parsing vs Native Tool Calling Comparison

1.2 Native Tool Calling 5-Stage Lifecycle Architecture

준비

1. 1subgraph
2. 2Pydantic / Python Handler
3. 3JSON Schema Generator
4. 4Tools Array (JSON Schemas)
5. 5end

반복 실행

1. 1API Request Payload Inject
2. 2User Goal & History

실패 시 다시 실행 단계로 돌아갑니다.

정리

1. 1LLM Provider Engine
2. 2Tool Execution Needed?
3. 3Standard Content Text Output
4. 4Grammar-Guided Constrained Sampler
5. 5Native tool_calls JSON Output
6. 6Agent Runtime Parser
7. 7Pydantic Schema Validation
8. 8Diagnostic Error Generation
9. 9Tool Handler Execution
10. 10Inject Role: tool (Error Feedback)
11. 11Inject Role: tool (Success Result)
12. 12style

Detailed 5-Stage Operation of LLM Tool Calling

1. Tool Schema Injection: The host application converts available tools into a JSON Schema wrapper and injects it into LLM request parameters (tools).
2. Payload Transmission: Combines user input, conversation history, and tool schemas into a single request sent to the LLM API.
3. Constrained Decoding & Tool Call Generation: The model analyzes prompt context and user goals to generate a structured JSON tool_calls object (id, name, arguments).
4. Host Validation & Execution: The agent runtime receives the tool_calls array, identifies the target tool handler, validates parameter values, and executes the call.
5. Context Feedback Loop: Execution results (success text or JSON error) are formatted as a role: "tool" (or role: "user" + tool_call_id) message, appended to context history, and sent back to request model re-reasoning.

Context Overhead Caution: Indiscriminately injecting dozens of tool schemas into the tools array incurs thousands of tokens of prompt cost per request and degrades the model's tool selection accuracy due to name and description collisions.

Schema Optimization Tip: When defining tool configurations, describe the description clearly and specifically. Models prioritize the description over function names when selecting tools.

---

2. Tool Execution Control Loop and Sequence Diagrams

Tool execution does not end with a single call. If tool execution returns an exception or unexpected data, a Self-Correction Loop must operate where the LLM observes the error message to self-correct argument values or invoke fallback tools.

2.1 Agent Tool Execution Full Sequence Diagram

준비

1. 1User

반복 실행

1. 1Agent
2. 2Tool Registry

3. Top-K Selected Tool Schemas

정리

1. 1LLM Engine
2. 2MCP Server

2.2 Self-Correction Loop Mechanism

반복 실행

1. 1LLM_Inference
2. 2ToolCall_Generated
3. 3Final_Answer
4. 4Schema_Validation
5. 5state
6. 6Syntax_Check
7. 7Type_Check
8. 8Diagnostic_Syntax_Error
9. 9Business_Rule_Check
10. 10Diagnostic_Type_Error
11. 11Validated
12. 12Diagnostic_Rule_Error
13. 13LLM_Self_Correction
14. 14Retry_Check

실패 시 다시 실행 단계로 돌아갑니다.

정리

1. 1Max_Retry_Exceeded
2. 2Tool_Execution
3. 3Executing
4. 4Execution_Success
5. 5Execution_Failure

2.3 Agent Tool Failure Handling Strategies Comparison

Max Retries Bound: The self-correction loop must enforce a max_retries limit (e.g., 3 retries). If the model repeatedly produces errors due to an invalid schema, you must cut off token explosion and infinite waiting.

---

3. Model Context Protocol (MCP) Standard and Architecture

As the agent ecosystem expands, standardizing the protocol between host applications (Claude Desktop, Cursor, Custom Agent Framework) and individual tools (PostgreSQL, GitHub, Slack, Local Filesystem) becomes critical. Model Context Protocol (MCP) is an open standard protocol designed to solve this $M \times N$ interface fragmentation problem.

3.1 MCP 3 Core Primitives Comparison

MCP provides three core primitives on top of the JSON-RPC 2.0 protocol.

3.2 MCP Stdio vs SSE Transport Layer Architecture

준비

1. 1subgraph
2. 2Host Application (Claude / Cursor / Custom Agent)
3. 3MCP Client Manager
4. 4end
5. 5direction

반복 실행

1. 1Stdio Transport Client
2. 2Child Process stdin (NDJSON)
3. 3Child Process stdout (NDJSON)

실패 시 다시 실행 단계로 돌아갑니다.

정리

1. 1SSE Transport Client
2. 2HTTP POST Channel (/messages)
3. 3SSE Event Stream Channel (/sse)
4. 4Local MCP Server Process
5. 5Remote MCP HTTP/SSE Server
6. 6LocalDB
7. 7Cloud Enterprise APIs / PostgreSQL

3.3 MCP Stdio vs SSE Transport Layer Detailed Comparison

JSON-RPC 2.0 Standard Specification: MCP follows the JSON-RPC 2.0 specification for all message exchanges, featuring jsonrpc: "2.0", method, params, and id fields. This provides seamless compatibility even if the client and server are implemented in different languages.

Local Development and Remote Distributed Operation: It is recommended to start development using Stdio transport for zero-overhead iteration, and switch to SSE/HTTP transport in production deployment to scale out as microservices.

---

4. Practical MCP Python Implementation (Server & Client)

In this section, we construct production-grade Python code for an MCP server and an agent client that connects to it, dynamically discovering and executing tools.

4.1 MCP Handshake and Message Exchange Sequence

반복 실행

1. 1Client
2. 2Transport
3. 3Server

3. JSON-RPC response: initialize (serverInfo, capabilities)

4.2 MCP Tool Server Implementation (mcp_server.py)

Copy1"""2MCP Server Implementation using FastMCP / Async patterns.3This server provides enterprise database query and user role modification tools.4"""5 6import asyncio7import logging8from typing import Dict, Any, List9from pydantic import BaseModel, Field10from mcp.server.fastmcp import FastMCP11 12logging.basicConfig(level=logging.INFO)13logger = logging.getLogger("EnterpriseMCPServer")14 15# Initialize FastMCP Server Instance16mcp = FastMCP("Enterprise-Data-MCP-Server")17 18# ==========================================19# Pydantic Input Schemas20# ==========================================21class UserQueryInput(BaseModel):22    user_id: str = Field(23        description="Unique UUID key of the target user",24        json_schema_extra={"example": "usr_102938"}25    )26    include_deleted: bool = Field(27        default=False,28        description="Whether to include deleted user accounts"29    )30 31class UserRoleUpdateInput(BaseModel):32    user_id: str = Field(description="ID of the user whose role will be updated")33    target_role: str = Field(description="New role to assign (admin, operator, viewer)")34    reason: str = Field(min_length=10, description="Reason for role change (minimum 10 characters required)")35 36# ==========================================37# MCP Tool Registration38# ==========================================39@mcp.tool(40    name="get_user_profile",41    description="Retrieves detailed user profile data and recent activity logs from database by user ID."42)43async def get_user_profile(user_id: str, include_deleted: bool = False) -> Dict[str, Any]:44    # Defensive validation (secondary server-side check)45    input_data = UserQueryInput(user_id=user_id, include_deleted=include_deleted)46    logger.info(f"User Query Executed for ID: {input_data.user_id}")47 48    # Simulated DB Lookup49    if input_data.user_id == "usr_404":50        return {"status": "error", "message": f"User '{input_data.user_id}' not found."}51 52    return {53        "status": "success",54        "data": {55            "user_id": input_data.user_id,56            "name": "Jane Doe",57            "email": "[email protected]",58            "role": "operator",59            "is_active": True,60        }61    }62 63@mcp.tool(64    name="update_user_role",65    description="Updates system access permissions (Role) for a user. Mandatory reason required as this is a high-risk tool."66)67async def update_user_role(user_id: str, target_role: str, reason: str) -> Dict[str, Any]:68    # Pydantic schema-based data validation69    try:70        payload = UserRoleUpdateInput(user_id=user_id, target_role=target_role, reason=reason)71    except Exception as e:72        return {"status": "error", "message": f"Schema Validation Failed: {str(e)}"}73 74    if payload.target_role not in ["admin", "operator", "viewer"]:75        return {"status": "error", "message": f"Invalid role '{payload.target_role}'. Must be one of [admin, operator, viewer]."}76 77    logger.warning(f"SECURITY EVENT: User {payload.user_id} role changed to {payload.target_role} (Reason: {payload.reason})")78 79    return {80        "status": "success",81        "result": {82            "user_id": payload.user_id,83            "updated_role": payload.target_role,84            "audit_logged": True85        }86    }87 88if __name__ == "__main__":89    # Run MCP Server in Stdio mode90    mcp.run(transport="stdio")

4.3 MCP Agent Client Integration (mcp_client.py)

Copy1"""2MCP Client Manager that connects to stdio MCP Server and exposes JSON-RPC tools to LLM.3"""4 5import sys6import json7import asyncio8from typing import List, Dict, Any, Optional9from contextlib import AsyncExitStack10from mcp import ClientSession, StdioServerParameters11from mcp.client.stdio import stdio_client12 13class AgentMCPClientManager:14    """Client class managing lifecycles with MCP servers and transforming tool schemas into LLM standard formats."""15 16    def __init__(self, server_script_path: str):17        self.server_script_path = server_script_path18        self.session: Optional[ClientSession] = None19        self._exit_stack = AsyncExitStack()20 21    async def connect(self):22        """Launches MCP server as Python child process and opens stdio channels."""23        server_params = StdioServerParameters(24            command=sys.executable,25            args=[self.server_script_path],26            env=None27        )28 29        # Lifecycle management of stdio transport and session using AsyncExitStack30        read_stream, write_stream = await self._exit_stack.enter_async_context(31            stdio_client(server_params)32        )33        self.session = await self._exit_stack.enter_async_context(34            ClientSession(read_stream, write_stream)35        )36        await self.session.initialize()37        print("[MCP Client] Connection successfully established to MCP Server.")38 39    async def get_llm_tool_definitions(self) -> List[Dict[str, Any]]:40        """Calls tools/list on MCP server and converts them to OpenAI/Anthropic JSON Schema specs."""41        if not self.session:42            raise RuntimeError("MCP Session is not initialized.")43 44        mcp_tools_response = await self.session.list_tools()45        llm_tools = []46 47        for tool in mcp_tools_response.tools:48            # Transform MCP Tool Schema -> OpenAI Function Calling Schema49            input_schema = getattr(tool, "inputSchema", getattr(tool, "input_schema", {}))50            llm_tools.append({51                "type": "function",52                "function": {53                    "name": tool.name,54                    "description": tool.description or "",55                    "parameters": input_schema56                }57            })58        return llm_tools59 60    async def call_tool(self, tool_name: str, arguments: Dict[str, Any]) -> Dict[str, Any]:61        """Sends tools/call message to MCP server and returns the result."""62        if not self.session:63            raise RuntimeError("MCP Session is not initialized.")64 65        result = await self.session.call_tool(tool_name, arguments)66 67        # Parse MCP TextContent / ImageContent responses68        output_chunks = []69        for content in result.content:70            if getattr(content, "type", None) == "text":71                output_chunks.append(content.text)72 73        is_error = getattr(result, "isError", getattr(result, "is_error", False))74        return {75            "is_error": bool(is_error),76            "raw_output": "\n".join(output_chunks)77        }78 79    async def close(self):80        """Safely releases MCP client resources."""81        await self._exit_stack.aclose()82        self.session = None83        print("[MCP Client] Connection safely closed.")

Security Risk Warning (Command & Tool Injection): When a Stdio MCP server runs as a subprocess, executing unvalidated client input via os.system or shell commands can lead to total system compromise. Always sanitize parameters and enforce Pydantic validation.

Resource Management with AsyncExitStack: Using AsyncExitStack when building MCP clients ensures that subprocess stdin/stdout streams and sessions are safely torn down even during unexpected exceptions or network drops.

---

5. Tool Registry Pattern and Dynamic Tool Selection

In enterprise environments, the number of tools registered to an agent grows to 50–100+. Injecting all tool schemas into every prompt leads to three critical issues:

1. Token Cost Bloat: Tool definitions alone consume tens of thousands of tokens.
2. Tool Confusion: When unrelated tools share similar parameters, models call wrong tools.
3. Latency Penalty: Increased prompt length degrades Time to First Token (TTFT).

To address these challenges, we adopt the Tool Registry pattern and Semantic Dynamic Tool Selection.

5.1 Static Tool Injection vs Dynamic Semantic Registry Comparison

5.2 Dynamic Tool Selection Cosine Similarity Filtering Flow

subgraph

Tool 1: get_user_profile

Generate Text Embedding

Tool 2: update_user_role

Generate Text Embedding

Tool N: 100+ Enterprise Tools

Generate Text Embedding

VDB

end

User Input Goal: 'Check Jane's profile and permissions'

Query Text Embedder

Query Vector (V_q)

Cosine Similarity Calculator

Similarity Score Matrix

Top-K Ranking Filter (K=3)

Selected Schemas (Top-3)

Construct Final LLM Request

LLM Inference Call

style

5.3 Semantic Dynamic Tool Registry Implementation (tool_registry.py)

Copy1"""2Dynamic Tool Registry with Semantic Tool Filtering.3"""4 5import math6from typing import List, Dict, Any, Callable, Awaitable, Optional7from pydantic import BaseModel, Field, ConfigDict8 9class RegisteredTool(BaseModel):10    model_config = ConfigDict(arbitrary_types_allowed=True)11 12    name: str13    description: str14    parameters_schema: Dict[str, Any]15    handler: Callable[..., Awaitable[Dict[str, Any]]] = Field(exclude=True)16    tags: List[str] = Field(default_factory=list)17 18class DynamicToolRegistry:19    """Registry that manages tool definitions and dynamically injects Top-K tools matching user queries."""20 21    def __init__(self):22        self._tools: Dict[str, RegisteredTool] = {}23 24    def register_tool(25        self,26        name: str,27        description: str,28        parameters_schema: Dict[str, Any],29        handler: Callable[..., Awaitable[Dict[str, Any]]],30        tags: Optional[List[str]] = None31    ):32        tool = RegisteredTool(33            name=name,34            description=description,35            parameters_schema=parameters_schema,36            handler=handler,37            tags=tags or []38        )39        self._tools[name] = tool40 41    def _dummy_embedding(self, text: str) -> List[float]:42        """Simple embedding vector generator (Use OpenAI text-embedding-3 or similar in production)"""43        vocab = ["user", "profile", "role", "update", "delete", "database", "search", "order", "email"]44        tokens = text.lower().replace("_", " ").split()45        vector = [1.0 if word in tokens else 0.0 for word in vocab]46        norm = math.sqrt(sum(v * v for v in vector)) or 1.047        return [v / norm for v in vector]48 49    def _cosine_similarity(self, vec1: List[float], vec2: List[float]) -> float:50        return sum(a * b for a, b in zip(vec1, vec2))51 52    def select_tools(self, query: str, top_k: int = 3) -> List[Dict[str, Any]]:53        """Calculates similarity with user query and filters Top-K relevant tools."""54        if len(self._tools) <= top_k:55            return [56                {57                    "type": "function",58                    "function": {59                        "name": t.name,60                        "description": t.description,61                        "parameters": t.parameters_schema62                    }63                }64                for t in self._tools.values()65            ]66 67        query_vec = self._dummy_embedding(query)68        scored_tools = []69 70        for tool in self._tools.values():71            tool_text = f"{tool.name} {tool.description} {' '.join(tool.tags)}"72            tool_vec = self._dummy_embedding(tool_text)73            score = self._cosine_similarity(query_vec, tool_vec)74            scored_tools.append((score, tool))75 76        # Sort descending by cosine similarity77        scored_tools.sort(key=lambda x: x[0], reverse=True)78        selected = scored_tools[:top_k]79 80        return [81            {82                "type": "function",83                "function": {84                    "name": t.name,85                    "description": t.description,86                    "parameters": t.parameters_schema87                }88            }89            for _, t in selected90        ]91 92    async def execute_tool(self, name: str, arguments: Dict[str, Any]) -> Dict[str, Any]:93        if name not in self._tools:94            raise KeyError(f"Tool '{name}' is not registered in the Tool Registry.")95        return await self._tools[name].handler(**arguments)

Top-K Balancing Dilemma: Setting top_k too small (e.g., 1–2) may omit necessary tools during complex multi-step tasks, while setting it too large (e.g., 20+) diminishes token savings from dynamic filtering. Typically, 3–5 tools represent the sweet spot.

Hybrid Filtering (Tag + Semantic Search): Combining tag-based hard filtering (e.g., selecting tools with category database) with vector-based semantic filtering significantly boosts selection accuracy.

---

6. Pydantic v2-Based Strict Schema Validation and Self-Correction

The tool_calls.arguments generated by an LLM is a raw JSON string. Models frequently omit required fields, pass strings instead of integers, or truncate trailing closing brackets. Thus, host agent runtimes require a defensive validation pipeline built with Pydantic v2.

6.1 Pydantic v2 Defensive Validation Pipeline Flow

1. JSON Syntax Parse

Generate Syntax Fix Diagnostic Prompt

Feedback Loop (role: 'tool')

2. Type & Schema Validation

Generate Schema Fix Diagnostic Prompt

3. Business Rule Validation

6.2 Pydantic Validation Error Classification & Self-Correction Feedback Map

6.3 Defensive Schema Parser and Self-Correction Loop (schema_validator.py)

Copy1"""2Pydantic v2 Defensive Argument Parsing & LLM Diagnostic Error Feedback Loop.3"""4 5import json6from typing import Type, TypeVar, Tuple, Optional, Any, Dict7from pydantic import BaseModel, ValidationError, Field, field_validator8 9T = TypeVar("T", bound=BaseModel)10 11class DatabaseQueryArgs(BaseModel):12    table_name: str = Field(description="Target table name to query")13    limit: int = Field(default=10, ge=1, le=100, description="Number of records to return (1-100)")14    filters: Dict[str, Any] = Field(default_factory=dict, description="Search filter conditions")15 16    @field_validator("table_name")17    @classmethod18    def validate_table_name(cls, v: str) -> str:19        allowed_tables = ["users", "orders", "audit_logs", "products"]20        clean_v = v.strip().lower()21        if clean_v not in allowed_tables:22            raise ValueError(f"Unauthorized table '{v}' access attempt. Allowed tables: {allowed_tables}")23        return clean_v24 25class SafeToolCallExecutor:26    """Executor that safely parses LLM-generated JSON argument strings and validates them with Pydantic v2."""27 28    @staticmethod29    def parse_and_validate(30        raw_arguments_json: str,31        schema_cls: Type[T]32    ) -> Tuple[Optional[T], Optional[str]]:33        """34        Parses JSON string and performs Pydantic schema validation.35        Returns (ValidatedModel, None) on success.36        Returns (None, DiagnosticErrorMessage) on failure to guide LLM re-inference.37        """38        # 1. JSON Syntax Validation39        try:40            parsed_dict = json.loads(raw_arguments_json)41        except json.JSONDecodeError as e:42            error_hint = (43                f"[Tool Argument Syntax Error] Input arguments are not in valid JSON format.\n"44                f"Error Details: {str(e)}\n"45                f"Raw Input: {raw_arguments_json}\n"46                f"Action Required: Rewrite the argument values in valid JSON syntax."47            )48            return None, error_hint49 50        # 2. Pydantic v2 Schema & Business Rule Validation51        try:52            validated_model = schema_cls.model_validate(parsed_dict)53            return validated_model, None54        except ValidationError as ve:55            # Structure detailed Pydantic diagnostic errors56            formatted_errors = []57            for err in ve.errors():58                loc = " -> ".join(str(p) for p in err["loc"])59                msg = err["msg"]60                formatted_errors.append(f"- Field '{loc}': {msg}")61 62            error_hint = (63                f"[Tool Argument Schema Validation Failed]\n"64                f"Failed to validate tool execution arguments. Review the errors below and retry with valid values:\n"65                + "\n".join(formatted_errors) + "\n"66                f"Requested Schema Structure: {json.dumps(schema_cls.model_json_schema(), indent=2, ensure_ascii=False)}"67            )68            return None, error_hint

Importance of Diagnostic Messages: Simply returning "Validation Failed" to an LLM leads to repeated errors. Feedback must clearly detail the exact path of the failing field, the root cause, and the allowed schema (model_json_schema()) so the LLM can self-correct in a single turn.

Pydantic v2 Performance Benefits: Pydantic v2 rewrote its core parsing engine in Rust, delivering 5–15x faster parsing speeds compared to v1, minimizing runtime latency in agent loops.

---

7. Defensive Tool Integration Engineering Checklist

An engineering checklist designed to prevent security vulnerabilities and service outages when exposing tools in production environments.

7.1 Enterprise Tool Integration Security & Reliability Matrix

7.2 Human-in-the-Loop & Permission Boundary Mechanism

준비

1. 1LLM Tool Call Request
2. 2Tool Risk Level Analysis

반복 실행

1. 1Direct System Execution
2. 2Interrupt Agent Loop & Request HITL Approval
3. 3Human Operator Action

실패 시 다시 실행 단계로 돌아갑니다.

정리

1. 1Return Rejection Error to LLM Context
2. 2Return Execution Result (role: 'tool')
3. 3style

7.3 Defensive Tool Integration Checklist

Argument Injection Prevention

• Path Traversal Prevention: Enforce os.path.abspath and sandbox routines on file read/write tools to block ../ path manipulation.
• Command & SQL Injection Prevention: Ban shell=True on shell-executing tools and enforce parameterized queries.

Async Timeout and Circuit Breaker

• Hard Timeout Enforcement: Enforce hard timeout wrappers (asyncio.wait_for(..., timeout=15.0)) on all tool handler calls.
• Resilience: Ensure failures or timeouts in external 3rd-party API calls do not block the entire agent execution loop.

Human-in-the-Loop & Permission Boundaries

• Read-Only vs Mutating Segregation: Enforce manual Human-in-the-Loop (HITL) approval steps for operations that modify system state or perform destructive actions (DB deletion, sending external emails, payments, etc.).

Idempotency Guarantees

• Idempotency Tracking: Ensure idempotency_key or unique request identifiers are processed to prevent duplicate executions during network retries or LLM self-correction retries.

Diagnostic Messages vs Security Masking

• Error Sanitization: Verify that internal database access credentials, API keys, and stack traces are sanitized from tool error texts returned to the LLM.
• Actionable Guidance: Ensure sufficient business diagnostic hints are included so the LLM can diagnose root causes and self-correct.

Prohibit High-Risk Tool Automation: Irreversible operations such as dropping DB tables, approving financial transfers, or sending bulk emails should never be left to autonomous LLM execution. Always require human verification through a HITL approval layer.

---

8. Summary and Next Episode Preview

8.1 Episode 3 Core Summary Matrix

In the next episode, Part 4: Multi-Agent Orchestration: LangGraph and Subagent Handoffs, we will move beyond single-agent limitations to analyze multi-agent orchestration patterns where specialized agents divide responsibilities and collaborate based on Graph State.

If you have any questions or feedback regarding tool integration when building agent systems, please feel free to leave a comment or issue.