Tools

Tools are Python functions an agent can call. Dreadnode uses type annotations and Pydantic to generate the schema the model sees, so well-typed function signatures become well-shaped tool calls.

import typing as t

from dreadnode import tool


@tool
def lookup_indicator(
    indicator: t.Annotated[str, "IP, domain, or hash to investigate"],
) -> dict[str, str]:
    """Look up an indicator in an intel source."""
    return {"indicator": indicator, "verdict": "unknown"}

The docstring becomes the tool description. typing.Annotated metadata becomes the parameter description. The return type drives serialization.

Before writing a Python tool

Python tools are powerful, but they’re not always the right shape. Most capabilities are best served by teaching a workflow in a skill and letting the agent reach for tools it already has. Before adding @tool, work down this ladder:

1. Bash + an existing CLI. If the workflow can be expressed as a shell pipeline against a tool the agent already knows (rg, jq, gh, kubectl, vendor CLIs), the cheapest capability is a skill that teaches the pipeline. The agent has a bash tool that runs the command out-of-process under a timeout — no schema to author, no Python to keep in sync with the CLI, and every command is visible in the transcript.
2. An MCP server. Reach for MCP when the agent will call the same operation many times in a run, when the CLI is awkward (stateful sessions, GUI helpers, structured outputs that don’t survive a pipe), or when the implementation lives in a non-Python runtime. MCP isolates the work in its own process and exposes a typed surface to the agent.
3. A Python @tool. Last fallback. Reach here when the logic is genuinely Python-native — parsing a Pydantic structure, manipulating an in-process object, glue that’s tighter than spawning a subprocess.

A capability that ships ten thin Python wrappers around CLIs you could have called from bash is a maintenance liability — the wrappers go stale, the schemas drift, and every call still spawns a subprocess underneath. If you do write Python tools, follow the Async tools rule below — blocking sync work in a @tool is the single most common cause of stalled TUI sessions.

Where tools live

Capability tools come from Python files declared in the manifest:

tools:
  - tools/intel.py

If tools: is omitted, the runtime auto-discovers any *.py in the tools/ directory. Set tools: [] to disable entirely.

The loader collects from each file:

• module-level @tool-decorated functions
• module-level Tool instances
• module-level Toolset instances
• Toolset subclasses that construct with no arguments

Async tools

Define a tool as async def and the runtime awaits the call automatically. No additional decorator argument needed.

import httpx
import typing as t

from dreadnode import tool


@tool
async def fetch_indicator(
    indicator: t.Annotated[str, "Indicator to look up"],
) -> dict[str, str]:
    """Fetch indicator metadata from the intel API."""
    async with httpx.AsyncClient() as client:
        response = await client.get(f"https://intel.example.com/{indicator}")
        response.raise_for_status()
        return response.json()

Use async def whenever the tool does I/O — network calls, subprocesses, database queries, large file reads, anything that waits on the kernel. Sync @tool functions are reserved for pure-CPU work that returns in well under a second.

If you need to call a subprocess, use asyncio.create_subprocess_exec (see dreadnode.tools.execute for a worked example), not the standard-library blocking variants:

# Don't — blocks the agent runtime for the duration of the subprocess.
@tool
def scan(target: str) -> str:
    result = subprocess.run(["nmap", target], capture_output=True, text=True, timeout=600)
    return result.stdout

# Do — yields back to the event loop while waiting on the child.
@tool
async def scan(target: str) -> str:
    proc = await asyncio.create_subprocess_exec(
        "nmap", target,
        stdout=asyncio.subprocess.PIPE,
        stderr=asyncio.subprocess.STDOUT,
    )
    stdout, _ = await asyncio.wait_for(proc.communicate(), timeout=600)
    return stdout.decode(errors="replace")

The runtime offloads sync tools to a worker thread, so a blocking sync @tool won’t deadlock the agent — but it still gives up one of the thread pool’s slots, can’t be cancelled cleanly, and competes for the GIL with the TUI’s renderer. Async is the supported shape for I/O; the offload is a safety net so a misbehaving third-party tool doesn’t take the whole session down.

Error handling

By default, @tool catches every exception and surfaces it to the model as a structured error so it can recover. Override the policy with catch:

@tool(catch=[ConnectionError, TimeoutError])
def network_lookup(host: str) -> dict[str, str]:
    """Catch only the listed exceptions; everything else aborts the turn."""
    ...

@tool(catch=False)
def must_succeed(name: str) -> dict[str, str]:
    """Propagate everything — turn fails if this raises."""
    ...

When the runtime catches an exception, the tool result becomes an ErrorModel carrying the exception type and message. The agent sees enough to retry or change approach.

Truncating output

Long tool outputs eat context. truncate caps the serialized return value:

@tool(truncate=4000)
def list_files(path: str) -> str:
    """Returns at most 4000 characters of output."""
    ...

Truncation happens after serialization, before the result is handed to the model.

Automatic output offload

Even with truncate unset, the runtime guards against runaway tool output. When a serialized return value exceeds 30,000 characters, the agent loop writes the full content to ~/.dreadnode/tool-output/<YYYYMMDD-HHMMSS>-<tool-call-id>.txt (or whatever configure(cache=...) resolves to) and replaces the in-context result with a middle-out summary — the first 15K characters, a [... N lines truncated — full output saved to <absolute-path>] ... marker, then the last 15K. The agent sees the absolute path and can read the file with the standard file-read tool. Span metadata records only the cache-relative path (e.g. tool-output/<file>.txt) so the platform never receives absolute filesystem paths.

This is automatic; tools don’t need to opt in. Set truncate= explicitly when you want a tighter cap or know the model never needs the long-tail content.

Stateful toolsets

Use Toolset when a group of tools shares state — an HTTP session, a cache, a client:

import typing as t

import dreadnode


class IntelTools(dreadnode.Toolset):
    def __init__(self) -> None:
        self.cache: dict[str, str] = {}

    @dreadnode.tool_method
    def lookup(
        self,
        indicator: t.Annotated[str, "Indicator to investigate"],
    ) -> dict[str, str]:
        """Look up an indicator."""
        if indicator in self.cache:
            return {"indicator": indicator, "verdict": self.cache[indicator]}
        verdict = "unknown"
        self.cache[indicator] = verdict
        return {"indicator": indicator, "verdict": verdict}

Every method decorated with @dreadnode.tool_method becomes a tool. The instance is constructed once per capability load — state lives for the runtime’s lifetime.

@tool_method accepts the same catch and truncate arguments as @tool.

Toolset subclasses must construct with no arguments — the loader calls MyToolset() directly and skips any class that raises TypeError. Take constructor parameters and your Toolset will be silently dropped from the capability.

Async resources in toolsets

The loader instantiates Toolset subclasses synchronously and never enters an async context. So if your tools need an async resource (an httpx.AsyncClient, a database connection pool, a long-lived MCP client), construct it lazily on first use — not in __init__:

import httpx
import typing as t
from pydantic import PrivateAttr

import dreadnode


class HttpTools(dreadnode.Toolset):
    _client: httpx.AsyncClient | None = PrivateAttr(default=None)

    def _ensure_client(self) -> httpx.AsyncClient:
        if self._client is None:
            self._client = httpx.AsyncClient(timeout=30)
        return self._client

    @dreadnode.tool_method
    async def fetch(
        self,
        url: t.Annotated[str, "URL to fetch"],
    ) -> str:
        """Fetch a URL and return the body."""
        response = await self._ensure_client().get(url)
        response.raise_for_status()
        return response.text

Use PrivateAttr for runtime-only state — Pydantic skips it during validation, which keeps the toolset constructible with no args.

Reference

The full @tool, Tool, and Toolset API — including Component, Context injection, and serialization details — lives at dreadnode.tools.