Tools are the mechanisms through which agents interact with external systems, APIs, databases, and services. In Azcore, tools extend agents' capabilities beyond language model reasoning, enabling them to take actions and retrieve information.
🔧 Tool Basics
What are Tools?
Tools in Azcore are functions that agents can invoke to perform specific actions. They are based on LangChain's BaseTool interface and can:
- Retrieve information (search, query databases)
- Perform calculations
- Interact with APIs
- Manipulate files
- Control external systems
Tool Structure
Every tool has:
- Name: Unique identifier
- Description: What the tool does (used by LLM to decide when to use it)
- Parameters: Input arguments with types
- Implementation: The actual function logic
🛠️ Creating Tools
Method 1: @tool Decorator (Recommended)
The simplest way to create tools using LangChain's @tool decorator:
from langchain_core.tools import tool
@tool
def search_web(query: str) -> str:
"""
Search the web for information.
Args:
query: The search query
Returns:
Search results as a string
"""
# Implementation
results = perform_web_search(query)
return f"Search results for '{query}': {results}"
@tool
def calculate(expression: str) -> float:
"""
Evaluate a mathematical expression.
Args:
expression: Mathematical expression to evaluate
Returns:
Result of the calculation
"""
try:
result = eval(expression)
return float(result)
except Exception as e:
return f"Error: {str(e)}"
@tool
def fetch_weather(location: str, units: str = "celsius") -> str:
"""
Get current weather for a location.
Args:
location: City name or coordinates
units: Temperature units (celsius or fahrenheit)
Returns:
Current weather information
"""
weather_data = get_weather_api(location, units)
return f"Weather in {location}: {weather_data['temp']}°{units[0].upper()}, {weather_data['condition']}"
Method 2: BaseTool Class
For more complex tools with state or configuration:
from langchain_core.tools import BaseTool
from typing import Optional, Type
from pydantic import BaseModel, Field
class DatabaseQueryInput(BaseModel):
"""Input schema for database query tool."""
query: str = Field(description="SQL query to execute")
limit: int = Field(default=100, description="Maximum number of results")
class DatabaseQueryTool(BaseTool):
"""Tool for querying a database."""
name: str = "database_query"
description: str = "Execute SQL queries against the database"
args_schema: Type[BaseModel] = DatabaseQueryInput
# Tool configuration
database_url: str = Field(description="Database connection URL")
def __init__(self, database_url: str):
super().__init__(database_url=database_url)
self.connection = create_db_connection(database_url)
def _run(self, query: str, limit: int = 100) -> str:
"""Synchronous execution."""
try:
results = self.connection.execute(query, limit=limit)
return format_results(results)
except Exception as e:
return f"Query error: {str(e)}"
async def _arun(self, query: str, limit: int = 100) -> str:
"""Asynchronous execution."""
try:
results = await self.connection.execute_async(query, limit=limit)
return format_results(results)
except Exception as e:
return f"Query error: {str(e)}"
# Create instance
db_tool = DatabaseQueryTool(database_url="postgresql://localhost/mydb")
Method 3: StructuredTool
For dynamic tool creation:
from langchain_core.tools import StructuredTool
def process_data(data: str, operation: str) -> str:
"""Process data with specified operation."""
if operation == "uppercase":
return data.upper()
elif operation == "lowercase":
return data.lower()
elif operation == "reverse":
return data[::-1]
return data
# Create structured tool
data_processor = StructuredTool.from_function(
func=process_data,
name="data_processor",
description="Process text data with various operations (uppercase, lowercase, reverse)"
)
🎯 Tool Patterns
Information Retrieval Tools
@tool
def search_documentation(query: str) -> str:
"""
Search the documentation for information.
Args:
query: Search query
Returns:
Relevant documentation sections
"""
docs = load_documentation()
results = docs.search(query)
return format_doc_results(results)
@tool
def get_user_info(user_id: str) -> str:
"""
Retrieve user information.
Args:
user_id: Unique user identifier
Returns:
User information as JSON string
"""
user = database.get_user(user_id)
return json.dumps({
"id": user.id,
"name": user.name,
"email": user.email,
"role": user.role
})
Action Tools
@tool
def send_email(to: str, subject: str, body: str) -> str:
"""
Send an email.
Args:
to: Recipient email address
subject: Email subject
body: Email body content
Returns:
Success or error message
"""
try:
email_service.send(
to=to,
subject=subject,
body=body
)
return f"Email sent successfully to {to}"
except Exception as e:
return f"Failed to send email: {str(e)}"
@tool
def create_ticket(title: str, description: str, priority: str) -> str:
"""
Create a support ticket.
Args:
title: Ticket title
description: Detailed description
priority: Priority level (low, medium, high)
Returns:
Ticket ID and confirmation
"""
ticket = ticket_system.create(
title=title,
description=description,
priority=priority
)
return f"Ticket created: {ticket.id}"
Computation Tools
@tool
def calculate_statistics(data: str) -> str:
"""
Calculate statistics for a dataset.
Args:
data: Comma-separated numbers
Returns:
Statistical summary
"""
numbers = [float(x) for x in data.split(",")]
mean = sum(numbers) / len(numbers)
sorted_nums = sorted(numbers)
median = sorted_nums[len(sorted_nums) // 2]
std_dev = (sum((x - mean) ** 2 for x in numbers) / len(numbers)) ** 0.5
return f"""Statistics:
Mean: {mean:.2f}
Median: {median:.2f}
Std Dev: {std_dev:.2f}
Min: {min(numbers):.2f}
Max: {max(numbers):.2f}
"""
@tool
def convert_units(value: float, from_unit: str, to_unit: str) -> str:
"""
Convert between units.
Args:
value: Value to convert
from_unit: Source unit
to_unit: Target unit
Returns:
Converted value
"""
conversion = get_conversion_factor(from_unit, to_unit)
result = value * conversion
return f"{value} {from_unit} = {result} {to_unit}"
🔗 Using Tools with Agents
Adding Tools to Agents
from azcore.agents.agent_factory import AgentFactory
from langchain_openai import ChatOpenAI
# Create tools
tools = [
search_web,
calculate,
fetch_weather,
send_email
]
# Create agent with tools
llm = ChatOpenAI(model="gpt-4")
factory = AgentFactory(default_llm=llm)
agent = factory.create_react_agent(
name="assistant",
tools=tools,
prompt="You are a helpful assistant with access to various tools."
)
Adding Tools to Teams
from azcore.agents.team_builder import TeamBuilder
# Create specialized team with domain tools
security_tools = [
monitor_cameras,
check_access_logs,
send_alert,
lock_door
]
security_team = (TeamBuilder("security_team")
.with_llm(llm)
.with_tools(security_tools)
.with_prompt("You are a security monitoring team.")
.build())
Dynamic Tool Management
# Add tools dynamically
agent.add_tool(new_tool)
# Remove tools
agent.remove_tool("tool_name")
# Check available tools
print(f"Agent has {len(agent.tools)} tools:")
for tool in agent.tools:
print(f" - {tool.name}: {tool.description}")
🎓 Advanced Tool Patterns
Tools with State
class SessionAwareTool(BaseTool):
"""Tool that maintains session state."""
name: str = "session_tool"
description: str = "Tool with session awareness"
def __init__(self):
super().__init__()
self.sessions = {}
def _run(self, session_id: str, action: str) -> str:
"""Execute with session state."""
if session_id not in self.sessions:
self.sessions[session_id] = {"count": 0, "history": []}
session = self.sessions[session_id]
session["count"] += 1
session["history"].append(action)
return f"Session {session_id}: Action {action} (#{session['count']})"
Async Tools
@tool
async def async_api_call(endpoint: str) -> str:
"""
Make an asynchronous API call.
Args:
endpoint: API endpoint
Returns:
API response
"""
async with aiohttp.ClientSession() as session:
async with session.get(endpoint) as response:
data = await response.json()
return json.dumps(data)
Tools with Validation
@tool
def validated_tool(email: str, amount: float) -> str:
"""
Tool with input validation.
Args:
email: Email address
amount: Amount in dollars
Returns:
Result or error message
"""
# Validate email
import re
email_pattern = r'^[\w\.-]+@[\w\.-]+\.\w+$'
if not re.match(email_pattern, email):
return f"Error: Invalid email format: {email}"
# Validate amount
if amount <= 0:
return f"Error: Amount must be positive: {amount}"
if amount > 10000:
return f"Error: Amount exceeds maximum: {amount}"
# Process
return f"Processed ${amount:.2f} for {email}"
Tools with Error Handling
@tool
def robust_tool(query: str) -> str:
"""
Tool with comprehensive error handling.
Args:
query: Search query
Returns:
Results or error message
"""
try:
# Validate input
if not query or len(query.strip()) == 0:
return "Error: Query cannot be empty"
# Process
results = external_api.search(query)
if not results:
return f"No results found for: {query}"
return format_results(results)
except ConnectionError:
return "Error: Unable to connect to search service"
except TimeoutError:
return "Error: Search request timed out"
except Exception as e:
return f"Error: Unexpected error occurred: {str(e)}"
🚀 Tool Composition
Tool Chains
@tool
def fetch_data(source: str) -> str:
"""Fetch data from source."""
return database.get(source)
@tool
def transform_data(data: str) -> str:
"""Transform data format."""
return json.loads(data)
@tool
def analyze_data(data: str) -> str:
"""Analyze transformed data."""
return perform_analysis(data)
# Agent uses tools in sequence
agent = factory.create_react_agent(
name="data_pipeline",
tools=[fetch_data, transform_data, analyze_data],
prompt="""Use tools in sequence:
1. Fetch data
2. Transform to required format
3. Analyze the results
"""
)
Conditional Tool Usage
@tool
def check_cache(key: str) -> str:
"""Check if data is in cache."""
if key in cache:
return f"Cache hit: {cache[key]}"
return "Cache miss"
@tool
def fetch_from_api(key: str) -> str:
"""Fetch data from API if not cached."""
data = api.get(key)
cache[key] = data
return data
# Agent checks cache first, then API
agent = factory.create_react_agent(
name="smart_fetcher",
tools=[check_cache, fetch_from_api],
prompt="""Check cache first. If cache miss, fetch from API."""
)
📊 Tool Performance
Tool Timing
import time
from functools import wraps
def timed_tool(func):
"""Decorator to measure tool execution time."""
@wraps(func)
def wrapper(*args, **kwargs):
start = time.time()
result = func(*args, **kwargs)
duration = time.time() - start
print(f"Tool {func.__name__} took {duration:.3f}s")
return result
return wrapper
@tool
@timed_tool
def slow_operation(data: str) -> str:
"""Operation that takes time."""
time.sleep(2)
return f"Processed: {data}"
Tool Caching
from functools import lru_cache
@tool
def expensive_calculation(input_data: str) -> str:
"""
Expensive calculation with caching.
Args:
input_data: Input data
Returns:
Calculation result
"""
@lru_cache(maxsize=128)
def _calculate(data: str) -> str:
# Expensive operation
time.sleep(1)
return f"Result for {data}"
return _calculate(input_data)
🎯 Best Practices
1. Clear Tool Descriptions
# ✅ GOOD: Detailed description
@tool
def search_products(
query: str,
category: str = "all",
max_results: int = 10
) -> str:
"""
Search for products in the catalog.
Use this tool when users ask about products, inventory, or pricing.
Args:
query: Search keywords (e.g., "laptop", "red shoes")
category: Product category filter (e.g., "electronics", "clothing")
max_results: Maximum number of results to return (1-50)
Returns:
JSON list of products with name, price, and availability
Examples:
- "laptop under $1000" with category="electronics"
- "red dress" with category="clothing"
"""
pass
# ❌ BAD: Vague description
@tool
def search(query: str) -> str:
"""Search for stuff."""
pass
2. Input Validation
# ✅ GOOD: Validate inputs
@tool
def process_order(order_id: str, action: str) -> str:
"""
Process an order.
Args:
order_id: Order ID (format: ORD-XXXXX)
action: Action to take (cancel, refund, ship)
Returns:
Action result
"""
# Validate order_id format
if not re.match(r'^ORD-\d{5}$', order_id):
return f"Error: Invalid order ID format: {order_id}"
# Validate action
valid_actions = ["cancel", "refund", "ship"]
if action not in valid_actions:
return f"Error: Invalid action. Must be one of: {valid_actions}"
# Process
return execute_order_action(order_id, action)
3. Error Messages
# ✅ GOOD: Helpful error messages
@tool
def fetch_report(report_type: str, date: str) -> str:
"""Fetch a report."""
try:
return get_report(report_type, date)
except ValueError as e:
return f"Error: Invalid date format '{date}'. Please use YYYY-MM-DD format."
except KeyError:
return f"Error: Report type '{report_type}' not found. Available types: sales, inventory, customers"
except Exception as e:
return f"Error: Unable to fetch report: {str(e)}"
4. Return Structured Data
# ✅ GOOD: Return structured, parseable data
@tool
def get_order_status(order_id: str) -> str:
"""Get order status."""
order = database.get_order(order_id)
return json.dumps({
"order_id": order.id,
"status": order.status,
"items": order.items,
"total": order.total,
"estimated_delivery": order.delivery_date
}, indent=2)
# ❌ BAD: Return unstructured text
@tool
def get_order(order_id: str) -> str:
"""Get order."""
order = database.get_order(order_id)
return f"Order {order.id} is {order.status}"
🚀 Complete Tool Example
from langchain_core.tools import tool
from typing import Optional
import json
import re
from datetime import datetime
@tool
def customer_support_tool(
action: str,
customer_id: str,
details: Optional[str] = None
) -> str:
"""
Comprehensive customer support tool.
Handles various customer support actions including account lookup,
order management, and ticket creation.
Args:
action: Action to perform (lookup, order_status, create_ticket, process_refund)
customer_id: Customer ID (format: CUST-XXXXX)
details: Additional details required for some actions (JSON string)
Returns:
JSON string with action results
Examples:
- action="lookup", customer_id="CUST-12345"
- action="order_status", customer_id="CUST-12345", details='{"order_id": "ORD-67890"}'
- action="create_ticket", customer_id="CUST-12345", details='{"issue": "Login problem"}'
"""
# Validate customer_id
if not re.match(r'^CUST-\d{5}$', customer_id):
return json.dumps({
"error": f"Invalid customer ID format: {customer_id}",
"expected_format": "CUST-XXXXX"
})
# Parse details if provided
parsed_details = {}
if details:
try:
parsed_details = json.loads(details)
except json.JSONDecodeError:
return json.dumps({"error": "Invalid JSON in details field"})
try:
# Route to appropriate action
if action == "lookup":
customer = database.get_customer(customer_id)
return json.dumps({
"customer_id": customer.id,
"name": customer.name,
"email": customer.email,
"status": customer.status,
"since": customer.created_at
}, indent=2)
elif action == "order_status":
order_id = parsed_details.get("order_id")
if not order_id:
return json.dumps({"error": "order_id required in details"})
order = database.get_order(order_id)
return json.dumps({
"order_id": order.id,
"status": order.status,
"items": order.items,
"total": order.total,
"tracking": order.tracking_number
}, indent=2)
elif action == "create_ticket":
issue = parsed_details.get("issue")
if not issue:
return json.dumps({"error": "issue description required in details"})
ticket = ticket_system.create(
customer_id=customer_id,
issue=issue,
timestamp=datetime.now()
)
return json.dumps({
"ticket_id": ticket.id,
"status": "created",
"priority": ticket.priority
}, indent=2)
elif action == "process_refund":
order_id = parsed_details.get("order_id")
amount = parsed_details.get("amount")
if not order_id or not amount:
return json.dumps({
"error": "order_id and amount required in details"
})
refund = payment_system.process_refund(
customer_id=customer_id,
order_id=order_id,
amount=amount
)
return json.dumps({
"refund_id": refund.id,
"amount": amount,
"status": "processed"
}, indent=2)
else:
return json.dumps({
"error": f"Unknown action: {action}",
"valid_actions": [
"lookup",
"order_status",
"create_ticket",
"process_refund"
]
})
except Exception as e:
return json.dumps({
"error": f"Tool execution failed: {str(e)}",
"action": action,
"customer_id": customer_id
})
🎓 Summary
Azcore tools provide:
- Simple Creation: Use
@tooldecorator for quick tool creation - Flexible Patterns: Support various tool types and use cases
- Integration: Seamless integration with agents and teams
- Error Handling: Robust error handling and validation
- Async Support: Both synchronous and asynchronous execution
- Composability: Tools can be chained and combined
Tools are the primary way agents interact with external systems, making them essential for building practical, production-ready AI applications.