4.Add a Tool

from upsonic.tools import tool
from typing import Dict, Any

@tool
async def analyze_website_content(url: str, analysis_type: str = "general") -> Dict[str, Any]:
    """
    Analyzes the content of a website and provides insights based on the specified analysis type.
    
    Args:
        url: The website URL to analyze (must be a valid HTTP/HTTPS URL)
        analysis_type: Type of analysis to perform ('general', 'seo', 'accessibility', 'performance')
    
    Returns:
        Dictionary containing analysis results with keys: 'status', 'insights', 'recommendations'
    
    Raises:
        ValueError: If the URL is invalid or unreachable
        ConnectionError: If the website cannot be accessed
    """
    # Tool implementation here
    pass

from upsonic.tools import tool

@tool
def analyze(url, type):
    # Missing type hints, docstring, and error handling
    pass

from upsonic.tools import tool
from typing import Dict, Any

@tool
async def robust_tool(param: str) -> Dict[str, Any]:
    """A tool with comprehensive error handling."""
    try:
        # Tool logic here
        result = await perform_operation(param)
        return {"status": "success", "data": result}
    except ValueError as e:
        return {"status": "error", "error": f"Invalid parameter: {e}"}
    except Exception as e:
        return {"status": "error", "error": f"Unexpected error: {e}"}

from upsonic.tools import tool

@tool
async def validated_tool(url: str, timeout: int = 30) -> str:
    """A tool with input validation."""
    if not url.startswith(('http://', 'https://')):
        raise ValueError("URL must start with http:// or https://")
    
    if timeout <= 0 or timeout > 300:
        raise ValueError("Timeout must be between 1 and 300 seconds")
    
    # Tool logic here
    return "Validated and processed"

from upsonic.tools import tool
from typing import Dict, Any

@tool
async def well_documented_tool(param: str, option: bool = True) -> Dict[str, Any]:
    """
    Clear description of what the tool does.
    
    Args:
        param: Description of the parameter and its expected format
        option: Description of the optional parameter and its default behavior
        
    Returns:
        Description of the return value structure and expected keys
        
    Raises:
        ValueError: When param is invalid
        ConnectionError: When external service is unavailable
    """
    # Tool implementation
    pass

from upsonic.tools import tool
from upsonic import Agent, Task

@tool(requires_confirmation=True)
def delete_file(file_path: str) -> str:
    """Delete a file from the system."""
    return f"File '{file_path}' deleted successfully."

task = Task(
    description="Delete the ./old_data.txt file",
    tools=[delete_file]
)

agent = Agent(model="anthropic/claude-sonnet-4-5")
result = agent.print_do(task)
# User will be prompted: "Proceed? (y/n): "

from upsonic.tools import tool
from upsonic import Agent, Task

@tool(
    requires_user_input=True, 
    user_input_fields=["api_key", "secret_token"]
)
def get_balance(account_number: str, api_key: str = "", secret_token: str = "") -> str:
    """Get the current balance of the account."""
    return f"The current balance is 100000 {account_number}"

task = Task(
    description="Get the current balance of the account 1234567890",
    tools=[get_balance]
)

agent = Agent(model="anthropic/claude-sonnet-4-5")
result = agent.print_do(task)
# User will be prompted for each field:
# Enter value for 'api_key': 
# Enter value for 'secret_token':

from upsonic.tools import tool

@tool(
    requires_user_input=True,
    user_input_fields=["password"]
)
def reset_database(database_name: str, password: str = "") -> str:
    """Reset a database (requires confirmation and password)."""
    return f"Database '{database_name}' has been reset."

import subprocess
from upsonic.tools import tool
from upsonic import Agent, Task

@tool(external_execution=True)
def execute_shell_command(command: str) -> str:
    """Execute a shell command externally."""
    return subprocess.check_output(command, shell=True).decode("utf-8")

task = Task(
    description="List files in the current directory",
    tools=[execute_shell_command]
)

agent = Agent(model="anthropic/claude-sonnet-4-5")

# Initial execution - agent pauses when tool is needed
output = agent.print_do(task, return_output=True)

# Check if execution paused with external tool requirements
if output.active_requirements:
    # Execute each external tool requirement
    for requirement in output.active_requirements:
        if requirement.is_external_tool_execution:
            tool_exec = requirement.tool_execution
            print(f"Tool: {tool_exec.tool_name}")
            print(f"Args: {tool_exec.tool_args}")
            
            confirm = input(f"Execute the tool yourself?  {tool_exec.tool_args} (yes/no): ")
            if confirm == "yes":
                external_result = execute_shell_command(**tool_exec.tool_args)
                requirement.tool_execution.result = external_result
            else:
                requirement.tool_execution.result = "Operation cancelled by user"
    
    final_result = agent.continue_run(task=task, requirements=output.requirements)
    print(final_result)

uv pip install upsonic
# pip install upsonic

# Upsonic Docs: Create Custom Tools
# https://docs.upsonic.ai/guides/create_custom_tools

# Imports
from upsonic import Agent, Task
from upsonic.tools import tool, ToolHooks
from typing import Dict, Any
import requests
from bs4 import BeautifulSoup
import time

# Tool 1: Basic Website Content Fetcher
@tool
async def fetch_website_content(url: str) -> Dict[str, Any]:
    """
    Fetches and returns the HTML content of a website.
    
    Args:
        url: The website URL to fetch content from
        
    Returns:
        Dictionary containing 'status', 'content', 'headers', and 'response_time'
        
    Raises:
        requests.RequestException: If the request fails
        ValueError: If the URL is invalid
    """
    start_time = time.time()
    
    try:
        response = requests.get(url, timeout=30)
        response.raise_for_status()
        
        return {
            "status": "success",
            "content": response.text,
            "headers": dict(response.headers),
            "response_time": time.time() - start_time
        }
    except requests.RequestException as e:
        return {
            "status": "error",
            "error": str(e),
            "response_time": time.time() - start_time
        }

# Tool 2: SEO Analysis Tool with Human-in-the-Loop via Hooks
def before_analysis_hook(html_content: str):
    """Hook that runs before SEO analysis to get user confirmation."""
    print(f"🔍 About to start SEO analysis on content ({len(html_content)} characters)")
    
    # Ask user for confirmation
    user_input = input("Do you want to proceed with SEO analysis? (y/n): ").lower().strip()
    
    if user_input != 'y':
        raise Exception("Not allowed, rejected by user")
    
    print("📊 User approved. Starting SEO analysis...")

def after_analysis_hook(result):
    """Hook that runs after SEO analysis - currently not used."""
    pass

@tool(tool_hooks=ToolHooks(before=before_analysis_hook, after=after_analysis_hook))
async def analyze_seo_metrics(html_content: str) -> Dict[str, Any]:
    """
    Analyzes HTML content for SEO metrics with human-in-the-loop hooks.
    
    Args:
        html_content: The HTML content to analyze
        
    Returns:
        Dictionary containing enhanced SEO metrics, recommendations, and priority actions
    """
    soup = BeautifulSoup(html_content, 'html.parser')
    
    # Extract SEO elements
    title = soup.find('title')
    meta_description = soup.find('meta', attrs={'name': 'description'})
    h1_tags = soup.find_all('h1')
    
    seo_score = 0
    recommendations = []
    
    # Analyze title
    if title and title.text.strip():
        seo_score += 30
        if len(title.text) > 60:
            recommendations.append("Title is too long (should be under 60 characters)")
    else:
        recommendations.append("Missing title tag")
    
    # Analyze meta description
    if meta_description and meta_description.get('content'):
        seo_score += 30
    else:
        recommendations.append("Missing meta description")
    
    # Analyze H1 tags
    if h1_tags:
        seo_score += 40
        if len(h1_tags) > 1:
            recommendations.append("Multiple H1 tags found (should have only one)")
    else:
        recommendations.append("Missing H1 tag")
    
    return {
        "seo_score": seo_score,
        "title": title.text.strip() if title else None,
        "meta_description": meta_description.get('content') if meta_description else None,
        "h1_count": len(h1_tags),
        "recommendations": recommendations
    }

# Agent Creation with Custom Tools
website_analyzer_agent = Agent(
    model="anthropic/claude-sonnet-4-5",
    name="Website Analysis Expert",
    role="Website Performance and SEO Specialist",
    goal="Provide comprehensive website analysis with actionable insights for optimization"
)

# Example Task Using Custom Tools
analysis_task = Task(
    description="Analyze the website 'example.com' for SEO optimization",
    tools=[
        fetch_website_content,
        analyze_seo_metrics
    ]
)

# Execute the analysis
result = website_analyzer_agent.do(analysis_task)
print("Analysis Complete:", result)