Skip to main content

​
Overview

The Tools API defines the interface for implementing runtime tools in AXON. Every tool implements the BaseTool abstract class, whether it’s a stub for testing or a real backend with I/O. 
from axon.runtime.tools.base_tool import BaseTool, ToolResult
import httpx

class WebSearchBrave(BaseTool):
    TOOL_NAME = "WebSearch"
    IS_STUB = False
    DEFAULT_TIMEOUT = 10.0
    
    def validate_config(self) -> None:
        if "api_key" not in self.config:
            raise ValueError("Missing api_key in config")
    
    async def execute(self, query: str, **kwargs) -> ToolResult:
        max_results = kwargs.get("max_results", 5)
        
        async with httpx.AsyncClient() as client:
            response = await client.get(
                "https://api.brave.com/search",
                headers={"X-API-Key": self.config["api_key"]},
                params={"q": query, "count": max_results}
            )
            data = response.json()
        
        return ToolResult(
            success=True,
            data=data["results"],
            metadata={"provider": "brave", "query": query}
        )

​
Class: BaseTool

Abstract base class for all AXON runtime tools.

​
Class Variables

Subclasses must set these:
​
TOOL_NAME
str
required
Unique identifier matching the IRToolSpec.name
​
IS_STUB
bool
required
True for stub implementations, False for real backends
​
DEFAULT_TIMEOUT
float
default:"30.0"
Default timeout in seconds for tool execution

​
Constructor

BaseTool(config: dict[str, Any] | None = None)
​
config
dict
Configuration dictionary (API keys, endpoints, etc.)

​
Abstract Methods

Subclasses must implement:

​
validate_config() -> None

Validate that self.config has the required keys. 
def validate_config(self) -> None:
    required = ["api_key", "endpoint"]
    for key in required:
        if key not in self.config:
            raise ValueError(f"Missing required config key: {key}")
    
    if not self.config["api_key"]:
        raise ValueError("api_key cannot be empty")

​
async execute(query: str, **kwargs) -> ToolResult

Execute the tool with the given query.
​
query
str
required
The primary input (search query, code, file path, etc.)
​
**kwargs
Any
Tool-specific parameters (e.g., max_results, timeout)
Returns: ToolResult with success/data/error 
async def execute(self, query: str, **kwargs) -> ToolResult:
    try:
        result = await self._do_work(query, **kwargs)
        return ToolResult(success=True, data=result)
    except Exception as e:
        return ToolResult(
            success=False,
            data=None,
            error=str(e),
            metadata={"exception_type": type(e).__name__}
        )

​
Properties

​
get_tool_name: str

Tool name (forwards to TOOL_NAME). 
tool = WebSearchBrave(config={"api_key": "..."})
print(tool.get_tool_name)  # "WebSearch"

​
get_is_stub: bool

Whether this implementation is a stub. 
print(tool.get_is_stub)  # False

​
Class: ToolResult

Standardized result returned by every tool execution. 
class ToolResult:
    success: bool
    data: Any
    error: str | None
    metadata: dict[str, Any]
    
    def to_dict(self) -> dict[str, Any]:
        ...
​
success
bool
required
True if the tool completed without error
​
data
Any
required
The payload — its shape depends on the tool
​
error
str
Human-readable error message on failure
​
metadata
dict
Extra info (provider, timings, warnings, etc.)
Example: 
result = ToolResult(
    success=True,
    data={"results": [...]},
    metadata={
        "provider": "serper",
        "query_time_ms": 234,
        "result_count": 10
    }
)

print(result.to_dict())
# {
#   "success": true,
#   "data": {"results": [...]},
#   "metadata": {...}
# }

​
Built-in Tools

AXON includes stub implementations for common tools:

​
WebSearch

from axon.runtime.tools.stubs.web_search_stub import WebSearchStub

tool = WebSearchStub()
result = await tool.execute("AXON programming language")

print(result.data)
# [
#   {"title": "Result 1", "url": "...", "snippet": "..."},
#   {"title": "Result 2", "url": "...", "snippet": "..."}
# ]

​
Calculator

from axon.runtime.tools.stubs.calculator_tool import CalculatorTool

tool = CalculatorTool()
result = await tool.execute("2 + 2 * 3")

print(result.data)  # {"result": 8, "expression": "2 + 2 * 3"}

​
FileReader

from axon.runtime.tools.stubs.file_reader_stub import FileReaderStub

tool = FileReaderStub()
result = await tool.execute("/path/to/file.txt")

print(result.data)  # {"content": "...", "path": "...", "size": 1234}

​
DateTime

from axon.runtime.tools.stubs.datetime_tool import DateTimeTool

tool = DateTimeTool()
result = await tool.execute("current_time")

print(result.data)  # {"timestamp": "2024-01-15T10:30:00Z", ...}

​
Creating Custom Tools

​
Example: API Call Tool

from axon.runtime.tools.base_tool import BaseTool, ToolResult
import httpx
import json

class APICallTool(BaseTool):
    TOOL_NAME = "APICall"
    IS_STUB = False
    DEFAULT_TIMEOUT = 15.0
    
    def validate_config(self) -> None:
        # No required config - can be used with any API
        pass
    
    async def execute(self, query: str, **kwargs) -> ToolResult:
        """
        Execute an API call.
        
        Args:
            query: JSON string with {"url": "...", "method": "GET", "data": {...}}
            **kwargs: Additional options
        """
        try:
            request_data = json.loads(query)
            url = request_data["url"]
            method = request_data.get("method", "GET").upper()
            body = request_data.get("data")
            headers = request_data.get("headers", {})
            
            timeout = kwargs.get("timeout", self.DEFAULT_TIMEOUT)
            
            async with httpx.AsyncClient(timeout=timeout) as client:
                if method == "GET":
                    response = await client.get(url, headers=headers)
                elif method == "POST":
                    response = await client.post(url, json=body, headers=headers)
                elif method == "PUT":
                    response = await client.put(url, json=body, headers=headers)
                elif method == "DELETE":
                    response = await client.delete(url, headers=headers)
                else:
                    raise ValueError(f"Unsupported HTTP method: {method}")
                
                return ToolResult(
                    success=response.is_success,
                    data={
                        "status_code": response.status_code,
                        "body": response.json() if response.headers.get("content-type", "").startswith("application/json") else response.text,
                        "headers": dict(response.headers)
                    },
                    error=None if response.is_success else f"HTTP {response.status_code}",
                    metadata={
                        "url": url,
                        "method": method,
                        "duration_ms": response.elapsed.total_seconds() * 1000
                    }
                )
        
        except json.JSONDecodeError as e:
            return ToolResult(
                success=False,
                data=None,
                error=f"Invalid JSON query: {e}"
            )
        except Exception as e:
            return ToolResult(
                success=False,
                data=None,
                error=str(e),
                metadata={"exception_type": type(e).__name__}
            )

​
Example: Database Query Tool

from axon.runtime.tools.base_tool import BaseTool, ToolResult
import asyncpg

class PostgresQueryTool(BaseTool):
    TOOL_NAME = "PostgresQuery"
    IS_STUB = False
    DEFAULT_TIMEOUT = 30.0
    
    def validate_config(self) -> None:
        required = ["host", "database", "user", "password"]
        for key in required:
            if key not in self.config:
                raise ValueError(f"Missing required config: {key}")
    
    async def execute(self, query: str, **kwargs) -> ToolResult:
        """
        Execute a Postgres query.
        
        Args:
            query: SQL query string
            **kwargs: Query parameters
        """
        conn = None
        try:
            # Connect to database
            conn = await asyncpg.connect(
                host=self.config["host"],
                database=self.config["database"],
                user=self.config["user"],
                password=self.config["password"],
            )
            
            # Execute query
            if query.strip().upper().startswith("SELECT"):
                rows = await conn.fetch(query)
                results = [dict(row) for row in rows]
                
                return ToolResult(
                    success=True,
                    data={
                        "rows": results,
                        "count": len(results)
                    },
                    metadata={
                        "query_type": "SELECT",
                        "row_count": len(results)
                    }
                )
            else:
                # For INSERT/UPDATE/DELETE
                result = await conn.execute(query)
                
                return ToolResult(
                    success=True,
                    data={"result": result},
                    metadata={"query_type": "MUTATION"}
                )
        
        except Exception as e:
            return ToolResult(
                success=False,
                data=None,
                error=f"Database error: {e}",
                metadata={"exception_type": type(e).__name__}
            )
        
        finally:
            if conn:
                await conn.close()

​
Tool Registration

Register custom tools with the ToolDispatcher: 
from axon.runtime.tools.dispatcher import ToolDispatcher
from axon.runtime.tools.registry import ToolRegistry

# Create registry and dispatcher
registry = ToolRegistry()
dispatcher = ToolDispatcher(registry)

# Register custom tools
api_tool = APICallTool()
registry.register(api_tool)

db_tool = PostgresQueryTool(config={
    "host": "localhost",
    "database": "mydb",
    "user": "user",
    "password": "pass"
})
registry.register(db_tool)

# Use with executor
from axon.runtime import Executor

executor = Executor(
    client=my_client,
    tool_dispatcher=dispatcher
)

​
Testing Tools

​
Unit Testing

import pytest
from your_tools import WebSearchBrave

@pytest.mark.asyncio
async def test_web_search_success():
    tool = WebSearchBrave(config={"api_key": "test_key"})
    result = await tool.execute("AXON language")
    
    assert result.success
    assert isinstance(result.data, list)
    assert len(result.data) > 0
    assert "title" in result.data[0]

@pytest.mark.asyncio
async def test_web_search_error():
    tool = WebSearchBrave(config={"api_key": "invalid"})
    result = await tool.execute("test")
    
    assert not result.success
    assert result.error is not None

def test_config_validation():
    with pytest.raises(ValueError, match="Missing api_key"):
        tool = WebSearchBrave(config={})

​
Stub Testing

from axon.runtime.tools.base_tool import BaseTool, ToolResult

class MockWebSearch(BaseTool):
    TOOL_NAME = "WebSearch"
    IS_STUB = True
    
    def validate_config(self) -> None:
        pass
    
    async def execute(self, query: str, **kwargs) -> ToolResult:
        return ToolResult(
            success=True,
            data=[
                {"title": "Mock Result 1", "url": "https://example.com"},
                {"title": "Mock Result 2", "url": "https://example.org"},
            ],
            metadata={"is_stub": True}
        )

# Use mock in tests
tool = MockWebSearch()
result = await tool.execute("anything")
assert result.success
assert result.metadata["is_stub"] is True

​
Complete Example

import asyncio
from axon.runtime.tools.base_tool import BaseTool, ToolResult
from axon.runtime.tools.registry import ToolRegistry
from axon.runtime.tools.dispatcher import ToolDispatcher
import httpx

# 1. Define custom tool
class WeatherTool(BaseTool):
    TOOL_NAME = "Weather"
    IS_STUB = False
    
    def validate_config(self) -> None:
        if "api_key" not in self.config:
            raise ValueError("Missing api_key")
    
    async def execute(self, query: str, **kwargs) -> ToolResult:
        location = query
        
        async with httpx.AsyncClient() as client:
            response = await client.get(
                "https://api.weatherapi.com/v1/current.json",
                params={"key": self.config["api_key"], "q": location}
            )
            data = response.json()
        
        return ToolResult(
            success=True,
            data={
                "location": data["location"]["name"],
                "temperature_c": data["current"]["temp_c"],
                "condition": data["current"]["condition"]["text"]
            },
            metadata={"provider": "weatherapi.com"}
        )

# 2. Register and use
async def main():
    # Setup
    registry = ToolRegistry()
    weather = WeatherTool(config={"api_key": "your_key"})
    registry.register(weather)
    
    # Execute
    result = await weather.execute("San Francisco")
    
    if result.success:
        print(f"Location: {result.data['location']}")
        print(f"Temperature: {result.data['temperature_c']}°C")
        print(f"Condition: {result.data['condition']}")
    else:
        print(f"Error: {result.error}")

asyncio.run(main())

​
Next Steps

Executor API

Use tools in program execution

Memory API

Implement semantic memory storage

Build docs developers (and LLMs) love

Get started for freeTalk to us