Tools

from agnt5 import tool

@tool(auto_schema=True)
def search_web(query: str, max_results: int = 10) -> List[Dict[str, str]]:
    """Search the web for information.

    Args:
        query: The search query string
        max_results: Maximum number of results to return

    Returns:
        List of search results with title, url, and snippet
    """
    # Implementation
    return search_results

{
  "name": "search_web",
  "description": "Search the web for information.",
  "input_schema": {
    "type": "object",
    "properties": {
      "query": {"type": "string", "description": "The search query string"},
      "max_results": {"type": "integer", "default": 10}
    },
    "required": ["query"]
  }
}

from agnt5 import Agent, tool, LanguageModel

@tool(auto_schema=True)
def calculate_area(length: float, width: float) -> float:
    """Calculate the area of a rectangle.

    Args:
        length: Length in meters
        width: Width in meters

    Returns:
        Area in square meters
    """
    return length * width

lm = LanguageModel()
agent = Agent(
    name="math_assistant",
    model=lm,
    tools=[calculate_area],
    instructions="Help users with geometry calculations."
)

result = await agent.run("What's the area of a 5m by 3m room?")
# Agent automatically calls calculate_area(5.0, 3.0)

from agnt5 import worker
from agnt5.tools import HostedTool

# Define worker function
@worker.handler
def analyze_data(data: Dict) -> Dict:
    """Worker function for complex data analysis."""
    # Heavy computation here
    return analysis_results

# Create hosted tool
analysis_tool = HostedTool(
    name="analyze_data",
    description="Perform complex data analysis",
    endpoint="agnt5://data-analysis-service/analyze_data"
)

# Use with agent
agent = Agent(name="analyst", tools=[analysis_tool])

from agnt5.tools import MCPTool

# Connect to MCP server
filesystem_tool = MCPTool(
    name="filesystem",
    mcp_server_url="http://localhost:3000/mcp",
    capabilities=["read_file", "write_file", "list_directory"]
)

agent = Agent(name="file_assistant", tools=[filesystem_tool])

from agnt5.tools import OpenAPITool

# Create tools from OpenAPI spec
github_tools = OpenAPITool.from_spec(
    spec_url="https://api.github.com/openapi.json",
    operations=["get_repo", "list_issues", "create_issue"]
)

agent = Agent(name="github_bot", tools=github_tools)

from agnt5 import Tool

search_tool = Tool(
    name="search",
    description="Search for information",
    input_schema={
        "type": "object",
        "properties": {
            "query": {"type": "string", "minLength": 1},
            "filters": {"type": "object"}
        },
        "required": ["query"]
    },
    handler=search_function
)

@tool(auto_schema=True, confirmation=True)
def delete_database(database_name: str) -> Dict[str, str]:
    """Delete a database permanently.

    Args:
        database_name: Name of the database to delete

    Returns:
        Status of deletion operation

    Warning:
        This operation is irreversible and will delete all data.
    """
    # Requires human approval before execution
    pass

# Agent proposes deletion but waits for approval
agent = Agent(name="admin", tools=[delete_database])
result = await agent.run("Clean up the test database")
# User receives confirmation prompt before tool executes

@tool(auto_schema=True)
def search_papers(query: str, year_from: int = 2020) -> List[Dict]:
    """Search academic papers."""
    pass

@tool(auto_schema=True)
def download_pdf(url: str) -> bytes:
    """Download PDF document."""
    pass

@tool(auto_schema=True)
def extract_text(pdf_data: bytes) -> str:
    """Extract text from PDF."""
    pass

# Agent orchestrates multiple tools
research_agent = Agent(
    name="researcher",
    tools=[search_papers, download_pdf, extract_text],
    instructions="Search papers, download them, and extract key findings."
)

result = await research_agent.run("Survey recent work on transformers")
# Agent chains: search_papers → download_pdf → extract_text

@tool(auto_schema=True)
def fetch_stock_price(symbol: str) -> Dict[str, Any]:
    """Fetch current stock price.

    Args:
        symbol: Stock ticker symbol (e.g., 'AAPL', 'GOOGL')

    Returns:
        Stock price data

    Raises:
        ValueError: If symbol is invalid
        ConnectionError: If market data service is unavailable
    """
    try:
        price_data = market_api.get_price(symbol)
        return {
            "symbol": symbol,
            "price": price_data.current,
            "change": price_data.change
        }
    except InvalidSymbolError:
        raise ValueError(f"Invalid stock symbol: {symbol}")
    except MarketAPIError as e:
        raise ConnectionError(f"Market data unavailable: {e}")

# Agent handles tool errors gracefully
agent = Agent(name="stock_advisor", tools=[fetch_stock_price])

# Base toolset
base_tools = [search_tool, calculate_tool]

# Add specialized tools based on user role
if user.role == "admin":
    admin_tools = [delete_user_tool, modify_permissions_tool]
    all_tools = base_tools + admin_tools
else:
    all_tools = base_tools

agent = Agent(
    name="assistant",
    tools=all_tools,
    instructions=f"You are assisting a {user.role}."
)

from agnt5 import tool, Context

@tool(auto_schema=True)
async def store_memory(ctx: Context, key: str, value: str) -> Dict[str, str]:
    """Store information in long-term memory.

    Args:
        ctx: Execution context (automatically provided)
        key: Memory key
        value: Content to store

    Returns:
        Confirmation of storage
    """
    # Access context for durable storage
    await ctx.memory.set(key, value)

    return {
        "status": "stored",
        "key": key,
        "timestamp": ctx.now()
    }

# Context is automatically injected when tool is called
agent = Agent(name="memory_agent", tools=[store_memory])

from typing import List, Dict, Optional

@tool(auto_schema=True)
def analyze_sentiment(
    text: str,
    language: str = "en",
    return_scores: bool = False
) -> Dict[str, Any]:
    """Analyze sentiment of text.

    Args:
        text: Text to analyze (minimum 10 characters)
        language: ISO language code (en, es, fr, de)
        return_scores: Include detailed confidence scores

    Returns:
        Sentiment analysis with label (positive/negative/neutral)
        and optional confidence scores
    """
    # Type hints + docstring = complete schema
    pass

# Dangerous operations should require confirmation
@tool(auto_schema=True, confirmation=True)
def execute_code(code: str, language: str = "python") -> Dict[str, str]:
    """Execute arbitrary code in a sandboxed environment.

    Warning:
        Code execution can be dangerous. Requires explicit user approval.
    """
    pass

@tool(auto_schema=True, confirmation=True)
def send_email_blast(recipients: List[str], subject: str, body: str) -> Dict:
    """Send email to multiple recipients.

    Warning:
        Bulk email requires confirmation to prevent spam.
    """
    pass