The Claude tool use pattern I copy-paste into every AI project
← Back
April 2, 2026Claude7 min read

The Claude tool use pattern I copy-paste into every AI project

Published April 2, 20267 min read

I have built eight different projects on top of Claude's API this year. Each time, I rewrote the same tool-calling scaffolding from scratch — until I extracted it into a reusable class on project five. The pattern I landed on handles structured output, tool dispatch, retries on malformed responses, and multi-turn conversation state. I now copy it into every new project on day one. 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 because:

  • Claude sometimes returns a partial tool call if the response is cut off by max_tokens
  • Tool inputs occasionally fail your own validation (wrong type, missing field)
  • Multi-turn conversations require threading messages correctly — get it wrong and Claude loses context
  • You need to handle the case where Claude calls a tool you did not define (happens with aggressive system prompts)

Solving each of these ad-hoc across eight projects taught me the exact shape of a robust abstraction.

The ToolRunner class

python
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:

python
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:

python
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:

python
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 adding a dependency on a backoff library.

Tool-level errors (your handler raises an exception) are caught in _dispatch and returned as {"error": "..."} JSON to Claude. This is intentional — it lets Claude recover gracefully ("I tried to look up the user but got an error. Let me try a different approach.") rather than 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 always return structured JSON — no parsing of free text required.

python
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 do not include

The class deliberately omits streaming, parallel tool calls, and token counting. Not because those are unimportant — but because they require project-specific decisions. Adding them to the base class adds complexity that not every project needs. 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 were 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 flexible enough to handle every production use case I have hit so far.

Copy it, adapt it to your stack, and spend those two hours on something that matters.

Share this
← All Posts7 min read