Function Calling: How Tool Use Works Under the Hood

import outlines
from pydantic import BaseModel
from typing import Literal

class WeatherArgs(BaseModel):
    location: str
    units: Literal["fahrenheit", "celsius"] = "celsius"

model = outlines.models.transformers("mistralai/Mistral-7B-v0.1")
generator = outlines.generate.json(model, WeatherArgs)

# The generator will ONLY produce output that parses to WeatherArgs
result = generator("What is the weather in Tokyo?")
# result is guaranteed to be a valid WeatherArgs instance

import litellm
import json

# Same call works for both providers
response = litellm.completion(
    model="anthropic/claude-opus-4-5",  # or "gpt-4o" — same code
    messages=[{"role": "user", "content": "What is 847 * 293?"}],
    tools=openai_tools,  # LiteLLM uses OpenAI format, converts for Anthropic
)

# Same response format regardless of provider
tool_calls = response.choices[0].message.tool_calls
for tc in tool_calls:
    name = tc.function.name
    args = json.loads(tc.function.arguments)
    print(f"Tool: {name}, Args: {args}")

import asyncio
import anthropic
import json
from typing import Any

client = anthropic.Anthropic()

async def execute_tool_async(tool_name: str, tool_input: dict) -> Any:
    """Execute a single tool call asynchronously."""
    # Route to the appropriate tool function
    if tool_name == "get_weather":
        return await get_weather_async(tool_input["location"])
    elif tool_name == "calculate":
        return await calculate_async(tool_input["expression"])
    elif tool_name == "search_web":
        return await search_web_async(tool_input["query"])
    else:
        return {"error": "UnknownTool", "message": f"Tool '{tool_name}' is not registered"}

async def handle_parallel_tool_calls(response) -> list[dict]:
    """Execute all tool calls in a single response in parallel."""
    tool_use_blocks = [
        block for block in response.content
        if block.type == "tool_use"
    ]

    if not tool_use_blocks:
        return []

    # Launch all tool executions simultaneously
    tasks = [
        execute_tool_async(block.name, block.input)
        for block in tool_use_blocks
    ]

    # Wait for all to complete (or fail)
    results = await asyncio.gather(*tasks, return_exceptions=True)

    # Build tool result messages
    tool_results = []
    for block, result in zip(tool_use_blocks, results):
        if isinstance(result, Exception):
            tool_results.append({
                "type": "tool_result",
                "tool_use_id": block.id,
                "content": json.dumps({
                    "error": type(result).__name__,
                    "message": str(result)
                }),
                "is_error": True
            })
        else:
            tool_results.append({
                "type": "tool_result",
                "tool_use_id": block.id,
                "content": json.dumps(result)
            })

    return tool_results

from pydantic import BaseModel, Field, ValidationError
from typing import Optional, Literal
import json

# Define the argument schema with Pydantic
class CalculateArgs(BaseModel):
    expression: str = Field(
        description="Mathematical expression to evaluate",
        min_length=1,
        max_length=200
    )
    precision: Optional[int] = Field(
        default=2,
        ge=0,
        le=10,
        description="Decimal places in the result"
    )

class GetWeatherArgs(BaseModel):
    location: str = Field(min_length=2, max_length=100)
    units: Literal["fahrenheit", "celsius"] = "celsius"

# Map tool names to their argument schemas
TOOL_SCHEMAS: dict[str, type[BaseModel]] = {
    "calculate": CalculateArgs,
    "get_weather": GetWeatherArgs,
}

def validate_and_execute(tool_name: str, raw_input: dict) -> dict:
    """Validate tool arguments and execute, returning structured errors on failure."""
    if tool_name not in TOOL_SCHEMAS:
        return {
            "error": "UnknownTool",
            "message": f"Tool '{tool_name}' is not registered. Available tools: {list(TOOL_SCHEMAS.keys())}"
        }

    schema = TOOL_SCHEMAS[tool_name]
    try:
        validated_args = schema(**raw_input)
    except ValidationError as e:
        # Return structured error — the model can read this and self-correct
        errors = []
        for error in e.errors():
            errors.append({
                "field": ".".join(str(loc) for loc in error["loc"]),
                "error": error["msg"],
                "received": error.get("input"),
                "expected": error.get("ctx", {}).get("expected")
            })
        return {
            "error": "ValidationError",
            "message": f"Invalid arguments for tool '{tool_name}'",
            "schema": schema.model_json_schema(),
            "validation_errors": errors
        }

    # All arguments valid — execute the tool
    try:
        if tool_name == "calculate":
            result = eval(validated_args.expression, {"__builtins__": {}})
            return {"result": round(float(result), validated_args.precision)}
        elif tool_name == "get_weather":
            return {"weather": f"72°F, sunny in {validated_args.location}"}  # Mock
    except Exception as e:
        return {
            "error": "ExecutionError",
            "message": str(e)
        }

# Example: model passes wrong type
result = validate_and_execute("calculate", {"expression": 847, "precision": 2})
print(json.dumps(result, indent=2))
# {
#   "error": "ValidationError",
#   "message": "Invalid arguments for tool 'calculate'",
#   "validation_errors": [{"field": "expression", "error": "Input should be a valid string", ...}]
# }

import json
from typing import Any

def truncate_tool_result(
    result: Any,
    max_tokens: int = 2000,
    strategy: str = "tail"
) -> str:
    """Truncate a tool result to fit within a token budget."""
    result_str = json.dumps(result) if not isinstance(result, str) else result

    # Rough approximation: 4 chars ≈ 1 token
    max_chars = max_tokens * 4

    if len(result_str) <= max_chars:
        return result_str

    if strategy == "tail":
        # Keep the end — often more relevant for search results
        truncated = result_str[-max_chars:]
        return f"[Truncated {len(result_str) - max_chars} characters from start]\n{truncated}"

    elif strategy == "head":
        # Keep the beginning
        truncated = result_str[:max_chars]
        return f"{truncated}\n[Truncated {len(result_str) - max_chars} characters from end]"

    elif strategy == "summary":
        # Use an LLM to summarize — most expensive but highest quality
        summary = summarize_with_llm(result_str, target_tokens=max_tokens)
        return f"[Summary of {len(result_str)//4} token result]\n{summary}"

    elif strategy == "paginate":
        # Return page 1 with instructions to request more
        page = result_str[:max_chars]
        remaining_pages = len(result_str) // max_chars
        return f"{page}\n[Result continues — {remaining_pages} more pages available. Call tool with page=2 to continue.]"

    return result_str[:max_chars]

import asyncio
import anthropic
import json
from pydantic import BaseModel, ValidationError
from typing import Any

client = anthropic.Anthropic()

async def run_agent_with_full_tool_handling(
    user_message: str,
    tools: list,
    max_result_tokens: int = 2000,
    max_retries: int = 3
) -> str:
    """
    Production agent loop with:
    - Parallel tool execution
    - Pydantic argument validation
    - Tool result truncation
    - Structured error responses
    - Retry counting
    """
    messages = [{"role": "user", "content": user_message}]
    retry_counts: dict[str, int] = {}

    while True:
        response = client.messages.create(
            model="claude-opus-4-5",
            max_tokens=2048,
            tools=tools,
            messages=messages
        )

        if response.stop_reason == "end_turn":
            return next(
                block.text for block in response.content
                if hasattr(block, "text")
            )

        # Collect all tool calls from this response
        tool_use_blocks = [b for b in response.content if b.type == "tool_use"]

        if not tool_use_blocks:
            break

        messages.append({"role": "assistant", "content": response.content})

        # Execute all tool calls in parallel
        async def process_one_tool(block) -> dict:
            tool_id = block.id
            tool_name = block.name

            # Track retries per tool
            retry_counts[tool_id] = retry_counts.get(tool_id, 0)

            if retry_counts[tool_id] >= max_retries:
                return {
                    "type": "tool_result",
                    "tool_use_id": tool_id,
                    "content": json.dumps({
                        "error": "MaxRetriesExceeded",
                        "message": f"Tool '{tool_name}' failed after {max_retries} attempts"
                    }),
                    "is_error": True
                }

            # Validate and execute
            result = validate_and_execute(tool_name, block.input)

            # Truncate if the result is large
            result_str = json.dumps(result)
            if len(result_str) > max_result_tokens * 4:
                result_str = truncate_tool_result(result, max_tokens=max_result_tokens)

            is_error = "error" in result if isinstance(result, dict) else False
            if is_error:
                retry_counts[tool_id] = retry_counts.get(tool_id, 0) + 1

            return {
                "type": "tool_result",
                "tool_use_id": tool_id,
                "content": result_str,
                "is_error": is_error
            }

        tool_results = await asyncio.gather(
            *[process_one_tool(block) for block in tool_use_blocks]
        )

        messages.append({"role": "user", "content": list(tool_results)})

    return "Agent loop ended without final response"