The tool module contract

Everything the loader and the agent loop read off your module, enumerated from the code. Two attributes are required. The rest have defaults, and this page says what each default is and what it costs you.

Sources: tools/registry.py (_spec_from_module, ToolSpec), core/plugins.py (_register_plugin_module), core/loop.py (execute_with_heartbeat, _tool_budget, _expr_for, _voice_for), core/voice_patterns.py (build_soul_patterns), tools/__init__.py (ToolContext, ToolResult).

Required

SCHEMA — a dict

The tool as the model sees it, in OpenAI Responses API function-tool form.

SCHEMA = {
    "type": "function",
    "name": "hn_top",
    "description": "Get the current top stories on Hacker News, with their titles and links.",
    "parameters": {
        "type": "object",
        "properties": {"count": {"type": "integer", "description": "How many (1-10)."}},
        "required": [],
        "additionalProperties": False,
    },
}

If SCHEMA is missing or is not a dict, the module is skipped silently. No log line, no row in Settings — _spec_from_module returns None and the loader moves on. This is deliberate: it is how a helper module sitting beside your tool avoids being registered as one.

SCHEMA["name"] becomes the registered name. If you omit it, the loader falls back to the .py file's stem for a folder plugin, or to the entry-point name for an entry-point plugin.

The description is the whole of your user interface. The model reads that sentence and nothing else when it decides whether your tool is the right one.

execute(args, ctx) — an async def

async def execute(args: dict, ctx) -> str:
    ...

• It must be a coroutine function. The loop does asyncio.create_task(tool.execute(args, ctx)). A plain def raises TypeError: a coroutine was expected, got '<your return value>' out of the tool runner — it is not converted into a tool failure.
• args is the arguments dict the model produced. Validate it yourself: nothing between the model and you enforces your parameters schema. Treat args as possibly None.
• ctx is a ToolContext (below).

What you return:

The exception row is the one to design around. If your tool can fail in an interesting way, catch it yourself and return a sentence. An uncaught exception costs you the diagnosis and gives her five useless words to relay.

Your output is capped at 16,000 characters before it reaches the model (core/loop._MAX_TOOL_OUTPUT_CHARS). Past that it is cut, with a note appended telling her to summarise what she has or fetch something more specific. Return a summary rather than a dump.

Optional, and what the default costs

RISK — "read" (default), "write", "exec" or "network"

Risk is not an approval level. Nothing in the loop asks the user for permission because your RISK says exec. It decides three things:

1. Whether the call is an action. write, exec and network are (core/loop.py: is_action = _spec.risk in ("write", "exec", "network")). An action draws a step row in the work panel and writes a row to the audit log with detail="executed" or "executed:failed". A read tool draws a status line naming the tool, and no audit row.
2. Its minimum time budget. A tool with RISK = "exec" is given at least the length of this turn's approval window plus 5 s, so the loop cannot cancel it while a card is still on screen (_tool_budget).
3. Nothing else in practice. schemas_for(allow_risk=...) can filter on risk, but every caller in the tree passes all four values, so risk never removes your tool from the list.

Declare the risk that is true. It is what the audit log records and what the panel shows the person watching.

check() — a zero-argument function returning a bool

Answered before the tool is offered to the model, and cached for 30 seconds (registry._CHECK_TTL). Default: always available.

def check() -> bool:
    return bool(os.getenv("MY_SERVICE_TOKEN"))

• It runs on every loop iteration, inside the event loop. Keep it cheap: no network, no tree walk. A built-in that must probe a daemon does a socket pre-check first.
• If it raises, the LAST cached answer stands and a warning is logged; only a raise with nothing cached reads as unavailable. A check() that flaps therefore does not flap the tool out of the schema — which is deliberate, and is why yours must stay cheap and bounded.
• check() gates whether the tool is offered, not whether it can run. dispatchable() deliberately does not consult it, so a name the model already had earlier in the turn still resolves.

ANNOUNCE, HEARTBEAT, COMPLETE, FAIL

Her voice around your tool. Defaults are empty, and empty means silence — see the caveat below.

ANNOUNCE  = "Let me pull up the front page..."   # said once, before the call
HEARTBEAT = ["Still fetching the stories..."]    # said while it runs
COMPLETE  = "Here's what's on top right now:"    # said after ok=True
FAIL      = "It didn't answer just now."         # said after ok=False

• HEARTBEAT is a list. The first line fires at about 9 s, then about every 21 s (_HEARTBEAT_FIRST / _HEARTBEAT_EVERY). When the list runs out, the last line repeats.
• Heartbeats are deferred, not skipped, while your tool is blocked waiting for a human.
• COMPLETE is said only on ok=True; FAIL only on an actual failure. An action the user declined or a card still unanswered says neither.
• All four can be overridden per install in soul/default.md under ## Tool voice patterns, without touching your file.

There is no generic fallback for you. core/loop._GENERIC_VOICE exists, but a plugin present at startup never reaches it: the engine builds one entry per registered tool at boot (build_soul_patterns), so your tool already has an entry — four empty strings if you declared nothing — and _voice_for returns that. narrate() drops an empty phrase, so she simply says nothing around your tool. Write the four lines, or accept silence.

And your lines may still be suppressed. When the answering model is a reasoning model (reasoning_effort set and not off), the loop skips the canned English entirely, because the model narrates its own tool use in the user's language. Heartbeats become a wordless hum in a voice turn and nothing at all in a text one. Write the lines anyway — they are what a non-reasoning configuration hears — but do not put information in them that appears nowhere else.

EXPRESSIONS — a dict

Her face while your tool runs.

EXPRESSIONS = {"focus": "thinking", "fail": "confused"}

Only focus and fail are emitted. focus when the call starts (default thinking), fail when it fails (default sad). There is no success emission on purpose: the face for a finished turn comes from the audio tag she actually speaks, so voice and face have one source. A done key is read by nothing.

The 14 valid values (core/emotions.VALID_EMOTIONS):

neutral  happy  excited  sad  crying  angry  surprised
embarrassed  thinking  sleepy  affectionate  confused  scared  determined

TIMEOUT — an int, default 30

Seconds of compute. core/loop.TOOL_TIMEOUT is 30; TIMEOUT = 45 raises yours to 45.

ARG_TIMEOUT / MAX_TIMEOUT

Set ARG_TIMEOUT = True if your schema takes a timeout argument and the budget should follow it. The loop then uses the model's requested value plus approval headroom, capped at MAX_TIMEOUT (default 600).

INTERACTIVE — a bool, default False

Set it when your tool blocks waiting for a person to answer something. The loop then awaits execute() directly with no compute timeout and no heartbeats, so a human taking two minutes to type does not get cancelled at 30 seconds.

BUILT_IN — a bool, default False

Leave it alone. True means the model provider runs the tool server-side and the loop never calls your execute() at all.

TOOLSET

Ignored for plugins. _register_plugin_module rebuilds the spec with toolset = f"plugin:{plugin_name}" before registering, whatever you wrote. The comment in the code says why: letting a module keep its own plugin:* toolset let it pick a name the Settings switch did not govern, so turning the plugin off turned nothing off.

ctx — the ToolContext

Defined in tools/__init__.py. The fields a plugin is likely to want:

Staying inside the jail

If your tool takes a path from the model, resolve it the way the built-ins do — this follows symlinks and .. and refuses anything that leaves ctx.workdir:

from kotoba.core.path_security import PathSecurityError, validate_within_dir

try:
    target = validate_within_dir(args["path"], ctx.workdir)
except PathSecurityError:
    return "That path is outside my workspace."

Saying "I refused", and saying "I failed"

Returning a string is ok=True. A refusal sentence and a failure sentence are both strings, so without a witness the loop grades them exactly like a finished job — green mark, executed audit row, and a retry told it is already done. There are two witnesses, and they mean different things:

from kotoba.core.loop import note_tool_refusal, note_tool_failure

# Nothing happened. Your tool read its own arguments and declined.
# Grey mark, no audit row, no FAIL line — the outcome is a decision, not a stumble.
note_tool_refusal(ctx)
return "I won't do that: the file you named is outside the workspace."

# It really ran and did not land. Red mark, an `executed:failed` audit row, your FAIL line.
note_tool_failure("the upstream service returned 502")
return "The service is down right now."

note_tool_refusal is keyed on ctx.call_id. note_tool_failure is scoped to the running block instead, and is a no-op if called outside one.

The minimum viable plugin

Everything above is optional except two things:

SCHEMA = {"type": "function", "name": "ping", "description": "Reply with pong.",
          "parameters": {"type": "object", "properties": {}, "required": []}}


async def execute(args, ctx):
    return "pong"

That loads, registers as plugin:<folder name>, gets risk="read", is always available, says nothing out loud, and works.