Tools API Reference

from aakit import tool, async_tool, ToolRegistry, Agent
import aiohttp
from typing import List, Dict, Optional
from datetime import datetime

# Create a tool registry
registry = ToolRegistry("web_tools")

@registry.register
@tool(
    description="Search the web for information",
    returns="List of search results",
    examples=[
        "web_search('python tutorials')",
        "web_search('machine learning', max_results=10)"
    ],
    tags=["web", "search"],
    cache=True
)
async def web_search(
    query: str,
    max_results: int = 5,
    language: str = "en"
) -> List[Dict[str, str]]:
    """
    Search the web and return relevant results.
    
    Args:
        query: The search query
        max_results: Maximum number of results to return
        language: Language code for results
        
    Returns:
        List of dictionaries containing title, url, and snippet
    """
    # Simulated search results
    results = []
    for i in range(max_results):
        results.append({
            "title": f"Result {i+1} for {query}",
            "url": f"https://example.com/{i+1}",
            "snippet": f"Content about {query}..."
        })
    return results

@registry.register
@async_tool(
    description="Extract content from a webpage",
    max_concurrency=3,
    timeout=30
)
async def extract_content(url: str) -> Dict[str, str]:
    """Extract main content from a webpage."""
    async with aiohttp.ClientSession() as session:
        async with session.get(url) as response:
            html = await response.text()
            # Content extraction logic here
            return {
                "title": "Page Title",
                "content": "Main content...",
                "published": datetime.now().isoformat()
            }

@registry.register
@tool(
    description="Summarize text content",
    parameters={
        "text": {
            "type": "string",
            "description": "Text to summarize",
            "maxLength": 10000
        },
        "max_length": {
            "type": "integer",
            "description": "Maximum summary length",
            "default": 200,
            "minimum": 50,
            "maximum": 500
        }
    }
)
def summarize(text: str, max_length: int = 200) -> str:
    """Create a concise summary of the provided text."""
    # Simple summarization (in practice, use NLP)
    words = text.split()[:max_length]
    return " ".join(words) + "..."

# Compose tools into a research pipeline
from aakit import compose_tools

research_pipeline = compose_tools(
    [web_search, extract_content, summarize],
    name="research_pipeline",
    description="Search, extract, and summarize web content",
    composition_type="sequential"
)

# Create an agent with the tools
agent = Agent(
    name="research_assistant",
    instruction="You help users research topics by searching and summarizing web content",
    model="gpt-4",
    tools=registry.get_all_tools() + [research_pipeline]
)

# Tool validation
from aakit import validate_tool

for tool_name, tool_func in registry.items():
    result = validate_tool(tool_func)
    if result.is_valid:
        print(f"✓ {tool_name} is valid")
    else:
        print(f"✗ {tool_name} has errors: {result.errors}")

# Export tools for MCP
mcp_catalog = registry.to_mcp_catalog()
print(f"Exported {len(mcp_catalog['tools'])} tools to MCP format")