03_TOOL_SYSTEM

Shared from "Study" on Inkdown

Tool System - Comprehensive Deep Dive

Overview

The Tool system is one of the most powerful features of the OpenAI Agents SDK. Tools are functions or capabilities that agents can call to perform actions beyond just generating text. Think of tools as "skills" or "abilities" that you give to your agents - they can search the web, read files, execute code, call APIs, interact with databases, and much more.

Core Concepts

What is a Tool?

A tool is a callable function that an agent can invoke. When an agent needs to perform an action (like searching the web or reading a file), it can call a tool instead of trying to do everything through text generation.

Why Tools Matter:

Action Capability - Agents can actually do things, not just talk about them
Accuracy - Tools can provide precise, factual information
Integration - Connect agents to external systems (databases, APIs, etc.)
Efficiency - Delegating to specialized tools is often faster/better

Python

from agents import function_tool

@function_tool
def calculate_sum(a: int, b: int) -> int:
    """Calculate the sum of two numbers."""
    return a + b

agent = Agent(
    name="calculator",
    instructions="Help with math calculations",
    tools=[calculate_sum],
)

Python

from typing import Optional, List
from datetime import datetime
from pydantic import BaseModel

@function_tool
def search_database(
    query: str,
    limit: int = 10,
    filters: Optional[List[str]] = None,
    date: Optional[datetime] = None,
) -> str:
    """Search the database with optional filters."""
    # Implementation
    return f"Results for {query}"

Python

from agents import function_tool, RunContextWrapper

@function_tool
def get_user_data(context: RunContextWrapper[MyContext], user_id: str) -> str:
    """Get user data from the context."""
    # Access context
    current_user = context.context.user_id
    return f"Data for user {user_id}"

Python

from agents import function_tool, ToolContext

@function_tool
def detailed_tool(context: ToolContext, param: str) -> str:
    """Tool with detailed context."""
    print(f"Tool name: {context.tool_name}")
    print(f"Tool call ID: {context.tool_call_id}")
    print(f"Tool arguments: {context.tool_arguments}")
    print(f"Tool call: {context.tool_call}")
    return f"Processed {param}"

Python

import asyncio

@function_tool
async def async_operation(query: str) -> str:
    """Perform an async operation."""
    await asyncio.sleep(1)  # Simulate async work
    return f"Async result for {query}"

Python

from agents import function_tool, tool_namespace

@tool_namespace("math")
@function_tool
def add(a: int, b: int) -> int:
    """Add two numbers."""
    return a + b

@tool_namespace("math")
@function_tool
def multiply(a: int, b: int) -> int:
    """Multiply two numbers."""
    return a * b

@tool_namespace("string")
@function_tool
def concatenate(a: str, b: str) -> str:
    """Concatenate two strings."""
    return a + b

Python

@function_tool
def admin_function(context: RunContextWrapper[MyContext]) -> str:
    """Admin-only function."""
    return "Admin operation"

# Set enablement condition
admin_function.is_enabled = lambda ctx, agent: ctx.context.user_role == "admin"

Python

@function_tool
def search_web(query: str) -> str:
    """
    Search the web for information about the given query.
    
    Returns a summary of the top 5 search results.
    Use this when you need current information or facts not in your training data.
    """
    # Implementation
    return "Search results"

Python

from agents import ToolErrorFunction

def custom_error_formatter(
    context: RunContextWrapper,
    error: Exception,
) -> str:
    """Custom error formatter."""
    return f"Custom error: {str(error)}. Please try again."

@function_tool(failure_error_function=custom_error_formatter)
def tool_with_custom_errors(param: str) -> str:
    """Tool with custom error handling."""
    raise ValueError("Something went wrong")

Python

from agents import ToolOutputText, ToolOutputImage, ToolOutputFileContent

@function_tool
def text_tool() -> ToolOutputText:
    """Return text output."""
    return ToolOutputText(text="Plain text result")

@function_tool
def image_tool() -> ToolOutputImage:
    """Return image output."""
    return ToolOutputImage(
        image_url="https://example.com/image.png",
        detail="high"
    )

@function_tool
def file_tool() -> ToolOutputFileContent:
    """Return file content."""
    return ToolOutputFileContent(
        file_data="base64encodedcontent",
        filename="result.txt"
    )

Python

from agents import Agent, FileSearchTool, WebSearchTool, CodeInterpreterTool

agent = Agent(
    name="researcher",
    instructions="Research topics using available tools",
    tools=[
        FileSearchTool(),
        WebSearchTool(),
        CodeInterpreterTool(),
    ],
)

Python

from agents import Agent, MCPServerStdio, MCPServerStdioParams

mcp_server = MCPServerStdio(
    params=MCPServerStdioParams(
        command="python",
        args=["-m", "my_mcp_server"],
    )
)

agent = Agent(
    name="mcp_agent",
    instructions="Use MCP tools",
    mcp_servers=[mcp_server],
)

Python

from agents import MCPConfig

mcp_config = MCPConfig(
    convert_schemas_to_strict=True,  # Convert to strict JSON schema
    failure_error_function=default_tool_error_function,  # Error handling
)

agent = Agent(
    name="mcp_agent",
    instructions="Use MCP tools",
    mcp_servers=[mcp_server],
    mcp_config=mcp_config,
)

Python

from agents import ToolFilter, ToolFilterCallable

def filter_tools(context: ToolFilterContext) -> ToolFilter:
    """Filter MCP tools."""
    return ToolFilterStatic(include=["tool1", "tool2"])

agent = Agent(
    name="mcp_agent",
    instructions="Use MCP tools",
    mcp_servers=[mcp_server],
    mcp_config=MCPConfig(
        tool_filter=filter_tools,
    ),
)

Python

from agents import Agent

specialist = Agent(
    name="specialist",
    instructions="Handle specialized tasks",
)

generalist = Agent(
    name="generalist",
    instructions="Delegate to specialists when needed",
    tools=[specialist.as_tool("specialist_tool", "Call the specialist")],
)

Python

specialist_tool = specialist.as_tool(
    tool_name="specialist",
    tool_description="Call the specialist agent",
    max_turns=5,  # Limit nested agent turns
    needs_approval=True,  # Require approval
    parameters=SpecialistInput,  # Structured input
)

Python

def on_agent_stream(event: AgentToolStreamEvent) -> None:
    """Handle streaming events from nested agent."""
    print(f"Event: {event.event}")

specialist_tool = specialist.as_tool(
    tool_name="specialist",
    tool_description="Call the specialist",
    on_stream=on_agent_stream,
)

Python

from agents import (
    LocalShellTool,
    ShellToolLocalEnvironment,
    ShellToolContainerNetworkPolicyDisabled,
)

tool = LocalShellTool(
    environment=ShellToolLocalEnvironment(),
    network_policy=ShellToolContainerNetworkPolicyDisabled(),
)

Python

from agents import Tool, ToolContext

class MyCustomTool(Tool):
    name = "my_custom_tool"
    description = "A custom tool implementation"
    
    async def on_invoke_tool(self, context: ToolContext, args: str) -> Any:
        """Implement the tool logic."""
        # Parse args
        # Execute logic
        return "Result"

agent = Agent(
    name="custom_agent",
    instructions="Use custom tools",
    tools=[MyCustomTool()],
)

Python

from agents import Tool, ToolContext
from pydantic import BaseModel

class MyToolArgs(BaseModel):
    param1: str
    param2: int

class MyCustomTool(Tool):
    name = "my_custom_tool"
    description = "A custom tool with schema"
    params_json_schema = MyToolArgs.model_json_schema()
    
    async def on_invoke_tool(self, context: ToolContext, args: str) -> Any:
        parsed = MyToolArgs.model_validate_json(args)
        # Use parsed.args
        return "Result"

Python

from agents import tool_input_guardrail, ToolInputGuardrailFunctionOutput

@tool_input_guardrail
def validate_search_args(
    context: RunContextWrapper,
    tool_name: str,
    args_json: str,
) -> ToolInputGuardrailFunctionOutput:
    """Validate search tool arguments."""
    args = json.loads(args_json)
    if len(args.get("query", "")) < 3:
        return ToolInputGuardrailFunctionOutput(
            output_info="Query too short",
            tripwire_triggered=True,
        )
    return ToolInputGuardrailFunctionOutput(
        output_info="Valid",
        tripwire_triggered=False,
    )

@function_tool(tool_input_guardrails=[validate_search_args])
def search(query: str) -> str:
    """Search with input validation."""
    return f"Results for {query}"

Python

from agents import tool_output_guardrail, ToolOutputGuardrailFunctionOutput

@tool_output_guardrail
def validate_output(
    context: RunContextWrapper,
    tool_name: str,
    output: str,
) -> ToolOutputGuardrailFunctionOutput:
    """Validate tool output."""
    if "sensitive" in output.lower():
        return ToolOutputGuardrailFunctionOutput(
            output_info="Sensitive content detected",
            tripwire_triggered=True,
        )
    return ToolOutputGuardrailFunctionOutput(
        output_info="Valid",
        tripwire_triggered=False,
    )

@function_tool(tool_output_guardrails=[validate_output])
def process_data(data: str) -> str:
    """Process data with output validation."""
    return f"Processed {data}"

Python

# 1. Model requests tool call
tool_call = ResponseFunctionToolCall(
    id="call_123",
    name="my_tool",
    arguments='{"param": "value"}',
)

# 2. Find the tool
tool = find_tool(tool_call.name)

# 3. Check if enabled
if not is_tool_enabled(tool, context):
    skip_tool(tool_call)

# 4. Check if approval needed
if tool.needs_approval:
    approval = await request_approval(tool_call)
    if not approval:
        record_rejection(tool_call)
        return

# 5. Run input guardrails
guardrail_result = await run_tool_input_guardrails(tool, tool_call.arguments)
if guardrail_result.tripwire_triggered:
    handle_guardrail_tripwire(guardrail_result)
    return

# 6. Execute the tool
try:
    result = await tool.execute(tool_call.arguments, context)
except Exception as e:
    result = handle_tool_error(e, tool)

# 7. Run output guardrails
guardrail_result = await run_tool_output_guardrails(tool, result)
if guardrail_result.tripwire_triggered:
    handle_guardrail_tripwire(guardrail_result)
    result = guardrail_result.output_info

# 8. Record the result
record_tool_output(tool_call, result)

# 9. Return to model
return result

Python

from agents import Agent, StopAtTools

agent = Agent(
    name="direct_agent",
    instructions="Use tools directly",
    tool_use_behavior="stop_on_first_tool",  # First tool result is final
)

agent = Agent(
    name="controlled_agent",
    instructions="Controlled tool use",
    tool_use_behavior=StopAtTools(
        stop_at_tool_names=["final_tool"]
    ),
)

Python

from agents import ToolsToFinalOutputResult, ToolsToFinalOutputFunction

async def custom_tool_handler(
    context: RunContextWrapper,
    tool_results: list[FunctionToolResult],
) -> ToolsToFinalOutputResult:
    """Custom tool result handling."""
    # Check if any tool produced an error
    if any("error" in r.output for r in tool_results):
        return ToolsToFinalOutputResult(
            is_final_output=True,
            final_output="An error occurred",
        )
    
    # Check if we got the answer
    if any("answer" in r.output for r in tool_results):
        return ToolsToFinalOutputResult(
            is_final_output=True,
            final_output="Got the answer",
        )
    
    # Continue with more turns
    return ToolsToFinalOutputResult(is_final_output=False)

agent = Agent(
    name="custom_agent",
    instructions="Custom tool handling",
    tool_use_behavior=custom_tool_handler,
)

Python

from agents import function_tool

@function_tool(timeout=30)  # 30 second timeout
def slow_operation(param: str) -> str:
    """Operation that might be slow."""
    time.sleep(60)  # This will timeout
    return "Done"

Python

# Good
@function_tool
def process_data(
    user_id: str,
    limit: int,
    filters: Optional[List[str]] = None,
) -> str:
    ...

# Avoid
@function_tool
def process_data(user_id, limit, filters=None):  # No type hints
    ...

Python

# Good
@function_tool
def calculate_discount(price: float, discount_percent: float) -> float:
    """
    Calculate the discounted price.
    
    Args:
        price: The original price.
        discount_percent: The discount percentage (0-100).
    
    Returns:
        The discounted price.
    
    Example:
        calculate_discount(100.0, 20) returns 80.0
    """
    return price * (1 - discount_percent / 100)

# Avoid
@function_tool
def calculate_discount(price: float, discount_percent: float) -> float:
    """Calculate discount."""  # Too vague
    ...

Python

# Good
@function_tool
def api_call(endpoint: str) -> str:
    """Call an API endpoint."""
    try:
        response = requests.get(endpoint)
        response.raise_for_status()
        return response.text
    except requests.RequestException as e:
        return f"API error: {str(e)}"

# Avoid
@function_tool
def api_call(endpoint: str) -> str:
    """Call an API endpoint."""
    return requests.get(endpoint).text  # No error handling

Python

# Good
@function_tool
def divide(a: float, b: float) -> float:
    """Divide two numbers."""
    if b == 0:
        raise ValueError("Cannot divide by zero")
    return a / b

# Avoid
@function_tool
def divide(a: float, b: float) -> float:
    """Divide two numbers."""
    return a / b  # Will crash on b=0

Python

@function_tool
def query_database(table: str, filters: dict) -> str:
    """Query a database table with filters."""
    # Connect to database
    # Execute query
    # Return results as string
    return json.dumps(results)

Python

@function_tool
def read_file(path: str) -> str:
    """Read a file's contents."""
    with open(path, 'r') as f:
        return f.read()

@function_tool
def write_file(path: str, content: str) -> str:
    """Write content to a file."""
    with open(path, 'w') as f:
        f.write(content)
    return "File written"

Python

@function_tool
def process_csv(file_path: str, operation: str) -> str:
    """Process a CSV file."""
    df = pd.read_csv(file_path)
    
    if operation == "summarize":
        return df.describe().to_string()
    elif operation == "filter":
        return df.head().to_string()
    else:
        return "Unknown operation"

03_TOOL_SYSTEM

Tool System - Comprehensive Deep Dive

Overview

Core Concepts

What is a Tool?

03_TOOL_SYSTEM

Tool System - Comprehensive Deep Dive

Overview

Core Concepts

What is a Tool?

Tool Types

Function Tools

Basic Function Tools

Function Tool Parameters

Context-Aware Function Tools

Tool Context

Async Function Tools

Tool Namespaces

Dynamic Tool Enablement

Tool Descriptions

Tool Error Handling

Custom Error Formatting

Tool Output Types

Hosted Tools

OpenAI Hosted Tools

Image Generation Tool

MCP Tools

Model Context Protocol

MCP Configuration

MCP Tool Filtering

Agent as Tool

Exposing Agents as Tools

Agent Tool Configuration

Agent Tool Streaming

Shell Tools

Local Shell Tool

Shell Tool Configuration

Computer Tools

Computer Interaction

Custom Tools

Custom Tool Implementation

Custom Tool with Schema

Tool Guardrails

Tool Input Guardrails

Tool Output Guardrails

Tool Metadata

Tool Origin

Tool Identity

Tool Execution Flow

Complete Tool Execution Flow

Tool Use Behavior

Agent-Level Tool Use Behavior

Custom Tool Use Behavior

Tool Timeout

Setting Tool Timeouts

Tool Best Practices

1. Clear Names

2. Type Hints

3. Docstrings

4. Error Handling

5. Input Validation

Common Tool Patterns

1. Database Query Tool

2. API Integration Tool

3. File Operations Tool

4. Data Processing Tool

Summary