Tool & Agent Integration Patterns

Building Blocks of Intelligence

Tools and agents are the fundamental building blocks of AgentRouter. Like Lego pieces that snap together perfectly, they can be combined in countless ways to create powerful, intelligent systems. This document explains how to create custom tools, integrate specialized agents, and leverage the seamless interoperability between them.

Tool & Agent Architecture

🎯 Tool Creation with @tool Decorator

Simple Tool Definition

from agentrouter import tool

@tool(schema={
    "type": "function",
    "function": {
        "name": "calculate",
        "description": "Perform mathematical calculations",
        "parameters": {
            "type": "object",
            "properties": {
                "expression": {
                    "type": "string",
                    "description": "Mathematical expression to evaluate"
                }
            },
            "required": ["expression"]
        }
    }
})
def calculate(expression: str) -> str:
    """Safely evaluate mathematical expressions"""
    try:
        result = eval(expression, {"__builtins__": {}}, {})
        return f"Result: {result}"
    except Exception as e:
        return f"Error: {str(e)}"

Advanced Tool with Multiple Parameters

@tool(schema={
    "type": "function",
    "function": {
        "name": "process_data",
        "description": "Process data with multiple parameters",
        "parameters": {
            "type": "object",
            "properties": {
                "data": {
                    "type": "string",
                    "description": "Data to process"
                },
                "format": {
                    "type": "string",
                    "enum": ["json", "csv", "text"],
                    "description": "Output format"
                },
                "validate": {
                    "type": "boolean",
                    "description": "Whether to validate data",
                    "default": True
                }
            },
            "required": ["data"]
        }
    }
})
def process_data(data: str, format: str = "json", validate: bool = True) -> dict:
    """Process data with validation and formatting"""
    # Implementation here
    result = {
        "data": data,
        "format": format,
        "validated": validate,
        "status": "processed"
    }
    return result

Worker Agents as Tools

Automatic Tool Conversion

When a worker agent is attached to a manager or another worker, it's automatically converted to a tool:

# Define a specialized worker agent
data_analyst = WorkerAgent(
    name="data_analyst",
    api_key="key",
    role="Processes raw datasets, calculates statistical metrics, identifies trends, and produces visualization reports",  # REQUIRED: Concise role
    goal="Analyze data and provide insights",
    backstory="Expert in statistical analysis and data visualization",
    instruction="Use statistical methods to find patterns and anomalies"
)

# Attach to manager - automatically becomes a tool
manager = ManagerAgent(
    name="manager",
    api_key="key"
)

manager.attach_worker(data_analyst)  # Worker registered as tool

# The Planning Agent can now call 'data_analyst' as a tool

Worker Agent Tool Schema

When converted to a tool, worker agents have this schema:

{
    "type": "function",
    "function": {
        "name": "data_analyst",  # Worker name becomes tool name
        "description": "Analyze data and provide insights",  # From goal
        "parameters": {
            "type": "object",
            "properties": {
                "task": {
                    "type": "string",
                    "description": "The task to perform"
                },
                "additional_context": {
                    "type": "string",
                    "description": "Additional context for the task"
                }
            },
            "required": ["task"]
        }
    }
}

🔄 Tool Registration & Discovery

Dynamic Tool Registration

# Register tools at initialization
manager = ManagerAgent(
    name="manager",
    api_key="key",
    tools=[calculate, process_data]  # Function references
)

# Or register after initialization
manager.register_tool(new_tool_function)

# Register a worker as a tool
manager.attach_worker(specialized_agent)

Tool Discovery & Validation

The system automatically:

1. Discovers tool metadata from the decorator
2. Validates function signature matches schema
3. Type-checks parameters at runtime
4. Handles errors gracefully with retry logic

Tool vs Agent Decision Matrix

When to Use Tools

Use Case	Example	Reasoning
Simple Calculations	add(2, 3)	Stateless, immediate result
Data Transformations	format_date("2024-01-15")	Pure function
External API Calls	fetch_data("endpoint")	Single operation
File Operations	read_file("data.csv")	System interaction
Validation	validate_email("user@example.com")	Quick check

When to Use Agents

Use Case	Example	Reasoning
Multi-step Workflows	Process refund request	Requires multiple decisions
Complex Analysis	Sentiment analysis on reviews	Needs context understanding
Creative Tasks	Write marketing copy	Requires creativity
Decision Making	Approve loan application	Complex evaluation
Orchestration	Coordinate team tasks	Managing multiple operations

🔧 Practical Tool Patterns

1. Data Processing Tools

@tool(schema={
    "type": "function",
    "function": {
        "name": "parse_csv",
        "description": "Parse CSV data",
        "parameters": {
            "type": "object",
            "properties": {
                "csv_data": {
                    "type": "string",
                    "description": "CSV data as string"
                },
                "delimiter": {
                    "type": "string",
                    "description": "CSV delimiter",
                    "default": ","
                }
            },
            "required": ["csv_data"]
        }
    }
})
def parse_csv(csv_data: str, delimiter: str = ",") -> list:
    """Parse CSV data into list of dictionaries"""
    import csv
    import io
    
    reader = csv.DictReader(io.StringIO(csv_data), delimiter=delimiter)
    return list(reader)

2. Validation Tools

@tool(schema={
    "type": "function",
    "function": {
        "name": "validate_json",
        "description": "Validate JSON structure",
        "parameters": {
            "type": "object",
            "properties": {
                "json_string": {
                    "type": "string",
                    "description": "JSON string to validate"
                }
            },
            "required": ["json_string"]
        }
    }
})
def validate_json(json_string: str) -> dict:
    """Validate and parse JSON string"""
    import json
    
    try:
        parsed = json.loads(json_string)
        return {
            "valid": True,
            "data": parsed,
            "error": None
        }
    except json.JSONDecodeError as e:
        return {
            "valid": False,
            "data": None,
            "error": str(e)
        }

3. Calculation Tools

@tool(schema={
    "type": "function",
    "function": {
        "name": "calculate_statistics",
        "description": "Calculate basic statistics",
        "parameters": {
            "type": "object",
            "properties": {
                "numbers": {
                    "type": "array",
                    "items": {"type": "number"},
                    "description": "List of numbers"
                }
            },
            "required": ["numbers"]
        }
    }
})
def calculate_statistics(numbers: list) -> dict:
    """Calculate mean, median, min, max"""
    if not numbers:
        return {"error": "Empty list"}
    
    import statistics
    
    return {
        "mean": statistics.mean(numbers),
        "median": statistics.median(numbers),
        "min": min(numbers),
        "max": max(numbers),
        "count": len(numbers)
    }

🔐 Tool Security & Validation

Input Validation Best Practices

@tool(schema=schema)
def secure_tool(user_input: str) -> dict:
    """Tool with input validation"""
    # Validate input length
    if len(user_input) > 10000:
        return {"error": "Input too long"}
    
    # Sanitize input
    sanitized = user_input.strip()
    
    # Validate format
    if not sanitized:
        return {"error": "Empty input"}
    
    # Process safely
    try:
        result = process_safely(sanitized)
        return {"success": True, "result": result}
    except Exception as e:
        return {"success": False, "error": str(e)}

Error Handling in Tools

@tool(schema=schema)
def robust_tool(param: str) -> dict:
    """Tool with comprehensive error handling"""
    try:
        # Attempt operation
        result = perform_operation(param)
        return {
            "success": True,
            "data": result
        }
    except ValueError as e:
        # Handle specific error
        return {
            "success": False,
            "error": f"Invalid value: {e}",
            "error_type": "validation"
        }
    except Exception as e:
        # Handle unexpected errors
        logger.error(f"Tool error: {e}")
        return {
            "success": False,
            "error": "Operation failed",
            "error_type": "unknown"
        }

🎯 Real-World Integration Example

E-Commerce Assistant with Tools and Agents

# Define specialized tools
@tool(schema=inventory_schema)
def check_inventory(product_id: str) -> dict:
    """Validates product availability, returns current stock levels and warehouse location"""
    # Simulate inventory check
    return {
        "product_id": product_id,
        "available": True,
        "quantity": 50
    }

@tool(schema=pricing_schema)
def calculate_discount(amount: float, code: str) -> float:
    """Applies promotional codes, calculates final price after discount validation"""
    discounts = {"SAVE10": 0.1, "SAVE20": 0.2}
    discount_rate = discounts.get(code, 0)
    return amount * (1 - discount_rate)

# Define specialized agents
customer_service = WorkerAgent(
    name="customer_service",
    role="Resolves complaints, processes returns, answers product questions, and escalates critical issues",  # REQUIRED: Clear role
    goal="Handle customer inquiries and complaints",
    instruction="Be helpful and empathetic"
)

order_processor = WorkerAgent(
    name="order_processor",
    role="Validates payments, updates inventory, confirms shipping details, and tracks order fulfillment",  # REQUIRED: Focused role
    goal="Process and fulfill orders",
    instruction="Ensure all orders are validated before processing"
)

# Create the main manager
ecommerce_manager = ManagerAgent(
    name="ecommerce_assistant",
    api_key="key",
    max_iterations=30
)

# Register tools
ecommerce_manager.register_tool(check_inventory)
ecommerce_manager.register_tool(calculate_discount)

# Attach workers
ecommerce_manager.attach_worker(customer_service)
ecommerce_manager.attach_worker(order_processor)

# The manager can now:
# 1. Use tools for simple operations (inventory, pricing)
# 2. Delegate to agents for complex tasks (customer service, order processing)

💡 Best Practices

1. Tool Naming Convention

# Good: Clear, action-oriented names
@tool(schema={"function": {"name": "fetch_user_profile"}})
@tool(schema={"function": {"name": "calculate_shipping_cost"}})
@tool(schema={"function": {"name": "validate_credit_card"}})

# Bad: Vague or noun-based names
# @tool(name="user")
# @tool(name="shipping")
# @tool(name="credit")

2. Schema Validation

# Always define complete schemas
schema = {
    "type": "function",
    "function": {
        "name": "process_order",
        "description": "Process customer order with validation",
        "parameters": {
            "type": "object",
            "properties": {
                "order_id": {
                    "type": "string",
                    "description": "Unique order identifier"
                },
                "items": {
                    "type": "array",
                    "description": "List of order items",
                    "items": {
                        "type": "object",
                        "properties": {
                            "product_id": {"type": "string"},
                            "quantity": {"type": "integer", "minimum": 1}
                        },
                        "required": ["product_id", "quantity"]
                    }
                }
            },
            "required": ["order_id", "items"]
        }
    }
}

3. Return Consistent Formats

@tool(schema=schema)
def consistent_tool(param: str) -> dict:
    """Always return consistent structure"""
    # Always return the same structure
    return {
        "success": bool,
        "data": Any,      # Actual result when successful
        "error": str,     # Error message when failed
        "metadata": dict  # Optional additional info
    }

4. Worker Agent Best Practices

# Create focused, specialized workers
analyst = WorkerAgent(
    name="data_analyst",
    role="Performs regression analysis, detects outliers, calculates correlations, and creates predictive models",  # REQUIRED: Concise role
    goal="Analyze data and provide statistical insights",  # Clear goal
    backstory="Expert in data science and statistics",  # Context
    instruction="Use appropriate statistical methods",  # Guidance
    max_iterations=20  # Appropriate for analysis tasks
)

# Inherit configuration from parent
analyst.config.api_timeout = 60.0  # Inherit from manager

---

Next: Nested Worker Hierarchies

Learn about multi-level agent delegation and complex workflows

Explore Nested Workers →