Introduction: Tool use transforms LLMs from text generators into capable agents that can search the web, query databases, execute code, and interact with APIs. But implementing tool use well is tricky—models hallucinate tool calls, pass invalid arguments, and struggle with multi-step tool chains. The difference between a demo and production system lies in robust tool definitions, proper error handling, and intelligent orchestration. This guide covers practical tool use patterns: defining tools with clear schemas, implementing reliable tool execution, handling errors gracefully, building multi-tool workflows, and creating production-ready tool orchestration systems.
Tool Definition and Registry
from dataclasses import dataclass, field
from typing import Any, Callable, Optional
from enum import Enum
import json
class ParameterType(Enum):
"""Types for tool parameters."""
STRING = "string"
INTEGER = "integer"
NUMBER = "number"
BOOLEAN = "boolean"
ARRAY = "array"
OBJECT = "object"
@dataclass
class ToolParameter:
"""A parameter for a tool."""
name: str
param_type: ParameterType
description: str
required: bool = True
enum_values: list[str] = None
default: Any = None
@dataclass
class Tool:
"""A tool that can be called by an LLM."""
name: str
description: str
parameters: list[ToolParameter]
handler: Callable
requires_confirmation: bool = False
timeout_seconds: int = 30
def to_openai_schema(self) -> dict:
"""Convert to OpenAI function calling schema."""
properties = {}
required = []
for param in self.parameters:
prop = {
"type": param.param_type.value,
"description": param.description
}
if param.enum_values:
prop["enum"] = param.enum_values
properties[param.name] = prop
if param.required:
required.append(param.name)
return {
"type": "function",
"function": {
"name": self.name,
"description": self.description,
"parameters": {
"type": "object",
"properties": properties,
"required": required
}
}
}
def to_anthropic_schema(self) -> dict:
"""Convert to Anthropic tool schema."""
properties = {}
required = []
for param in self.parameters:
prop = {
"type": param.param_type.value,
"description": param.description
}
if param.enum_values:
prop["enum"] = param.enum_values
properties[param.name] = prop
if param.required:
required.append(param.name)
return {
"name": self.name,
"description": self.description,
"input_schema": {
"type": "object",
"properties": properties,
"required": required
}
}
class ToolRegistry:
"""Registry of available tools."""
def __init__(self):
self.tools: dict[str, Tool] = {}
def register(self, tool: Tool) -> None:
"""Register a tool."""
self.tools[tool.name] = tool
def get(self, name: str) -> Optional[Tool]:
"""Get a tool by name."""
return self.tools.get(name)
def list_tools(self) -> list[Tool]:
"""List all registered tools."""
return list(self.tools.values())
def to_openai_tools(self) -> list[dict]:
"""Get all tools in OpenAI format."""
return [tool.to_openai_schema() for tool in self.tools.values()]
def to_anthropic_tools(self) -> list[dict]:
"""Get all tools in Anthropic format."""
return [tool.to_anthropic_schema() for tool in self.tools.values()]
# Example tool implementations
async def search_web(query: str, num_results: int = 5) -> dict:
"""Search the web for information."""
# Simulated web search
return {
"results": [
{"title": f"Result {i}", "snippet": f"Information about {query}"}
for i in range(num_results)
]
}
async def get_weather(location: str, units: str = "celsius") -> dict:
"""Get current weather for a location."""
# Simulated weather API
return {
"location": location,
"temperature": 22,
"units": units,
"conditions": "partly cloudy"
}
async def calculate(expression: str) -> dict:
"""Evaluate a mathematical expression."""
try:
# Safe evaluation
result = eval(expression, {"__builtins__": {}}, {})
return {"result": result, "expression": expression}
except Exception as e:
return {"error": str(e)}
# Create and register tools
def create_default_tools() -> ToolRegistry:
"""Create registry with default tools."""
registry = ToolRegistry()
registry.register(Tool(
name="search_web",
description="Search the web for current information on any topic",
parameters=[
ToolParameter("query", ParameterType.STRING, "The search query"),
ToolParameter("num_results", ParameterType.INTEGER, "Number of results to return", required=False)
],
handler=search_web
))
registry.register(Tool(
name="get_weather",
description="Get current weather conditions for a location",
parameters=[
ToolParameter("location", ParameterType.STRING, "City name or location"),
ToolParameter("units", ParameterType.STRING, "Temperature units", required=False, enum_values=["celsius", "fahrenheit"])
],
handler=get_weather
))
registry.register(Tool(
name="calculate",
description="Evaluate a mathematical expression",
parameters=[
ToolParameter("expression", ParameterType.STRING, "Mathematical expression to evaluate")
],
handler=calculate
))
return registryTool Execution Engine
from dataclasses import dataclass
from typing import Any, Optional
import asyncio
import json
import traceback
@dataclass
class ToolCall:
"""A request to call a tool."""
id: str
name: str
arguments: dict
@dataclass
class ToolResult:
"""Result of a tool execution."""
call_id: str
tool_name: str
success: bool
result: Any = None
error: str = None
execution_time_ms: float = 0
class ToolExecutor:
"""Execute tool calls safely."""
def __init__(self, registry: ToolRegistry):
self.registry = registry
async def execute(self, call: ToolCall) -> ToolResult:
"""Execute a single tool call."""
import time
start = time.time()
tool = self.registry.get(call.name)
if not tool:
return ToolResult(
call_id=call.id,
tool_name=call.name,
success=False,
error=f"Unknown tool: {call.name}"
)
try:
# Validate arguments
validation_error = self._validate_arguments(tool, call.arguments)
if validation_error:
return ToolResult(
call_id=call.id,
tool_name=call.name,
success=False,
error=validation_error
)
# Execute with timeout
result = await asyncio.wait_for(
tool.handler(**call.arguments),
timeout=tool.timeout_seconds
)
execution_time = (time.time() - start) * 1000
return ToolResult(
call_id=call.id,
tool_name=call.name,
success=True,
result=result,
execution_time_ms=execution_time
)
except asyncio.TimeoutError:
return ToolResult(
call_id=call.id,
tool_name=call.name,
success=False,
error=f"Tool execution timed out after {tool.timeout_seconds}s"
)
except Exception as e:
return ToolResult(
call_id=call.id,
tool_name=call.name,
success=False,
error=f"Execution error: {str(e)}"
)
def _validate_arguments(self, tool: Tool, arguments: dict) -> Optional[str]:
"""Validate tool arguments."""
# Check required parameters
for param in tool.parameters:
if param.required and param.name not in arguments:
return f"Missing required parameter: {param.name}"
# Check parameter types
for param in tool.parameters:
if param.name in arguments:
value = arguments[param.name]
if param.param_type == ParameterType.STRING and not isinstance(value, str):
return f"Parameter {param.name} must be a string"
if param.param_type == ParameterType.INTEGER and not isinstance(value, int):
return f"Parameter {param.name} must be an integer"
if param.param_type == ParameterType.NUMBER and not isinstance(value, (int, float)):
return f"Parameter {param.name} must be a number"
if param.param_type == ParameterType.BOOLEAN and not isinstance(value, bool):
return f"Parameter {param.name} must be a boolean"
if param.enum_values and value not in param.enum_values:
return f"Parameter {param.name} must be one of: {param.enum_values}"
return None
async def execute_batch(self, calls: list[ToolCall]) -> list[ToolResult]:
"""Execute multiple tool calls in parallel."""
tasks = [self.execute(call) for call in calls]
return await asyncio.gather(*tasks)
class RetryingToolExecutor:
"""Tool executor with retry logic."""
def __init__(
self,
executor: ToolExecutor,
max_retries: int = 3,
retry_delay: float = 1.0
):
self.executor = executor
self.max_retries = max_retries
self.retry_delay = retry_delay
async def execute(self, call: ToolCall) -> ToolResult:
"""Execute with retries."""
last_result = None
for attempt in range(self.max_retries):
result = await self.executor.execute(call)
if result.success:
return result
last_result = result
# Don't retry certain errors
if "Unknown tool" in (result.error or ""):
return result
if "Missing required parameter" in (result.error or ""):
return result
# Wait before retry
if attempt < self.max_retries - 1:
await asyncio.sleep(self.retry_delay * (attempt + 1))
return last_resultLLM Tool Integration
from dataclasses import dataclass
from typing import Any, Optional
import json
@dataclass
class ToolUseResponse:
"""Response from tool-using LLM."""
content: str
tool_calls: list[ToolCall]
tool_results: list[ToolResult]
finished: bool
class ToolUsingLLM:
"""LLM client with tool use capabilities."""
def __init__(
self,
client: Any,
registry: ToolRegistry,
model: str = "gpt-4o-mini",
max_tool_rounds: int = 5
):
self.client = client
self.registry = registry
self.executor = ToolExecutor(registry)
self.model = model
self.max_tool_rounds = max_tool_rounds
async def chat(
self,
messages: list[dict],
system_prompt: str = None
) -> ToolUseResponse:
"""Chat with tool use."""
all_messages = []
if system_prompt:
all_messages.append({"role": "system", "content": system_prompt})
all_messages.extend(messages)
all_tool_calls = []
all_tool_results = []
for round_num in range(self.max_tool_rounds):
# Call LLM with tools
response = await self.client.chat.completions.create(
model=self.model,
messages=all_messages,
tools=self.registry.to_openai_tools(),
tool_choice="auto"
)
message = response.choices[0].message
# Check if model wants to use tools
if not message.tool_calls:
return ToolUseResponse(
content=message.content or "",
tool_calls=all_tool_calls,
tool_results=all_tool_results,
finished=True
)
# Add assistant message
all_messages.append({
"role": "assistant",
"content": message.content,
"tool_calls": [
{
"id": tc.id,
"type": "function",
"function": {
"name": tc.function.name,
"arguments": tc.function.arguments
}
}
for tc in message.tool_calls
]
})
# Execute tool calls
for tc in message.tool_calls:
call = ToolCall(
id=tc.id,
name=tc.function.name,
arguments=json.loads(tc.function.arguments)
)
all_tool_calls.append(call)
result = await self.executor.execute(call)
all_tool_results.append(result)
# Add tool result to messages
all_messages.append({
"role": "tool",
"tool_call_id": tc.id,
"content": json.dumps(result.result if result.success else {"error": result.error})
})
# Max rounds reached
return ToolUseResponse(
content="Maximum tool use rounds reached",
tool_calls=all_tool_calls,
tool_results=all_tool_results,
finished=False
)
class AnthropicToolUsingLLM:
"""Anthropic Claude with tool use."""
def __init__(
self,
client: Any,
registry: ToolRegistry,
model: str = "claude-3-5-sonnet-20241022",
max_tool_rounds: int = 5
):
self.client = client
self.registry = registry
self.executor = ToolExecutor(registry)
self.model = model
self.max_tool_rounds = max_tool_rounds
async def chat(
self,
messages: list[dict],
system_prompt: str = None
) -> ToolUseResponse:
"""Chat with tool use."""
all_tool_calls = []
all_tool_results = []
current_messages = list(messages)
for round_num in range(self.max_tool_rounds):
response = await self.client.messages.create(
model=self.model,
max_tokens=4096,
system=system_prompt or "",
messages=current_messages,
tools=self.registry.to_anthropic_tools()
)
# Check stop reason
if response.stop_reason == "end_turn":
# Extract text content
text_content = ""
for block in response.content:
if block.type == "text":
text_content += block.text
return ToolUseResponse(
content=text_content,
tool_calls=all_tool_calls,
tool_results=all_tool_results,
finished=True
)
# Process tool use blocks
assistant_content = []
tool_use_blocks = []
for block in response.content:
if block.type == "text":
assistant_content.append({"type": "text", "text": block.text})
elif block.type == "tool_use":
assistant_content.append({
"type": "tool_use",
"id": block.id,
"name": block.name,
"input": block.input
})
tool_use_blocks.append(block)
current_messages.append({"role": "assistant", "content": assistant_content})
# Execute tools and add results
tool_results_content = []
for block in tool_use_blocks:
call = ToolCall(
id=block.id,
name=block.name,
arguments=block.input
)
all_tool_calls.append(call)
result = await self.executor.execute(call)
all_tool_results.append(result)
tool_results_content.append({
"type": "tool_result",
"tool_use_id": block.id,
"content": json.dumps(result.result if result.success else {"error": result.error})
})
current_messages.append({"role": "user", "content": tool_results_content})
return ToolUseResponse(
content="Maximum tool use rounds reached",
tool_calls=all_tool_calls,
tool_results=all_tool_results,
finished=False
)Tool Selection and Routing
from dataclasses import dataclass
from typing import Any, Optional
@dataclass
class ToolSelectionResult:
"""Result of tool selection."""
selected_tools: list[str]
confidence: float
reasoning: str
class ToolSelector:
"""Select appropriate tools for a query."""
def __init__(self, client: Any, registry: ToolRegistry, model: str = "gpt-4o-mini"):
self.client = client
self.registry = registry
self.model = model
async def select_tools(
self,
query: str,
max_tools: int = 3
) -> ToolSelectionResult:
"""Select tools needed for a query."""
tools_description = "\n".join([
f"- {tool.name}: {tool.description}"
for tool in self.registry.list_tools()
])
prompt = f"""Given this user query, select which tools (if any) would be helpful.
Available tools:
{tools_description}
User query: {query}
Respond in JSON format:
{{
"selected_tools": ["tool1", "tool2"],
"reasoning": "Brief explanation of why these tools are needed"
}}
If no tools are needed, return an empty list."""
response = await self.client.chat.completions.create(
model=self.model,
messages=[{"role": "user", "content": prompt}],
response_format={"type": "json_object"}
)
import json
result = json.loads(response.choices[0].message.content)
# Validate selected tools exist
valid_tools = [
t for t in result.get("selected_tools", [])
if self.registry.get(t)
][:max_tools]
return ToolSelectionResult(
selected_tools=valid_tools,
confidence=0.9 if valid_tools else 0.5,
reasoning=result.get("reasoning", "")
)
class FilteredToolRegistry:
"""Registry that filters tools based on selection."""
def __init__(self, base_registry: ToolRegistry):
self.base_registry = base_registry
self.allowed_tools: set[str] = set()
def set_allowed_tools(self, tool_names: list[str]) -> None:
"""Set which tools are allowed."""
self.allowed_tools = set(tool_names)
def to_openai_tools(self) -> list[dict]:
"""Get filtered tools in OpenAI format."""
return [
tool.to_openai_schema()
for tool in self.base_registry.list_tools()
if tool.name in self.allowed_tools
]
def get(self, name: str) -> Optional[Tool]:
"""Get tool if allowed."""
if name in self.allowed_tools:
return self.base_registry.get(name)
return None
class SmartToolRouter:
"""Route queries to appropriate tool configurations."""
def __init__(
self,
client: Any,
registry: ToolRegistry,
selector: ToolSelector
):
self.client = client
self.registry = registry
self.selector = selector
self.executor = ToolExecutor(registry)
async def process(
self,
query: str,
system_prompt: str = None
) -> ToolUseResponse:
"""Process query with smart tool routing."""
# First, select relevant tools
selection = await self.selector.select_tools(query)
if not selection.selected_tools:
# No tools needed, direct response
response = await self.client.chat.completions.create(
model="gpt-4o-mini",
messages=[
{"role": "system", "content": system_prompt or "You are a helpful assistant."},
{"role": "user", "content": query}
]
)
return ToolUseResponse(
content=response.choices[0].message.content,
tool_calls=[],
tool_results=[],
finished=True
)
# Create filtered registry
filtered = FilteredToolRegistry(self.registry)
filtered.set_allowed_tools(selection.selected_tools)
# Use tool-enabled LLM with filtered tools
tool_llm = ToolUsingLLM(
self.client,
filtered,
max_tool_rounds=3
)
return await tool_llm.chat(
[{"role": "user", "content": query}],
system_prompt
)Production Tool Service
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from typing import Optional
app = FastAPI()
# Initialize components
registry = create_default_tools()
executor = ToolExecutor(registry)
tool_llm = None # Initialize with client
tool_selector = None
smart_router = None
class ChatRequest(BaseModel):
messages: list[dict]
system_prompt: Optional[str] = None
max_tool_rounds: int = 5
class ToolCallRequest(BaseModel):
name: str
arguments: dict
class SelectToolsRequest(BaseModel):
query: str
max_tools: int = 3
@app.post("/v1/chat")
async def chat_with_tools(request: ChatRequest):
"""Chat with tool use."""
response = await tool_llm.chat(
request.messages,
request.system_prompt
)
return {
"content": response.content,
"tool_calls": [
{
"id": tc.id,
"name": tc.name,
"arguments": tc.arguments
}
for tc in response.tool_calls
],
"tool_results": [
{
"call_id": tr.call_id,
"tool_name": tr.tool_name,
"success": tr.success,
"result": tr.result,
"error": tr.error
}
for tr in response.tool_results
],
"finished": response.finished
}
@app.post("/v1/chat/smart")
async def smart_chat(request: ChatRequest):
"""Chat with smart tool routing."""
query = request.messages[-1]["content"] if request.messages else ""
response = await smart_router.process(query, request.system_prompt)
return {
"content": response.content,
"tools_used": [tc.name for tc in response.tool_calls],
"finished": response.finished
}
@app.post("/v1/tools/execute")
async def execute_tool(request: ToolCallRequest):
"""Execute a single tool."""
call = ToolCall(
id="direct_call",
name=request.name,
arguments=request.arguments
)
result = await executor.execute(call)
return {
"success": result.success,
"result": result.result,
"error": result.error,
"execution_time_ms": result.execution_time_ms
}
@app.post("/v1/tools/select")
async def select_tools(request: SelectToolsRequest):
"""Select appropriate tools for a query."""
result = await tool_selector.select_tools(
request.query,
request.max_tools
)
return {
"selected_tools": result.selected_tools,
"confidence": result.confidence,
"reasoning": result.reasoning
}
@app.get("/v1/tools")
async def list_tools():
"""List available tools."""
return {
"tools": [
{
"name": tool.name,
"description": tool.description,
"parameters": [
{
"name": p.name,
"type": p.param_type.value,
"description": p.description,
"required": p.required
}
for p in tool.parameters
]
}
for tool in registry.list_tools()
]
}
@app.get("/v1/tools/{tool_name}")
async def get_tool(tool_name: str):
"""Get tool details."""
tool = registry.get(tool_name)
if not tool:
raise HTTPException(status_code=404, detail="Tool not found")
return {
"name": tool.name,
"description": tool.description,
"parameters": [
{
"name": p.name,
"type": p.param_type.value,
"description": p.description,
"required": p.required,
"enum_values": p.enum_values
}
for p in tool.parameters
],
"timeout_seconds": tool.timeout_seconds,
"requires_confirmation": tool.requires_confirmation
}
@app.get("/health")
async def health():
return {"status": "healthy"}References
- OpenAI Function Calling: https://platform.openai.com/docs/guides/function-calling
- Anthropic Tool Use: https://docs.anthropic.com/en/docs/tool-use
- LangChain Tools: https://python.langchain.com/docs/modules/tools/
- ReAct Pattern: https://arxiv.org/abs/2210.03629
Conclusion
Tool use is what transforms LLMs from impressive text generators into practical agents that can take action in the world. Start with clear tool definitions—good descriptions and parameter schemas help the model understand when and how to use each tool. Implement robust execution with proper validation, timeouts, and error handling. Use retry logic for transient failures but fail fast on validation errors. Multi-round tool use enables complex workflows where the model can chain multiple tools together, using results from one tool to inform the next. Smart tool selection reduces latency and improves accuracy by only presenting relevant tools for each query. For production systems, log all tool calls for debugging and auditing, implement rate limiting to prevent runaway tool use, and consider requiring confirmation for destructive operations. The key insight is that tool use quality depends as much on your tool definitions and execution infrastructure as on the LLM itself—invest in both.