BaseToolRunner Class — anthropic-sdk-python Architecture

Architecture documentation for the BaseToolRunner class in _beta_runner.py from the anthropic-sdk-python codebase.

Class python ExtensibilityTools

Entity Profile

ExtensibilityTools→ BaseToolRunner Class — anthropic-sdk-python Architecture

Dependency Diagram

graph TD
  74dfae5d_02b1_0e09_5e30_62a5a1ac1ac2["BaseToolRunner"]
  0da6889b_9352_8c02_9f21_a1deb1bdab0f["BetaMessage"]
  74dfae5d_02b1_0e09_5e30_62a5a1ac1ac2 -->|extends| 0da6889b_9352_8c02_9f21_a1deb1bdab0f
  34260b96_4096_ba59_bee2_0b2a94476dcc["_beta_runner.py"]
  74dfae5d_02b1_0e09_5e30_62a5a1ac1ac2 -->|defined in| 34260b96_4096_ba59_bee2_0b2a94476dcc
  dc2a0a3b_8495_0090_d41b_efe0dd8e4f77["__init__()"]
  74dfae5d_02b1_0e09_5e30_62a5a1ac1ac2 -->|method| dc2a0a3b_8495_0090_d41b_efe0dd8e4f77
  adc89f46_a31c_f374_60f3_d6ae19ce9e15["set_messages_params()"]
  74dfae5d_02b1_0e09_5e30_62a5a1ac1ac2 -->|method| adc89f46_a31c_f374_60f3_d6ae19ce9e15
  4970c546_0fad_9c03_1498_9ae6f7952dcf["append_messages()"]
  74dfae5d_02b1_0e09_5e30_62a5a1ac1ac2 -->|method| 4970c546_0fad_9c03_1498_9ae6f7952dcf
  21042fa5_2057_e026_e344_8ecf6abf02cc["_should_stop()"]
  74dfae5d_02b1_0e09_5e30_62a5a1ac1ac2 -->|method| 21042fa5_2057_e026_e344_8ecf6abf02cc

Relationship Graph

Source Code

src/anthropic/lib/tools/_beta_runner.py lines 63–117

class BaseToolRunner(Generic[AnyFunctionToolT, ResponseFormatT]):
    def __init__(
        self,
        *,
        params: ParseMessageCreateParamsBase[ResponseFormatT],
        options: RequestOptions,
        tools: Iterable[AnyFunctionToolT],
        max_iterations: int | None = None,
        compaction_control: CompactionControl | None = None,
    ) -> None:
        self._tools_by_name = {tool.name: tool for tool in tools}
        self._params: ParseMessageCreateParamsBase[ResponseFormatT] = {
            **params,
            "messages": [message for message in params["messages"]],
        }
        self._options = options
        self._messages_modified = False
        self._cached_tool_call_response: BetaMessageParam | None = None
        self._max_iterations = max_iterations
        self._iteration_count = 0
        self._compaction_control = compaction_control

    def set_messages_params(
        self,
        params: ParseMessageCreateParamsBase[ResponseFormatT]
        | Callable[[ParseMessageCreateParamsBase[ResponseFormatT]], ParseMessageCreateParamsBase[ResponseFormatT]],
    ) -> None:
        """
        Update the parameters for the next API call. This invalidates any cached tool responses.

        Args:
            params (ParsedMessageCreateParamsBase[ResponseFormatT] | Callable): Either new parameters or a function to mutate existing parameters
        """
        if callable(params):
            params = params(self._params)
        self._params = params

    def append_messages(self, *messages: BetaMessageParam | ParsedBetaMessage[ResponseFormatT]) -> None:
        """Add one or more messages to the conversation history.

        This invalidates the cached tool response, i.e. if tools were already called, then they will
        be called again on the next loop iteration.
        """
        message_params: List[BetaMessageParam] = [
            {"role": message.role, "content": message.content} if isinstance(message, BetaMessage) else message
            for message in messages
        ]
        self._messages_modified = True
        self.set_messages_params(lambda params: {**params, "messages": [*self._params["messages"], *message_params]})
        self._cached_tool_call_response = None

    def _should_stop(self) -> bool:
        if self._max_iterations is not None and self._iteration_count >= self._max_iterations:
            return True
        return False