The Claude tool use pattern I copy-paste into every AI project
I've built eight different projects on top of Claude's API this year. Each time, I rewrote the same tool-calling scaffolding from scratch. On project five I finally pulled it out into a reusable class.
The pattern handles structured output, tool dispatch, retries on malformed responses, and multi-turn conversation state. I copy it into every new project on day one now. Here it is.
The problem with naive tool use
The Anthropic docs show the simplest possible tool use example: define a tool, call the API, handle tool_use blocks in the response. That works for demos. In production it falls apart:
- Claude sometimes returns a partial tool call when the response is cut off by
max_tokens - Tool inputs occasionally fail your own validation (wrong type, missing field)
- Multi-turn conversations need messages threaded correctly. Get it wrong and Claude loses context
- You need to handle Claude calling a tool you never defined (happens with aggressive system prompts, I still don't fully understand why)
Solving each of these ad-hoc across eight projects is how I landed on the shape below.
The ToolRunner class
import json
import time
import logging
from typing import Any, Callable
from dataclasses import dataclass, field
import anthropic
logger = logging.getLogger(__name__)
@dataclass
class Tool:
name: str
description: str
input_schema: dict
handler: Callable[[dict], Any]
@dataclass
class ToolRunner:
client: anthropic.Anthropic
model: str = "claude-opus-4-5"
max_tokens: int = 4096
max_retries: int = 3
retry_delay: float = 1.0
tools: list[Tool] = field(default_factory=list)
messages: list[dict] = field(default_factory=list)
system: str = ""
def register(self, tool: Tool) -> "ToolRunner":
"""Register a tool. Returns self for chaining."""
self.tools.append(tool)
return self
def _tool_definitions(self) -> list[dict]:
return [
{
"name": t.name,
"description": t.description,
"input_schema": t.input_schema,
}
for t in self.tools
]
def _find_tool(self, name: str) -> Tool | None:
return next((t for t in self.tools if t.name == name), None)
def run(self, user_message: str) -> str:
"""Run a single turn. Returns Claude's final text response."""
self.messages.append({"role": "user", "content": user_message})
return self._agentic_loop()
def _agentic_loop(self) -> str:
for attempt in range(self.max_retries):
try:
response = self.client.messages.create(
model=self.model,
max_tokens=self.max_tokens,
system=self.system,
tools=self._tool_definitions(),
messages=self.messages,
)
except anthropic.APIError as e:
if attempt == self.max_retries - 1:
raise
logger.warning("API error (attempt %d): %s", attempt + 1, e)
time.sleep(self.retry_delay * (attempt + 1))
continue
# Append assistant turn
self.messages.append({"role": "assistant", "content": response.content})
if response.stop_reason == "end_turn":
# Extract final text block
for block in response.content:
if block.type == "text":
return block.text
return ""
if response.stop_reason == "tool_use":
tool_results = []
for block in response.content:
if block.type != "tool_use":
continue
result = self._dispatch(block.name, block.input)
tool_results.append({
"type": "tool_result",
"tool_use_id": block.id,
"content": json.dumps(result),
})
self.messages.append({"role": "user", "content": tool_results})
# Loop continues — Claude will respond to tool results
elif response.stop_reason == "max_tokens":
logger.warning("Hit max_tokens, response may be truncated")
# Still try to extract text if there is any
for block in response.content:
if block.type == "text":
return block.text
return ""
raise RuntimeError(f"Agentic loop did not terminate after {self.max_retries} attempts")
def _dispatch(self, name: str, inputs: dict) -> Any:
tool = self._find_tool(name)
if not tool:
logger.error("Claude called unknown tool: %s", name)
return {"error": f"Unknown tool: {name}"}
try:
return tool.handler(inputs)
except Exception as e:
logger.exception("Tool %s raised: %s", name, e)
return {"error": str(e)}
Registering and running tools
Here is the pattern for wiring it up with real tools:
import anthropic
from tool_runner import Tool, ToolRunner
def get_user_from_db(inputs: dict) -> dict:
user_id = inputs["user_id"]
# real DB call here
return {"id": user_id, "name": "Alice", "plan": "pro"}
def send_email(inputs: dict) -> dict:
to = inputs["to"]
subject = inputs["subject"]
body = inputs["body"]
# real email call here
return {"sent": True, "message_id": "msg_123"}
runner = ToolRunner(
client=anthropic.Anthropic(),
model="claude-opus-4-5",
system="You are a customer support agent. Use tools to look up users and send emails.",
)
runner.register(Tool(
name="get_user",
description="Look up a user by their ID",
input_schema={
"type": "object",
"properties": {
"user_id": {"type": "string", "description": "The user's UUID"},
},
"required": ["user_id"],
},
handler=get_user_from_db,
))
runner.register(Tool(
name="send_email",
description="Send an email to a user",
input_schema={
"type": "object",
"properties": {
"to": {"type": "string"},
"subject": {"type": "string"},
"body": {"type": "string"},
},
"required": ["to", "subject", "body"],
},
handler=send_email,
))
result = runner.run("Look up user abc-123 and send them a renewal reminder email")
print(result)
The multi-turn pattern
The messages list on the dataclass persists across calls to run(). That means follow-up turns just work:
runner = ToolRunner(client=client, system="You are a helpful assistant.")
runner.register(search_tool)
# First turn
response1 = runner.run("What is the weather in Mumbai today?")
# Second turn — Claude remembers the first turn
response2 = runner.run("What about tomorrow?")
If you want a stateless runner (fresh conversation each call), just clear the messages between runs:
runner.messages.clear()
What the retry logic actually handles
The retry in _agentic_loop handles API-level errors: rate limits, transient 5xx, network timeouts. The retry_delay * (attempt + 1) gives you linear backoff without pulling in a backoff library.
Tool-level errors (your handler raises an exception) get caught in _dispatch and returned as {"error": "..."} JSON to Claude. That's intentional. It lets Claude recover ("I tried to look up the user but got an error, let me try a different approach") instead of crashing the loop.
Structured output via tools
One underused pattern: use a dummy tool to force structured output. Define a tool with your desired output schema, tell Claude "you must call extract_result when done", and Claude will return structured JSON. No parsing of free text required.
runner.register(Tool(
name="extract_result",
description="Call this with your final structured answer",
input_schema={
"type": "object",
"properties": {
"sentiment": {"type": "string", "enum": ["positive", "negative", "neutral"]},
"confidence": {"type": "number"},
"summary": {"type": "string"},
},
"required": ["sentiment", "confidence", "summary"],
},
handler=lambda inputs: inputs, # just return the inputs
))
runner.system = "Analyse the sentiment of the review. You MUST call extract_result with your answer."
result = runner.run("The product is fine but the shipping was terrible.")
# result is a JSON string — parse it
import json
structured = json.loads(result) # or inspect runner.messages[-2] for the tool call directly
What I deliberately left out
The class skips streaming, parallel tool calls, and token counting. Those all need project-specific decisions, and baking them into the base class adds complexity most projects don't need. When a project needs streaming I subclass ToolRunner and override _agentic_loop.
Two hours saved, every project
Before extracting this pattern, the first two hours of any Claude project went to debugging message threading, handling edge cases in tool dispatch, and wiring up retries. Now those two hours go to building the actual product. The class is small enough to read in five minutes and has covered every production use case I've hit so far.
Copy it, adapt it to your stack, move on.