Tool Use and Function Calling: The Hardware Ports of LLMs

Before 2023, teaching an LLM to call tools required fragile regex parsing of unstructured strings like "Action: run_sql[query]". If the model output a misplaced bracket or comma, the parser crashed.

Modern frontier models (OpenAI, Anthropic Claude, Google Gemini) feature native Function Calling / Tool Use:

• Tools are declared using standard JSON Schema or Pydantic models.
• When the model decides to invoke a tool, it suspends normal text generation and emits a structured tool_calls payload containing the function name and validated JSON arguments.

1. The Tool Calling Lifecycle

sequenceDiagram
    participant App as Application / Python Runtime
    participant Model as LLM (OpenAI / Gemini)
    participant DB as Production Database (Snowflake / Postgres)

    App->>Model: Prompt: 'Show total sales for customer Acme Corp' + [Tool Schemas: execute_sql]

    Note over Model: Model analyzes question against Tool Schemas.<br/>Decides execute_sql is needed.
    Model-->>App: Emits tool_calls: {name: 'execute_sql', args: {query: 'SELECT SUM(amount) FROM sales WHERE client = "Acme Corp"'}}

    Note over App: App inspects tool_call payload.<br/>Runs query against real DB.
    App->>DB: Execute SQL
    DB-->>App: Result: '$240,000'

    App->>Model: Returns Tool Message: {role: 'tool', content: '$240,000'}

    Note over Model: Model digests query output.
    Model-->>App: Final Response: 'Acme Corp has generated a total of $240,000 in sales.'

2. Declaring Tools with Pydantic

In Python, the cleanest way to define tools is using Pydantic schemas:

from pydantic import BaseModel, Field

class ExecuteSQLQueryArgs(BaseModel):
    query: str = Field(
        description="The syntactically correct Snowflake SQL query to execute. Must be read-only (SELECT)."
    )
    timeout_seconds: int = Field(
        default=30,
        description="Maximum execution timeout before query is aborted."
    )

class InspectTablePartitionsArgs(BaseModel):
    table_name: str = Field(
        description="Full name of the target table, e.g., 'CORE_DW.FCT_ORDERS'"
    )
    filter_column: str = Field(
        description="Partition column name to inspect for data skew"
    )

3. Hands-on: Building a Function Calling Dispatcher

import json
from typing import Dict, Any, Callable

class ToolRegistry:
    def __init__(self):
        self._tools: Dict[str, Callable] = {}
        self._schemas: list = []

    def register(self, name: str, description: str, schema: dict):
        def decorator(func: Callable):
            self._tools[name] = func
            self._schemas.append({
                "type": "function",
                "function": {
                    "name": name,
                    "description": description,
                    "parameters": schema
                }
            })
            return func
        return decorator

    def get_schemas(self):
        return self._schemas

    def dispatch(self, tool_name: str, args_json: str) -> str:
        if tool_name not in self._tools:
            return json.dumps({"error": f"Tool '{tool_name}' not found."})
        try:
            kwargs = json.loads(args_json)
            result = self._tools[tool_name](**kwargs)
            return json.dumps({"status": "SUCCESS", "result": result})
        except Exception as e:
            return json.dumps({"status": "ERROR", "message": str(e)})

# Initialize Registry
registry = ToolRegistry()

@registry.register(
    name="get_cluster_size",
    description="Returns current node count and memory allocation for a PySpark cluster.",
    schema={
        "type": "object",
        "properties": {
            "cluster_id": {"type": "string", "description": "The unique identifier of the cluster"}
        },
        "required": ["cluster_id"]
    }
)
def get_cluster_size(cluster_id: str):
    return {"cluster_id": cluster_id, "worker_nodes": 8, "total_cores": 64, "memory_gb": 256}

if __name__ == "__main__":
    print("Exported JSON Schemas for LLM:")
    print(json.dumps(registry.get_schemas(), indent=2))

    # Simulate LLM emitting tool_calls payload
    simulated_call = {
        "function_name": "get_cluster_size",
        "arguments": '{"cluster_id": "spark-cluster-prod-01"}'
    }

    print("
Executing Tool Dispatch:")
    execution_result = registry.dispatch(
        simulated_call["function_name"],
        simulated_call["arguments"]
    )
    print("Tool Output returned to LLM:", execution_result)

Key Takeaways

• Native Tool Calling replaces fragile regex parsing with validated JSON arguments.
• The model does not execute code: The model only emits the intent to run a tool; your secure application code executes the tool and passes the observation back.
• Always sandbox tool execution: Verify that generated queries are read-only (SELECT) to prevent destructive operations.