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

This post covers the five stages of Tool Calling, MCP connections, and argument validation with Pydantic v2.

---

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

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

2.2 Self-Correction Loop Mechanism

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

Within MCP Client Host Runtime, Host Application (Claude / Cursor / Custom Agent) → MCP Client Manager. The manager connects separately to Stdio Transport Client and SSE Transport Client.

Both servers belong to MCP Server Layer. These rows preserve the Stdio/SSE transport model described in this article.

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)

This section connects a Python MCP server and client, retrieves the tool list, and executes a tool.

4.1 MCP Handshake and Message Exchange Sequence

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

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

LLM Raw arguments String enters 1. JSON Syntax Parse.

Each of the three diagnostics feeds Feedback Loop (role: 'tool') → LLM Re-Inference Attempt.

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

Validation feedback should include the failing field path, the cause, and the allowed schema (model_json_schema()). These details give the model a basis for correcting its arguments.

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

↳Approve risky tools before execution

Read the diagram as text

An LLM Tool Call Request enters Tool Risk Level Analysis. Safe / Read-Only calls such as get_user go to Direct System Execution. Mutating / High-Risk calls such as delete_db interrupt the Agent Loop and request HITL Approval. Human Operator Action sends Approve to Direct System Execution and Reject to Return Rejection Error to LLM Context. Both execution and rejection feed Return Execution Result with role: tool.

An LLM Tool Call Request enters Tool Risk Level Analysis. Safe / Read-Only calls such as get_user go to Direct System Execution. Mutating / High-Risk calls such as delete_db interrupt the Agent Loop and request HITL Approval. Human Operator Action sends Approve to Direct System Execution and Reject to Return Rejection Error to LLM Context. Both execution and rejection feed Return Execution Result with role: tool.

Open full size ↗

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

The next episode, Part 4: Multi-Agent Orchestration: LangGraph and Subagent Handoffs, covers control and Graph State transfers between specialized agents.

When reporting a tool-integration issue, include the validation error and a reproducible input with credentials removed.