BaseSyncToolRunner Class — anthropic-sdk-python Architecture

Source Code

src/anthropic/lib/tools/_beta_runner.py lines 120–346

class BaseSyncToolRunner(BaseToolRunner[BetaRunnableTool, ResponseFormatT], Generic[RunnerItemT, ResponseFormatT], ABC):
    def __init__(
        self,
        *,
        params: ParseMessageCreateParamsBase[ResponseFormatT],
        options: RequestOptions,
        tools: Iterable[BetaRunnableTool],
        client: Anthropic,
        max_iterations: int | None = None,
        compaction_control: CompactionControl | None = None,
    ) -> None:
        super().__init__(
            params=params,
            options=options,
            tools=tools,
            max_iterations=max_iterations,
            compaction_control=compaction_control,
        )
        self._client = client

        self._iterator = self.__run__()
        self._last_message: (
            Callable[[], ParsedBetaMessage[ResponseFormatT]] | ParsedBetaMessage[ResponseFormatT] | None
        ) = None

    def __next__(self) -> RunnerItemT:
        return self._iterator.__next__()

    def __iter__(self) -> Iterator[RunnerItemT]:
        for item in self._iterator:
            yield item

    @abstractmethod
    @contextmanager
    def _handle_request(self) -> Iterator[RunnerItemT]:
        raise NotImplementedError()
        yield  # type: ignore[unreachable]

    def _check_and_compact(self) -> bool:
        """
        Check token usage and compact messages if threshold exceeded.
        Returns True if compaction was performed, False otherwise.
        """
        if self._compaction_control is None or not self._compaction_control["enabled"]:
            return False

        message = self._get_last_message()
        tokens_used = 0
        if message is not None:
            total_input_tokens = (
                message.usage.input_tokens
                + (message.usage.cache_creation_input_tokens or 0)
                + (message.usage.cache_read_input_tokens or 0)
            )
            tokens_used = total_input_tokens + message.usage.output_tokens

        threshold = self._compaction_control.get("context_token_threshold", DEFAULT_THRESHOLD)

        if tokens_used < threshold:
            return False

        # Perform compaction
        log.info(f"Token usage {tokens_used} has exceeded the threshold of {threshold}. Performing compaction.")

        model = self._compaction_control.get("model", self._params["model"])

        messages = list(self._params["messages"])

        if messages[-1]["role"] == "assistant":
            # Remove tool_use blocks from the last message to avoid 400 error
            # (tool_use requires tool_result, which we don't have yet)
            non_tool_blocks = [
                block
                for block in messages[-1]["content"]
                if isinstance(block, dict) and block.get("type") != "tool_use"
            ]

            if non_tool_blocks:
                messages[-1]["content"] = non_tool_blocks
            else:
                messages.pop()