Tying it all together in a single loop with Claude Tool Use — covering stop_reason branching, parallel calls, and turning errors into data

Imagine a scenario where you need to fetch competitor prices from three sources simultaneously while also collecting related news. The traditional approach requires you to hand-code API calls → result parsing → aggregation → next-step decisions. By the time you handle branching logic, partial failure recovery, and retry policies, you end up with far more infrastructure code than business logic.

Claude API's Tool Use approaches this problem differently. You simply hand Claude the tool definitions, and Claude decides on its own which tools to call, when, and in what order. The client executes those decisions and returns the results. And because Claude can request multiple independent tools in a single response simultaneously, you can process several external APIs in parallel within a single iteration (official docs).

This article covers the actual structure of the agentic loop, implementing parallel tool calls, the pattern of treating errors as data, and the tradeoffs you encounter in production. We'll build the loop from scratch using a competitor price monitoring pipeline as our example.

The Agentic Loop Skeleton — stop_reason Drives Every Branch

When I first used Tool Use, the most confusing part was "how do you end the loop." The answer lies in the stop_reason field in Claude's response. The actual values in the Messages API are five: end_turn, max_tokens, stop_sequence, tool_use, and pause_turn.

When Claude returns stop_reason: "tool_use", execute the tools, return the results, then call again. When stop_reason: "end_turn" arrives, it signals that Claude has completed its final response, so you stop the loop.

There is one important rule here. Do not rebuild the message history from scratch each time — keep appending to the cumulative array. Append the entire assistant response including the tool_use blocks, then immediately insert a user turn containing the tool_results. Without this cumulative structure, Claude loses the context of previous tool calls.

The basic skeleton in code looks like this:

import anthropic
 
client = anthropic.Anthropic()
MAX_ITERATIONS = 25  # Example value. Adjust the actual ceiling to fit your pipeline's characteristics.
 
def run_agentic_loop(initial_prompt: str, tools: list) -> str:
    messages = [{"role": "user", "content": initial_prompt}]
 
    for iteration in range(MAX_ITERATIONS):
        response = client.messages.create(
            model="claude-sonnet-4-5",
            max_tokens=4096,
            tools=tools,
            messages=messages,
        )
 
        messages.append({"role": "assistant", "content": response.content})
 
        if response.stop_reason == "end_turn":
            for block in response.content:
                if block.type == "text":
                    return block.text
            return ""
 
        elif response.stop_reason == "tool_use":
            tool_use_blocks = [b for b in response.content if b.type == "tool_use"]
            # execute_tools_parallel is defined in the next section
            tool_results = asyncio.run(execute_tools_parallel(tool_use_blocks))
            messages.append({"role": "user", "content": tool_results})
 
        else:
            # max_tokens, stop_sequence, pause_turn, etc.
            raise RuntimeError(f"Unexpected stop_reason: {response.stop_reason}")
 
    raise RuntimeError(f"Max iterations ({MAX_ITERATIONS}) reached")

Attempting to judge stop_reason from natural language (e.g., parsing whether the response text contains "done") is unreliable. Always branch on this field.

Parallel Tool Calls — Execute N Tools Simultaneously in One Iteration

Claude can return multiple independent tools at once as an array of tool_use blocks in a single response. When the client executes them in parallel and sends the results back in a single user message, tools that would have taken multiple iterations in sequential calling are compressed into one round. Note that this feature is supported starting from the Claude 3 family, including Claude 3.5 Sonnet (official docs). To be clear: the Claude API round-trips themselves don't disappear — the tool calls and result delivery are consolidated into a single round.

Each tool_result is linked to its corresponding call via tool_use_id. Claude performs its synthesized reasoning after receiving all results. As shown above, a minimum of two Claude API round-trips is required.

The parallel execution code is handled with asyncio.gather. Here we unify the approach by catching exceptions inside each individual tool function and converting them into is_error.

import asyncio
 
async def execute_single_tool(block) -> dict:
    try:
        result_content = await dispatch_tool(block.name, block.input)
        return {
            "type": "tool_result",
            "tool_use_id": block.id,
            "content": result_content,
        }
    except Exception as e:
        return {
            "type": "tool_result",
            "tool_use_id": block.id,
            "is_error": True,
            "content": f"Tool execution failed: {type(e).__name__}: {str(e)}",
        }
 
async def execute_tools_parallel(tool_use_blocks: list) -> list:
    tasks = [execute_single_tool(block) for block in tool_use_blocks]
    return await asyncio.gather(*tasks)

Since execute_single_tool already absorbs all exceptions internally, return_exceptions=True and subsequent isinstance(result, Exception) checks are effectively dead code. Keeping exception handling in a single place makes the code easier to read and reduces mistakes.

Real-World Code — Competitor Price Monitoring Pipeline

Now let's assemble an actual pipeline from tool definitions to the full loop. The scenario is collecting data from three sources simultaneously (price API, news search, market statistics) to produce a comprehensive analysis.

import anthropic
import asyncio
import json
 
client = anthropic.Anthropic()
 
TOOLS = [
    {
        "name": "fetch_price",
        "description": "Fetches the price of a specific product from a competitor",
        "input_schema": {
            "type": "object",
            "properties": {
                "competitor": {
                    "type": "string",
                    "description": "Competitor identifier (e.g., alpha, beta, gamma)",
                },
                "product_id": {
                    "type": "string",
                    "description": "Product ID to query",
                },
            },
            "required": ["competitor", "product_id"],
        },
    },
    {
        "name": "search_news",
        "description": "Searches for recent news articles by keyword",
        "input_schema": {
            "type": "object",
            "properties": {
                "query": {"type": "string", "description": "Search keyword"},
                "max_results": {
                    "type": "integer",
                    "description": "Maximum number of results to return",
                    "default": 5,
                },
            },
            "required": ["query"],
        },
    },
    {
        "name": "get_market_data",
        "description": "Fetches market statistics and trend data by category",
        "input_schema": {
            "type": "object",
            "properties": {
                "category": {"type": "string", "description": "Product category"},
            },
            "required": ["category"],
        },
    },
]
 
 
# Conceptual example. In a real environment, call external APIs with httpx.AsyncClient or similar.
async def dispatch_tool(name: str, inputs: dict) -> str:
    if name == "fetch_price":
        await asyncio.sleep(0.3)
        return json.dumps({"competitor": inputs["competitor"], "price": 29900, "currency": "KRW"})
 
    elif name == "search_news":
        await asyncio.sleep(0.5)
        return json.dumps({"articles": [{"title": f"Latest news related to {inputs['query']}", "sentiment": "neutral"}]})
 
    elif name == "get_market_data":
        await asyncio.sleep(0.4)
        return json.dumps({"category": inputs["category"], "market_size": "1.2 trillion KRW", "growth_rate": "8.3%"})
 
    raise ValueError(f"Unknown tool: {name}")
 
 
async def execute_single_tool(block) -> dict:
    try:
        content = await dispatch_tool(block.name, block.input)
        return {"type": "tool_result", "tool_use_id": block.id, "content": content}
    except Exception as e:
        return {
            "type": "tool_result",
            "tool_use_id": block.id,
            "is_error": True,
            "content": f"{type(e).__name__}: {str(e)}",
        }
 
 
async def execute_tools_parallel(tool_use_blocks: list) -> list:
    return await asyncio.gather(*(execute_single_tool(b) for b in tool_use_blocks))
 
 
SYSTEM_PROMPT = (
    "You are a competitive landscape analysis agent. Call multiple tools in parallel to collect data.\n"
    "If a tool fails with is_error: (1) if it looks like a parameter error, fix and retry at most once, "
    "(2) if it looks like an external API outage, determine whether an alternative source is available, "
    "(3) if recovery is difficult, explicitly note the missing data and complete the analysis with what remains. "
    "Infinite retries are prohibited."
)
 
 
async def run_pipeline(product_id: str, category: str) -> str:
    prompt = (
        f"For product ID '{product_id}', query prices from 3 competitors (alpha, beta, gamma), "
        f"collect the latest news and market data for the '{category}' category, "
        "then provide a comprehensive competitive landscape analysis."
    )
 
    messages = [{"role": "user", "content": prompt}]
    MAX_ITERATIONS = 25
 
    for iteration in range(MAX_ITERATIONS):
        response = client.messages.create(
            model="claude-sonnet-4-5",
            max_tokens=4096,
            system=SYSTEM_PROMPT,
            tools=TOOLS,
            messages=messages,
        )
 
        messages.append({"role": "assistant", "content": response.content})
 
        if response.stop_reason == "end_turn":
            for block in response.content:
                if block.type == "text":
                    return block.text
            return ""
 
        elif response.stop_reason == "tool_use":
            tool_use_blocks = [b for b in response.content if b.type == "tool_use"]
            print(f"[{iteration+1}] Claude requested parallel execution of {len(tool_use_blocks)} tools")
 
            tool_results = await execute_tools_parallel(tool_use_blocks)
            messages.append({"role": "user", "content": tool_results})
 
        else:
            raise RuntimeError(f"Unexpected stop_reason: {response.stop_reason}")
 
    raise RuntimeError(f"Max iterations ({MAX_ITERATIONS}) exceeded")
 
 
if __name__ == "__main__":
    result = asyncio.run(run_pipeline(product_id="PROD-001", category="consumer electronics"))
    print(result)

One point worth noting: the entire loop is defined as async def, with asyncio.run() called only once at the top level. Calling asyncio.run() on every iteration — as in an early draft — crashes with RuntimeError: This event loop is already running in environments that already have an event loop running, such as FastAPI handlers or Jupyter notebooks. If you plan to integrate into such environments, either await run_pipeline(...) directly, or if you must call it from a synchronous context, consider a workaround like nest_asyncio.apply().

Treating Errors as Data, Not Exceptions

In an agentic pipeline, handling tool failures as Python exceptions kills the entire loop. Instead, package the is_error: true flag and the error message into the tool_result and return it to Claude, allowing Claude to use it as context to decide its next action.

There is an important practical tip here. Passing an error to Claude does not mean Claude will automatically retry or devise an alternative strategy. Without recovery instructions in the system prompt, it commonly just summarizes the error and ends with end_turn. You need to explicitly state rules like "on failure, fix parameters and retry at most once," "if it's an external API outage, evaluate alternative sources," and "if recovery is difficult, note the missing data and complete the analysis with the rest" — as in the SYSTEM_PROMPT above — to reliably reproduce the recovery flow you expect.

The error return pattern itself is straightforward:

import httpx
 
# Wrong approach — kills the entire loop
async def bad_tool_handler(block):
    result = await call_external_api(block.input)  # raises exception on failure
    return result
 
# Correct approach — passes the error as data to Claude
async def good_tool_handler(block) -> dict:
    try:
        content = await call_external_api(block.input)
        return {
            "type": "tool_result",
            "tool_use_id": block.id,
            "content": content,
        }
    except httpx.TimeoutException:
        return {
            "type": "tool_result",
            "tool_use_id": block.id,
            "is_error": True,
            "content": (
                f"API timeout for {block.name} with input {block.input}. "
                "Last successful call was over 2 hours ago. "
                "Consider retrying once or noting the missing data."
            ),
        }
    except Exception as e:
        return {
            "type": "tool_result",
            "tool_use_id": block.id,
            "is_error": True,
            "content": f"Unexpected error: {type(e).__name__}: {str(e)}",
        }

Write error messages with enough specificity for Claude to make a judgment. timeout after 5s, last successful call was 2h ago is far more useful for deciding the next action than error occurred.

Tradeoffs You Encounter in Production

Honestly, agentic pipelines are trickier than they appear. The concept is simple, but production environments add several layers of complexity.

Pros and Cons at a Glance

The Individual Accuracy Trap

Even if a single agent has 95% accuracy, five stages collaborating in series — under the simple assumption that each stage's success is independent — drops the overall success rate to roughly 77% (0.95⁵). In practice, coordination overhead and context loss compound this, making it worse. The more complex the pipeline, the higher the reliability bar you need to set for each stage.

Things to Watch in Production

Set an iteration ceiling on the loop: I personally use 10 iterations for conversational flows and 25 for batch jobs as my ceiling, but these are values from my own experience, not industry standards. The right approach is to instrument the average number of iterations in your pipeline, then set a safety margin above that. Whatever value you choose, always set some ceiling — an infinite loop causes API costs to grow exponentially.

Per-tool timeouts: Set per-tool timeouts so that one slow tool doesn't block the entire asyncio.gather.

async def execute_single_tool_with_timeout(block, timeout_seconds: float = 10.0) -> dict:
    try:
        content = await asyncio.wait_for(
            dispatch_tool(block.name, block.input),
            timeout=timeout_seconds,
        )
        return {"type": "tool_result", "tool_use_id": block.id, "content": content}
    except asyncio.TimeoutError:
        return {
            "type": "tool_result",
            "tool_use_id": block.id,
            "is_error": True,
            "content": f"Tool '{block.name}' timed out after {timeout_seconds}s",
        }

Prompt caching for tool definitions: Since tool definitions are included in the context of every request, applying Anthropic's prompt caching can reduce the cost of repeated calls. This is a key cost optimization strategy for large-scale pipelines.

Observability: Instrument average iterations per session, per-tool latency, and is_error frequency. A sudden increase in average iterations can signal a tool schema change or a downstream API change. Patterns integrating LLM-specific monitoring tools like LangSmith and Langfuse are becoming established practice.

Recently Added Features (as of 2026)

Once you have a solid understanding of the basic loop, it becomes easier to see where recently added features help. The beta/general availability status of the items below may change over time, so verify the current status and exact field names in the Anthropic official docs before use.

Programmatic Tool Calling: An approach where Claude writes and executes code in a single inference pass to orchestrate multiple tools without individual API round-trips. It can filter and aggregate multiple tool results in code, then load only the necessary portions back into Claude's context to reduce token usage. Since field names required for activation vary by release, it's safest to reference the official documentation schema directly.

Tool Search: Instead of preloading hundreds of tool definitions into context, agents search for and discover tools on demand. This is a direct optimization for pipelines where a large number of tools puts pressure on the context window.

Claude Agent SDK: The official library, renamed from the Claude Code SDK. Its async design allows multiple agent conversations to run concurrently, and it unifies file reading, shell commands, web search, and MCP server calls under a single interface.

Temporal + LangGraph combination: The pattern of combining Temporal (workflow durability, retries, state persistence) with LangGraph (LLM logic, tool calls, memory) is becoming established in production systems. Claude API is integrated as LangGraph's LLM backend. A directly implemented loop is sufficient for simple pipelines, but for long-running workflows or cases requiring checkpoints, this combination is worth considering.

Closing Thoughts

Having built pipelines with Claude Tool Use many times, the biggest thing I've come to feel is that the reliability of an agentic loop ultimately comes down not to the tools, but to how precisely you've instructed what to do next when something goes wrong. The skeleton rules — stop_reason branching, parallel execution, is_error returns — can be learned in a day. But without recovery instructions in the system prompt, Claude will politely summarize the error and give up; and without an iteration ceiling, your API bill will pay the price.

When setting up a new pipeline, I spend far more time on the failure-response clauses in the system prompt, the iteration ceiling, and is_error frequency instrumentation than on the while loop skeleton itself. Explicitly specifying in plain language how you want existing tools to behave when they fail has contributed far more to production stability than attaching a few new tools.

References

• Tool use with Claude — Official Overview
• How tool use works
• Parallel tool use
• Handling stop_reason
• Prompt caching
• Claude Agent SDK