Overview
Custom tools allow you to add your own functionality to the SDK without needing to create a full MCP server. They work seamlessly with all LLM providers (Anthropic, OpenAI, Gemini) and can be combined with existing MCP tools.
Basic Usage
from observee_agents import chat_with_tools_stream
import asyncio
# Define custom tool handler
async def custom_tool_handler(tool_name: str, tool_input: dict) -> str:
"""Handle custom tool executions"""
if tool_name == "add_numbers":
return str(tool_input.get("a", 0) + tool_input.get("b", 0))
elif tool_name == "get_time":
from datetime import datetime
return datetime.now().strftime("%I:%M %p")
else:
return f"Unknown tool: {tool_name}"
# Define custom tools in OpenAI format
custom_tools = [
{
"type": "function",
"function": {
"name": "add_numbers",
"description": "Add two numbers together",
"parameters": {
"type": "object",
"properties": {
"a": {"type": "number", "description": "First number"},
"b": {"type": "number", "description": "Second number"}
},
"required": ["a", "b"]
}
}
},
{
"type": "function",
"function": {
"name": "get_time",
"description": "Get the current time",
"parameters": {
"type": "object",
"properties": {}
}
}
}
]
# Use custom tools
async def example():
async for chunk in chat_with_tools_stream(
message="What's 5 + 3? Also, what time is it?",
provider="anthropic",
custom_tools=custom_tools,
custom_tool_handler=custom_tool_handler,
observee_api_key="obs_your_key_here"
):
if chunk["type"] == "content":
print(chunk["content"], end="", flush=True)
elif chunk["type"] == "tool_result":
print(f"\n🔧 [Tool: {chunk['tool_name']} = {chunk['result']}]")
asyncio.run(example())
The custom tool handler is an async function that receives:
tool_name: The name of the tool being calledtool_input: Dictionary of input parameters
It should return a string with the result.
async def custom_tool_handler(tool_name: str, tool_input: dict) -> str:
if tool_name == "calculate_tax":
income = tool_input.get("income", 0)
rate = tool_input.get("rate", 0.2)
return str(income * rate)
elif tool_name == "fetch_weather":
# Implement weather fetching logic
city = tool_input.get("city", "Unknown")
return f"Weather in {city}: Sunny, 72°F"
else:
return f"Unknown tool: {tool_name}"
Custom tools use the OpenAI function calling format:
{
"type": "function",
"function": {
"name": "tool_name",
"description": "What this tool does",
"parameters": {
"type": "object",
"properties": {
"param1": {
"type": "string",
"description": "Description of param1"
},
"param2": {
"type": "number",
"description": "Description of param2"
}
},
"required": ["param1"] # List of required parameters
}
}
}
Examples
Math Operations
# Math tool handler
async def math_handler(tool_name: str, tool_input: dict) -> str:
if tool_name == "add":
return str(tool_input.get("a", 0) + tool_input.get("b", 0))
elif tool_name == "multiply":
return str(tool_input.get("a", 0) * tool_input.get("b", 0))
elif tool_name == "power":
base = tool_input.get("base", 0)
exponent = tool_input.get("exponent", 1)
return str(base ** exponent)
return "Unknown operation"
# Math tools definition
math_tools = [
{
"type": "function",
"function": {
"name": "add",
"description": "Add two numbers",
"parameters": {
"type": "object",
"properties": {
"a": {"type": "number"},
"b": {"type": "number"}
},
"required": ["a", "b"]
}
}
},
{
"type": "function",
"function": {
"name": "power",
"description": "Raise a number to a power",
"parameters": {
"type": "object",
"properties": {
"base": {"type": "number"},
"exponent": {"type": "number"}
},
"required": ["base", "exponent"]
}
}
}
]
# Utility handler
async def utility_handler(tool_name: str, tool_input: dict) -> str:
if tool_name == "random_number":
import random
min_val = tool_input.get("min", 0)
max_val = tool_input.get("max", 100)
return str(random.randint(min_val, max_val))
elif tool_name == "format_date":
from datetime import datetime
date_str = tool_input.get("date", "")
input_format = tool_input.get("input_format", "%Y-%m-%d")
output_format = tool_input.get("output_format", "%B %d, %Y")
try:
date = datetime.strptime(date_str, input_format)
return date.strftime(output_format)
except:
return "Invalid date format"
elif tool_name == "word_count":
text = tool_input.get("text", "")
return str(len(text.split()))
return "Unknown utility"
# Utility tools
utility_tools = [
{
"type": "function",
"function": {
"name": "random_number",
"description": "Generate a random number",
"parameters": {
"type": "object",
"properties": {
"min": {"type": "integer", "description": "Minimum value"},
"max": {"type": "integer", "description": "Maximum value"}
}
}
}
},
{
"type": "function",
"function": {
"name": "word_count",
"description": "Count words in text",
"parameters": {
"type": "object",
"properties": {
"text": {"type": "string", "description": "Text to count words in"}
},
"required": ["text"]
}
}
}
]
# Custom tools work alongside MCP tools
async def combined_example():
async for chunk in chat_with_tools_stream(
message="Search for AI news and tell me what time it is",
provider="openai",
custom_tools=custom_tools, # Your custom tools
custom_tool_handler=custom_tool_handler,
enable_filtering=True, # MCP tools still available
observee_api_key="obs_your_key_here"
):
# Process response...
Non-Streaming Usage
Custom tools also work with the synchronous API:
from observee_agents import chat_with_tools
result = chat_with_tools(
message="Calculate 15 + 27",
provider="anthropic",
custom_tools=custom_tools,
custom_tool_handler=custom_tool_handler,
observee_api_key="obs_your_key_here"
)
print(result["content"])
Best Practices
- Error Handling: Always handle unknown tool names gracefully
- Type Validation: Validate input parameters in your handler
- Async Operations: Use async/await for I/O operations in handlers
- Clear Descriptions: Provide detailed tool descriptions for better LLM usage
- Return Strings: Always return string values from handlers
Common Use Cases
- Business Logic: Implement company-specific calculations or rules
- API Wrappers: Create simple wrappers for internal APIs
- Data Processing: Add custom data transformation tools
- Prototyping: Quickly test tool ideas before creating MCP servers
Limitations
- Custom tools are local to your application
- They don’t appear in
list_tools() results
- No built-in authentication or permissions
- Limited to the OpenAI function format or when filtering=True
Next Steps
