After three projects, you will notice the loop is identical: append assistant content, branch on stop_reason, dispatch tool_use blocks (parallel where possible), append tool_results, repeat. Extract this into a Tool Runner — a class or function that takes (tools, handlers, max_iters) and runs the loop.
What a Tool Runner owns
It owns: tool registry, dispatcher (with concurrency), iteration limit, error handling, observability hooks. It does not own: tool definitions or handlers (those live with the domain code), the prompt (caller's job), or the model selection (caller's job).
cwkPippa-shaped Tool Runner
cwkPippa's Claude variant uses the Agent SDK's built-in tool support, but the ChatGPT and Gemini variants implement a Tool Runner pattern. Same registry, same dispatch shape, just different upstream APIs. The pattern survives moving between providers.
Principle: Extract the loop after the third tool. Earlier is premature; later is rewriting the same code.
Code
Minimal Tool Runner·python
import asyncio, json
from dataclasses import dataclass, field
from typing import Callable, Awaitable
@dataclass
class ToolRunner:
client: AsyncAnthropic
model: str
tools: list[dict]
handlers: dict[str, Callable[..., Awaitable[dict]]]
max_iters: int = 10
on_tool_call: Callable[[str, dict], None] | None = None
async def _dispatch(self, blocks):
async def one(b):
if self.on_tool_call:
self.on_tool_call(b.name, b.input)
try:
out = await self.handlers[b.name](**b.input)
return {"type": "tool_result", "tool_use_id": b.id, "content": json.dumps(out)}
except Exception as e:
return {"type": "tool_result", "tool_use_id": b.id, "is_error": True, "content": str(e)}
return await asyncio.gather(*(one(b) for b in blocks))
async def run(self, system: str, user: str) -> str:
messages = [{"role": "user", "content": user}]
for _ in range(self.max_iters):
resp = await self.client.messages.create(
model=self.model, max_tokens=2048, system=system,
tools=self.tools, messages=messages,
)
messages.append({"role": "assistant", "content": resp.content})
if resp.stop_reason != "tool_use":
return next(b.text for b in resp.content if b.type == "text")
blocks = [b for b in resp.content if b.type == "tool_use"]
messages.append({"role": "user", "content": await self._dispatch(blocks)})
raise RuntimeError("max_iters reached")
Extract the tool loop in your project into a Tool Runner. Add three observability hooks: pre-call, post-call, and on-error. Wire them to your existing logging — no new dependencies.
Hint
If extracting the runner is hard, your handlers depend on the loop's state; that is the bug to fix first.
Progress
Progress is local-only — sign in to sync across devices.