BaseAsyncToolRunner Class — anthropic-sdk-python Architecture

Source Code

src/anthropic/lib/tools/_beta_runner.py lines 367–598

class BaseAsyncToolRunner(
    BaseToolRunner[BetaAsyncRunnableTool, ResponseFormatT], Generic[RunnerItemT, ResponseFormatT], ABC
):
    def __init__(
        self,
        *,
        params: ParseMessageCreateParamsBase[ResponseFormatT],
        options: RequestOptions,
        tools: Iterable[BetaAsyncRunnableTool],
        client: AsyncAnthropic,
        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[[], Coroutine[None, None, ParsedBetaMessage[ResponseFormatT]]]
            | ParsedBetaMessage[ResponseFormatT]
            | None
        ) = None

    async def __anext__(self) -> RunnerItemT:
        return await self._iterator.__anext__()

    async def __aiter__(self) -> AsyncIterator[RunnerItemT]:
        async for item in self._iterator:
            yield item

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

    async 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 = await 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"
            ]