Implementation:Anthropics Anthropic sdk python BetaToolRunner

Knowledge Sources	Anthropic Python SDK Anthropic Tool Use Docs
Domains	Tool_Use, LLM, Function_Calling
Last Updated	2026-02-15 00:00 GMT

Overview

BetaToolRunner implements an automated agent loop that orchestrates the full tool use cycle: sending API requests, detecting tool calls, executing tool functions, submitting results, and repeating until the model produces a final response. It is an iterator that yields a ParsedBetaMessage for each API round-trip, and provides an until_done() convenience method to drain the loop and return the final message.

Class Hierarchy

BaseToolRunner (Generic, stores params/tools/options)
  |
  +-- BaseSyncToolRunner (adds sync iteration, __run__, generate_tool_call_response)
  |     |
  |     +-- BetaToolRunner (non-streaming, uses client.beta.messages.parse)
  |     +-- BetaStreamingToolRunner (streaming, uses client.beta.messages.stream)
  |
  +-- BaseAsyncToolRunner (adds async iteration, __run__, generate_tool_call_response)
        |
        +-- BetaAsyncToolRunner (non-streaming async)
        +-- BetaAsyncStreamingToolRunner (streaming async)

Source Location

File: src/anthropic/lib/tools/_beta_runner.py, lines 63-617

• BaseToolRunner: lines 63-117
• BaseSyncToolRunner: lines 120-347
• BetaToolRunner: lines 349-355
• BaseAsyncToolRunner: lines 367-598
• BetaAsyncToolRunner: lines 601-607

Import

from anthropic.lib.tools import BetaToolRunner
# Or use via the client convenience method:
# runner = client.beta.tools.run(...)

Constructor Parameters

Parameter	Type	Default	Description
params	ParseMessageCreateParamsBase[ResponseFormatT]	(required)	Message creation parameters (model, max_tokens, messages, tools, etc.)
options	RequestOptions	(required)	HTTP request options (extra_headers, extra_query, extra_body, timeout)
tools	Iterable[BetaRunnableTool]	(required)	Tool instances (BetaFunctionTool or BetaBuiltinFunctionTool)
client	Anthropic	(required)	The Anthropic client instance for API calls
max_iterations	None	None	Maximum number of API round-trips. None means unlimited.
compaction_control	None	None	Configuration for automatic context compaction

CompactionControl

Defined in src/anthropic/lib/tools/_beta_compaction_control.py:

class CompactionControl(TypedDict, total=False):
    enabled: Required[bool]                  # Whether compaction is active
    context_token_threshold: int             # Default: 100,000 tokens
    model: str                               # Default: same as runner model
    summary_prompt: str                       # Default: built-in continuation summary prompt

Core Loop: __run__()

The core iteration logic is in BaseSyncToolRunner.__run__() (lines 242-262):

def __run__(self) -> Iterator[RunnerItemT]:
    while not self._should_stop():
        with self._handle_request() as item:
            yield item
            message = self._get_last_message()
            assert message is not None

        self._iteration_count += 1

        # If compaction was performed, skip tool call generation this iteration
        if not self._check_and_compact():
            response = self.generate_tool_call_response()
            if response is None:
                log.debug("Tool call was not requested, exiting from tool runner loop.")
                return

            if not self._messages_modified:
                self.append_messages(message, response)

        self._messages_modified = False
        self._cached_tool_call_response = None

Loop steps:

1. _should_stop(): Checks if max_iterations has been reached (line 114-117)
2. _handle_request(): Makes the API call. For BetaToolRunner, this calls client.beta.messages.parse() (lines 352-355)
3. yield item: Yields the ParsedBetaMessage to the caller
4. _check_and_compact(): If compaction is enabled and token threshold is exceeded, summarizes and replaces history (lines 158-240)
5. generate_tool_call_response(): Iterates over tool_use blocks, calls each tool, collects results (lines 274-334)
6. append_messages(): Adds the assistant response and tool results to conversation history (lines 100-112)
7. Loop termination: If generate_tool_call_response() returns None (no tool calls), the loop exits

Key Methods

until_done()

def until_done(self) -> ParsedBetaMessage[ResponseFormatT]:

Consumes the iterator completely and returns the final message (lines 264-272). This is the simplest way to use the runner when intermediate results are not needed.

generate_tool_call_response()

def generate_tool_call_response(self) -> BetaMessageParam | None:

Processes the last assistant message's content blocks (lines 274-334):

1. Filters for tool_use blocks
2. Looks up each tool by name in self._tools_by_name
3. Calls tool.call(tool_use.input)
4. On success: appends {"type": "tool_result", "tool_use_id": ..., "content": result}
5. On tool not found: appends an error result and emits a UserWarning
6. On exception: logs the error and appends {"type": "tool_result", ..., "content": repr(exc), "is_error": True}
7. Returns {"role": "user", "content": results} or None if no tool calls were found

append_messages()

def append_messages(self, *messages: BetaMessageParam | ParsedBetaMessage[ResponseFormatT]) -> None:

Adds messages to the conversation history and invalidates cached tool responses (lines 100-112).

set_messages_params()

def set_messages_params(
    self,
    params: ParseMessageCreateParamsBase[ResponseFormatT]
    | Callable[[ParseMessageCreateParamsBase[ResponseFormatT]], ParseMessageCreateParamsBase[ResponseFormatT]],
) -> None:

Updates the parameters for the next API call. Accepts either new parameters directly or a function that mutates existing parameters (lines 85-98).

Usage Example

Basic automated loop:

import anthropic
from anthropic import beta_tool

@beta_tool
def get_weather(city: str, unit: str = "celsius") -> str:
    """Get the current weather for a city.

    Args:
        city: The city name
        unit: Temperature unit (celsius or fahrenheit)
    """
    return f"The weather in {city} is 22 degrees {unit}"

client = anthropic.Anthropic()

# Using the runner directly
from anthropic.lib.tools import BetaToolRunner

runner = BetaToolRunner(
    params={
        "model": "claude-sonnet-4-20250514",
        "max_tokens": 1024,
        "messages": [{"role": "user", "content": "What's the weather in London and Paris?"}],
        "tools": [get_weather.to_dict()],
    },
    options={},
    tools=[get_weather],
    client=client,
    max_iterations=10,
)

# Option A: Drain and get final message
final_message = runner.until_done()
print(final_message.content)

# Option B: Observe intermediate messages
for message in runner:
    print(f"Round {runner._iteration_count}: stop_reason={message.stop_reason}")

With compaction control:

runner = BetaToolRunner(
    params={
        "model": "claude-sonnet-4-20250514",
        "max_tokens": 4096,
        "messages": [{"role": "user", "content": "Research this topic thoroughly..."}],
        "tools": [search_tool.to_dict(), read_tool.to_dict()],
    },
    options={},
    tools=[search_tool, read_tool],
    client=client,
    max_iterations=50,
    compaction_control={
        "enabled": True,
        "context_token_threshold": 80_000,
        "model": "claude-sonnet-4-20250514",
    },
)

final = runner.until_done()

Compaction Behavior

When compaction is triggered (lines 158-240):

1. Checks total token usage (input + output + cache tokens) against threshold
2. Removes any tool_use blocks from the last assistant message (to avoid orphaned tool_use without tool_result)
3. Appends the summary prompt as a user message
4. Calls the API with an X-Stainless-Helper: compaction header
5. Replaces the entire message history with a single user message containing the summary
6. Returns True, causing the main loop to skip tool call generation for this iteration

Error Handling

The runner handles tool execution errors gracefully within _generate_tool_call_response() (lines 288-334):

• Tool not found: Emits a UserWarning and returns an error result suggesting the tool should be registered via beta_tool()
• Execution exception: Catches any Exception, logs it via log.exception(), and returns {"is_error": True, "content": repr(exc)}
• No tool calls: Returns None, which terminates the loop

Async Variants

The async variants mirror the sync API:

• BetaAsyncToolRunner (line 601): Uses await client.beta.messages.parse()
• BetaAsyncStreamingToolRunner (line 610): Uses async with client.beta.messages.stream()

Both implement __aiter__, __anext__, and async until_done().

Dependencies

• httpx: For RequestOptions timeout type and underlying HTTP transport
• pydantic: For ParsedBetaMessage response models

Related Pages

Implements Principle

• Principle:Anthropics_Anthropic_sdk_python_Automated_Tool_Loop

Requires Environment

• Environment:Anthropics_Anthropic_sdk_python_Python_SDK_Core_Environment