Guides Python tool development for Atomic Agents: BaseTool templates, Pydantic schemas, config, error handling, and agent integration.
From atomic-agentsnpx claudepluginhub brainblend-ai/atomic-agents --plugin atomic-agentsThis skill uses the workspace's default tool permissions.
examples/api-tool.pySearches, retrieves, and installs Agent Skills from prompts.chat registry using MCP tools like search_skills and get_skill. Activates for finding skills, browsing catalogs, or extending Claude.
Searches prompts.chat for AI prompt templates by keyword or category, retrieves by ID with variable handling, and improves prompts via AI. Use for discovering or enhancing prompts.
Compares coding agents like Claude Code and Aider on custom YAML-defined codebase tasks using git worktrees, measuring pass rate, cost, time, and consistency.
Tools extend agent capabilities by providing access to external services, APIs, databases, and computations. They follow a consistent pattern with input/output schemas and error handling.
┌─────────────────────────────────────┐
│ BaseTool │
├─────────────────────────────────────┤
│ input_schema: BaseIOSchema │
│ output_schema: BaseIOSchema │
│ config: BaseToolConfig │
├─────────────────────────────────────┤
│ run(params) -> Output | Error │
└─────────────────────────────────────┘
from atomic_agents.lib.base.base_tool import BaseTool, BaseToolConfig
from atomic_agents.lib.base.base_io_schema import BaseIOSchema
from pydantic import Field
from typing import Optional
import os
# ============================================================
# Schemas
# ============================================================
class MyToolInputSchema(BaseIOSchema):
"""Input for the tool."""
query: str = Field(..., description="The query to process")
class MyToolOutputSchema(BaseIOSchema):
"""Successful output."""
result: str = Field(..., description="The result")
class MyToolErrorSchema(BaseIOSchema):
"""Error output."""
error: str = Field(..., description="Error message")
code: Optional[str] = Field(default=None, description="Error code")
# ============================================================
# Configuration
# ============================================================
class MyToolConfig(BaseToolConfig):
"""Tool configuration."""
api_key: str = Field(
default_factory=lambda: os.getenv("MY_API_KEY", ""),
description="API key"
)
timeout: int = Field(default=30, description="Timeout in seconds")
# ============================================================
# Tool
# ============================================================
class MyTool(BaseTool):
"""Tool description."""
input_schema = MyToolInputSchema
output_schema = MyToolOutputSchema
def __init__(self, config: MyToolConfig = None):
super().__init__(config or MyToolConfig())
self.config: MyToolConfig = self.config
def run(self, params: MyToolInputSchema) -> MyToolOutputSchema | MyToolErrorSchema:
try:
# Tool logic here
result = f"Processed: {params.query}"
return MyToolOutputSchema(result=result)
except Exception as e:
return MyToolErrorSchema(error=str(e), code="ERROR")
# Convenience instance
tool = MyTool()
class APIToolConfig(BaseToolConfig):
"""Configuration with environment variables."""
api_key: str = Field(
default_factory=lambda: os.getenv("SERVICE_API_KEY", ""),
description="API key for the service"
)
base_url: str = Field(
default="https://api.service.com/v1",
description="Base URL for API"
)
timeout: int = Field(
default=30,
ge=1,
le=300,
description="Request timeout in seconds"
)
max_retries: int = Field(
default=3,
ge=0,
le=10,
description="Maximum retry attempts"
)
Always return error schemas instead of raising exceptions:
def run(self, params: InputSchema) -> OutputSchema | ErrorSchema:
# Validate configuration
if not self.config.api_key:
return ErrorSchema(
error="API key not configured",
code="CONFIG_ERROR"
)
try:
# Make external call
response = requests.get(
f"{self.config.base_url}/endpoint",
params={"q": params.query},
headers={"Authorization": f"Bearer {self.config.api_key}"},
timeout=self.config.timeout
)
response.raise_for_status()
data = response.json()
return OutputSchema(result=data["result"])
except requests.Timeout:
return ErrorSchema(error="Request timed out", code="TIMEOUT")
except requests.HTTPError as e:
return ErrorSchema(error=f"HTTP error: {e}", code="HTTP_ERROR")
except Exception as e:
return ErrorSchema(error=str(e), code="UNKNOWN_ERROR")
from my_tools import search_tool, SearchInputSchema
# Call tool directly
result = search_tool.run(SearchInputSchema(query="atomic agents"))
from typing import Union
from atomic_agents.agents.base_agent import AtomicAgent, AgentConfig
# Define tool selection schema
class ToolCallSchema(BaseIOSchema):
tool_name: Literal["search", "calculate", "none"] = Field(
..., description="Which tool to use"
)
tool_input: Union[SearchInput, CalculateInput, None] = Field(
..., description="Input for the selected tool"
)
# Agent decides which tool to use
agent = AtomicAgent[UserQuerySchema, ToolCallSchema](config=config)
# Orchestration loop
user_input = UserQuerySchema(query="What is 2+2?")
tool_decision = agent.run(user_input)
if tool_decision.tool_name == "calculate":
result = calculator_tool.run(tool_decision.tool_input)
elif tool_decision.tool_name == "search":
result = search_tool.run(tool_decision.tool_input)
Download tools from Atomic Forge:
atomic download calculator
atomic download searxng
atomic download youtube-transcript
Available tools:
See references/ for:
api-integration.md - Patterns for REST API toolsdatabase-tools.md - Database integration patternsSee examples/ for:
simple-tool.py - Minimal tool implementationapi-tool.py - External API integration