Add custom functionality to agents with tools in Python
Tools enable agents to perform actions beyond text generation, such as retrieving data, calling APIs, or executing code. The Agent Framework provides a flexible tool system with automatic schema generation and safety controls.
The @tool decorator converts any Python function into a tool:
from agent_framework import toolfrom typing import Annotatedfrom pydantic import Field@tool(approval_mode="always_require")def get_weather( location: Annotated[str, Field(description="The city name")]) -> str: """Get the current weather for a location.""" # Implementation return f"The weather in {location} is sunny and 72°F."
Production Safety: Always use approval_mode="always_require" for tools in production environments. Only use "never_require" for samples and testing.
import aiohttp@tool(approval_mode="always_require")async def fetch_data(url: Annotated[str, Field(description="URL to fetch")]) -> str: """Fetch data from a URL.""" async with aiohttp.ClientSession() as session: async with session.get(url) as response: return await response.text()
get_time_schema = { "type": "object", "properties": { "timezone": { "type": "string", "description": "The timezone (e.g., UTC, America/New_York)", "default": "UTC" } }}@tool( name="get_current_time", description="Get the current time in a timezone.", schema=get_time_schema, approval_mode="always_require")def get_current_time(timezone: str = "UTC") -> str: """Get the current time.""" from datetime import datetime from zoneinfo import ZoneInfo return f"The current time in {timezone} is {datetime.now(ZoneInfo(timezone)).isoformat()}"
# Always require approval (recommended for production)@tool(approval_mode="always_require")def delete_file(path: str) -> str: """Delete a file.""" os.remove(path) return f"Deleted {path}"# Never require approval (use only for safe operations in testing)@tool(approval_mode="never_require")def get_time() -> str: """Get the current time.""" return datetime.now().isoformat()# Require approval for specific conditions@tool(approval_mode="conditional")def transfer_money(amount: float, to_account: str) -> str: """Transfer money between accounts.""" # Approval logic determined by middleware return f"Transferred ${amount} to {to_account}"
from agent_framework import tool# Specialist agentresearcher = client.as_agent( name="Researcher", instructions="You are a research specialist.")@tool(approval_mode="always_require")async def research_topic(topic: str) -> str: """Research a topic in depth.""" result = await researcher.run(f"Research: {topic}") return result.text# Main agent that uses the researchercoordinator = client.as_agent( name="Coordinator", instructions="Coordinate tasks and delegate to specialists.", tools=[research_topic])response = await coordinator.run( "Research quantum computing and summarize it")
from agent_framework.exceptions import ToolException@tool( approval_mode="always_require", max_exceptions=3 # Retry up to 3 times on error)def flaky_api_call(endpoint: str) -> str: """Call an external API that might fail.""" try: response = requests.get(endpoint) response.raise_for_status() return response.text except requests.HTTPError as e: # Framework will retry automatically raise ToolException(f"API call failed: {e}")
Declare tools without providing implementations (for external execution):
from agent_framework import FunctionTool# Declare tool schema onlyweather_tool = FunctionTool( name="get_weather", description="Get weather for a location", parameters={ "type": "object", "properties": { "location": {"type": "string", "description": "City name"} }, "required": ["location"] })agent = client.as_agent( name="Agent", instructions="Use the weather tool.", tools=[weather_tool])# Tool calls are included in response but not executedresponse = await agent.run("What's the weather in NYC?")for call in response.tool_calls: print(f"Tool: {call.name}, Args: {call.arguments}") # Execute externally and feed results back
The function name and docstring are sent to the model to help it understand when to use the tool. Be clear and specific:
@tool(approval_mode="always_require")def search_knowledge_base( query: Annotated[str, Field(description="Search query")]) -> str: """Search the company knowledge base for relevant information. Use this tool when the user asks questions about company policies, procedures, or internal documentation. """ return search_results
Provide detailed parameter descriptions
Use Pydantic Field to add descriptions that help the model understand parameter expectations:
location: Annotated[ str, Field(description="City name (e.g., 'San Francisco', 'London')")]
Return structured, parseable data
Return data in a format that’s easy for the agent to parse and use:
def tool( func: Callable | None = None, *, name: str | None = None, description: str | None = None, schema: type[BaseModel] | dict | None = None, approval_mode: Literal["always_require", "never_require", "conditional"] = "always_require", max_invocations: int | None = None, max_exceptions: int | None = None) -> Callable: """Convert a function into a tool. Args: func: The function to convert (auto-provided when used as decorator) name: Override tool name (defaults to function name) description: Override tool description (defaults to docstring) schema: Explicit parameter schema (Pydantic model or JSON schema dict) approval_mode: When to require approval before execution max_invocations: Maximum number of times this tool can be called per run max_exceptions: Number of exceptions before giving up Returns: A FunctionTool that can be passed to an agent """
import aiohttpfrom datetime import datetime@tool(approval_mode="always_require")async def search_web(query: str) -> str: """Search the web for information.""" # Web search implementation return search_results@tool(approval_mode="always_require")def get_current_time(timezone: str = "UTC") -> str: """Get the current time in a timezone.""" from zoneinfo import ZoneInfo return datetime.now(ZoneInfo(timezone)).isoformat()@tool(approval_mode="always_require")async def save_note(content: str) -> str: """Save a note to storage.""" await db.notes.insert({"content": content, "timestamp": datetime.now()}) return "Note saved"agent = client.as_agent( name="Assistant", instructions="You are a helpful assistant with web search, time, and note-taking capabilities.", tools=[search_web, get_current_time, save_note])response = await agent.run( "Search for Python news and save a summary")