Working with Tools

Built-in Tools

AgenticFleet provides several built-in tools:

from agentic_fleet.tools import (
    WebSearch,
    FileReader,
    CodeExecutor,
    APIClient,
    DatabaseTool
)

# Create tools
web_search = WebSearch()
file_reader = FileReader()
code_executor = CodeExecutor()

Using Tools

With Agents

from agentic_fleet import Agent

# Create agent with tools
agent = Agent(
    name="assistant",
    tools=[WebSearch(), FileReader()]
)

# Use tool through agent
result = await agent.use_tool(
    "web_search",
    query="AgenticFleet documentation"
)

Standalone Usage

# Create tool
web_search = WebSearch()

# Use tool directly
results = await web_search.run(
    query="AgenticFleet features"
)

Custom Tools

Basic Tool

from agentic_fleet.tools import BaseTool
from typing import Dict, Any

class CustomTool(BaseTool):
    name = "custom_tool"
    description = "A custom tool for specific tasks"
    
    async def _run(self, **kwargs) -> Dict[str, Any]:
        # Implement tool logic
        result = await self.process(**kwargs)
        return {"result": result}
    
    async def process(self, **kwargs):
        # Process inputs
        pass

Advanced Tool

from agentic_fleet.tools import AdvancedTool
from typing import Optional

class AnalysisTool(AdvancedTool):
    name = "analysis_tool"
    description = "Analyzes data using advanced algorithms"
    version = "1.0.0"
    
    def __init__(
        self,
        model: str = "default",
        threshold: float = 0.5
    ):
        self.model = model
        self.threshold = threshold
    
    async def setup(self):
        # Initialize resources
        await self.load_model()
    
    async def _run(
        self,
        data: Dict[str, Any],
        options: Optional[Dict] = None
    ) -> Dict[str, Any]:
        # Run analysis
        results = await self.analyze(data)
        return {"results": results}
    
    async def cleanup(self):
        # Clean up resources
        await self.unload_model()

Tool Configuration

from agentic_fleet.tools import ToolConfig

config = ToolConfig(
    name="web_search",
    api_key="your-api-key",
    rate_limit=100,
    timeout=30
)

tool = WebSearch(config)

Tool Chains

from agentic_fleet.tools import ToolChain

# Create tool chain
chain = ToolChain([
    WebSearch(),
    FileReader(),
    CodeExecutor()
])

# Execute chain
results = await chain.execute(
    initial_input="Research and implement"
)

Error Handling

from agentic_fleet.exceptions import ToolError

try:
    result = await tool.run(input_data="test")
except ToolError as e:
    print(f"Tool error: {e}")
    # Handle error

Tool Validation

from agentic_fleet.tools import validate_tool

# Validate tool
validation_result = validate_tool(
    tool,
    test_input={"query": "test"}
)

# Check validation
if validation_result.is_valid:
    print("Tool is valid")
else:
    print(f"Validation errors: {validation_result.errors}")

Tool Registry

from agentic_fleet.tools import ToolRegistry

# Create registry
registry = ToolRegistry()

# Register tool
registry.register(CustomTool())

# Get tool
tool = registry.get("custom_tool")

Tool Monitoring

from agentic_fleet.monitoring import ToolMonitor

# Create monitor
monitor = ToolMonitor()

# Track tool usage
with monitor.track(tool):
    result = await tool.run(input_data="test")

# Get metrics
metrics = monitor.get_metrics(tool.name)

Best Practices

  1. Tool Design

    class WellDesignedTool(BaseTool):
        # Clear name and description
        name = "well_designed_tool"
        description = "Detailed description of purpose"
        
        # Validate inputs
        async def validate(self, **kwargs):
            if "required_param" not in kwargs:
                raise ValueError("Missing required parameter")
    
  2. Resource Management

    async with tool:
        result = await tool.run()
    
  3. Error Handling

    try:
        result = await tool.run()
    except ToolError as e:
        logger.error(f"Tool error: {e}")
        # Implement fallback
    

Advanced Features

Tool Composition

from agentic_fleet.tools import compose_tools

# Compose tools
combined_tool = compose_tools(
    [tool1, tool2],
    name="combined_tool"
)

Tool Versioning

from agentic_fleet.tools import versioned_tool

@versioned_tool(version="1.0.0")
class VersionedTool(BaseTool):
    name = "versioned_tool"

Tool Caching

from agentic_fleet.tools import cached_tool

@cached_tool(ttl=3600)
class CachedTool(BaseTool):
    name = "cached_tool"

Deployment

from agentic_fleet.deployment import deploy_tool

# Deploy tool
deployment = await deploy_tool(
    tool,
    platform="kubernetes"
)

# Scale deployment
await deployment.scale(replicas=3)