Essentials
Tools
Essentials
Tools
Understanding and creating tools in AgenticFleet
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
-
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")
-
Resource Management
async with tool: result = await tool.run()
-
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)