100%

Page 11 of 22 · 292 min left

Chapter 6 · Tool and Function Calling

Tool Contracts and Execution

Page 11 of 22
22 min read

Nadia in Dhaka builds a help-desk assistant. A customer asks, "Is the old returns policy still valid?" The model cannot open the document store itself. It can only ask your program: "please call lookup_document with this id." What happens next decides whether the feature is safe. This page builds the part of EvidenceDesk that receives such a request, checks it, runs it and returns a clear result.

You already know Pydantic from Git, APIs and Backend for AI: Beginner and result schemas from Reliable Output Contracts. Here the same ideas guard a different door: the one between a model's words and your code.

What you will learn

Why a model's tool call is a request, and who runs it, and how a registry stores each tool: name, description, argument schema and handler.
How to build a read-only calculator without eval, and a document lookup over the corpus.
How to validate arguments with Pydantic v2, reject unknown tool names, bound a slow tool and return every outcome as explicit data.
How to map different provider formats into one internal ToolCall, and how to test the whole loop offline with a fake client.

A model's call is a request, not an action

With tool calling, the model selects a tool and proposes arguments. Your application executes it, or refuses. These are two separate steps, and the model only takes part in the first. Nothing the model writes has run until your code runs it.

We keep the two steps apart in code, with the checks in between:

Output

model --request--> [ known name? ] --> [ valid arguments? ] --> [ run, with a time limit ]
                                                                  |
model <--result--  [ envelope: ok + data, or ok=false + error ] <--+

We start with a request as plain data. Here the "model" is only a dictionary we wrote. Real models are less predictable, which is exactly why we test.

Python

# a recorded model request: just data. Nothing has run.
request = {"id": "c1", "name": "calculator", "arguments": {"expression": "12 * (3 + 4)"}}
print("requested:", request["name"], request["arguments"])

Output

requested: calculator {'expression': '12 * (3 + 4)'}

Two corrections to common folklore. A prompt is not a guardrail: "only call safe tools" in a system prompt is a hope, not a control, so enforce names and argument shapes in code, outside the model. And tool output is untrusted input: a result may hold text from a file or web page, and that text is never an instruction. The pages Tool Permissions and Failure Handling and Prompt Injection and Access Boundaries build these defences.

Provider note (as of 2026-10-12): the OWASP Top 10 for LLM Applications, 2025 edition, lists "Excessive Agency" and names "excessive functionality," "excessive permissions" and "excessive autonomy" as root causes. A newer 2026 edition exists, and risk numbers changed between editions, so cite risks by name and edition. The UK National Cyber Security Centre (blog, 2025-12-08) says prompt injection may never be totally mitigated like SQL injection, and advises "deterministic (non-LLM) safeguards that constrain the actions of the system." This page follows that advice: code decides, not the model. Check the current documentation and the current edition.

The module: registry, calculator, lookup, dispatch

Create the file below as kit/evidencedesk/tools.py. Work in the folder that contains kit/. The kit is the finished project folder of the earlier pages (its evidencedesk/, tests/, corpus/ and data/ are the ones you built), so from here on paths start with kit/. Make the package importable once per terminal:

Bash

export PYTHONPATH=kit

Python

# file: kit/evidencedesk/tools.py
"""EvidenceDesk tools: registry, validation, dispatch (read-only tools)."""
from __future__ import annotations

import ast
import json
import operator
from concurrent.futures import ThreadPoolExecutor, TimeoutError as FutureTimeout
from dataclasses import dataclass
from pathlib import Path
from typing import Any, Callable

from pydantic import BaseModel, ConfigDict, Field, ValidationError


class ToolError(Exception):
    """An expected failure inside a tool. The code is safe to show to the model."""

    def __init__(self, code: str, message: str):
        super().__init__(message)
        self.code, self.message = code, message


class ToolArgs(BaseModel):
    """Base class for every tool's arguments: unknown fields are rejected."""

    model_config = ConfigDict(extra="forbid")


@dataclass(frozen=True)
class ToolCall:
    """The one internal shape of a model's request (a request, not an action)."""

    id: str
    name: str
    arguments: dict[str, Any]


@dataclass(frozen=True)
class ToolResult:
    """Result envelope: ok=True carries data, ok=False carries an error."""

    call_id: str
    name: str
    ok: bool
    data: Any = None
    error: dict[str, Any] | None = None

    def as_dict(self) -> dict[str, Any]:
        body = {"data": self.data} if self.ok else {"error": self.error}
        return {"call_id": self.call_id, "name": self.name, "ok": self.ok, **body}

    def as_message(self) -> dict[str, str]:
        content = json.dumps(self.as_dict(), sort_keys=True, ensure_ascii=False)
        return {"role": "tool", "tool_call_id": self.call_id, "content": content}


@dataclass(frozen=True)
class Tool:
    name: str
    description: str
    args_model: type[BaseModel]
    handler: Callable[[Any], Any]

    def spec(self) -> dict[str, Any]:
        schema = self.args_model.model_json_schema()
        schema.pop("title", None)
        return {"name": self.name, "description": self.description, "parameters": schema}


def _fail(call: ToolCall, code: str, message: str, **extra: Any) -> ToolResult:
    return ToolResult(call.id, call.name[:40], False, error={"code": code, "message": message, **extra})


class ToolRegistry:
    def __init__(self) -> None:
        self._tools: dict[str, Tool] = {}

    def register(self, tool: Tool) -> None:
        if tool.name in self._tools:
            raise ValueError(f"tool already registered: {tool.name}")
        self._tools[tool.name] = tool

    def names(self) -> list[str]:
        return sorted(self._tools)

    def specs(self) -> list[dict[str, Any]]:
        return [self._tools[n].spec() for n in self.names()]

    def execute(self, call: ToolCall, timeout: float = 2.0) -> ToolResult:
        tool = self._tools.get(call.name)
        if tool is None:
            return _fail(call, "unknown_tool", "no such tool is registered")
        try:
            args = tool.args_model.model_validate(call.arguments)
        except ValidationError as exc:
            problems = [{"loc": ".".join(str(p) for p in e["loc"]), "msg": e["msg"]} for e in exc.errors()]
            return _fail(call, "invalid_arguments", "arguments rejected", problems=problems)
        pool = ThreadPoolExecutor(max_workers=1)
        try:
            data = pool.submit(tool.handler, args).result(timeout=timeout)
        except FutureTimeout:
            return _fail(call, "timeout", f"tool did not finish within {timeout} s")
        except ToolError as exc:
            return _fail(call, exc.code, exc.message)
        except Exception:  # the exception text may hold internals: do not pass it on
            return _fail(call, "internal_error", "the tool failed")
        finally:
            pool.shutdown(wait=False)
        return ToolResult(call.id, call.name, True, data=data)


# --- calculator: read-only, deterministic, no eval() ---------------------------------------------
_BINARY = {ast.Add: operator.add, ast.Sub: operator.sub, ast.Mult: operator.mul,
           ast.Div: operator.truediv, ast.FloorDiv: operator.floordiv, ast.Mod: operator.mod}
_UNARY = {ast.UAdd: operator.pos, ast.USub: operator.neg}


def safe_eval(expression: str) -> float | int:
    """Evaluate + - * / // % on plain numbers by walking the syntax tree."""

    def walk(node: ast.AST) -> float | int:
        if isinstance(node, ast.Expression):
            return walk(node.body)
        if isinstance(node, ast.Constant) and type(node.value) in (int, float):
            return node.value
        if isinstance(node, ast.BinOp) and type(node.op) in _BINARY:
            return _BINARY[type(node.op)](walk(node.left), walk(node.right))
        if isinstance(node, ast.UnaryOp) and type(node.op) in _UNARY:
            return _UNARY[type(node.op)](walk(node.operand))
        raise ToolError("bad_expression", f"not allowed: {type(getattr(node, 'op', node)).__name__}")

    try:
        tree = ast.parse(expression, mode="eval")
        value = walk(tree)
    except SyntaxError:
        raise ToolError("bad_expression", "not a valid arithmetic expression") from None
    except ZeroDivisionError:
        raise ToolError("division_by_zero", "division by zero") from None
    return round(value, 10) if isinstance(value, float) else value


class CalculatorArgs(ToolArgs):
    expression: str = Field(min_length=1, max_length=100)


calculator_tool = Tool(
    name="calculator",
    description="Evaluate an arithmetic expression with + - * / // % and parentheses.",
    args_model=CalculatorArgs,
    handler=lambda args: {"value": safe_eval(args.expression)},
)


# --- document lookup over the corpus front matter ----------------------------------------------
class LookupArgs(ToolArgs):
    doc_id: str = Field(pattern=r"^[a-z0-9-]{3,40}$")


def read_front_matter(path: Path) -> dict[str, str]:
    lines = path.read_text(encoding="utf-8").splitlines()
    meta: dict[str, str] = {}
    for line in lines[1:]:
        if line.strip() == "---":
            break
        key, _, value = line.partition(":")
        meta[key.strip()] = value.strip()
    return meta


def make_lookup_tool(corpus_dir: Path) -> Tool:

    def handler(args: LookupArgs) -> dict[str, str]:
        for path in sorted(corpus_dir.glob("*.md")):
            meta = read_front_matter(path)
            if meta.get("id") == args.doc_id:
                return {k: meta[k] for k in ("id", "title", "version", "status", "lang")}
        raise ToolError("not_found", "no document has that id")

    return Tool("lookup_document", "Look up a help-centre document by its id.", LookupArgs, handler)


def default_registry(corpus_dir: Path) -> ToolRegistry:
    registry = ToolRegistry()
    registry.register(calculator_tool)
    registry.register(make_lookup_tool(corpus_dir))
    return registry


# --- provider shapes -> one ToolCall (dict in, dataclass out) ----------------------------------
def _arguments(raw: Any) -> dict[str, Any]:
    if isinstance(raw, str):  # some APIs send the arguments as a JSON string
        try:
            raw = json.loads(raw)
        except json.JSONDecodeError:
            raise ToolError("malformed_call", "arguments are not valid JSON") from None
    if not isinstance(raw, dict):
        raise ToolError("malformed_call", "arguments are not an object")
    return raw


def from_openai(item: dict[str, Any]) -> ToolCall:
    return ToolCall(item["call_id"], item["name"], _arguments(item["arguments"]))


def from_anthropic(block: dict[str, Any]) -> ToolCall:
    return ToolCall(block["id"], block["name"], _arguments(block["input"]))


def from_gemini(step: dict[str, Any]) -> ToolCall:
    return ToolCall(step["id"], step["name"], _arguments(step["arguments"]))


def from_ollama(call: dict[str, Any], index: int = 0) -> ToolCall:
    fn = call["function"]  # this shape has no call id: make a stable one from the position
    return ToolCall(f"call_{index}", fn["name"], _arguments(fn["arguments"]))


# --- offline: a stand-in model and one step ----------------------------------------------------
class ScriptedToolModel:
    """Fake model: replays recorded requests. Real models are less predictable."""

    def __init__(self, recorded: list[dict[str, Any]]):
        self._recorded = list(recorded)

    def next_call(self, question: str) -> dict[str, Any] | None:
        return self._recorded.pop(0) if self._recorded else None


def run_one_step(model: ScriptedToolModel, registry: ToolRegistry, question: str) -> list[dict[str, Any]]:
    transcript: list[dict[str, Any]] = [{"role": "user", "content": question}]
    raw = model.next_call(question)
    if raw is None:
        return transcript
    transcript.append({"role": "assistant", "tool_request": raw})
    arguments = raw.get("arguments")
    call = ToolCall(str(raw.get("id", "")), str(raw.get("name", "")), arguments if isinstance(arguments, dict) else {})
    transcript.append(registry.execute(call).as_message())
    return transcript

Read it in this order: Tool and ToolRegistry; safe_eval and the two tools; execute, the gate; ToolResult, the envelope; the four from_* mappers; and ScriptedToolModel with run_one_step, the offline loop.

Define a tool registry

A tool is four things: a name the model can say, a description that tells the model when to use it, an arguments model that defines the allowed input, and a handler, the Python function that does the work. The registry is a dictionary of such tools and nothing else. It is also the complete list of what the model can reach: if a function is not registered, no model output can call it.

The argument models inherit from ToolArgs, which sets extra="forbid". The JSON schema that spec() produces comes from the same Pydantic class that later validates the input, so the contract shown to the model and the contract enforced by your code cannot drift apart.

Python

from pprint import pprint
from pathlib import Path
from evidencedesk.tools import ToolCall, default_registry

registry = default_registry(Path("kit/corpus"))
print(registry.names())
spec = registry.specs()[0]
print(spec["description"])
pprint(spec["parameters"], width=76)

Output

['calculator', 'lookup_document']
Evaluate an arithmetic expression with + - * / // % and parentheses.
{'additionalProperties': False,
 'properties': {'expression': {'maxLength': 100,
                               'minLength': 1,
                               'title': 'Expression',
                               'type': 'string'}},
 'required': ['expression'],
 'type': 'object'}

Notice additionalProperties: false in the schema. A schema helps the model choose well, but the model may ignore it, so your code validates again.

Create a read-only calculator tool

A calculator is a good first tool: deterministic, no side effects, easy to test with fixed cases. The tempting implementation is eval(expression). Do not do it. eval runs any Python, not only arithmetic. The text __import__('os').getcwd() is harmless here, but with a different method name the same line deletes files or reads secrets. The text comes from a model, which may have copied it from an attacker's document, so eval and exec never touch it.

safe_eval parses the text with the ast module into a tree and walks it. Only numbers, the operators + - * / // %, and a sign in front of a number are allowed. Everything else, including names, calls and ** (which can build gigantic numbers), is refused. The first output line shows the nodes the walker sees.

Python

import ast
from evidencedesk.tools import ToolError, safe_eval

tree = ast.parse("12 * (3 + 4)", mode="eval")
print([type(node).__name__ for node in ast.walk(tree)])
for text in ["12 * (3 + 4)", "7 // 2 + 0.5", "__import__('os').getcwd()", "2 ** 999999", "1 / 0", "1 +"]:
    try:
        print(f"{text!r:28} -> {safe_eval(text)}")
    except ToolError as exc:
        print(f"{text!r:28} -> refused: {exc.code} ({exc.message})")

Output

['Expression', 'BinOp', 'Constant', 'Mult', 'BinOp', 'Constant', 'Add', 'Constant']
'12 * (3 + 4)'               -> 84
'7 // 2 + 0.5'               -> 3.5
"__import__('os').getcwd()"  -> refused: bad_expression (not allowed: Call)
'2 ** 999999'                -> refused: bad_expression (not allowed: Pow)
'1 / 0'                      -> refused: division_by_zero (division by zero)
'1 +'                        -> refused: bad_expression (not a valid arithmetic expression)

Refusals are ToolError values with a short code, not crashes. The second tool, lookup_document, reads the front matter (id, title, version, status, language) of the corpus files and returns it. It is also read-only, and it does not return the document body.

Validate tool arguments with Pydantic

The model's arguments are untrusted input, like a JSON body from the internet. execute calls model_validate before the handler sees anything. Three failures matter: a missing field, an extra field (rejected because of extra="forbid") and an invalid value (wrong type, wrong length or pattern). Pydantic's error objects also contain a documentation URL and the input value, so we copy only loc and msg into the result.

Python

def show(name, **arguments):
    result = registry.execute(ToolCall("c1", name, arguments))
    print(name, arguments)
    print("  ->", result.error["code"] if not result.ok else result.data)
    for problem in (result.error or {}).get("problems", []):
        print("    ", problem["loc"], "|", problem["msg"])

show("calculator")
show("calculator", expression="1+1", mode="fast")
show("calculator", expression=42)
show("calculator", expression="2*21")

Output

calculator {}
  -> invalid_arguments
     expression | Field required
calculator {'expression': '1+1', 'mode': 'fast'}
  -> invalid_arguments
     mode | Extra inputs are not permitted
calculator {'expression': 42}
  -> invalid_arguments
     expression | Input should be a valid string
calculator {'expression': '2*21'}
  -> {'value': 42}

The last line shows success. Valid arguments prove only that the shape is right, not that the request is wise.

Dispatch only registered tools

Dispatch is a dictionary lookup on the exact name: no getattr, no globals()[name], no prefix matching. An unknown name returns unknown_tool, and the name is cut to 40 characters in the answer. The block also tries the lookup tool with a good id, an unknown id, and a path-traversal-looking id, which the pattern rule rejects before any file is touched.

Timeouts and unexpected errors

Tools fail and tools hang. execute runs the handler in a worker thread and waits for a limited time; if the wait ends, you get a timeout result and the feature moves on. If the handler raises an unexpected exception, the result says only "the tool failed", because exception text can contain paths or secrets. One honest limit: Python cannot kill a thread, so a timed-out handler may keep running. That is acceptable for a read-only tool, and a strong reason to keep early tools read-only. The slow tool below waits on an event, so the demo never depends on the clock.

Python

import threading
from evidencedesk.tools import Tool, ToolArgs

show("delete_everything", path="/")
show("lookup_document", doc_id="kc-warranty-v2")
show("lookup_document", doc_id="kc-nope")
show("lookup_document", doc_id="../../etc/passwd")

release = threading.Event()   # lets the slow handler stop when we are done


class NoArgs(ToolArgs):
    pass


registry.register(Tool("slow", "waits for a long time", NoArgs, lambda args: release.wait(5)))
timed_out = registry.execute(ToolCall("c2", "slow", {}), timeout=0.05)
release.set()
print("slow ->", timed_out.error)

Output

delete_everything {'path': '/'}
  -> unknown_tool
lookup_document {'doc_id': 'kc-warranty-v2'}
  -> {'id': 'kc-warranty-v2', 'title': 'Warranty', 'version': '2', 'status': 'current', 'lang': 'en'}
lookup_document {'doc_id': 'kc-nope'}
  -> not_found
lookup_document {'doc_id': '../../etc/passwd'}
  -> invalid_arguments
     doc_id | String should match pattern '^[a-z0-9-]{3,40}$'
slow -> {'code': 'timeout', 'message': 'tool did not finish within 0.05 s'}

Return tool results as explicit data

Every outcome, success or failure, becomes the same envelope: call_id, name, ok, and either data or error. The call_id ties the result to the request. as_message() turns the envelope into the JSON text that goes back into the model flow. Keys are sorted, so the text is stable and easy to compare in tests.

Python

ok = registry.execute(ToolCall("c3", "lookup_document", {"doc_id": "kc-returns-v2"}))
bad = registry.execute(ToolCall("c4", "calculator", {"expression": "2 ** 9"}))
import json
from pprint import pprint

for result in (ok, bad):
    pprint(json.loads(result.as_message()["content"]), width=76)

Output

{'call_id': 'c3',
 'data': {'id': 'kc-returns-v2',
          'lang': 'en',
          'status': 'superseded',
          'title': 'Returns policy (old)',
          'version': '2'},
 'name': 'lookup_document',
 'ok': True}
{'call_id': 'c4',
 'error': {'code': 'bad_expression', 'message': 'not allowed: Pow'},
 'name': 'calculator',
 'ok': False}

The second result is a failure the model can understand and react to. The first is a success that returns a superseded document: the tool did its job, but the application must still check the data. Passing validation is not a correct answer. The message content is this dictionary written as JSON text.

Implement a fake tool-calling client

ScriptedToolModel replays requests that we recorded earlier, so the execution loop is tested without a network, a key or a model. It is a stand-in: it never decides anything, which is the point of a repeatable test. run_one_step asks it once, executes the request through the registry and returns the transcript, ready to be sent back to a real model.

Python

from pprint import pprint
from evidencedesk.tools import ScriptedToolModel, run_one_step

model = ScriptedToolModel([{"id": "c5", "name": "lookup_document", "arguments": {"doc_id": "kc-warranty-v2"}}])
for message in run_one_step(model, registry, "Which document is the warranty?"):
    pprint(message, width=76)
print(len(run_one_step(ScriptedToolModel([]), registry, "Hello")), "message when the model asks for nothing")

Output

{'content': 'Which document is the warranty?', 'role': 'user'}
{'role': 'assistant',
 'tool_request': {'arguments': {'doc_id': 'kc-warranty-v2'},
                  'id': 'c5',
                  'name': 'lookup_document'}}
{'content': '{"call_id": "c5", "data": {"id": "kc-warranty-v2", "lang": '
            '"en", "status": "current", "title": "Warranty", "version": '
            '"2"}, "name": "lookup_document", "ok": true}',
 'role': 'tool',
 'tool_call_id': 'c5'}
1 message when the model asks for nothing

The transcript has three messages: question, request and tool message. If the model asks for nothing, nothing runs.

Compare provider tool-calling formats

Providers describe a tool request differently: arguments may be a JSON string or an object, and a call id may be missing. Do not spread these differences through your code. Map each shape to one internal ToolCall at the edge. The mappers are plain dict-in, dataclass-out code, so we run them on small dictionaries that we wrote by hand following the documented shapes. They are examples, not real responses.

Provider note (as of 2026-10-12): OpenAI (Responses API) declares a flat tool {"type": "function", "name", "description", "parameters"}; the response items of type function_call carry call_id, name and arguments as a JSON string; the result returns as a function_call_output item. Anthropic declares input_schema; the response has stop_reason == "tool_use" and tool_use blocks with id, name and input (already an object); the result returns as a tool_result block with tool_use_id. Gemini has two paths, the Interactions API (steps of type function_call with name, arguments, id), which its documentation recommends, and the older generateContent path. Ollama returns message.tool_calls with function.name and a parsed function.arguments; tool support depends on the model. Some Anthropic models reject a forced tool_choice. The type of Gemini arguments was not stated in the pages we read, so _arguments accepts a string or an object. Check the current documentation.

Python

import json
from evidencedesk.tools import from_anthropic, from_gemini, from_ollama, from_openai

# Hand-written examples that follow the documented shapes. They are NOT real responses.
args = {"expression": "12 * (3 + 4)"}
openai_item = {"type": "function_call", "call_id": "call_1", "name": "calculator", "arguments": json.dumps(args)}
anthropic_block = {"type": "tool_use", "id": "toolu_1", "name": "calculator", "input": args}
gemini_step = {"type": "function_call", "id": "fc_1", "name": "calculator", "arguments": args}
ollama_call = {"function": {"name": "calculator", "arguments": args}}

for call in (from_openai(openai_item), from_anthropic(anthropic_block), from_gemini(gemini_step), from_ollama(ollama_call)):
    print(call)

Output

ToolCall(id='call_1', name='calculator', arguments={'expression': '12 * (3 + 4)'})
ToolCall(id='toolu_1', name='calculator', arguments={'expression': '12 * (3 + 4)'})
ToolCall(id='fc_1', name='calculator', arguments={'expression': '12 * (3 + 4)'})
ToolCall(id='call_0', name='calculator', arguments={'expression': '12 * (3 + 4)'})

Four different shapes, four identical results. The fourth call has no id in the documented example, so from_ollama makes one from the position. A malformed argument string raises ToolError("malformed_call") instead of passing bad data along. Real SDK calls look like the next block. It is marked # not run here: it needs a key and a network, and we show no output because we did not run it. The SDK object is only the carrier; your registry does the checking. Treat the names as VERIFY — checked 2026-10-12.

Python

# not run here: needs a provider SDK, an API key and network access
# Responses-style call (OpenAI SDK)
specs = [{"type": "function", **spec} for spec in registry.specs()]
reply = client.responses.create(model=MODEL, tools=specs, input=messages)
messages += reply.output
for item in reply.output:
    if item.type == "function_call":
        result = registry.execute(from_openai({"call_id": item.call_id, "name": item.name, "arguments": item.arguments}))
        messages.append({"type": "function_call_output", "call_id": item.call_id, "output": result.as_message()["content"]})
# the Messages-style call of the other SDK differs only in shape: from_anthropic() maps its tool_use blocks

Exercise: a safe one-step tool call

The test file below simulates a model that chooses a read-only tool and checks each rule of this page: missing, extra and invalid arguments, unknown names, refused expressions, a timeout, an exception that must not leak, the provider mappings and a full one-step run. Save it as kit/tests/test_tools.py and run it.

Python

# file: kit/tests/test_tools.py
import json
import threading
from pathlib import Path

import pytest

from evidencedesk.tools import (ScriptedToolModel, Tool, ToolArgs, ToolCall, ToolError, ToolRegistry, default_registry,
                                from_anthropic, from_gemini, from_ollama, from_openai, run_one_step, safe_eval)

CORPUS = Path(__file__).resolve().parents[1] / "corpus"


@pytest.fixture
def registry():
    return default_registry(CORPUS)


def call(name, **arguments):
    return ToolCall("c1", name, arguments)


def test_calculator_ok(registry):
    result = registry.execute(call("calculator", expression="12 * (3 + 4) - 10 / 4"))
    assert result.ok and result.data == {"value": 81.5}


@pytest.mark.parametrize("bad", ["__import__('os')", "2 ** 3", "abs(-1)", "x + 1", "'a' * 3", "True + 1", "1 +"])
def test_safe_eval_refuses_everything_but_arithmetic(bad):
    with pytest.raises(ToolError):
        safe_eval(bad)


def test_missing_extra_and_invalid_arguments(registry):
    missing = registry.execute(call("calculator"))
    extra = registry.execute(call("calculator", expression="1+1", mode="fast"))
    invalid = registry.execute(call("calculator", expression=42))
    for r in (missing, extra, invalid):
        assert r.error["code"] == "invalid_arguments"
    assert missing.error["problems"][0]["loc"] == "expression"
    assert extra.error["problems"][0]["loc"] == "mode"
    assert all("url" not in p for p in invalid.error["problems"])


def test_unknown_tool_is_rejected(registry):
    result = registry.execute(call("delete_everything", path="/"))
    assert result.error["code"] == "unknown_tool"


def test_lookup_document(registry):
    ok = registry.execute(call("lookup_document", doc_id="kc-warranty-v2"))
    assert ok.data["title"] == "Warranty" and ok.data["status"] == "current"
    assert registry.execute(call("lookup_document", doc_id="kc-nope")).error["code"] == "not_found"
    assert registry.execute(call("lookup_document", doc_id="../../etc/passwd")).error["code"] == "invalid_arguments"


def test_timeout_and_unexpected_exception():
    release = threading.Event()

    class NoArgs(ToolArgs):
        pass

    def boom(_):
        raise RuntimeError("secret internal detail")

    registry = ToolRegistry()
    registry.register(Tool("slow", "waits", NoArgs, lambda _: release.wait(5)))
    registry.register(Tool("boom", "fails", NoArgs, boom))
    try:
        assert registry.execute(call("slow"), timeout=0.05).error["code"] == "timeout"
    finally:
        release.set()
    failed = registry.execute(call("boom"))
    assert failed.error == {"code": "internal_error", "message": "the tool failed"}






def test_provider_shapes_map_to_one_call():
    args = {"expression": "1+1"}
    calls = [
        from_openai({"type": "function_call", "call_id": "a", "name": "calculator", "arguments": json.dumps(args)}),
        from_anthropic({"type": "tool_use", "id": "a", "name": "calculator", "input": args}),
        from_gemini({"type": "function_call", "id": "a", "name": "calculator", "arguments": args}),
        from_ollama({"function": {"name": "calculator", "arguments": args}}),
    ]
    assert {(c.name, json.dumps(c.arguments)) for c in calls} == {("calculator", json.dumps(args))}
    with pytest.raises(ToolError):
        from_openai({"call_id": "a", "name": "calculator", "arguments": "{not json"})


def test_safe_one_step(registry):
    model = ScriptedToolModel([{"id": "c9", "name": "lookup_document", "arguments": {"doc_id": "kc-returns-v2"}}])
    transcript = run_one_step(model, registry, "Which returns document is old?")
    body = json.loads(transcript[-1]["content"])
    assert body["ok"] and body["data"]["status"] == "superseded"
    assert len(transcript) == 3

Bash

python -m pytest -q kit/tests/test_tools.py | sed 's/ in [0-9.]*s//'

Output

..............                                                           [100%]
14 passed

This is the pass line; sed removes the seconds because they change on every run. What these tests cannot show: that a real model will pick the right tool or write sensible arguments. They only prove that your code survives whatever it is given.

Where this shows up at work

Assistants that touch company data (order status, ticket lookup, quotes) use this pattern. Kwame in Accra, who reviews such a feature, asks two questions: what did the model ask for, and what actually ran? A registry with an envelope answers both from the logs, and unknown-tool results are the first clue that something is probing the system.

Common mistakes

Using eval on model text. It runs any Python, and the text may come from an attacker. Parse with ast and allow only the nodes you need.
Trusting the schema you showed the model. The model may ignore it. Validate in your own code.
Returning raw exceptions or Pydantic errors. They can leak paths, values and links. Return a code and loc/msg only.
Dispatching by getattr or globals(). Then the model's text chooses which function runs. Use an explicit registry.
Thinking a timeout stops the work. It stops the waiting; keep early tools read-only.

Try it yourself

Easy. Predict the result of the calculator for -5 + 10 % 4, then run it through the registry.
Medium. Revisit Reliable Output Contracts: write a Pydantic model Envelope with a validator that requires data when ok is true and an error when it is false. Check one good and one bad envelope.
Hard. Add a third read-only tool, count_documents, whose only argument is status, either "current" or "superseded". Register it and show that "draft" is rejected.

Answers

Easy. The modulo binds tighter than addition: 10 % 4 is 2, and -5 + 2 is -3.

Python

result = registry.execute(ToolCall("e1", "calculator", {"expression": "-5 + 10 % 4"}))
print(result.ok, result.data)

Output

True {'value': -3}

Medium. The validator checks a business rule that the types alone cannot express.

Python

from pydantic import BaseModel, ValidationError, model_validator


class Envelope(BaseModel):
    call_id: str
    name: str
    ok: bool
    data: object = None
    error: dict | None = None

    @model_validator(mode="after")
    def data_xor_error(self):
        broken = (self.data is None or self.error is not None) if self.ok else self.error is None
        if broken:
            raise ValueError("an ok result needs data and no error; a failure needs an error")
        return self


print(Envelope(**ok.as_dict()).ok, Envelope(**bad.as_dict()).ok)
try:
    Envelope(call_id="x", name="calculator", ok=True, error={"code": "x"})
except ValidationError as exc:
    print(exc.errors()[0]["msg"])

Output

True False
Value error, an ok result needs data and no error; a failure needs an error

Hard. The handler reuses read_front_matter. The Literal type makes Pydantic reject every other status before the handler runs.

Python

from pathlib import Path
from typing import Literal
from evidencedesk.tools import read_front_matter


class CountArgs(ToolArgs):
    status: Literal["current", "superseded"]


def count_documents(args):
    metas = [read_front_matter(p) for p in Path("kit/corpus").glob("*.md")]
    return {"status": args.status, "count": sum(m.get("status") == args.status for m in metas)}


registry.register(Tool("count_documents", "Count documents with a given status.", CountArgs, count_documents))
for status in ("current", "superseded", "draft"):
    result = registry.execute(ToolCall("h1", "count_documents", {"status": status}))
    print(status, "->", result.data if result.ok else result.error["problems"])

Output

current -> {'status': 'current', 'count': 12}
superseded -> {'status': 'superseded', 'count': 1}
draft -> [{'loc': 'status', 'msg': "Input should be 'current' or 'superseded'"}]

Pass checklist:

Easy: the printed value is -3 and ok is true.
Medium: the good envelopes validate, and the bad one raises a validation error with your message.
Hard: the counts are 12 and 1, and "draft" produces an invalid_arguments problem with loc status.
No answer uses eval, prints a Pydantic URL or reads outside kit/corpus; any new tool has a pytest case.

Summary

A model proposes a tool call; your code validates and executes it, or refuses.
A registry of name, description, argument model and handler is the complete list of reachable actions.
Arguments are untrusted: validate with Pydantic v2 and extra="forbid", report only loc and msg, and reject unknown names.
Do arithmetic with ast, never eval. Bound every call with a timeout and return every outcome as an explicit envelope.
Map provider formats into one ToolCall and test the loop with a scripted fake client.

Next: Tool Permissions and Failure Handling adds the rules around execution: which tools a task may use, confirmation for consequential actions, call limits and audit events.

AI Engineering with LLMs: Beginner

100%

Page 11 of 22 · 292 min left

Chapter 6 · Tool and Function Calling

Tool Contracts and Execution

Page 11 of 22
22 min read

What you will learn

Why a model's tool call is a request, and who runs it, and how a registry stores each tool: name, description, argument schema and handler.
How to build a read-only calculator without eval, and a document lookup over the corpus.
How to validate arguments with Pydantic v2, reject unknown tool names, bound a slow tool and return every outcome as explicit data.
How to map different provider formats into one internal ToolCall, and how to test the whole loop offline with a fake client.

A model's call is a request, not an action

We keep the two steps apart in code, with the checks in between:

Output

model --request--> [ known name? ] --> [ valid arguments? ] --> [ run, with a time limit ]
                                                                  |
model <--result--  [ envelope: ok + data, or ok=false + error ] <--+

We start with a request as plain data. Here the "model" is only a dictionary we wrote. Real models are less predictable, which is exactly why we test.

Python

# a recorded model request: just data. Nothing has run.
request = {"id": "c1", "name": "calculator", "arguments": {"expression": "12 * (3 + 4)"}}
print("requested:", request["name"], request["arguments"])

Output

requested: calculator {'expression': '12 * (3 + 4)'}

Provider note (as of 2026-10-12): the OWASP Top 10 for LLM Applications, 2025 edition, lists "Excessive Agency" and names "excessive functionality," "excessive permissions" and "excessive autonomy" as root causes. A newer 2026 edition exists, and risk numbers changed between editions, so cite risks by name and edition. The UK National Cyber Security Centre (blog, 2025-12-08) says prompt injection may never be totally mitigated like SQL injection, and advises "deterministic (non-LLM) safeguards that constrain the actions of the system." This page follows that advice: code decides, not the model. Check the current documentation and the current edition.

The module: registry, calculator, lookup, dispatch

Bash

export PYTHONPATH=kit

Python

# file: kit/evidencedesk/tools.py
"""EvidenceDesk tools: registry, validation, dispatch (read-only tools)."""
from __future__ import annotations

import ast
import json
import operator
from concurrent.futures import ThreadPoolExecutor, TimeoutError as FutureTimeout
from dataclasses import dataclass
from pathlib import Path
from typing import Any, Callable

from pydantic import BaseModel, ConfigDict, Field, ValidationError


class ToolError(Exception):
    """An expected failure inside a tool. The code is safe to show to the model."""

    def __init__(self, code: str, message: str):
        super().__init__(message)
        self.code, self.message = code, message


class ToolArgs(BaseModel):
    """Base class for every tool's arguments: unknown fields are rejected."""

    model_config = ConfigDict(extra="forbid")


@dataclass(frozen=True)
class ToolCall:
    """The one internal shape of a model's request (a request, not an action)."""

    id: str
    name: str
    arguments: dict[str, Any]


@dataclass(frozen=True)
class ToolResult:
    """Result envelope: ok=True carries data, ok=False carries an error."""

    call_id: str
    name: str
    ok: bool
    data: Any = None
    error: dict[str, Any] | None = None

    def as_dict(self) -> dict[str, Any]:
        body = {"data": self.data} if self.ok else {"error": self.error}
        return {"call_id": self.call_id, "name": self.name, "ok": self.ok, **body}

    def as_message(self) -> dict[str, str]:
        content = json.dumps(self.as_dict(), sort_keys=True, ensure_ascii=False)
        return {"role": "tool", "tool_call_id": self.call_id, "content": content}


@dataclass(frozen=True)
class Tool:
    name: str
    description: str
    args_model: type[BaseModel]
    handler: Callable[[Any], Any]

    def spec(self) -> dict[str, Any]:
        schema = self.args_model.model_json_schema()
        schema.pop("title", None)
        return {"name": self.name, "description": self.description, "parameters": schema}


def _fail(call: ToolCall, code: str, message: str, **extra: Any) -> ToolResult:
    return ToolResult(call.id, call.name[:40], False, error={"code": code, "message": message, **extra})


class ToolRegistry:
    def __init__(self) -> None:
        self._tools: dict[str, Tool] = {}

    def register(self, tool: Tool) -> None:
        if tool.name in self._tools:
            raise ValueError(f"tool already registered: {tool.name}")
        self._tools[tool.name] = tool

    def names(self) -> list[str]:
        return sorted(self._tools)

    def specs(self) -> list[dict[str, Any]]:
        return [self._tools[n].spec() for n in self.names()]

    def execute(self, call: ToolCall, timeout: float = 2.0) -> ToolResult:
        tool = self._tools.get(call.name)
        if tool is None:
            return _fail(call, "unknown_tool", "no such tool is registered")
        try:
            args = tool.args_model.model_validate(call.arguments)
        except ValidationError as exc:
            problems = [{"loc": ".".join(str(p) for p in e["loc"]), "msg": e["msg"]} for e in exc.errors()]
            return _fail(call, "invalid_arguments", "arguments rejected", problems=problems)
        pool = ThreadPoolExecutor(max_workers=1)
        try:
            data = pool.submit(tool.handler, args).result(timeout=timeout)
        except FutureTimeout:
            return _fail(call, "timeout", f"tool did not finish within {timeout} s")
        except ToolError as exc:
            return _fail(call, exc.code, exc.message)
        except Exception:  # the exception text may hold internals: do not pass it on
            return _fail(call, "internal_error", "the tool failed")
        finally:
            pool.shutdown(wait=False)
        return ToolResult(call.id, call.name, True, data=data)


# --- calculator: read-only, deterministic, no eval() ---------------------------------------------
_BINARY = {ast.Add: operator.add, ast.Sub: operator.sub, ast.Mult: operator.mul,
           ast.Div: operator.truediv, ast.FloorDiv: operator.floordiv, ast.Mod: operator.mod}
_UNARY = {ast.UAdd: operator.pos, ast.USub: operator.neg}


def safe_eval(expression: str) -> float | int:
    """Evaluate + - * / // % on plain numbers by walking the syntax tree."""

    def walk(node: ast.AST) -> float | int:
        if isinstance(node, ast.Expression):
            return walk(node.body)
        if isinstance(node, ast.Constant) and type(node.value) in (int, float):
            return node.value
        if isinstance(node, ast.BinOp) and type(node.op) in _BINARY:
            return _BINARY[type(node.op)](walk(node.left), walk(node.right))
        if isinstance(node, ast.UnaryOp) and type(node.op) in _UNARY:
            return _UNARY[type(node.op)](walk(node.operand))
        raise ToolError("bad_expression", f"not allowed: {type(getattr(node, 'op', node)).__name__}")

    try:
        tree = ast.parse(expression, mode="eval")
        value = walk(tree)
    except SyntaxError:
        raise ToolError("bad_expression", "not a valid arithmetic expression") from None
    except ZeroDivisionError:
        raise ToolError("division_by_zero", "division by zero") from None
    return round(value, 10) if isinstance(value, float) else value


class CalculatorArgs(ToolArgs):
    expression: str = Field(min_length=1, max_length=100)


calculator_tool = Tool(
    name="calculator",
    description="Evaluate an arithmetic expression with + - * / // % and parentheses.",
    args_model=CalculatorArgs,
    handler=lambda args: {"value": safe_eval(args.expression)},
)


# --- document lookup over the corpus front matter ----------------------------------------------
class LookupArgs(ToolArgs):
    doc_id: str = Field(pattern=r"^[a-z0-9-]{3,40}$")


def read_front_matter(path: Path) -> dict[str, str]:
    lines = path.read_text(encoding="utf-8").splitlines()
    meta: dict[str, str] = {}
    for line in lines[1:]:
        if line.strip() == "---":
            break
        key, _, value = line.partition(":")
        meta[key.strip()] = value.strip()
    return meta


def make_lookup_tool(corpus_dir: Path) -> Tool:

    def handler(args: LookupArgs) -> dict[str, str]:
        for path in sorted(corpus_dir.glob("*.md")):
            meta = read_front_matter(path)
            if meta.get("id") == args.doc_id:
                return {k: meta[k] for k in ("id", "title", "version", "status", "lang")}
        raise ToolError("not_found", "no document has that id")

    return Tool("lookup_document", "Look up a help-centre document by its id.", LookupArgs, handler)


def default_registry(corpus_dir: Path) -> ToolRegistry:
    registry = ToolRegistry()
    registry.register(calculator_tool)
    registry.register(make_lookup_tool(corpus_dir))
    return registry


# --- provider shapes -> one ToolCall (dict in, dataclass out) ----------------------------------
def _arguments(raw: Any) -> dict[str, Any]:
    if isinstance(raw, str):  # some APIs send the arguments as a JSON string
        try:
            raw = json.loads(raw)
        except json.JSONDecodeError:
            raise ToolError("malformed_call", "arguments are not valid JSON") from None
    if not isinstance(raw, dict):
        raise ToolError("malformed_call", "arguments are not an object")
    return raw


def from_openai(item: dict[str, Any]) -> ToolCall:
    return ToolCall(item["call_id"], item["name"], _arguments(item["arguments"]))


def from_anthropic(block: dict[str, Any]) -> ToolCall:
    return ToolCall(block["id"], block["name"], _arguments(block["input"]))


def from_gemini(step: dict[str, Any]) -> ToolCall:
    return ToolCall(step["id"], step["name"], _arguments(step["arguments"]))


def from_ollama(call: dict[str, Any], index: int = 0) -> ToolCall:
    fn = call["function"]  # this shape has no call id: make a stable one from the position
    return ToolCall(f"call_{index}", fn["name"], _arguments(fn["arguments"]))


# --- offline: a stand-in model and one step ----------------------------------------------------
class ScriptedToolModel:
    """Fake model: replays recorded requests. Real models are less predictable."""

    def __init__(self, recorded: list[dict[str, Any]]):
        self._recorded = list(recorded)

    def next_call(self, question: str) -> dict[str, Any] | None:
        return self._recorded.pop(0) if self._recorded else None


def run_one_step(model: ScriptedToolModel, registry: ToolRegistry, question: str) -> list[dict[str, Any]]:
    transcript: list[dict[str, Any]] = [{"role": "user", "content": question}]
    raw = model.next_call(question)
    if raw is None:
        return transcript
    transcript.append({"role": "assistant", "tool_request": raw})
    arguments = raw.get("arguments")
    call = ToolCall(str(raw.get("id", "")), str(raw.get("name", "")), arguments if isinstance(arguments, dict) else {})
    transcript.append(registry.execute(call).as_message())
    return transcript

Define a tool registry

Python

from pprint import pprint
from pathlib import Path
from evidencedesk.tools import ToolCall, default_registry

registry = default_registry(Path("kit/corpus"))
print(registry.names())
spec = registry.specs()[0]
print(spec["description"])
pprint(spec["parameters"], width=76)

Output

['calculator', 'lookup_document']
Evaluate an arithmetic expression with + - * / // % and parentheses.
{'additionalProperties': False,
 'properties': {'expression': {'maxLength': 100,
                               'minLength': 1,
                               'title': 'Expression',
                               'type': 'string'}},
 'required': ['expression'],
 'type': 'object'}

Notice additionalProperties: false in the schema. A schema helps the model choose well, but the model may ignore it, so your code validates again.

Create a read-only calculator tool

Python

import ast
from evidencedesk.tools import ToolError, safe_eval

tree = ast.parse("12 * (3 + 4)", mode="eval")
print([type(node).__name__ for node in ast.walk(tree)])
for text in ["12 * (3 + 4)", "7 // 2 + 0.5", "__import__('os').getcwd()", "2 ** 999999", "1 / 0", "1 +"]:
    try:
        print(f"{text!r:28} -> {safe_eval(text)}")
    except ToolError as exc:
        print(f"{text!r:28} -> refused: {exc.code} ({exc.message})")

Output

['Expression', 'BinOp', 'Constant', 'Mult', 'BinOp', 'Constant', 'Add', 'Constant']
'12 * (3 + 4)'               -> 84
'7 // 2 + 0.5'               -> 3.5
"__import__('os').getcwd()"  -> refused: bad_expression (not allowed: Call)
'2 ** 999999'                -> refused: bad_expression (not allowed: Pow)
'1 / 0'                      -> refused: division_by_zero (division by zero)
'1 +'                        -> refused: bad_expression (not a valid arithmetic expression)

Validate tool arguments with Pydantic

Python

def show(name, **arguments):
    result = registry.execute(ToolCall("c1", name, arguments))
    print(name, arguments)
    print("  ->", result.error["code"] if not result.ok else result.data)
    for problem in (result.error or {}).get("problems", []):
        print("    ", problem["loc"], "|", problem["msg"])

show("calculator")
show("calculator", expression="1+1", mode="fast")
show("calculator", expression=42)
show("calculator", expression="2*21")

Output

calculator {}
  -> invalid_arguments
     expression | Field required
calculator {'expression': '1+1', 'mode': 'fast'}
  -> invalid_arguments
     mode | Extra inputs are not permitted
calculator {'expression': 42}
  -> invalid_arguments
     expression | Input should be a valid string
calculator {'expression': '2*21'}
  -> {'value': 42}

The last line shows success. Valid arguments prove only that the shape is right, not that the request is wise.

Dispatch only registered tools

Timeouts and unexpected errors

Python

import threading
from evidencedesk.tools import Tool, ToolArgs

show("delete_everything", path="/")
show("lookup_document", doc_id="kc-warranty-v2")
show("lookup_document", doc_id="kc-nope")
show("lookup_document", doc_id="../../etc/passwd")

release = threading.Event()   # lets the slow handler stop when we are done


class NoArgs(ToolArgs):
    pass


registry.register(Tool("slow", "waits for a long time", NoArgs, lambda args: release.wait(5)))
timed_out = registry.execute(ToolCall("c2", "slow", {}), timeout=0.05)
release.set()
print("slow ->", timed_out.error)

Output

delete_everything {'path': '/'}
  -> unknown_tool
lookup_document {'doc_id': 'kc-warranty-v2'}
  -> {'id': 'kc-warranty-v2', 'title': 'Warranty', 'version': '2', 'status': 'current', 'lang': 'en'}
lookup_document {'doc_id': 'kc-nope'}
  -> not_found
lookup_document {'doc_id': '../../etc/passwd'}
  -> invalid_arguments
     doc_id | String should match pattern '^[a-z0-9-]{3,40}$'
slow -> {'code': 'timeout', 'message': 'tool did not finish within 0.05 s'}

Return tool results as explicit data

Python

ok = registry.execute(ToolCall("c3", "lookup_document", {"doc_id": "kc-returns-v2"}))
bad = registry.execute(ToolCall("c4", "calculator", {"expression": "2 ** 9"}))
import json
from pprint import pprint

for result in (ok, bad):
    pprint(json.loads(result.as_message()["content"]), width=76)

Output

{'call_id': 'c3',
 'data': {'id': 'kc-returns-v2',
          'lang': 'en',
          'status': 'superseded',
          'title': 'Returns policy (old)',
          'version': '2'},
 'name': 'lookup_document',
 'ok': True}
{'call_id': 'c4',
 'error': {'code': 'bad_expression', 'message': 'not allowed: Pow'},
 'name': 'calculator',
 'ok': False}

Implement a fake tool-calling client

Python

from pprint import pprint
from evidencedesk.tools import ScriptedToolModel, run_one_step

model = ScriptedToolModel([{"id": "c5", "name": "lookup_document", "arguments": {"doc_id": "kc-warranty-v2"}}])
for message in run_one_step(model, registry, "Which document is the warranty?"):
    pprint(message, width=76)
print(len(run_one_step(ScriptedToolModel([]), registry, "Hello")), "message when the model asks for nothing")

Output

{'content': 'Which document is the warranty?', 'role': 'user'}
{'role': 'assistant',
 'tool_request': {'arguments': {'doc_id': 'kc-warranty-v2'},
                  'id': 'c5',
                  'name': 'lookup_document'}}
{'content': '{"call_id": "c5", "data": {"id": "kc-warranty-v2", "lang": '
            '"en", "status": "current", "title": "Warranty", "version": '
            '"2"}, "name": "lookup_document", "ok": true}',
 'role': 'tool',
 'tool_call_id': 'c5'}
1 message when the model asks for nothing

The transcript has three messages: question, request and tool message. If the model asks for nothing, nothing runs.

Compare provider tool-calling formats

Provider note (as of 2026-10-12): OpenAI (Responses API) declares a flat tool {"type": "function", "name", "description", "parameters"}; the response items of type function_call carry call_id, name and arguments as a JSON string; the result returns as a function_call_output item. Anthropic declares input_schema; the response has stop_reason == "tool_use" and tool_use blocks with id, name and input (already an object); the result returns as a tool_result block with tool_use_id. Gemini has two paths, the Interactions API (steps of type function_call with name, arguments, id), which its documentation recommends, and the older generateContent path. Ollama returns message.tool_calls with function.name and a parsed function.arguments; tool support depends on the model. Some Anthropic models reject a forced tool_choice. The type of Gemini arguments was not stated in the pages we read, so _arguments accepts a string or an object. Check the current documentation.

Python

import json
from evidencedesk.tools import from_anthropic, from_gemini, from_ollama, from_openai

# Hand-written examples that follow the documented shapes. They are NOT real responses.
args = {"expression": "12 * (3 + 4)"}
openai_item = {"type": "function_call", "call_id": "call_1", "name": "calculator", "arguments": json.dumps(args)}
anthropic_block = {"type": "tool_use", "id": "toolu_1", "name": "calculator", "input": args}
gemini_step = {"type": "function_call", "id": "fc_1", "name": "calculator", "arguments": args}
ollama_call = {"function": {"name": "calculator", "arguments": args}}

for call in (from_openai(openai_item), from_anthropic(anthropic_block), from_gemini(gemini_step), from_ollama(ollama_call)):
    print(call)

Output

ToolCall(id='call_1', name='calculator', arguments={'expression': '12 * (3 + 4)'})
ToolCall(id='toolu_1', name='calculator', arguments={'expression': '12 * (3 + 4)'})
ToolCall(id='fc_1', name='calculator', arguments={'expression': '12 * (3 + 4)'})
ToolCall(id='call_0', name='calculator', arguments={'expression': '12 * (3 + 4)'})

Python

# not run here: needs a provider SDK, an API key and network access
# Responses-style call (OpenAI SDK)
specs = [{"type": "function", **spec} for spec in registry.specs()]
reply = client.responses.create(model=MODEL, tools=specs, input=messages)
messages += reply.output
for item in reply.output:
    if item.type == "function_call":
        result = registry.execute(from_openai({"call_id": item.call_id, "name": item.name, "arguments": item.arguments}))
        messages.append({"type": "function_call_output", "call_id": item.call_id, "output": result.as_message()["content"]})
# the Messages-style call of the other SDK differs only in shape: from_anthropic() maps its tool_use blocks

Exercise: a safe one-step tool call

Python

# file: kit/tests/test_tools.py
import json
import threading
from pathlib import Path

import pytest

from evidencedesk.tools import (ScriptedToolModel, Tool, ToolArgs, ToolCall, ToolError, ToolRegistry, default_registry,
                                from_anthropic, from_gemini, from_ollama, from_openai, run_one_step, safe_eval)

CORPUS = Path(__file__).resolve().parents[1] / "corpus"


@pytest.fixture
def registry():
    return default_registry(CORPUS)


def call(name, **arguments):
    return ToolCall("c1", name, arguments)


def test_calculator_ok(registry):
    result = registry.execute(call("calculator", expression="12 * (3 + 4) - 10 / 4"))
    assert result.ok and result.data == {"value": 81.5}


@pytest.mark.parametrize("bad", ["__import__('os')", "2 ** 3", "abs(-1)", "x + 1", "'a' * 3", "True + 1", "1 +"])
def test_safe_eval_refuses_everything_but_arithmetic(bad):
    with pytest.raises(ToolError):
        safe_eval(bad)


def test_missing_extra_and_invalid_arguments(registry):
    missing = registry.execute(call("calculator"))
    extra = registry.execute(call("calculator", expression="1+1", mode="fast"))
    invalid = registry.execute(call("calculator", expression=42))
    for r in (missing, extra, invalid):
        assert r.error["code"] == "invalid_arguments"
    assert missing.error["problems"][0]["loc"] == "expression"
    assert extra.error["problems"][0]["loc"] == "mode"
    assert all("url" not in p for p in invalid.error["problems"])


def test_unknown_tool_is_rejected(registry):
    result = registry.execute(call("delete_everything", path="/"))
    assert result.error["code"] == "unknown_tool"


def test_lookup_document(registry):
    ok = registry.execute(call("lookup_document", doc_id="kc-warranty-v2"))
    assert ok.data["title"] == "Warranty" and ok.data["status"] == "current"
    assert registry.execute(call("lookup_document", doc_id="kc-nope")).error["code"] == "not_found"
    assert registry.execute(call("lookup_document", doc_id="../../etc/passwd")).error["code"] == "invalid_arguments"


def test_timeout_and_unexpected_exception():
    release = threading.Event()

    class NoArgs(ToolArgs):
        pass

    def boom(_):
        raise RuntimeError("secret internal detail")

    registry = ToolRegistry()
    registry.register(Tool("slow", "waits", NoArgs, lambda _: release.wait(5)))
    registry.register(Tool("boom", "fails", NoArgs, boom))
    try:
        assert registry.execute(call("slow"), timeout=0.05).error["code"] == "timeout"
    finally:
        release.set()
    failed = registry.execute(call("boom"))
    assert failed.error == {"code": "internal_error", "message": "the tool failed"}






def test_provider_shapes_map_to_one_call():
    args = {"expression": "1+1"}
    calls = [
        from_openai({"type": "function_call", "call_id": "a", "name": "calculator", "arguments": json.dumps(args)}),
        from_anthropic({"type": "tool_use", "id": "a", "name": "calculator", "input": args}),
        from_gemini({"type": "function_call", "id": "a", "name": "calculator", "arguments": args}),
        from_ollama({"function": {"name": "calculator", "arguments": args}}),
    ]
    assert {(c.name, json.dumps(c.arguments)) for c in calls} == {("calculator", json.dumps(args))}
    with pytest.raises(ToolError):
        from_openai({"call_id": "a", "name": "calculator", "arguments": "{not json"})


def test_safe_one_step(registry):
    model = ScriptedToolModel([{"id": "c9", "name": "lookup_document", "arguments": {"doc_id": "kc-returns-v2"}}])
    transcript = run_one_step(model, registry, "Which returns document is old?")
    body = json.loads(transcript[-1]["content"])
    assert body["ok"] and body["data"]["status"] == "superseded"
    assert len(transcript) == 3

Bash

python -m pytest -q kit/tests/test_tools.py | sed 's/ in [0-9.]*s//'

Output

..............                                                           [100%]
14 passed

Where this shows up at work

Common mistakes

Using eval on model text. It runs any Python, and the text may come from an attacker. Parse with ast and allow only the nodes you need.
Trusting the schema you showed the model. The model may ignore it. Validate in your own code.
Returning raw exceptions or Pydantic errors. They can leak paths, values and links. Return a code and loc/msg only.
Dispatching by getattr or globals(). Then the model's text chooses which function runs. Use an explicit registry.
Thinking a timeout stops the work. It stops the waiting; keep early tools read-only.

Try it yourself

Easy. Predict the result of the calculator for -5 + 10 % 4, then run it through the registry.
Medium. Revisit Reliable Output Contracts: write a Pydantic model Envelope with a validator that requires data when ok is true and an error when it is false. Check one good and one bad envelope.
Hard. Add a third read-only tool, count_documents, whose only argument is status, either "current" or "superseded". Register it and show that "draft" is rejected.

Answers

Easy. The modulo binds tighter than addition: 10 % 4 is 2, and -5 + 2 is -3.

Python

result = registry.execute(ToolCall("e1", "calculator", {"expression": "-5 + 10 % 4"}))
print(result.ok, result.data)

Output

True {'value': -3}

Medium. The validator checks a business rule that the types alone cannot express.

Python

from pydantic import BaseModel, ValidationError, model_validator


class Envelope(BaseModel):
    call_id: str
    name: str
    ok: bool
    data: object = None
    error: dict | None = None

    @model_validator(mode="after")
    def data_xor_error(self):
        broken = (self.data is None or self.error is not None) if self.ok else self.error is None
        if broken:
            raise ValueError("an ok result needs data and no error; a failure needs an error")
        return self


print(Envelope(**ok.as_dict()).ok, Envelope(**bad.as_dict()).ok)
try:
    Envelope(call_id="x", name="calculator", ok=True, error={"code": "x"})
except ValidationError as exc:
    print(exc.errors()[0]["msg"])

Output

True False
Value error, an ok result needs data and no error; a failure needs an error

Hard. The handler reuses read_front_matter. The Literal type makes Pydantic reject every other status before the handler runs.

Python

from pathlib import Path
from typing import Literal
from evidencedesk.tools import read_front_matter


class CountArgs(ToolArgs):
    status: Literal["current", "superseded"]


def count_documents(args):
    metas = [read_front_matter(p) for p in Path("kit/corpus").glob("*.md")]
    return {"status": args.status, "count": sum(m.get("status") == args.status for m in metas)}


registry.register(Tool("count_documents", "Count documents with a given status.", CountArgs, count_documents))
for status in ("current", "superseded", "draft"):
    result = registry.execute(ToolCall("h1", "count_documents", {"status": status}))
    print(status, "->", result.data if result.ok else result.error["problems"])

Output

current -> {'status': 'current', 'count': 12}
superseded -> {'status': 'superseded', 'count': 1}
draft -> [{'loc': 'status', 'msg': "Input should be 'current' or 'superseded'"}]

Pass checklist:

Easy: the printed value is -3 and ok is true.
Medium: the good envelopes validate, and the bad one raises a validation error with your message.
Hard: the counts are 12 and 1, and "draft" produces an invalid_arguments problem with loc status.
No answer uses eval, prints a Pydantic URL or reads outside kit/corpus; any new tool has a pytest case.

Summary

A model proposes a tool call; your code validates and executes it, or refuses.
A registry of name, description, argument model and handler is the complete list of reachable actions.
Arguments are untrusted: validate with Pydantic v2 and extra="forbid", report only loc and msg, and reject unknown names.
Do arithmetic with ast, never eval. Bound every call with a timeout and return every outcome as an explicit envelope.
Map provider formats into one ToolCall and test the loop with a scripted fake client.

Next: Tool Permissions and Failure Handling adds the rules around execution: which tools a task may use, confirmation for consequential actions, call limits and audit events.