Skip to main content
Every operation in Hiro League — pairing a device, listing channels, revoking access — is defined exactly once as a Tool. The CLI, the AI agent, and the HTTP API are all thin callers that invoke the same tool. There is no “CLI version” and “API version” of a feature.

​
Why this design

Before this design, the natural path was to write a CLI command that contained the logic, then separately write an agent tool that duplicated it, and later write an HTTP endpoint that duplicated it again. Each copy drifts independently. A bug fix in one doesn’t reach the others. A new parameter requires three changes. The specific problems this design resolves: Logic duplication across callers. The same operation (e.g. generate a pairing code) would need to be implemented three times: once in commands/device.py, once in the LangChain tool definition, and once in an HTTP handler. Any change to the operation requires finding and updating all three. No single enforcement point. With three separate implementations, there is no single place to add policy checks, audit logging, or rate limiting. Each caller has to remember to apply them independently — and they won’t. Inconsistent schemas. The CLI arg names, the agent tool parameter names, and the HTTP request fields would all drift apart over time, making the system harder to reason about. The design resolves these by:
  • Defining each operation as a Tool class with a single execute() method
  • Making CLI, agent, and HTTP all call execute() — they own rendering and transport, not logic
  • Providing a ToolRegistry as the single dispatch point for HTTP and future callers, where cross-cutting concerns live

​
Architecture overview

                    ┌─────────────────────────────┐
                    │         Tool classes         │
                    │  DeviceAddTool               │
                    │  DeviceListTool              │
                    │  DeviceRevokeTool            │
                    │  ...                         │
                    └──────────────┬──────────────┘
                                   │ .execute()
              ┌────────────────────┼────────────────────┐
              │                    │                    │
              ▼                    ▼                    ▼
     CLI (typer)           AI Agent              ToolRegistry
     commands/device.py    agent_manager.py      tools/registry.py
     parses flags          reads schema               │
     calls execute()       calls execute()            │ POST /invoke
     renders rich output   returns to LLM             ▼
                                              FastAPI server.py
                                              web UI / HTTP clients
The Tool is the unit of functionality. The three callers own only their transport layer.

​
The Tool abstraction

Every tool is a class that inherits from Tool (tools/base.py) and declares three things: 
class DeviceAddTool(Tool):
    name = "device_add"                          # snake_case, unique
    description = "Generate a short-lived pairing code to onboard a new mobile device"
    params = {
        "workspace": ToolParam(str, "Workspace name", required=False),
        "ttl_seconds": ToolParam(int, "Pairing code lifetime in seconds", required=False),
        "code_length": ToolParam(int, "Pairing code length in digits", required=False),
    }

    def execute(self, workspace=None, ttl_seconds=None, code_length=None) -> DeviceAddResult:
        # all logic lives here — no console output, no HTTP, no typer
        ...
name — used by the registry to dispatch /invoke calls, and by the agent to identify the tool. description — shown to the LLM when it decides which tool to call. Write it for an AI reader, not a human. params — flat dict of ToolParam. This is the single source of truth for CLI argument specs, LLM tool schemas, and the /invoke request shape. All three are derived from it automatically — you declare params once. execute() — pure logic. No console.print, no typer.Exit, no HTTP. Returns a typed dataclass result. Raises a typed exception on failure.

​
Result types

Each tool returns a dedicated dataclass: 
@dataclass
class DeviceAddResult:
    code: str
    expires_at: str
This keeps the return contract explicit and makes serialisation to JSON trivial (via dataclasses.asdict()).

​
Caller 1 — CLI

CLI commands (commands/device.py) are thin wrappers. They parse flags, call execute(), and render the result using rich. They contain no business logic. 
@device_app.command("add")
def device_add(workspace: str = ..., ttl_seconds: int = ...) -> None:
    result = DeviceAddTool().execute(workspace=workspace, ttl_seconds=ttl_seconds)
    console.print(f"  code: [bold]{result.code}[/bold]")
    console.print(f"  expires_at: [bold]{result.expires_at}[/bold]")
The CLI runs as a separate process from the server. It calls tools directly by importing them — no HTTP, no running server required. This means CLI commands work even when the server is stopped, which is correct for lifecycle operations like setup, start, and stop.

​
Caller 2 — AI agent

The agent (agent_manager.py) uses LangChain. The langchain_adapter.py module converts any Tool into a StructuredTool automatically, using the params dict to build the Pydantic schema: 
tools = to_langchain_list(all_tools())
The adapter reads tool.params to generate the LLM-facing schema, and wraps tool.execute() as the function LangChain calls when the model selects the tool. You never write a separate LangChain tool definition — the Tool class is the definition. The agent runs inside the server process, so it has direct in-process access to all tool instances. No HTTP boundary.

​
Caller 3 — HTTP API (/invoke)

The ToolRegistry (tools/registry.py) holds one instance of every registered tool and dispatches calls by name. It is the entry point for any caller that cannot import Python directly — primarily the web management UI.

​
Registry setup

At server startup (_server_process.py), the registry is built and injected into the FastAPI server: 
tool_registry = ToolRegistry()
tool_registry.register_all(all_tools())
set_tool_registry(tool_registry)

​
Endpoints

GET /tools — returns the schema of every registered tool. The web UI calls this to discover what operations are available and what parameters each one accepts. No hardcoding required. 
{
  "tools": [
    {
      "name": "device_add",
      "description": "Generate a short-lived pairing code...",
      "params": {
        "ttl_seconds": { "type": "int", "description": "...", "required": false }
      }
    }
  ]
}
POST /invoke — executes a tool by name with a flat params dict: 
// request
{ "tool": "device_add", "params": { "ttl_seconds": 120 } }

// response
{ "tool": "device_add", "result": { "code": "482910", "expires_at": "2026-03-05T..." } }
The registry filters params to only the keys the tool declares, so callers don’t need to be perfectly in sync with the tool signature. Unknown keys are silently ignored. Error responses:
SituationHTTP statusDetail
Tool name not registered404"Unknown tool: 'foo'. Available: [...]"
execute() raised an exception500"Tool 'device_add' raised: ..."
Registry not yet initialised503"Tool registry not initialised"

​
Policy function (future)

The registry accepts an optional policy callable. When provided, it is called before every invoke() dispatch, regardless of which caller triggered it. Raising any exception from the policy blocks the call. 
def my_policy(tool_name: str, params: dict, context: dict) -> None:
    if context.get("source") == "agent" and tool_name == "device_revoke":
        raise PermissionError("Agent is not allowed to revoke devices")
    if context.get("role") != "admin" and tool_name.startswith("workspace_"):
        raise PermissionError("Admin role required for workspace operations")

registry = ToolRegistry(policy=my_policy)
The context dict is populated by the caller — the HTTP endpoint injects the authenticated user and role, the agent injects its identity. The tools themselves never see or handle policy; it is entirely the registry’s concern. Planned policy capabilities:
  • Source-based rules — restrict which tools the agent can call vs. a human via web UI
  • Role-based access — read-only roles can call list tools but not add or revoke
  • Destructive op confirmation — require an explicit confirmed: true param for irreversible operations
  • Rate limiting — block repeated calls to device_add within a time window
  • Audit logging — record every tool call with caller identity and result

​
How to add a new tool

Follow these steps every time you add a new operation. The pattern is identical regardless of what the operation does.

​
Step 1 — Define the tool class

Create or add to a file in hirocli/tools/. Group related tools in one file (e.g. all channel operations in tools/channel.py). 
# hirocli/tools/channel.py

@dataclass
class ChannelAddResult:
    name: str
    enabled: bool

class ChannelAddTool(Tool):
    name = "channel_add"
    description = "Configure and register a new channel plugin"
    params = {
        "channel_name": ToolParam(str, "Channel name, e.g. 'telegram'"),
        "command": ToolParam(str, "Executable to run for this channel"),
        "enabled": ToolParam(bool, "Whether to enable the channel immediately", required=False),
        "workspace": ToolParam(str, "Workspace name", required=False),
    }

    def execute(self, channel_name, command, enabled=True, workspace=None) -> ChannelAddResult:
        # logic here — save config, no console output
        ...
        return ChannelAddResult(name=channel_name, enabled=enabled)
Rules for execute():
  • No console.print or typer.Exit
  • No sys.exit
  • Return a dataclass result
  • Raise a typed exception on failure (e.g. WorkspaceError, ValueError)

​
Step 2 — Export from tools/__init__.py and add to all_tools()

Add the import at the top of tools/__init__.py: 
from .channel import ChannelAddTool, ChannelListTool
Then add instances to the all_tools() list: 
def all_tools() -> list[Tool]:
    return [
        ...
        ChannelAddTool(),    # ← add here
        ChannelListTool(),
    ]
This single change automatically makes the tool available to both the HTTP registry and the AI agent. No other registration or adapter code needed.

​
Step 3 — Add the CLI command

​
Step 5 — Add the CLI command

In commands/channel.py (or a new commands file), write a thin wrapper: 
@channel_app.command("add")
def channel_add(
    channel_name: str = typer.Argument(...),
    command: str = typer.Option(..., "--command", "-c"),
    workspace: str = typer.Option(None, "--workspace", "-W"),
) -> None:
    """Configure and register a new channel plugin."""
    result = ChannelAddTool().execute(
        channel_name=channel_name,
        command=command,
        workspace=workspace,
    )
    console.print(f"[green]Channel '{result.name}' configured.[/green]")
The CLI command contains only: argument parsing, execute() call, result rendering.

​
Summary checklist

StepFileWhat you add
1tools/<domain>.pyTool class + result dataclass
2tools/__init__.pyExport the class + add instance to all_tools()
3commands/<domain>.pyThin CLI wrapper
Step 2 is the only registration step — all_tools() feeds both the HTTP registry and the AI agent automatically. Step 3 is rendering only — no logic.

​
Benefits summary

ConcernHow this design handles it
Single source of truthLogic lives in execute() once. CLI, agent, and HTTP all call it.
No schema driftparams dict is the single definition for CLI args, LLM schemas, and HTTP docs.
Uniform policy enforcementToolRegistry.invoke() is the only dispatch path for HTTP callers. Policy runs once, covers all tools.
Easy discoverabilityGET /tools exposes the full schema dynamically. Web UI never hardcodes tool names or params.
Offline CLICLI imports tools directly — no server required for cold operations.
In-process agentAgent calls execute() directly with no HTTP boundary or serialisation overhead.
Incremental growthAdding a tool is 3 mechanical steps. No new HTTP routes, no new adapter code, no new agent plumbing.