Week 3: Tool Calls as a Typed Protocol

A streamed tool call is only a candidate action. Before it can affect the world, its name must resolve, its arguments must satisfy a schema, and the decision to call must remain visible to the loop. This chapter uses Pydantic models as executable specifications, then exercises the existing Tool and ToolRegistry APIs offline. It turns the deltas from Week 2 into the exact error channel that Week 4 will use for filesystem and shell operations.

Tool-call messages

The model sends a name and JSON arguments; the harness sends schemas before the turn and a string result after the turn. Code remains local to the harness. This boundary has three distinct failure classes: an unknown name, invalid arguments, and a valid call that fails during execution. Keeping them distinct gives the model a correction channel and gives evaluation a failure taxonomy.

The library’s concrete API is slightly richer than the minimal pseudocode: subclasses expose name, description, kind, and schema, and implement execute(ToolInvocation) -> ToolResult. The difference is important because the result carries metadata and can be serialized into an event log.

from __future__ import annotations

import asyncio
import json
import re
import sys
from pathlib import Path
from typing import Any, Literal

from pydantic import BaseModel, Field, ValidationError

PROJECT_SRC = (Path.cwd() / "projects" / "agent-harness" / "src").resolve()
if not PROJECT_SRC.is_dir():
    raise RuntimeError(f"Expected the project source at {PROJECT_SRC}")
sys.path.insert(0, str(PROJECT_SRC))

from agent_harness.config import Config
from agent_harness.tools.base import (
    Tool,
    ToolInvocation,
    ToolKind,
    ToolRegistry,
    ToolResult,
)

Tool schemas

Pydantic gives one definition two uses: it validates the arguments that arrived from the model, and it generates the JSON Schema sent to the model. Prefer a small JSON-Schema subset with explicit descriptions and bounded choices. A schema should constrain the dangerous ambiguity without encoding an entire programming language in a prompt.

class ArithmeticArgs(BaseModel):
    operation: Literal["add", "multiply"] = Field(
        ...,
        description="The arithmetic operation. Choose add or multiply.",
    )
    left: float = Field(..., description="The first numeric operand.")
    right: float = Field(..., description="The second numeric operand.")
    round_digits: int = Field(
        2,
        ge=0,
        le=6,
        description="Digits after the decimal point in the result.",
    )


json_schema = ArithmeticArgs.model_json_schema(mode="serialization")
print(json.dumps({
    key: json_schema[key]
    for key in ("type", "properties", "required")
    if key in json_schema
}, indent=2))
assert json_schema["properties"]["operation"]["enum"] == ["add", "multiply"]
assert "round_digits" not in json_schema["required"]

operation is an enum rather than an unconstrained string, and round_digits has a bounded range with a default. Those choices affect both selection and recovery. The validator can reject an impossible operation before execution; the description tells a model what the accepted values mean. Neither choice proves that a model will select correctly, so the ablation experiment later measures a deterministic proxy rather than claiming a model result.

class ArithmeticTool(Tool):
    name = "arithmetic"
    description = "Add or multiply two numbers and return a formatted result."
    kind = ToolKind.READ
    schema = ArithmeticArgs

    async def execute(self, invocation: ToolInvocation) -> ToolResult:
        args = ArithmeticArgs(**invocation.params)
        if args.operation == "add":
            value = args.left + args.right
        else:
            value = args.left * args.right
        return ToolResult.success_result(
            f"{value:.{args.round_digits}f}",
            metadata={"operation": args.operation, "value": value},
        )


class EchoArgs(BaseModel):
    text: str = Field(..., min_length=1, description="Text to return verbatim.")


class EchoTool(Tool):
    name = "echo"
    description = "Return a non-empty text value verbatim."
    kind = ToolKind.READ
    schema = EchoArgs

    async def execute(self, invocation: ToolInvocation) -> ToolResult:
        args = EchoArgs(**invocation.params)
        return ToolResult.success_result(args.text)


assert ArithmeticTool.__abstractmethods__ == frozenset()
assert Tool.__abstractmethods__ == {"execute"}

Tool registry

Registration is local capability management. The model sees only get_schemas() for tools that survive the configuration allowlist; dispatch performs lookup, validation, and execution in that order. The stable order below is useful for reproducibility and for prompt diffs, even though a model should not rely on position.

workspace = (Path.cwd() / ".tmp" / "agent-harness-week3").resolve()
workspace.mkdir(parents=True, exist_ok=True)
config = Config(cwd=workspace)
registry = ToolRegistry(config)
registry.register(ArithmeticTool(config))
registry.register(EchoTool(config))

schemas = registry.get_schemas()
prompt_listing = "\n".join(
    f"- {schema['name']}: {schema['description']}"
    for schema in schemas
)
print(prompt_listing)
assert [schema["name"] for schema in schemas] == ["arithmetic", "echo"]
assert all("parameters" in schema for schema in schemas)

A dispatch result must be safe to append to the conversation even when the call is wrong. The registry therefore returns ToolResult.error_result for unknown names and validation failures instead of raising them into the agent loop. An exception inside a tool is contained as an internal-error result as well; Week 5 will test how that result affects subsequent turns.

valid = await registry.invoke(
    "arithmetic",
    {"operation": "multiply", "left": 3, "right": 4, "round_digits": 0},
    workspace,
)
invalid = await registry.invoke(
    "arithmetic",
    {"operation": "divide", "left": 3, "right": 4},
    workspace,
)
unknown = await registry.invoke("does_not_exist", {}, workspace)

print({
    "valid": {
        "success": valid.success,
        "output": valid.output,
        "metadata": valid.metadata,
    },
    "invalid": {
        "success": invalid.success,
        "error": invalid.error,
        "validation_errors": invalid.metadata["validation_errors"],
        "model_output": invalid.to_model_output(),
    },
    "unknown": {"success": unknown.success, "error": unknown.error},
})
assert valid.success and valid.output == "12"
assert invalid.success is False
assert "operation" in invalid.error
assert invalid.metadata["validation_errors"]
assert unknown.success is False and unknown.error == "Unknown tool: does_not_exist"

Parallel calls should be dispatched concurrently only when the tools are independent. Their completions still need to be associated with the original call index or ID. asyncio.gather preserves the order of the awaitables it receives, so sorting explicitly here makes the association obvious rather than relying on completion timing.

parallel_calls = [
    (0, "arithmetic", {"operation": "add", "left": 2, "right": 5}),
    (1, "echo", {"text": "second call"}),
]


async def dispatch(index: int, name: str, params: dict[str, Any]):
    result = await registry.invoke(name, params, workspace)
    return index, result


parallel_results = await asyncio.gather(
    *(dispatch(index, name, params) for index, name, params in parallel_calls)
)
ordered_results = sorted(parallel_results, key=lambda item: item[0])
print([(index, result.output) for index, result in ordered_results])
assert [index for index, _ in ordered_results] == [0, 1]
assert [result.output for _, result in ordered_results] == ["7.00", "second call"]

Before real filesystem tools exist, protocol tests benefit from tiny deterministic capabilities. The following seven stubs are notebook fixtures, not additions to the package’s default registry. They let a learner test naming, schema generation, allowlisting, and dispatch without network, clock, or repository state; the package’s real builtins arrive in Week 4.

class StubArgs(BaseModel):
    value: str = Field("", description="A deterministic fixture value.")


class ProtocolStubTool(Tool):
    kind = ToolKind.READ
    schema = StubArgs

    def __init__(self, config: Config, name: str, description: str) -> None:
        super().__init__(config)
        self.name = name
        self.description = description

    async def execute(self, invocation: ToolInvocation) -> ToolResult:
        args = StubArgs(**invocation.params)
        return ToolResult.success_result(f"{self.name}: {args.value}")


stub_descriptions = {
    "calculator": "Perform deterministic arithmetic.",
    "clock": "Return a fixed clock reading.",
    "echo": "Return text without changing it.",
    "json": "Format a small JSON value.",
    "status": "Report a deterministic status.",
    "plan": "Return a short planning fixture.",
    "search": "Return a deterministic search fixture.",
}
stub_registry = ToolRegistry(config)
for name, description in stub_descriptions.items():
    stub_registry.register(ProtocolStubTool(config, name, description))

stub_names = [tool.name for tool in stub_registry.get_tools()]
print(stub_names)
assert stub_names == list(stub_descriptions)
assert len(stub_registry.get_schemas()) == 7

Tool descriptions

A full model-selection study needs a fixed model, decoding configuration, task set, and held-out split. This notebook stays offline, so use a lexical selector only as a smoke test for information loss. It asks whether descriptions contain words that distinguish four tools, then compares full descriptions, names alone, and a deterministic shuffle. The result is a schema-review signal, not evidence about a frontier model.

selection_specs = {
    "read_file": "read file contents and line numbers",
    "write_file": "create or overwrite file content",
    "arithmetic": "add or multiply two numeric values",
    "echo": "return text verbatim",
}
tasks = [
    ("show file contents", "read_file"),
    ("create a file", "write_file"),
    ("multiply two numeric values", "arithmetic"),
    ("return this text", "echo"),
]
selection_names = list(selection_specs)


def words(text: str) -> set[str]:
    return set(re.findall(r"[a-z]+", text.lower()))


def choose_tool(query: str, descriptions: dict[str, str]) -> str:
    query_words = words(query)
    return max(
        descriptions,
        key=lambda name: (
            len(query_words & words(descriptions[name])),
            -selection_names.index(name),
        ),
    )


def accuracy(descriptions: dict[str, str]) -> float:
    correct = sum(choose_tool(query, descriptions) == expected for query, expected in tasks)
    return correct / len(tasks)


name_only = {name: name.replace("_", " ") for name in selection_specs}
rotated_names = selection_names[1:] + selection_names[:1]
shuffled = {
    name: selection_specs[rotated]
    for name, rotated in zip(selection_names, rotated_names)
}
accuracies = {
    "full": accuracy(selection_specs),
    "name_only": accuracy(name_only),
    "shuffled": accuracy(shuffled),
}
print(accuracies)
assert accuracies["full"] == 1.0
assert accuracies["full"] > accuracies["name_only"]

Enums create a second, narrower probe. Compare an enum-constrained schema with a free-form string schema on the same canned candidate calls. The experiment measures what the boundary catches, not how often a model would emit each candidate. A follow-up model study should report invalid-choice rate and recovery rate separately.

class FreeArithmeticArgs(BaseModel):
    operation: str
    left: float
    right: float


candidates = [
    {"operation": "add", "left": 1, "right": 2},
    {"operation": "divide", "left": 1, "right": 2},
    {"operation": "multiply", "left": 1, "right": 2},
    {"operation": "subtract", "left": 1, "right": 2},
]
enum_rejected = 0
for candidate in candidates:
    try:
        ArithmeticArgs(**candidate)
    except ValidationError:
        enum_rejected += 1
free_schema_accepts = sum(
    FreeArithmeticArgs(**candidate).operation in {"add", "multiply"}
    or True
    for candidate in candidates
)
semantic_invalid = sum(
    candidate["operation"] not in {"add", "multiply"}
    for candidate in candidates
)
print({
    "enum_rejected_at_boundary": enum_rejected,
    "free_schema_accepted": free_schema_accepts,
    "semantically_invalid_candidates": semantic_invalid,
})
assert enum_rejected == 2
assert free_schema_accepts == 4
assert semantic_invalid == 2

first_attempt = await registry.invoke(
    "arithmetic",
    {"operation": "divide", "left": 8, "right": 2},
    workspace,
)
corrected_attempt = await registry.invoke(
    "arithmetic",
    {"operation": "multiply", "left": 8, "right": 2, "round_digits": 0},
    workspace,
)
print({
    "first_success": first_attempt.success,
    "first_error": first_attempt.error,
    "corrected_success": corrected_attempt.success,
    "corrected_output": corrected_attempt.output,
})
assert first_attempt.success is False
assert corrected_attempt.success is True

The registry converts hallucinated capability into data: an unknown tool is an explicit error, malformed arguments preserve the validator’s message, and an execution exception becomes an internal-error result. That is specification error made measurable at the interface. It does not answer whether calling was wise, so the loop must retain the original request, tool call, and result for later evaluation.

Week 2 supplied the complete call arguments. Week 4 now makes the result contract precise enough for files and processes, and Week 5 will deliver errors back through multiple turns.