From raw OpenAI or Anthropic tool calling

You wrote the JSON schema by hand, kept a name -> function dict, and called fn(**json.loads(arguments)). Replace all three with a registry and an executor.

flowchart LR
  subgraph Before
    S1["hand-written schema"] --> D1["DISPATCH[name](**args)"]
    D1 --> X1["KeyError / TypeError<br/>crash the request"]
  end
  subgraph After
    S2["@registry.tool<br/>schema from the signature"] --> E2["executor.run_call"]
    E2 --> R2["ToolResult: value or a safe error"]
  end

Mapping

Before	After
a tools=[{...}] list written by hand	chat_tools(registry) / responses_tools(registry) / anthropic_tools(registry)
DISPATCH[name](**json.loads(args))	await executor.run_call(ToolCall(name, json.loads(args), id))
try: ... except Exception: return str(e)	result.error: a stable kind and a message safe for the model
your own retry loop	retries=, timeout= on the tool
the agent loop you wrote	run_tool_loop(model, registry, messages) with OpenAIModel / AnthropicModel

Before

def exchange_rate(currency: str) -> float:
    return {"USD": 0.92, "GBP": 1.17}[currency]


TOOLS = [  # written by hand, kept in step with the function by hand
    {
        "type": "function",
        "function": {
            "name": "exchange_rate",
            "description": "Today's EUR rate for a currency.",
            "parameters": {
                "type": "object",
                "properties": {"currency": {"type": "string"}},
                "required": ["currency"],
            },
        },
    }
]
DISPATCH = {"exchange_rate": exchange_rate}


def answer_tool_call_before(name: str, arguments: str) -> str:
    return json.dumps(DISPATCH[name](**json.loads(arguments)))  # KeyError, TypeError leak out

After

registry = Registry()


@registry.tool(timeout=5, retries=1)
def rate(currency: str) -> float:
    """Today's EUR rate for a currency."""
    return {"USD": 0.92, "GBP": 1.17}[currency]


executor = Executor(registry)


async def answer_tool_call(call_id: str, name: str, arguments: str) -> dict[str, Any]:
    """One OpenAI tool call in, one `role="tool"` message out; failures become text."""
    result = await executor.run_call(
        ToolCall(name=name, arguments=json.loads(arguments), id=call_id)
    )
    content = (
        json.dumps(result.value)
        if result.ok
        else json.dumps(result.error.model_dump() if result.error else {})
    )
    return {"role": "tool", "tool_call_id": call_id, "content": content}

print(chat_tools(registry)[0]["function"]["parameters"]["required"])  # generated
print(await answer_tool_call("c1", "rate", '{"currency": "USD"}'))
bad = await answer_tool_call("c2", "rate", '{"currency": 7}')
print(json.loads(bad["content"])["kind"])  # the model reads the error and retries
print([tool["name"] for tool in anthropic_tools(registry)])  # the same tool, Anthropic shape
blocks = tool_result_blocks(
    [await executor.run_call(ToolCall(name="rate", arguments={"currency": "GBP"}, id="tu_1"))]
)
print(blocks[0]["type"], blocks[0]["tool_use_id"])

Or let the adapter run the whole loop:

replies: list[Any] = [
    {"tool_calls": [ToolCall(name="rate", arguments={"currency": "USD"}, id="c3")]},
    "1 USD is 0.92 EUR.",
]

async def fake_model(request: LLMRequest) -> Any:  # in production: OpenAIModel(...)
    return replies.pop(0)

question = (Message(role="user", content="USD in EUR?"),)
done = await run_tool_loop(CallableModel(fake_model), registry, question, executor=executor)
print(done.finish, done.final.text)

Output

0.92
['currency']
{'role': 'tool', 'tool_call_id': 'c1', 'content': '0.92'}
tool-args-invalid
['rate']
tool_result tu_1
stop 1 USD is 0.92 EUR.

Gained and lost

Gained	Lost
one schema source: the signature and docstring	hand-tuned schema wording (use description=, Annotated and docstrings instead)
validation before the tool runs; errors the model can act on	nothing: json.loads stays in your code
approval, budgets, caching, ledger and events on every call
the same tools on Gemini, MCP and 18 frameworks

Checklist

• [ ] Every tool has a docstring and type hints (else ToolDefinitionError at start-up).
• [ ] One Executor, built at start-up.
• [ ] Each tool call becomes run_call; its ToolResult becomes the tool message.
• [ ] Risky tools have confirm= and the executor has an approval port.
• [ ] The old TOOLS list and DISPATCH dict are deleted.
• [ ] No spec.invoke outside tests.

Files

File	Does
examples/guide_migrate_raw.py	the old code, the new code, both renderings and a loop
tests/docs/test_migration.py	runs it offline and checks the output