Tools

The Daita tools system provides a universal abstraction for LLM-callable functions. Tools can come from plugins, MCP servers, or custom Python functions, and are automatically integrated into agents for autonomous use.

Overview

Tools are functions that agents can discover and execute to interact with external systems, process data, or perform operations. The unified tool system allows agents to:

1. Discover tools from multiple sources (plugins, MCP, custom functions)
2. Understand tool capabilities through structured schemas
3. Execute tools with type-safe parameter validation
4. Handle results consistently across all tool types

---

---

Key Features:

• Universal tool abstraction works with any source
• LLM-compatible schemas (OpenAI, Anthropic, etc.)
• Automatic type conversion and validation
• Async execution with timeout support
• Tool discovery and registration
• Provider-agnostic function calling format

Core Concepts

AgentTool Class

The AgentTool dataclass represents any callable function:

from daita.core.tools import AgentTool

tool = AgentTool(
    name="search_database",
    description="Search for records in the database",
    parameters={
        "query": {
            "type": "string",
            "description": "SQL query to execute",
            "required": True
        },
        "limit": {
            "type": "integer",
            "description": "Maximum results to return",
            "required": False
        }
    },
    handler=async_search_function,
    category="database",
    source="plugin",
    timeout_seconds=30
)

Tool Fields

Field	Type	Required	Description
name	str	Yes	Unique tool name
description	str	Yes	Human-readable tool description
parameters	Dict[str, Any]	Yes	JSON Schema parameter definition
handler	Callable	Yes	Async function that executes the tool
category	str	No	Tool category (database, storage, api, etc.)
source	str	No	Source of tool (plugin, mcp, custom) - default: "custom"
plugin_name	str	No	Plugin that provides this tool
timeout_seconds	int	No	Execution timeout in seconds

Creating Tools

Using the @tool Decorator

Create tools from any Python function using the @tool decorator:

from daita.core.tools import tool

# Simple function
@tool
async def calculate_total(price: float, quantity: int) -> float:
    """Calculate total cost of items."""
    return price * quantity

# Sync functions work too
@tool
def add_numbers(a: int, b: int) -> int:
    """Add two numbers together."""
    return a + b

# Execute tool
result = await calculate_total.execute({"price": 19.99, "quantity": 3})
print(result)  # 59.97

The @tool decorator automatically:

• Extracts parameter schemas from type hints and docstrings
• Handles both sync and async functions
• Converts functions to AgentTool instances

With Decorator Options

Customize tool metadata using decorator parameters:

from daita.core.tools import tool

@tool(
    name="product_search",
    description="Search the product catalog with advanced filters",
    timeout_seconds=10,
    category="search"
)
async def search_products(query: str, category: str = "all", max_results: int = 10):
    """
    Search for products in the catalog.

    Args:
        query: Search keywords or product name
        category: Product category to filter by (all, electronics, clothing, etc.)
        max_results: Maximum number of results to return (1-100)
    """
    # Search implementation
    return results

# Parameter schema is auto-extracted from type hints and docstring

The decorator automatically extracts parameter schemas from:

• Type hints (query: str, max_results: int)
• Default values (category: str = "all" makes it optional)
• Docstring (Args section provides descriptions)

Sync Functions

Sync functions are automatically wrapped for async execution:

from daita.core.tools import tool

@tool
def simple_calculation(x: int, y: int) -> int:
    """Add two numbers."""
    return x + y

# Automatically wrapped for async use
# Can be called with await
result = await simple_calculation.execute({"x": 5, "y": 3})

Tool Execution

Basic Execution

Execute tools by calling the execute() method:

from daita.core.tools import tool

@tool
async def my_function(param1: str, param2: int) -> str:
    """Example function."""
    return f"{param1}: {param2}"

# Execute with arguments
result = await my_function.execute({
    "param1": "value1",
    "param2": 42
})

Timeout Handling

Tools with timeouts raise RuntimeError if execution exceeds the limit:

import asyncio
from daita.core.tools import tool

@tool(timeout_seconds=5)
async def slow_operation(data: str) -> str:
    """Potentially slow operation."""
    await asyncio.sleep(10)  # Long operation
    return "done"

try:
    result = await slow_operation.execute({"data": "test"})
except RuntimeError as e:
    print(f"Timeout: {e}")
    # "Tool 'slow_operation' execution timed out after 5s"

Error Handling

Tool execution errors are propagated with context:

from daita.core.tools import tool

@tool
async def risky_operation(value: int) -> int:
    """Operation that might fail."""
    if value < 0:
        raise ValueError("Value must be positive")
    return value * 2

try:
    result = await risky_operation.execute({"value": -5})
except ValueError as e:
    print(f"Validation error: {e}")

LLM Integration Formats

OpenAI Function Calling

Convert tools to OpenAI function calling format:

tool = AgentTool.from_function(my_function)

# OpenAI format
openai_function = tool.to_openai_function()

# Returns:
{
    "type": "function",
    "function": {
        "name": "my_function",
        "description": "Function description",
        "parameters": {
            "type": "object",
            "properties": {...},
            "required": [...]
        }
    }
}

Anthropic Tool Format

Convert to Anthropic Claude tool format:

# Anthropic format
anthropic_tool = tool.to_anthropic_tool()

# Returns:
{
    "name": "my_function",
    "description": "Function description",
    "input_schema": {
        "type": "object",
        "properties": {...},
        "required": [...]
    }
}

Generic LLM Format

Use the generic format for any provider:

# Generic format (works with most providers)
llm_function = tool.to_llm_function()

# Returns:
{
    "name": "my_function",
    "description": "Function description",
    "parameters": {
        "type": "object",
        "properties": {...},
        "required": [...]
    }
}

Prompt Description

Generate human-readable descriptions for prompt injection:

tool = AgentTool.from_function(
    search_products,
    parameters={
        "query": {
            "type": "string",
            "description": "Search keywords",
            "required": True
        },
        "limit": {
            "type": "integer",
            "description": "Max results",
            "required": False
        }
    }
)

# Get prompt-friendly description
description = tool.to_prompt_description()

# Returns:
# search_products: Search the product catalog with filters
# Parameters:
#   - query (string) (required): Search keywords
#   - limit (integer) (optional): Max results

Tool Registry

ToolRegistry Class

The ToolRegistry manages collections of tools:

from daita.core.tools import ToolRegistry, AgentTool

# Create registry
registry = ToolRegistry()

# Register tools
tool1 = AgentTool.from_function(function1)
tool2 = AgentTool.from_function(function2)

registry.register(tool1)
registry.register(tool2)

# Or register many at once
registry.register_many([tool1, tool2, tool3])

# Get tool by name
tool = registry.get("function1")

# Execute tool through registry
result = await registry.execute("function1", {"arg": "value"})

# Registry info
print(f"Total tools: {registry.tool_count}")
print(f"Tool names: {registry.tool_names}")

Tool Lookup

Retrieve tools from the registry:

registry = ToolRegistry()
# ... register tools ...

# Get single tool
tool = registry.get("search_database")
if tool:
    print(f"Found: {tool.name}")

# List all tools
all_tools = registry.tools
for tool in all_tools:
    print(f"{tool.name} ({tool.source}): {tool.description}")

# Filter by source
plugin_tools = [t for t in registry.tools if t.source == "plugin"]
mcp_tools = [t for t in registry.tools if t.source == "mcp"]
custom_tools = [t for t in registry.tools if t.source == "custom"]

Agent Integration

Automatic Tool Registration

Tools from plugins and MCP servers are automatically registered:

from daita import SubstrateAgent
from daita.plugins import PostgreSQLPlugin

# Tools from multiple sources
db_plugin = PostgreSQLPlugin(host="localhost", database="mydb")

agent = SubstrateAgent(
    name="multi_tool_agent",
    tools=[db_plugin],  # Plugin tools
    mcp={                # MCP tools
        "command": "npx",
        "args": ["-y", "@modelcontextprotocol/server-filesystem", "/data"]
    }
)

# Start to initialize tools
await agent.start()

# All tools are now available
print(f"Available tools: {agent.tool_names}")

# Agent autonomously uses tools
answer = await agent.run("How many users are in the database?")

Manual Tool Registration

Register custom tools with an agent:

from daita import SubstrateAgent
from daita.core.tools import tool

# Create custom tool
@tool
async def custom_operation(data: str) -> str:
    """Custom business logic."""
    return data.upper()

@tool
async def another_operation(x: int, y: int) -> int:
    """Another custom operation."""
    return x * y

agent = SubstrateAgent(name="my_agent")

# Register single tool
agent.register_tool(custom_operation)

# Or register multiple
agent.register_tools([custom_operation, another_operation])

Autonomous Tool Usage

Once tools are registered, the agent autonomously decides when and how to use them:

from daita import SubstrateAgent
from daita.core.tools import tool

# Create custom tools
@tool
async def fetch_data(source: str) -> dict:
    """Fetch data from external source."""
    return {"users": [{"id": 1, "name": "Alice"}, {"id": 2, "name": "Bob"}]}

@tool
async def analyze_data(data: dict) -> dict:
    """Analyze data and return insights."""
    return {"count": len(data.get("users", [])), "insights": "Data looks good"}

agent = SubstrateAgent(name="agent")
agent.register_tools([fetch_data, analyze_data])

await agent.start()

# Agent autonomously chains tools to answer the question
answer = await agent.run("Fetch data from 'api/users' and analyze it")
# Agent will:
# 1. Call fetch_data with source='api/users'
# 2. Call analyze_data with the fetched data
# 3. Provide a natural language answer
print(answer)

Streaming Tool Execution

Monitor tool execution in real-time using streaming events:

from daita.core.streaming import AgentEvent, EventType

def monitor_tools(event: AgentEvent):
    if event.type == EventType.TOOL_CALL:
        print(f"🔧 Calling: {event.tool_name}")
        print(f"   Args: {event.tool_args}")

    elif event.type == EventType.TOOL_RESULT:
        print(f"   ✅ Result: {event.result}")

# Get real-time visibility into tool usage
answer = await agent.run(
    "Fetch and analyze user data",
    on_event=monitor_tools  # See tools being called in real-time
)

This provides transparency into which tools the agent is using, what arguments it's passing, and what results it receives - essential for debugging and understanding agent behavior.

Tool Discovery

Discover what tools are available to an agent:

from daita.plugins import PostgreSQLPlugin

db_plugin = PostgreSQLPlugin(host="localhost", database="mydb")

agent = SubstrateAgent(
    name="agent",
    tools=[db_plugin],
    mcp={
        "command": "npx",
        "args": ["-y", "@modelcontextprotocol/server-filesystem", "/data"]
    }
)

# Initialize tools
await agent.start()

# List all available tools
tools = agent.available_tools
for tool in tools:
    print(f"Tool: {tool.name}")
    print(f"  Source: {tool.source}")
    print(f"  Category: {tool.category}")
    print(f"  Description: {tool.description}")
    print()

# Get just the names
tool_names = agent.tool_names
print(f"Tools: {', '.join(tool_names)}")

Plugin Tools

Plugins can expose their capabilities as tools by implementing get_tools():

from daita.plugins.base import BasePlugin
from daita.core.tools import AgentTool

class MyCustomPlugin(BasePlugin):
    """Custom plugin with tools."""

    def __init__(self, api_key: str):
        self.api_key = api_key

    async def _search_api(self, query: str, limit: int = 10) -> list:
        """Internal search implementation."""
        # API call implementation
        return results

    def get_tools(self) -> list:
        """Expose plugin capabilities as tools."""
        search_tool = AgentTool(
            name="search_custom_api",
            description="Search the custom API",
            parameters={
                "query": {
                    "type": "string",
                    "description": "Search query",
                    "required": True
                },
                "limit": {
                    "type": "integer",
                    "description": "Max results",
                    "required": False
                }
            },
            handler=self._search_api,
            source="plugin",
            plugin_name="MyCustomPlugin",
            category="search"
        )

        return [search_tool]

# Use plugin with agent
plugin = MyCustomPlugin(api_key="key")

agent = SubstrateAgent(
    name="agent",
    tools=[plugin]
)

await agent.start()

# Plugin tools automatically registered - agent uses them autonomously
answer = await agent.run("Search the custom API for products")
print(answer)

MCP Tools

MCP tools are automatically converted to AgentTool format:

from daita.plugins.mcp import MCPServer, MCPTool
from daita.core.tools import AgentTool

# MCP tool from server
mcp_tool = MCPTool(
    name="read_file",
    description="Read a file from disk",
    input_schema={
        "type": "object",
        "properties": {
            "path": {
                "type": "string",
                "description": "File path"
            }
        }
    }
)

# Convert to AgentTool
registry = mcp_registry  # MCPToolRegistry instance
agent_tool = AgentTool.from_mcp_tool(mcp_tool, registry)

# Now can be used like any other tool
result = await agent_tool.execute({"path": "/data/file.txt"})

Advanced Examples

Complex Tool with Validation

async def process_order(
    order_id: str,
    items: list,
    customer_email: str,
    priority: str = "normal"
) -> dict:
    """
    Process a customer order.

    Validates order data and submits to processing queue.
    """
    # Validation
    if not order_id:
        raise ValueError("order_id is required")
    if not items or len(items) == 0:
        raise ValueError("items cannot be empty")
    if priority not in ["low", "normal", "high"]:
        raise ValueError("priority must be low, normal, or high")

    # Processing logic
    result = {
        "order_id": order_id,
        "status": "processing",
        "item_count": len(items),
        "priority": priority,
        "notification_sent": customer_email
    }

    return result

# Create tool with detailed schema
tool = AgentTool.from_function(
    process_order,
    parameters={
        "order_id": {
            "type": "string",
            "description": "Unique order identifier",
            "required": True
        },
        "items": {
            "type": "array",
            "description": "List of items in the order",
            "items": {
                "type": "object",
                "properties": {
                    "product_id": {"type": "string"},
                    "quantity": {"type": "integer"}
                }
            },
            "required": True
        },
        "customer_email": {
            "type": "string",
            "description": "Customer email for notifications",
            "required": True
        },
        "priority": {
            "type": "string",
            "description": "Order priority level",
            "enum": ["low", "normal", "high"],
            "required": False
        }
    },
    category="business",
    timeout_seconds=30
)

Tool Factory Pattern

def create_database_tool(table_name: str, connection_string: str) -> AgentTool:
    """Factory function to create database query tools."""

    async def query_table(filter_column: str, filter_value: str) -> list:
        """Query specific table with filters."""
        # Database query implementation
        return results

    return AgentTool.from_function(
        query_table,
        name=f"query_{table_name}",
        description=f"Query the {table_name} table",
        parameters={
            "filter_column": {
                "type": "string",
                "description": "Column to filter on",
                "required": True
            },
            "filter_value": {
                "type": "string",
                "description": "Value to filter for",
                "required": True
            }
        },
        category="database"
    )

# Create tools for different tables
users_tool = create_database_tool("users", conn_str)
orders_tool = create_database_tool("orders", conn_str)
products_tool = create_database_tool("products", conn_str)

# Register with agent
agent.register_tools([users_tool, orders_tool, products_tool])

Conditional Tool Loading

from daita import SubstrateAgent
from daita.core.tools import AgentTool

def create_agent_with_tools(environment: str) -> SubstrateAgent:
    """Create agent with environment-specific tools."""

    agent = SubstrateAgent(name=f"{environment}_agent")

    # Always include these tools
    base_tools = [
        AgentTool.from_function(validate_data),
        AgentTool.from_function(format_output)
    ]

    # Environment-specific tools
    if environment == "production":
        prod_tools = [
            AgentTool.from_function(log_to_datadog),
            AgentTool.from_function(send_alert)
        ]
        agent.register_tools(base_tools + prod_tools)

    elif environment == "development":
        dev_tools = [
            AgentTool.from_function(mock_external_api),
            AgentTool.from_function(debug_output)
        ]
        agent.register_tools(base_tools + dev_tools)

    return agent

# Create environment-specific agents
prod_agent = create_agent_with_tools("production")
dev_agent = create_agent_with_tools("development")

Best Practices

Tool Design

1. Clear names: Use descriptive, action-oriented names (e.g., search_users, calculate_total)
2. Detailed descriptions: Help LLMs understand when to use the tool
3. Parameter validation: Validate inputs early and provide clear error messages
4. Idempotency: Design tools to be safe to retry
5. Timeout appropriately: Set reasonable timeouts based on expected operation duration

# Good tool design
async def fetch_user_profile(user_id: str) -> dict:
    """
    Fetch complete user profile from the database.

    Returns user personal info, preferences, and account status.
    Safe to call multiple times - no side effects.
    """
    if not user_id or not user_id.strip():
        raise ValueError("user_id cannot be empty")

    # Implementation
    return user_data

tool = AgentTool.from_function(
    fetch_user_profile,
    parameters={
        "user_id": {
            "type": "string",
            "description": "Unique identifier for the user (UUID format)",
            "required": True
        }
    },
    category="user_management",
    timeout_seconds=10
)

Parameter Schemas

1. Use JSON Schema types: string, integer, number, boolean, array, object
2. Mark required fields: Set "required": True for mandatory parameters
3. Provide examples: Include example values in descriptions
4. Enumerate options: Use enum for fixed option sets
5. Document ranges: Specify min/max for numeric values

# Comprehensive parameter schema
parameters = {
    "query": {
        "type": "string",
        "description": "Search query (e.g., 'red shoes size 10')",
        "required": True
    },
    "category": {
        "type": "string",
        "description": "Product category filter",
        "enum": ["all", "electronics", "clothing", "books"],
        "required": False
    },
    "max_price": {
        "type": "number",
        "description": "Maximum price in USD (0.01 to 10000.00)",
        "required": False
    },
    "limit": {
        "type": "integer",
        "description": "Maximum results to return (1-100)",
        "required": False
    },
    "sort_by": {
        "type": "string",
        "description": "Sort order for results",
        "enum": ["relevance", "price_asc", "price_desc", "newest"],
        "required": False
    }
}

Error Handling

1. Validate early: Check parameters before expensive operations
2. Clear error messages: Explain what went wrong and how to fix it
3. Use appropriate exceptions: ValueError for validation, RuntimeError for execution failures
4. Log errors: Use logging for debugging without exposing internals

import logging

logger = logging.getLogger(__name__)

async def robust_tool(user_id: str, operation: str) -> dict:
    """Tool with comprehensive error handling."""

    # Input validation
    if not user_id:
        raise ValueError("user_id is required")

    if operation not in ["read", "update", "delete"]:
        raise ValueError(
            f"Invalid operation '{operation}'. "
            f"Must be one of: read, update, delete"
        )

    try:
        # Main operation
        result = await perform_operation(user_id, operation)
        return result

    except ConnectionError as e:
        logger.error(f"Database connection failed: {e}")
        raise RuntimeError("Unable to connect to database") from e

    except Exception as e:
        logger.error(f"Unexpected error in robust_tool: {e}")
        raise RuntimeError(f"Tool execution failed: {str(e)}") from e

Performance

1. Set appropriate timeouts: Prevent hanging operations
2. Use async: Leverage async/await for I/O operations
3. Cache when possible: Store expensive computation results
4. Batch operations: Combine multiple calls when possible

from functools import lru_cache

# Caching for expensive operations
@lru_cache(maxsize=100)
def get_config(config_key: str) -> dict:
    """Cached config lookup."""
    return load_config(config_key)

# Async for I/O
async def fetch_multiple_users(user_ids: list) -> list:
    """Fetch multiple users concurrently."""
    tasks = [fetch_user(uid) for uid in user_ids]
    return await asyncio.gather(*tasks)

# Batching
async def process_batch(items: list) -> list:
    """Process items in batch for efficiency."""
    return await bulk_process(items)

Troubleshooting

Tool Not Found

If tool isn't discovered:

# Check tool registration
agent = SubstrateAgent(name="agent")
tool = AgentTool.from_function(my_function)
agent.register_tool(tool)

# Verify registration
print(f"Registered tools: {agent.tool_names}")

Parameter Validation Errors

If tool execution fails with validation errors:

# Check parameter schema matches function signature
tool = AgentTool.from_function(
    my_function,
    parameters={
        # Ensure parameter names match function args
        "correct_param_name": {...}
    }
)

# Test execution with valid parameters
try:
    result = await tool.execute({"correct_param_name": "value"})
except Exception as e:
    print(f"Validation error: {e}")

Timeout Issues

If tools timeout frequently:

# Increase timeout
tool = AgentTool.from_function(
    slow_function,
    timeout_seconds=60  # Increase from default
)

# Or remove timeout for operations that may take long
tool = AgentTool.from_function(
    variable_duration_function,
    timeout_seconds=None  # No timeout
)

Next Steps

• MCP - Model Context Protocol integration
• Substrate Agent - Using tools with agents
• Workflows - Multi-tool orchestration
• Error Handling - Robust error management