Skip to main content

Tools System

Tools are executable capabilities that allow the agent to interact with the environment — reading files, running commands, searching the web, and more.

Overview

The tools system consists of three main components:

Tool Base Class

Abstract interface defining tool contract

Tool Registry

Dynamic registration and discovery

Tool Implementations

Concrete tool implementations

Tool Base Class

All tools inherit from the abstract Tool class in nanobot/agent/tools/base.py:
class Tool(ABC):
    """
    Abstract base class for agent tools.

    Tools are capabilities that the agent can use to interact with
    the environment, such as reading files, executing commands, etc.
    """

    @property
    @abstractmethod
    def name(self) -> str:
        """Tool name used in function calls."""
        pass

    @property
    @abstractmethod
    def description(self) -> str:
        """Description of what the tool does."""
        pass

    @property
    @abstractmethod
    def parameters(self) -> dict[str, Any]:
        """JSON Schema for tool parameters."""
        pass

    @abstractmethod
    async def execute(self, **kwargs: Any) -> str:
        """
        Execute the tool with given parameters.

        Returns:
            String result of the tool execution.
        """
        pass
See nanobot/agent/tools/base.py:7-53.

Built-in Tools

Nanobot ships with a comprehensive set of built-in tools:

Filesystem Tools

Implemented in nanobot/agent/tools/filesystem.py:
class ReadFileTool(Tool):
    """Tool to read file contents."""
    
    _MAX_CHARS = 128_000  # ~128 KB limit
    
    @property
    def name(self) -> str:
        return "read_file"
    
    @property
    def parameters(self) -> dict[str, Any]:
        return {
            "type": "object",
            "properties": {
                "path": {
                    "type": "string", 
                    "description": "The file path to read"
                }
            },
            "required": ["path"],
        }
    
    async def execute(self, path: str, **kwargs: Any) -> str:
        # Resolve path, check permissions
        file_path = _resolve_path(path, self._workspace, self._allowed_dir)
        # Read and return content
        content = file_path.read_text(encoding="utf-8")
        # Truncate if too large
        if len(content) > self._MAX_CHARS:
            return content[:self._MAX_CHARS] + "\n\n... (truncated)"
        return content
See nanobot/agent/tools/filesystem.py:26-73.
class WriteFileTool(Tool):
    """Tool to write content to a file."""
    
    @property
    def name(self) -> str:
        return "write_file"
    
    @property
    def parameters(self) -> dict[str, Any]:
        return {
            "type": "object",
            "properties": {
                "path": {
                    "type": "string", 
                    "description": "The file path to write to"
                },
                "content": {
                    "type": "string", 
                    "description": "The content to write"
                },
            },
            "required": ["path", "content"],
        }
    
    async def execute(self, path: str, content: str, **kwargs: Any) -> str:
        file_path = _resolve_path(path, self._workspace, self._allowed_dir)
        file_path.parent.mkdir(parents=True, exist_ok=True)
        file_path.write_text(content, encoding="utf-8")
        return f"Successfully wrote {len(content)} bytes to {file_path}"
See nanobot/agent/tools/filesystem.py:76-111.
class EditFileTool(Tool):
    """Tool to edit a file by replacing text."""
    
    @property
    def description(self) -> str:
        return "Edit a file by replacing old_text with new_text. The old_text must exist exactly in the file."
    
    async def execute(self, path: str, old_text: str, new_text: str, **kwargs: Any) -> str:
        file_path = _resolve_path(path, self._workspace, self._allowed_dir)
        content = file_path.read_text(encoding="utf-8")
        
        if old_text not in content:
            return self._not_found_message(old_text, content, path)
        
        count = content.count(old_text)
        if count > 1:
            return f"Warning: old_text appears {count} times. Please provide more context to make it unique."
        
        new_content = content.replace(old_text, new_text, 1)
        file_path.write_text(new_content, encoding="utf-8")
        return f"Successfully edited {file_path}"
Features intelligent error messages with diff output when text is not found.See nanobot/agent/tools/filesystem.py:114-192.
class ListDirTool(Tool):
    """Tool to list directory contents."""
    
    async def execute(self, path: str, **kwargs: Any) -> str:
        dir_path = _resolve_path(path, self._workspace, self._allowed_dir)
        items = []
        for item in sorted(dir_path.iterdir()):
            prefix = "📁 " if item.is_dir() else "📄 "
            items.append(f"{prefix}{item.name}")
        return "\n".join(items)
See nanobot/agent/tools/filesystem.py:195-238.

Shell Execution Tool

Implemented in nanobot/agent/tools/shell.py:
class ExecTool(Tool):
    """Tool to execute shell commands."""
    
    def __init__(
        self,
        timeout: int = 60,
        working_dir: str | None = None,
        deny_patterns: list[str] | None = None,
        restrict_to_workspace: bool = False,
        path_append: str = "",
    ):
        self.timeout = timeout
        self.deny_patterns = deny_patterns or [
            r"\brm\s+-[rf]{1,2}\b",     # rm -rf
            r"\bformat\b",               # format
            r"\bdd\s+if=",               # dd
            # ... more dangerous patterns
        ]
    
    async def execute(self, command: str, working_dir: str | None = None, **kwargs: Any) -> str:
        cwd = working_dir or self.working_dir or os.getcwd()
        
        # Safety checks
        guard_error = self._guard_command(command, cwd)
        if guard_error:
            return guard_error
        
        # Execute with timeout
        process = await asyncio.create_subprocess_shell(
            command,
            stdout=asyncio.subprocess.PIPE,
            stderr=asyncio.subprocess.PIPE,
            cwd=cwd,
            env=env,
        )
        
        try:
            stdout, stderr = await asyncio.wait_for(
                process.communicate(),
                timeout=self.timeout
            )
        except asyncio.TimeoutError:
            process.kill()
            return f"Error: Command timed out after {self.timeout} seconds"
        
        # Return combined output
        return stdout.decode("utf-8", errors="replace")
Features:
  • Configurable timeout
  • Safety guards against destructive commands
  • Optional workspace restriction
  • Combined stdout/stderr output
See nanobot/agent/tools/shell.py:12-159.

Web Tools

Implemented in nanobot/agent/tools/web.py:
class WebSearchTool(Tool):
    """Search the web using Brave Search API."""
    
    name = "web_search"
    description = "Search the web. Returns titles, URLs, and snippets."
    
    async def execute(self, query: str, count: int | None = None, **kwargs: Any) -> str:
        n = min(max(count or self.max_results, 1), 10)
        
        async with httpx.AsyncClient(proxy=self.proxy) as client:
            r = await client.get(
                "https://api.search.brave.com/res/v1/web/search",
                params={"q": query, "count": n},
                headers={
                    "Accept": "application/json", 
                    "X-Subscription-Token": self.api_key
                },
                timeout=10.0
            )
            r.raise_for_status()
        
        results = r.json().get("web", {}).get("results", [])[:n]
        
        lines = [f"Results for: {query}\n"]
        for i, item in enumerate(results, 1):
            lines.append(f"{i}. {item.get('title', '')}")
            lines.append(f"   {item.get('url', '')}")
            if desc := item.get("description"):
                lines.append(f"   {desc}")
        return "\n".join(lines)
See nanobot/agent/tools/web.py:47-106.
class WebFetchTool(Tool):
    """Fetch and extract content from a URL using Readability."""
    
    name = "web_fetch"
    description = "Fetch URL and extract readable content (HTML → markdown/text)."
    
    async def execute(
        self, 
        url: str, 
        extractMode: str = "markdown", 
        maxChars: int | None = None, 
        **kwargs: Any
    ) -> str:
        from readability import Document
        
        # Validate URL
        is_valid, error_msg = _validate_url(url)
        if not is_valid:
            return json.dumps({"error": f"URL validation failed: {error_msg}"})
        
        # Fetch with httpx
        async with httpx.AsyncClient(
            follow_redirects=True,
            max_redirects=MAX_REDIRECTS,
            timeout=30.0,
            proxy=self.proxy,
        ) as client:
            r = await client.get(url, headers={"User-Agent": USER_AGENT})
            r.raise_for_status()
        
        # Extract content based on content-type
        if "text/html" in r.headers.get("content-type", ""):
            doc = Document(r.text)
            content = self._to_markdown(doc.summary()) if extractMode == "markdown" else _strip_tags(doc.summary())
            text = f"# {doc.title()}\n\n{content}"
        else:
            text = r.text
        
        # Return as JSON
        return json.dumps({
            "url": url,
            "finalUrl": str(r.url),
            "status": r.status_code,
            "extractor": "readability",
            "text": text[:maxChars or self.max_chars]
        })
Features:
  • Readability extraction for clean content
  • Markdown or plain text output
  • URL validation and redirect following
  • Proxy support
See nanobot/agent/tools/web.py:109-181.

Coordination Tools

Allows the agent to send messages to specific chat channels.
class MessageTool(Tool):
    """Tool to send messages to specific channels."""
    
    async def execute(
        self, 
        content: str, 
        channel: str | None = None, 
        chat_id: str | None = None, 
        **kwargs: Any
    ) -> str:
        await self._send_callback(OutboundMessage(
            channel=channel or self._channel,
            chat_id=chat_id or self._chat_id,
            content=content,
        ))
        self._sent_in_turn = True
        return "Message sent successfully"
See nanobot/agent/tools/message.py.
Spawns a subagent to handle tasks in the background.
class SpawnTool(Tool):
    """Tool to spawn a subagent for background task execution."""
    
    async def execute(self, task: str, label: str | None = None, **kwargs: Any) -> str:
        return await self._manager.spawn(
            task=task,
            label=label,
            origin_channel=self._origin_channel,
            origin_chat_id=self._origin_chat_id,
            session_key=self._session_key,
        )
See nanobot/agent/tools/spawn.py:55-63.
Schedules tasks to run at specific times or intervals.
class CronTool(Tool):
    """Tool to schedule cron jobs."""
    
    async def execute(
        self, 
        schedule: str, 
        task: str, 
        label: str | None = None,
        **kwargs: Any
    ) -> str:
        await self._cron_service.add_job(
            schedule=schedule,
            task=task,
            label=label,
            channel=self._channel,
            chat_id=self._chat_id,
        )
        return f"Scheduled task: {label or task[:30]}"
See nanobot/agent/tools/cron.py.

Tool Registry

The ToolRegistry manages dynamic tool registration and execution:
class ToolRegistry:
    """
    Registry for agent tools.

    Allows dynamic registration and execution of tools.
    """

    def __init__(self):
        self._tools: dict[str, Tool] = {}

    def register(self, tool: Tool) -> None:
        """Register a tool."""
        self._tools[tool.name] = tool

    def get_definitions(self) -> list[dict[str, Any]]:
        """Get all tool definitions in OpenAI format."""
        return [tool.to_schema() for tool in self._tools.values()]

    async def execute(self, name: str, params: dict[str, Any]) -> str:
        """Execute a tool by name with given parameters."""
        tool = self._tools.get(name)
        if not tool:
            return f"Error: Tool '{name}' not found"
        
        # Validate parameters
        errors = tool.validate_params(params)
        if errors:
            return f"Error: Invalid parameters: {'; '.join(errors)}"
        
        # Execute tool
        result = await tool.execute(**params)
        return result
See nanobot/agent/tools/registry.py:8-66.

Parameter Validation

Tools automatically validate parameters against their JSON schema:
def validate_params(self, params: dict[str, Any]) -> list[str]:
    """Validate tool parameters against JSON schema. Returns error list (empty if valid)."""
    if not isinstance(params, dict):
        return [f"parameters must be an object, got {type(params).__name__}"]
    schema = self.parameters or {}
    return self._validate(params, {**schema, "type": "object"}, "")
Validation checks:
  • Type matching (string, integer, boolean, array, object)
  • Required fields
  • Enum values
  • Min/max bounds for numbers
  • Min/max length for strings
See nanobot/agent/tools/base.py:55-95.

Creating Custom Tools

1

Inherit from Tool Base Class

from nanobot.agent.tools.base import Tool
from typing import Any

class MyCustomTool(Tool):
    """My custom tool description."""
    pass
2

Implement Required Properties

@property
def name(self) -> str:
    return "my_tool"

@property
def description(self) -> str:
    return "Does something useful"

@property
def parameters(self) -> dict[str, Any]:
    return {
        "type": "object",
        "properties": {
            "input": {
                "type": "string",
                "description": "The input parameter"
            },
            "count": {
                "type": "integer",
                "description": "How many times to do it",
                "minimum": 1,
                "maximum": 10
            }
        },
        "required": ["input"]
    }
3

Implement Execute Method

async def execute(self, input: str, count: int = 1, **kwargs: Any) -> str:
    """Execute the tool logic."""
    try:
        # Your tool logic here
        result = await do_something(input, count)
        return f"Success: {result}"
    except Exception as e:
        return f"Error: {str(e)}"
4

Register the Tool

from nanobot.agent.tools.registry import ToolRegistry

registry = ToolRegistry()
registry.register(MyCustomTool())

MCP Tool Integration

Nanobot supports Model Context Protocol (MCP) for dynamic tool integration:
class MCPToolWrapper(Tool):
    """Wraps a single MCP server tool as a nanobot Tool."""

    def __init__(self, session, server_name: str, tool_def, tool_timeout: int = 30):
        self._session = session
        self._original_name = tool_def.name
        self._name = f"mcp_{server_name}_{tool_def.name}"
        self._description = tool_def.description or tool_def.name
        self._parameters = tool_def.inputSchema or {"type": "object", "properties": {}}
        self._tool_timeout = tool_timeout

    async def execute(self, **kwargs: Any) -> str:
        from mcp import types
        result = await asyncio.wait_for(
            self._session.call_tool(self._original_name, arguments=kwargs),
            timeout=self._tool_timeout,
        )
        parts = []
        for block in result.content:
            if isinstance(block, types.TextContent):
                parts.append(block.text)
        return "\n".join(parts) or "(no output)"
See nanobot/agent/tools/mcp.py:14-53.
MCP tools are automatically discovered and registered at agent startup.

Tool Schema Format

Tools are exposed to the LLM in OpenAI function calling format:
def to_schema(self) -> dict[str, Any]:
    """Convert tool to OpenAI function schema format."""
    return {
        "type": "function",
        "function": {
            "name": self.name,
            "description": self.description,
            "parameters": self.parameters,
        },
    }
Example output:
{
  "type": "function",
  "function": {
    "name": "read_file",
    "description": "Read the contents of a file at the given path.",
    "parameters": {
      "type": "object",
      "properties": {
        "path": {
          "type": "string",
          "description": "The file path to read"
        }
      },
      "required": ["path"]
    }
  }
}

Error Handling

Tools should return descriptive error messages as strings:
async def execute(self, **kwargs: Any) -> str:
    try:
        # Tool logic
        return "Success: result"
    except PermissionError as e:
        return f"Error: {e}"
    except FileNotFoundError:
        return f"Error: File not found: {path}"
    except Exception as e:
        return f"Error: {str(e)}"
The registry automatically appends a hint to error messages: “[Analyze the error above and try a different approach.]”

Security Considerations

Filesystem Tools: Can be restricted to workspace directory with restrict_to_workspace=TrueShell Tool: Includes deny patterns for dangerous commands (rm -rf, format, dd, etc.)Web Tools: Validate URLs and limit content size to prevent DoS

Agent Loop

How tools are executed in the loop

Skills

Teaching the agent to use tools

Architecture

Where tools fit in the system

Build docs developers (and LLMs) love