The agent uses a tool execution framework that allows you to build custom tools that the AI can invoke during conversations. This guide shows you how to create tools following the request_transfer pattern.
The tool system consists of three main components:
- Tool Definition - An async function that implements the tool logic
- Tool Builder - Dynamically generates tool schemas based on tenant configuration
- Tool Executor - Routes invocations to the correct tool implementation
Create a new file in src/apps/calls/tools/definitions/. Your tool must be an async function that:
- Takes
args: dict[str, Any], invocation: ToolInvocation, and optional deps: dict[str, Any]
- Returns a
ToolResult object
- Handles errors gracefully without raising exceptions
Here’s the complete structure from request_transfer.py:
import json
import logging
from typing import Any
from src.core.exceptions import ToolArgsParseError, ToolExecutionError
from src.core.logger import log_event
from src.models.openai.openai import ToolInvocation, ToolResult
async def request_transfer_tool(
args: dict[str, Any],
invocation: ToolInvocation,
deps: dict[str, Any] | None = None
) -> ToolResult:
"""
Tool to request a call transfer.
Expects args to contain:
- destination_id: str, the destination identifier
- reason: Optional[str], the reason for the transfer
"""
tenant_id = getattr(invocation.tenant_config, "tenant_id", None)
# Log the tool invocation
log_event(
logging.INFO,
"request_transfer_started",
tenant_id=tenant_id,
openai_sip_call_id=invocation.call_id,
function_call_id=invocation.function_call_id,
argument_keys=sorted(args.keys()),
)
# Validate required arguments
destination_id = args.get("destination_id")
if not isinstance(destination_id, str):
output = json.dumps({
"error": "invalid_arguments",
"detail": "destination_id argument is required and must be a string."
})
return ToolResult(
function_call_id=invocation.function_call_id,
ok=False,
error="invalid_arguments",
output=output,
)
# Execute the tool logic
try:
call_service = deps.get("call_service") if deps else None
target_uri = await resolve_destination(invocation.tenant_config, destination_id)
await call_service.refer_call(
call_id=invocation.call_id,
target_uri=target_uri,
idempotency_key=f"{invocation.call_id}-transfer-{target_uri}"
)
# Return success
output = json.dumps({
"message": "call_transfer_requested",
"target_uri": target_uri,
})
return ToolResult(
function_call_id=invocation.function_call_id,
ok=True,
output=output,
)
except Exception as e:
# Handle errors gracefully
output = json.dumps({
"error": "call_transfer_failed",
"detail": "Failed to transfer the call."
})
return ToolResult(
function_call_id=invocation.function_call_id,
ok=False,
error="call_transfer_failed",
output=output,
)
Always log tool invocations and results using log_event() for observability and debugging.
Register your tool with the ToolExecutor in tool_executor.py:
from src.apps.calls.tools.definitions.request_transfer import request_transfer_tool
from src.apps.calls.tools.definitions.your_tool import your_custom_tool
# Create the executor instance
tool_executor = ToolExecutor()
# Register tools
tool_executor.register("request_transfer", request_transfer_tool)
tool_executor.register("your_tool_name", your_custom_tool)
The registration maps a tool name (string) to your tool function. This name must match the name in your tool schema.
Add a builder method in ToolBuilder to generate the OpenAI function schema based on tenant configuration:
class ToolBuilder:
def __init__(self) -> None:
self._builders: list[Callable[[TenantConfig], FunctionTool | None]] = [
self._build_refer_tool,
self._build_your_custom_tool, # Add your builder here
]
def _build_your_custom_tool(self, cfg: TenantConfig) -> FunctionTool | None:
"""Build your custom tool schema based on tenant config."""
# Check if feature is enabled
if not cfg.features.your_feature.enabled:
return None
# Build the parameters schema
params = {
"type": "object",
"properties": {
"param1": {
"type": "string",
"description": "Description of parameter 1",
},
"param2": {
"type": "integer",
"description": "Description of parameter 2",
},
},
"required": ["param1"],
"additionalProperties": False,
}
# Return the function tool definition
return FunctionTool(
name="your_tool_name",
description="Clear description of what your tool does and when to use it.",
parameters=params,
)
The tool name in FunctionTool must exactly match the name used when registering in ToolExecutor.register().
Dependency Injection
Tools can receive dependencies through the deps parameter. Dependencies are injected in tool_executor.py:152-166:
# In ToolExecutor.execute()
deps: dict[str, Any] | None = None
if tool_name == "request_transfer":
deps = {}
deps["call_service"] = self._openai_call_service
result = await fn(args, invocation, deps)
For your custom tool, add a similar condition to inject required dependencies.
Best Practices
Validate Input Early
Check all required arguments at the start of your tool function. Return descriptive error messages in the ToolResult.output field.
Use Structured Output
Always return JSON-serialized data in the output field. This makes it easy for the AI to parse and understand the result.
Never Raise Exceptions
The ToolExecutor expects tools to return ToolResult in all cases. Catch exceptions and convert them to error results.
Log Comprehensively
Use log_event() to log:
- Tool invocation start with parameters
- Intermediate steps and decisions
- Success or failure with context
Make Tools Tenant-Aware
Use invocation.tenant_config to access tenant-specific settings and customize behavior per tenant.
Create tests in tests/apps/calls/tools/definitions/test_your_tool.py:
import pytest
from src.models.openai.openai import ToolInvocation, ToolResult
from src.apps.calls.tools.definitions.your_tool import your_custom_tool
@pytest.mark.asyncio
async def test_your_tool_success():
"""Test successful tool execution."""
args = {"param1": "value1", "param2": 42}
invocation = ToolInvocation(
call_id="test-call",
function_call_id="test-func-call",
name="your_tool_name",
# ... other required fields
)
deps = {"dependency_name": mock_dependency}
result = await your_custom_tool(args, invocation, deps)
assert result.ok is True
assert result.error is None
assert "expected_field" in result.output
@pytest.mark.asyncio
async def test_your_tool_invalid_args():
"""Test tool with invalid arguments."""
args = {} # Missing required param1
invocation = ToolInvocation(/* ... */)
result = await your_custom_tool(args, invocation, None)
assert result.ok is False
assert result.error == "invalid_arguments"
The complete execution flow from tool_executor.py:45-246:
- Parse Arguments - JSON string → Python dict (lines 100-149)
- Inject Dependencies - Add required services to
deps (lines 152-166)
- Execute Tool - Call the tool function with args, invocation, and deps (line 166)
- Handle Result - Log success or convert exceptions to error results (lines 168-246)
The ToolExecutor never raises exceptions. It always returns a ToolResult that the AI can understand, even when errors occur.
Common Patterns
Enum Parameters
Use enums to restrict parameter values:
params = {
"properties": {
"destination_id": {
"type": "string",
"enum": ["commercial", "planning", "comptabilite"],
"description": "Service to transfer to. Valid values: commercial, planning, comptabilite"
}
}
}
Optional Parameters
Mark parameters as optional by excluding them from required:
params = {
"properties": {
"required_param": {"type": "string"},
"optional_param": {"type": "string", "description": "Optional parameter"}
},
"required": ["required_param"], # optional_param not listed
}
Tenant-Specific Configuration
Access tenant config to customize tool behavior:
async def my_tool(args, invocation, deps):
cfg = invocation.tenant_config
feature_config = cfg.features.my_feature
if feature_config.require_confirmation:
# Add confirmation logic
pass
Next Steps
- Review
src/apps/calls/tools/definitions/request_transfer.py:1-276 for a complete working example
- See Prompt Engineering to instruct the AI on when to use your tool
- Check Testing for comprehensive testing strategies