Contents7 sections
Risks added when ordinary functions enter an agent
The fake model and behavioral tests now fix the search-read-finish path as a baseline. This makes it safe to replace dispatch_tool() with one tool registry without changing behavior.
search_local_sources() is simple in isolation. Inside an agent loop, the model may repeat requests or send oversized arguments, while the tool may time out or return excessive content. Write tools add a harder case: the client observes a timeout after the server already completed the write, so a retry duplicates the side effect.
A tool registry places one execution boundary around every tool.
Model request
│
▼
Schema check ─▶ Authorization ─▶ Timeout/concurrency ─▶ Handler
│
▼
Model result ◀─ Stable error ◀─ Truncate/redact ◀─ Raw resultLabel side effects first
| Level | Example | Automatic retry |
|---|---|---|
read | Search, read document | Bounded retry for transient faults |
reversible_write | Save an overwritable draft | Retry with idempotency and status lookup |
irreversible_write | Send message, make payment | No default retry; enter approval |
A timeout means the caller does not know the outcome. It does not prove that the operation failed. Reads can retry, while writes must query execution state first.
Keep descriptions and policy together in ToolSpec
from dataclasses import dataclass
from typing import Any, Callable
@dataclass(frozen=True)
class ToolSpec:
name: str
description: str
effect: str
timeout_seconds: float
max_attempts: int
max_output_chars: int
handler: Callable[..., Any]
class ToolRegistry:
def __init__(self) -> None:
self._tools: dict[str, ToolSpec] = {}
def register(self, spec: ToolSpec) -> None:
if spec.name in self._tools:
raise ValueError("duplicate_tool")
self._tools[spec.name] = spec
def get(self, name: str) -> ToolSpec | None:
return self._tools.get(name)The model-visible description, runtime policy, and handler come from the same registry entry, preventing schema changes from drifting away from the executor.
Normalize invocation results
def invoke_tool(registry: ToolRegistry, name: str, arguments: dict) -> dict:
spec = registry.get(name)
if spec is None:
return {"ok": False, "error": "unknown_tool", "retryable": False}
try:
value = spec.handler(**arguments)
return {"ok": True, "data": str(value)[:spec.max_output_chars]}
except PermissionError:
return {"ok": False, "error": "forbidden", "retryable": False}
except (ConnectionError, TimeoutError):
return {"ok": False, "error": "transient_failure", "retryable": True}
except TypeError:
return {"ok": False, "error": "invalid_arguments", "retryable": False}
except Exception:
return {"ok": False, "error": "tool_failure", "retryable": False}The model receives only stable error codes and retryable. Exception stacks stay in service logs. Even retryable=True remains subject to max_attempts and the overall task budget.
Give write tools business idempotency keys
For saving a brief, an idempotency key can combine task_id + path + content_hash. Repeating the same business intent returns the first result without writing again.
class IdempotencyStore:
def __init__(self) -> None:
self._results: dict[str, dict] = {}
def run_once(self, key: str, operation: Callable[[], dict]) -> dict:
if key in self._results:
return self._results[key]
result = operation()
if result.get("ok"):
self._results[key] = result
return resultA network request ID identifies one transport attempt rather than business intent. When deduplication must survive process restarts, store idempotency results in a database and commit them with output-file state.
Test from a failure matrix
| Failure | Read tool | Write tool | Expected result |
|---|---|---|---|
| Missing argument | Do not execute | Do not execute | invalid_arguments |
| Transient disconnect | Retry at most twice | Query status first | Count against budget |
| Authorization denial | Do not retry | Do not retry | forbidden |
| Timeout with unknown outcome | May query again | Never rewrite blindly | outcome_unknown |
| Oversized output | Truncate | Truncate receipt | Record original size |
Tests inspect the handler's actual call count. Asserting only the final error code cannot reveal that the executor wrote twice.
What changes after the reliable tool layer
The agent loop no longer calls arbitrary functions directly. Every action passes through the registry, making errors and budgets visible in one place. Chapter 7 keeps this registry, replaces demonstration data with a searchable corpus, and connects search_local_sources and get_document to one RAG pipeline.
REFERENCES
References
Series
AI AGENT Beginner's Guide