Tool Calling Guide

Tool Calling Guide

Complete guide to using tools (function calling) with cascadeflow.

📋 Table of Contents

Basic (Getting Started)

1. Introduction
2. Quick Start
3. Core Concepts
4. Defining Tools
5. Tool Execution
6. Multi-Turn Conversations

Advanced (Power Features)

7. Tool Streaming
8. Advanced Patterns
9. Best Practices
10. Troubleshooting

Basic Usage

Essential tool calling patterns for cascadeflow.

Introduction

Tools (also called "function calling") allow AI models to interact with external systems, APIs, and data sources. cascadeflow provides a complete tool system that works seamlessly with cascades, streaming, and quality validation.

What You Can Do With Tools

• 🌐 Call External APIs: Weather, stocks, search, etc.
• 🧮 Perform Calculations: Math, data analysis, conversions
• 💾 Access Databases: Query, update, retrieve data
• 🔧 Execute Code: Run scripts, process files, transform data
• 🤖 Multi-Step Workflows: Chain multiple tools together

Key Features

• ✅ Universal Format: Works with all providers (OpenAI, Anthropic, Groq)
• ✅ Automatic Conversion: Provider-specific format handling
• ✅ Quality Validation: Built-in tool call validation
• ✅ Cascade Compatible: Tools work with all cascade types
• ✅ Streaming Support: Watch tool calls form in real-time
• ✅ Parallel Execution: Run multiple tools concurrently
• ✅ Error Handling: Robust error recovery

Quick Start

3-Step Tool Setup

from cascadeflow import CascadeAgent, ModelConfig
from cascadeflow.tools import ToolConfig, ToolExecutor

# Step 1: Define your function
def get_weather(location: str, unit: str = "celsius") -> dict:
    """Get weather for a location."""
    # Your implementation here
    return {"temp": 22, "condition": "sunny"}

# Step 2: Create ToolConfig
tool_config = ToolConfig(
    name="get_weather",
    description="Get current weather for a location",
    parameters={
        "type": "object",
        "properties": {
            "location": {"type": "string"},
            "unit": {"type": "string", "enum": ["celsius", "fahrenheit"]}
        },
        "required": ["location"]
    },
    function=get_weather  # Link to actual function
)

# Step 3: Create executor and use with agent
executor = ToolExecutor([tool_config])

agent = CascadeAgent(models=[...])
result = await agent.run(
    "What's the weather in Paris?",
    tools=[{
        "name": "get_weather",
        "description": "Get current weather for a location",
        "parameters": {...}
    }]
)

# Execute tool calls
for tool_call in result.tool_calls:
    tool_result = await executor.execute(tool_call)
    print(f"Result: {tool_result.result}")

Core Concepts

Tool Definition vs Tool Schema

Important Distinction:

1. ToolConfig (Python object with function):

   • Used by ToolExecutor to run tools
   • Contains actual function reference
   • Never sent to model

2. Tool Schema (JSON dict):

   • Sent to model to describe available tools
   • Contains only name, description, parameters
   • No function reference

# ToolConfig - for execution
tool_config = ToolConfig(
    name="get_weather",
    description="Get weather",
    parameters={...},
    function=get_weather  # ← Actual function
)

# Tool Schema - for model
tool_schema = {
    "name": "get_weather",
    "description": "Get weather",
    "parameters": {...}
    # No function! Model doesn't see implementation
}

Tool Call Lifecycle

1. User Query
   ↓
2. Model Generates Tool Calls
   ↓
3. Parse Tool Calls (ToolCall objects)
   ↓
4. Execute with ToolExecutor
   ↓
5. Format Results (ToolResult objects)
   ↓
6. Feed Back to Model
   ↓
7. Model Generates Final Answer

Key Classes

Defining Tools

Method 1: Manual ToolConfig

from cascadeflow.tools import ToolConfig

def calculate(operation: str, x: float, y: float) -> dict:
    """Perform a calculation."""
    ops = {
        "add": x + y,
        "subtract": x - y,
        "multiply": x * y,
        "divide": x / y if y != 0 else None
    }
    return {"result": ops[operation], "operation": operation}

tool = ToolConfig(
    name="calculate",
    description="Perform basic math operations",
    parameters={
        "type": "object",
        "properties": {
            "operation": {
                "type": "string",
                "enum": ["add", "subtract", "multiply", "divide"]
            },
            "x": {"type": "number"},
            "y": {"type": "number"}
        },
        "required": ["operation", "x", "y"]
    },
    function=calculate
)

Method 2: Auto-Generate from Function

from cascadeflow.tools import ToolConfig

def get_stock_price(symbol: str, currency: str = "USD") -> dict:
    """Get current stock price for a symbol."""
    # Implementation...
    return {"symbol": symbol, "price": 150.25, "currency": currency}

# Automatically extracts: name, description, parameters, required fields
tool = ToolConfig.from_function(get_stock_price)

Method 3: Using Decorator

from cascadeflow.tools import tool

@tool
def search_documents(query: str, limit: int = 10) -> list:
    """Search documents by keyword."""
    # Implementation...
    return [{"title": "Doc 1", "score": 0.95}]

# search_documents is now a ToolConfig object!

Parameter Types

cascadeflow supports all JSON Schema types:

parameters = {
    "type": "object",
    "properties": {
        # String
        "name": {"type": "string"},
        
        # Number (int or float)
        "age": {"type": "integer"},
        "price": {"type": "number"},
        
        # Boolean
        "active": {"type": "boolean"},
        
        # Enum (constrained values)
        "status": {
            "type": "string",
            "enum": ["pending", "approved", "rejected"]
        },
        
        # Array
        "tags": {
            "type": "array",
            "items": {"type": "string"}
        },
        
        # Nested object
        "address": {
            "type": "object",
            "properties": {
                "street": {"type": "string"},
                "city": {"type": "string"}
            }
        }
    },
    "required": ["name", "age"]  # Required fields
}

Tool Execution

Basic Execution

from cascadeflow.tools import ToolExecutor, ToolCall, ToolCallFormat

# Create executor with tool configs
executor = ToolExecutor([tool1, tool2, tool3])

# Execute a tool call
tool_call = ToolCall(
    id="call_123",
    name="get_weather",
    arguments={"location": "Paris", "unit": "celsius"},
    provider_format=ToolCallFormat.OPENAI
)

result = await executor.execute(tool_call)

if result.success:
    print(f"Result: {result.result}")
else:
    print(f"Error: {result.error}")

Parallel Execution

Execute multiple tools concurrently for better performance:

tool_calls = [call1, call2, call3, call4]

# Execute up to 5 tools in parallel
results = await executor.execute_parallel(
    tool_calls,
    max_parallel=5
)

for result in results:
    if result.success:
        print(f"{result.name}: {result.result}")

Error Handling

result = await executor.execute(tool_call)

if not result.success:
    # Tool execution failed
    error_type = type(result.error).__name__
    error_msg = result.error
    
    if "not found" in error_msg:
        # Tool doesn't exist
        print(f"Unknown tool: {tool_call.name}")
    elif "arguments" in error_msg:
        # Invalid arguments
        print(f"Bad arguments: {tool_call.arguments}")
    else:
        # Other error
        print(f"Tool failed: {error_msg}")

Sync vs Async Functions

ToolExecutor handles both automatically:

# Synchronous function
def sync_tool(x: int) -> int:
    return x * 2

# Asynchronous function
async def async_tool(url: str) -> dict:
    async with httpx.AsyncClient() as client:
        response = await client.get(url)
        return response.json()

# Both work with ToolExecutor!
executor = ToolExecutor([
    ToolConfig.from_function(sync_tool),
    ToolConfig.from_function(async_tool)
])

Multi-Turn Conversations

Tools typically require multiple turns: request → execute → respond.

Complete Multi-Turn Example

async def tool_conversation(agent, executor, query, tools):
    """Run a complete tool conversation."""
    
    messages = [{"role": "user", "content": query}]
    max_turns = 3
    turn = 0
    
    while turn < max_turns:
        turn += 1
        
        # Get model response
        result = await agent.run(
            query=" ".join([m["content"] for m in messages if m["role"] == "user"]),
            tools=tools
        )
        
        # Check if model wants to use tools
        if result.tool_calls:
            # Execute tools
            tool_results = []
            for tool_call in result.tool_calls:
                tool_result = await executor.execute(tool_call)
                tool_results.append(tool_result)
            
            # Add assistant message with tool calls
            messages.append({
                "role": "assistant",
                "content": result.content or "",
                "tool_calls": result.tool_calls
            })
            
            # Add tool results
            for tool_result in tool_results:
                messages.append({
                    "role": "tool",
                    "tool_call_id": tool_result.call_id,
                    "name": tool_result.name,
                    "content": str(tool_result.result)
                })
            
            # Continue to next turn
            continue
        
        else:
            # Model generated final answer
            return result.content
    
    return "Max turns reached"

Message Format

Different providers use different formats for tool results:

OpenAI Format:

{
    "role": "tool",
    "tool_call_id": "call_123",
    "name": "get_weather",
    "content": "{'temp': 22, 'condition': 'sunny'}"
}

Anthropic Format:

{
    "role": "user",
    "content": [{
        "type": "tool_result",
        "tool_use_id": "toolu_123",
        "content": "{'temp': 22, 'condition': 'sunny'}"
    }]
}

cascadeflow handles this automatically:

# Automatically converts to correct format
tool_result.to_provider_message("openai")   # OpenAI format
tool_result.to_provider_message("anthropic") # Anthropic format

Advanced Usage

Power features for tool calling in production.

Tool Streaming

Watch tool calls form in real-time as the model generates them.

Basic Tool Streaming

from cascadeflow.streaming import ToolStreamEventType

async for event in agent.stream_events(query, tools=tools):
    match event.type:
        case ToolStreamEventType.TOOL_CALL_START:
            print(f"\n🔧 Calling: {event.tool_call['name']}")
        
        case ToolStreamEventType.TOOL_CALL_DELTA:
            # Arguments streaming in progressively
            print(event.delta, end='')
        
        case ToolStreamEventType.TOOL_CALL_COMPLETE:
            # Full tool call parsed
            tool = event.tool_call
            print(f"\n✓ Complete: {tool['name']}({tool['arguments']})")
        
        case ToolStreamEventType.TEXT_CHUNK:
            # Regular text response
            print(event.content, end='')

Progressive Argument Display

async for event in agent.stream_events(query, tools=tools):
    if event.type == ToolStreamEventType.TOOL_CALL_DELTA:
        # Show arguments as they arrive
        print(f"\rArguments: {event.partial_arguments}", end='')

See examples/streaming_tools.py for complete streaming example.

Advanced Patterns

Dynamic Tool Selection

Select tools based on query or context:

def get_tools_for_query(query: str) -> list:
    """Select relevant tools for query."""
    all_tools = {
        "weather": weather_tools,
        "calculation": math_tools,
        "search": search_tools
    }
    
    # Simple keyword matching (use embedding similarity in production)
    if "weather" in query.lower():
        return all_tools["weather"]
    elif any(word in query.lower() for word in ["calculate", "multiply", "add"]):
        return all_tools["calculation"]
    else:
        return all_tools["search"]

# Use with agent
tools = get_tools_for_query(user_query)
result = await agent.run(user_query, tools=tools)

Tool Chaining

Chain multiple tools together:

async def chain_tools(agent, executor, query):
    """Execute tools in sequence."""
    
    # Step 1: Get weather
    result1 = await agent.run(
        "Get weather for Paris",
        tools=[weather_tool]
    )
    weather_data = await executor.execute(result1.tool_calls[0])
    
    # Step 2: Use weather in next query
    result2 = await agent.run(
        f"Given weather is {weather_data.result}, should I bring an umbrella?",
        tools=[]
    )
    
    return result2.content

Conditional Tool Execution

Execute tools only if certain conditions are met:

async def conditional_execution(agent, executor, query, tools):
    """Execute tools with validation."""
    
    result = await agent.run(query, tools=tools)
    
    for tool_call in result.tool_calls:
        # Validate before execution
        if not validate_tool_call(tool_call):
            print(f"Skipping unsafe tool: {tool_call.name}")
            continue
        
        # Check budget
        if estimate_cost(tool_call) > budget:
            print(f"Tool too expensive: {tool_call.name}")
            continue
        
        # Execute
        tool_result = await executor.execute(tool_call)

Custom Tool Validators

Add custom validation logic:

def validate_tool_arguments(tool_call: ToolCall, tool_config: ToolConfig) -> bool:
    """Custom validation logic."""
    
    # Check required fields
    for required in tool_config.parameters.get("required", []):
        if required not in tool_call.arguments:
            return False
    
    # Custom business logic
    if tool_call.name == "transfer_money":
        amount = tool_call.arguments.get("amount", 0)
        if amount > 10000:
            return False  # Block large transfers
    
    return True

Best Practices

1. Clear Tool Descriptions

# ❌ Bad
description = "Gets weather"

# ✅ Good
description = "Get current weather information for a specific location, including temperature, conditions, and humidity"

2. Validate Inputs

def get_weather(location: str, unit: str = "celsius") -> dict:
    # Validate unit
    if unit not in ["celsius", "fahrenheit"]:
        raise ValueError(f"Invalid unit: {unit}. Must be 'celsius' or 'fahrenheit'.")
    
    # Validate location
    if not location or len(location) < 2:
        raise ValueError("Location must be at least 2 characters")
    
    # ... rest of implementation

3. Return Structured Data

# ❌ Bad - string response
def get_weather(location: str) -> str:
    return "It's 22°C and sunny"

# ✅ Good - structured dict
def get_weather(location: str) -> dict:
    return {
        "location": location,
        "temperature": 22,
        "unit": "celsius",
        "condition": "sunny",
        "humidity": 65,
        "timestamp": "2025-10-21T14:30:00Z"
    }

4. Handle Errors Gracefully

def call_external_api(url: str) -> dict:
    try:
        response = requests.get(url, timeout=5)
        response.raise_for_status()
        return response.json()
    
    except requests.Timeout:
        return {"error": "API timeout", "code": "TIMEOUT"}
    
    except requests.HTTPError as e:
        return {"error": f"HTTP {e.response.status_code}", "code": "HTTP_ERROR"}
    
    except Exception as e:
        return {"error": str(e), "code": "UNKNOWN"}

5. Add Timeouts

async def slow_tool(query: str) -> dict:
    """Tool with timeout protection."""
    try:
        async with asyncio.timeout(10):  # 10 second timeout
            result = await expensive_operation(query)
            return result
    
    except asyncio.TimeoutError:
        return {"error": "Operation timed out", "code": "TIMEOUT"}

6. Log Tool Executions

import logging

logger = logging.getLogger(__name__)

def get_weather(location: str) -> dict:
    logger.info(f"get_weather called with location={location}")
    
    try:
        result = fetch_weather_data(location)
        logger.info(f"get_weather succeeded: {result['condition']}")
        return result
    
    except Exception as e:
        logger.error(f"get_weather failed: {e}")
        raise

Troubleshooting

Tool Not Found

Error: Tool 'xyz' not found

Cause: Tool name in ToolCall doesn't match any ToolConfig

Solution:

# Check available tools
print(f"Available tools: {list(executor.tools.keys())}")

# Verify tool name matches exactly
tool_config = ToolConfig(name="get_weather", ...)  # Must match exactly

Invalid Arguments

Error: TypeError: missing required argument

Cause: Model didn't provide required parameters

Solution:

# Add validation
def get_weather(location: str, unit: str = "celsius") -> dict:
    if not location:
        raise ValueError("location is required")
    # ... rest

Tool Execution Timeout

Error: Tool hangs indefinitely

Solution:

async def safe_tool(param: str) -> dict:
    try:
        async with asyncio.timeout(30):
            return await slow_operation(param)
    except asyncio.TimeoutError:
        return {"error": "Timeout", "code": "TIMEOUT"}

Wrong Tool Schema Format

Error: KeyError: 'name' or tools not working

Cause: Using OpenAI format instead of universal format

Solution:

# ❌ Wrong - OpenAI format
tools = [{
    "type": "function",
    "function": {"name": "...", ...}
}]

# ✅ Correct - Universal format
tools = [{
    "name": "get_weather",
    "description": "...",
    "parameters": {...}
}]

Costs Higher Than Expected

Issue: Tool queries cost more than expected

Cause: Token-based pricing + multiple turns

Solution:

# Track per-turn costs
total_cost = 0
for turn in range(max_turns):
    result = await agent.run(query, tools=tools)
    total_cost += result.total_cost
    print(f"Turn {turn}: ${result.total_cost:.6f}")

print(f"Total: ${total_cost:.6f}")

Next Steps

• Examples: See examples/tool_execution.py for complete working code
• Streaming: Read Streaming Guide for real-time tool streaming
• API Reference: Check API docs for detailed class documentation

Questions? Open an issue on GitHub.