Module livekit.plugins.anthropic

Anthropic plugin for LiveKit Agents

See https://docs.livekit.io/agents/integrations/llm/anthropic/ for more information.

Classes

class AnthropicTool (*, id: str)

Expand source code

class AnthropicTool(ProviderTool, ABC):
    @abstractmethod
    def to_dict(self) -> dict[str, Any]: ...

    @property
    def beta_flag(self) -> str | None:
        return None

Helper class that provides a standard way to create an ABC using inheritance.

Ancestors

• livekit.agents.llm.tool_context.ProviderTool
• livekit.agents.llm.tool_context.Tool
• abc.ABC

Subclasses

• livekit.plugins.anthropic.tools.ComputerUse

Instance variables

prop beta_flag : str | None

Expand source code

@property
def beta_flag(self) -> str | None:
    return None

Methods

def to_dict(self) ‑> dict[str, typing.Any]

Expand source code

@abstractmethod
def to_dict(self) -> dict[str, Any]: ...

class ComputerTool (*, actions: PageActions, width: int = 1280, height: int = 720)

Expand source code

class ComputerTool(llm.Toolset):
    """Anthropic computer_use tool backed by browser PageActions.

    Usage::

        from livekit.plugins.browser import PageActions

        actions = PageActions(page=page)
        tool = ComputerTool(actions=actions, width=1280, height=720)
    """

    def __init__(
        self,
        *,
        actions: PageActions,
        width: int = 1280,
        height: int = 720,
    ) -> None:
        super().__init__(id="computer")
        self._actions = actions
        self._provider_tool = ComputerUse(
            display_width_px=width,
            display_height_px=height,
        )

    @property
    def tools(self) -> list[llm.Tool]:
        return [self._provider_tool]

    async def execute(self, action: str, **kwargs: Any) -> list[dict[str, Any]]:
        """Dispatch an Anthropic computer_use action and return screenshot content."""
        actions = self._actions

        match action:
            case "screenshot":
                pass
            case "left_click":
                x, y = _require_coordinate(kwargs)
                await actions.left_click(x, y, modifiers=kwargs.get("text"))
            case "right_click":
                x, y = _require_coordinate(kwargs)
                await actions.right_click(x, y)
            case "double_click":
                x, y = _require_coordinate(kwargs)
                await actions.double_click(x, y)
            case "triple_click":
                x, y = _require_coordinate(kwargs)
                await actions.triple_click(x, y)
            case "middle_click":
                x, y = _require_coordinate(kwargs)
                await actions.middle_click(x, y)
            case "mouse_move":
                x, y = _require_coordinate(kwargs)
                await actions.mouse_move(x, y)
            case "left_click_drag":
                sx, sy = _require_coordinate(kwargs, key="start_coordinate")
                ex, ey = _require_coordinate(kwargs)
                await actions.left_click_drag(start_x=sx, start_y=sy, end_x=ex, end_y=ey)
            case "left_mouse_down":
                x, y = _require_coordinate(kwargs)
                await actions.left_mouse_down(x, y)
            case "left_mouse_up":
                x, y = _require_coordinate(kwargs)
                await actions.left_mouse_up(x, y)
            case "scroll":
                x, y = _require_coordinate(kwargs)
                await actions.scroll(
                    x,
                    y,
                    direction=kwargs.get("scroll_direction", "down"),
                    amount=int(kwargs.get("scroll_amount", 3)),
                )
            case "type":
                await actions.type_text(_require(kwargs, "text"))
            case "key":
                await actions.key(_require(kwargs, "text"))
            case "hold_key":
                await actions.hold_key(
                    _require(kwargs, "text"),
                    duration=float(kwargs.get("duration", 0.5)),
                )
            case "wait":
                await actions.wait()
            case _:
                raise ValueError(f"Unknown computer_use action: {action!r}")

        await asyncio.sleep(_POST_ACTION_DELAY)

        frame = actions.last_frame
        if frame is None:
            return [{"type": "text", "text": "(no frame available yet)"}]
        return _screenshot_content(frame)

    def aclose(self) -> None:
        self._actions.aclose()

Anthropic computer_use tool backed by browser PageActions.

Usage::

from livekit.plugins.browser import PageActions

actions = PageActions(page=page)
tool = ComputerTool(actions=actions, width=1280, height=720)

Ancestors

• livekit.agents.llm.tool_context.Toolset
• abc.ABC

Instance variables

prop tools : list[llm.Tool]

Expand source code

@property
def tools(self) -> list[llm.Tool]:
    return [self._provider_tool]

Methods

def aclose(self) ‑> None

Expand source code

def aclose(self) -> None:
    self._actions.aclose()

async def execute(self, action: str, **kwargs: Any) ‑> list[dict[str, typing.Any]]

Expand source code

async def execute(self, action: str, **kwargs: Any) -> list[dict[str, Any]]:
    """Dispatch an Anthropic computer_use action and return screenshot content."""
    actions = self._actions

    match action:
        case "screenshot":
            pass
        case "left_click":
            x, y = _require_coordinate(kwargs)
            await actions.left_click(x, y, modifiers=kwargs.get("text"))
        case "right_click":
            x, y = _require_coordinate(kwargs)
            await actions.right_click(x, y)
        case "double_click":
            x, y = _require_coordinate(kwargs)
            await actions.double_click(x, y)
        case "triple_click":
            x, y = _require_coordinate(kwargs)
            await actions.triple_click(x, y)
        case "middle_click":
            x, y = _require_coordinate(kwargs)
            await actions.middle_click(x, y)
        case "mouse_move":
            x, y = _require_coordinate(kwargs)
            await actions.mouse_move(x, y)
        case "left_click_drag":
            sx, sy = _require_coordinate(kwargs, key="start_coordinate")
            ex, ey = _require_coordinate(kwargs)
            await actions.left_click_drag(start_x=sx, start_y=sy, end_x=ex, end_y=ey)
        case "left_mouse_down":
            x, y = _require_coordinate(kwargs)
            await actions.left_mouse_down(x, y)
        case "left_mouse_up":
            x, y = _require_coordinate(kwargs)
            await actions.left_mouse_up(x, y)
        case "scroll":
            x, y = _require_coordinate(kwargs)
            await actions.scroll(
                x,
                y,
                direction=kwargs.get("scroll_direction", "down"),
                amount=int(kwargs.get("scroll_amount", 3)),
            )
        case "type":
            await actions.type_text(_require(kwargs, "text"))
        case "key":
            await actions.key(_require(kwargs, "text"))
        case "hold_key":
            await actions.hold_key(
                _require(kwargs, "text"),
                duration=float(kwargs.get("duration", 0.5)),
            )
        case "wait":
            await actions.wait()
        case _:
            raise ValueError(f"Unknown computer_use action: {action!r}")

    await asyncio.sleep(_POST_ACTION_DELAY)

    frame = actions.last_frame
    if frame is None:
        return [{"type": "text", "text": "(no frame available yet)"}]
    return _screenshot_content(frame)

Dispatch an Anthropic computer_use action and return screenshot content.

class ComputerUse (display_width_px: int = 1280,
display_height_px: int = 720,
display_number: int = 1,
tool_version: str = 'computer_20251124')

Expand source code

@dataclass
class ComputerUse(AnthropicTool):
    display_width_px: int = 1280
    display_height_px: int = 720
    display_number: int = 1
    tool_version: str = "computer_20251124"

    def __post_init__(self) -> None:
        super().__init__(id="computer")

    @property
    def beta_flag(self) -> str | None:
        return _TOOL_VERSION_BETA_FLAGS.get(self.tool_version)

    def to_dict(self) -> dict[str, Any]:
        return {
            "type": self.tool_version,
            "name": "computer",
            "display_width_px": self.display_width_px,
            "display_height_px": self.display_height_px,
            "display_number": self.display_number,
        }

ComputerUse(display_width_px: 'int' = 1280, display_height_px: 'int' = 720, display_number: 'int' = 1, tool_version: 'str' = 'computer_20251124')

Ancestors

• livekit.plugins.anthropic.tools.AnthropicTool
• livekit.agents.llm.tool_context.ProviderTool
• livekit.agents.llm.tool_context.Tool
• abc.ABC

Instance variables

prop beta_flag : str | None

Expand source code

@property
def beta_flag(self) -> str | None:
    return _TOOL_VERSION_BETA_FLAGS.get(self.tool_version)

var display_height_px : int

var display_number : int

var display_width_px : int

var tool_version : str

Methods

def to_dict(self) ‑> dict[str, typing.Any]

Expand source code

def to_dict(self) -> dict[str, Any]:
    return {
        "type": self.tool_version,
        "name": "computer",
        "display_width_px": self.display_width_px,
        "display_height_px": self.display_height_px,
        "display_number": self.display_number,
    }

class LLM (*,
model: str | ChatModels = 'claude-3-5-sonnet-20241022',
api_key: NotGivenOr[str] = NOT_GIVEN,
base_url: NotGivenOr[str] = NOT_GIVEN,
user: NotGivenOr[str] = NOT_GIVEN,
client: anthropic.AsyncClient | None = None,
top_k: NotGivenOr[int] = NOT_GIVEN,
max_tokens: NotGivenOr[int] = NOT_GIVEN,
temperature: NotGivenOr[float] = NOT_GIVEN,
parallel_tool_calls: NotGivenOr[bool] = NOT_GIVEN,
tool_choice: NotGivenOr[ToolChoice] = NOT_GIVEN,
caching: "NotGivenOr[Literal['ephemeral']]" = NOT_GIVEN)

Expand source code

class LLM(llm.LLM):
    def __init__(
        self,
        *,
        model: str | ChatModels = "claude-3-5-sonnet-20241022",
        api_key: NotGivenOr[str] = NOT_GIVEN,
        base_url: NotGivenOr[str] = NOT_GIVEN,
        user: NotGivenOr[str] = NOT_GIVEN,
        client: anthropic.AsyncClient | None = None,
        top_k: NotGivenOr[int] = NOT_GIVEN,
        max_tokens: NotGivenOr[int] = NOT_GIVEN,
        temperature: NotGivenOr[float] = NOT_GIVEN,
        parallel_tool_calls: NotGivenOr[bool] = NOT_GIVEN,
        tool_choice: NotGivenOr[ToolChoice] = NOT_GIVEN,
        caching: NotGivenOr[Literal["ephemeral"]] = NOT_GIVEN,
    ) -> None:
        """
        Create a new instance of Anthropic LLM.

        ``api_key`` must be set to your Anthropic API key, either using the argument or by setting
        the ``ANTHROPIC_API_KEY`` environmental variable.

        model (str | ChatModels): The model to use. Defaults to "claude-3-5-sonnet-20241022".
        api_key (str, optional): The Anthropic API key. Defaults to the ANTHROPIC_API_KEY environment variable.
        base_url (str, optional): The base URL for the Anthropic API. Defaults to None.
        user (str, optional): The user for the Anthropic API. Defaults to None.
        client (anthropic.AsyncClient | None): The Anthropic client to use. Defaults to None.
        temperature (float, optional): The temperature for the Anthropic API. Defaults to None.
        parallel_tool_calls (bool, optional): Whether to parallelize tool calls. Defaults to None.
        tool_choice (ToolChoice, optional): The tool choice for the Anthropic API. Defaults to "auto".
        caching (Literal["ephemeral"], optional): If set to "ephemeral", caching will be enabled for the system prompt, tools, and chat history.
        """  # noqa: E501

        super().__init__()

        self._opts = _LLMOptions(
            model=model,
            user=user,
            temperature=temperature,
            parallel_tool_calls=parallel_tool_calls,
            tool_choice=tool_choice,
            caching=caching,
            top_k=top_k,
            max_tokens=max_tokens,
        )
        anthropic_api_key = api_key if is_given(api_key) else os.environ.get("ANTHROPIC_API_KEY")
        if not anthropic_api_key:
            raise ValueError(
                "Anthropic API key is required, either as argument or set"
                " ANTHROPIC_API_KEY environment variable"
            )

        self._client = client or anthropic.AsyncClient(
            api_key=anthropic_api_key,
            base_url=base_url if is_given(base_url) else None,
            http_client=httpx.AsyncClient(
                timeout=5.0,
                follow_redirects=True,
                limits=httpx.Limits(
                    max_connections=1000,
                    max_keepalive_connections=100,
                    keepalive_expiry=120,
                ),
            ),
        )

    @property
    def model(self) -> str:
        return self._opts.model

    @property
    def provider(self) -> str:
        return self._client._base_url.netloc.decode("utf-8")

    def chat(
        self,
        *,
        chat_ctx: ChatContext,
        tools: list[Tool] | None = None,
        conn_options: APIConnectOptions = DEFAULT_API_CONNECT_OPTIONS,
        parallel_tool_calls: NotGivenOr[bool] = NOT_GIVEN,
        tool_choice: NotGivenOr[ToolChoice] = NOT_GIVEN,
        extra_kwargs: NotGivenOr[dict[str, Any]] = NOT_GIVEN,
    ) -> LLMStream:
        extra = {}

        if is_given(extra_kwargs):
            extra.update(extra_kwargs)

        if is_given(self._opts.user):
            extra["user"] = self._opts.user

        if is_given(self._opts.temperature):
            extra["temperature"] = self._opts.temperature

        if is_given(self._opts.top_k):
            extra["top_k"] = self._opts.top_k

        extra["max_tokens"] = self._opts.max_tokens if is_given(self._opts.max_tokens) else 1024

        beta_flag: str | None = None
        if tools:
            from .tools import AnthropicTool

            tool_ctx = llm.ToolContext(tools)
            tool_schemas = tool_ctx.parse_function_tools("anthropic")

            for tool in tool_ctx.provider_tools:
                if isinstance(tool, AnthropicTool):
                    tool_schemas.append(tool.to_dict())
                    if tool.beta_flag:
                        beta_flag = tool.beta_flag

            extra["tools"] = tool_schemas

            tool_choice = (
                cast(ToolChoice, tool_choice) if is_given(tool_choice) else self._opts.tool_choice
            )
            if is_given(tool_choice):
                anthropic_tool_choice: dict[str, Any] | None = {"type": "auto"}
                if isinstance(tool_choice, dict) and tool_choice.get("type") == "function":
                    anthropic_tool_choice = {
                        "type": "tool",
                        "name": tool_choice["function"]["name"],
                    }
                elif isinstance(tool_choice, str):
                    if tool_choice == "required":
                        anthropic_tool_choice = {"type": "any"}
                    elif tool_choice == "none":
                        extra["tools"] = []
                        anthropic_tool_choice = None
                if anthropic_tool_choice is not None:
                    parallel_tool_calls = (
                        parallel_tool_calls
                        if is_given(parallel_tool_calls)
                        else self._opts.parallel_tool_calls
                    )
                    if is_given(parallel_tool_calls):
                        anthropic_tool_choice["disable_parallel_tool_use"] = not parallel_tool_calls
                    extra["tool_choice"] = anthropic_tool_choice

        anthropic_ctx, extra_data = chat_ctx.to_provider_format(format="anthropic")
        messages = cast(list[anthropic.types.MessageParam], anthropic_ctx)
        if extra_data.system_messages:
            extra["system"] = [
                anthropic.types.TextBlockParam(text=content, type="text")
                for content in extra_data.system_messages
            ]

        # add cache control
        if self._opts.caching == "ephemeral":
            if extra.get("system"):
                extra["system"][-1]["cache_control"] = CACHE_CONTROL_EPHEMERAL

            if extra.get("tools"):
                extra["tools"][-1]["cache_control"] = CACHE_CONTROL_EPHEMERAL

            seen_assistant = False
            for msg in reversed(messages):
                if (
                    msg["role"] == "assistant"
                    and (content := msg["content"])
                    and not seen_assistant
                ):
                    content[-1]["cache_control"] = CACHE_CONTROL_EPHEMERAL  # type: ignore
                    seen_assistant = True

                elif msg["role"] == "user" and (content := msg["content"]) and seen_assistant:
                    content[-1]["cache_control"] = CACHE_CONTROL_EPHEMERAL  # type: ignore
                    break

        if beta_flag:
            stream = self._client.beta.messages.create(
                betas=[beta_flag],
                messages=messages,  # type: ignore[arg-type]
                model=self._opts.model,
                stream=True,
                timeout=conn_options.timeout,
                **extra,
            )
        else:
            stream = self._client.messages.create(
                messages=messages,
                model=self._opts.model,
                stream=True,
                timeout=conn_options.timeout,
                **extra,
            )

        return LLMStream(
            self,
            anthropic_stream=stream,  # type: ignore[arg-type]
            chat_ctx=chat_ctx,
            tools=tools or [],
            conn_options=conn_options,
        )

Helper class that provides a standard way to create an ABC using inheritance.

Create a new instance of Anthropic LLM.

api_key must be set to your Anthropic API key, either using the argument or by setting the ANTHROPIC_API_KEY environmental variable.

model (str | ChatModels): The model to use. Defaults to "claude-3-5-sonnet-20241022". api_key (str, optional): The Anthropic API key. Defaults to the ANTHROPIC_API_KEY environment variable. base_url (str, optional): The base URL for the Anthropic API. Defaults to None. user (str, optional): The user for the Anthropic API. Defaults to None. client (anthropic.AsyncClient | None): The Anthropic client to use. Defaults to None. temperature (float, optional): The temperature for the Anthropic API. Defaults to None. parallel_tool_calls (bool, optional): Whether to parallelize tool calls. Defaults to None. tool_choice (ToolChoice, optional): The tool choice for the Anthropic API. Defaults to "auto". caching (Literal["ephemeral"], optional): If set to "ephemeral", caching will be enabled for the system prompt, tools, and chat history.

Ancestors

• livekit.agents.llm.llm.LLM
• abc.ABC
• EventEmitter
• typing.Generic

Instance variables

prop model : str

Expand source code

@property
def model(self) -> str:
    return self._opts.model

Get the model name/identifier for this LLM instance.

Returns

The model name if available, "unknown" otherwise.

Note

Plugins should override this property to provide their model information.

prop provider : str

Expand source code

@property
def provider(self) -> str:
    return self._client._base_url.netloc.decode("utf-8")

Get the provider name/identifier for this LLM instance.

Returns

The provider name if available, "unknown" otherwise.

Note

Plugins should override this property to provide their provider information.

Methods

def chat(self,
*,
chat_ctx: ChatContext,
tools: list[Tool] | None = None,
conn_options: APIConnectOptions = APIConnectOptions(max_retry=3, retry_interval=2.0, timeout=10.0),
parallel_tool_calls: NotGivenOr[bool] = NOT_GIVEN,
tool_choice: NotGivenOr[ToolChoice] = NOT_GIVEN,
extra_kwargs: NotGivenOr[dict[str, Any]] = NOT_GIVEN) ‑> livekit.plugins.anthropic.llm.LLMStream

Expand source code

def chat(
    self,
    *,
    chat_ctx: ChatContext,
    tools: list[Tool] | None = None,
    conn_options: APIConnectOptions = DEFAULT_API_CONNECT_OPTIONS,
    parallel_tool_calls: NotGivenOr[bool] = NOT_GIVEN,
    tool_choice: NotGivenOr[ToolChoice] = NOT_GIVEN,
    extra_kwargs: NotGivenOr[dict[str, Any]] = NOT_GIVEN,
) -> LLMStream:
    extra = {}

    if is_given(extra_kwargs):
        extra.update(extra_kwargs)

    if is_given(self._opts.user):
        extra["user"] = self._opts.user

    if is_given(self._opts.temperature):
        extra["temperature"] = self._opts.temperature

    if is_given(self._opts.top_k):
        extra["top_k"] = self._opts.top_k

    extra["max_tokens"] = self._opts.max_tokens if is_given(self._opts.max_tokens) else 1024

    beta_flag: str | None = None
    if tools:
        from .tools import AnthropicTool

        tool_ctx = llm.ToolContext(tools)
        tool_schemas = tool_ctx.parse_function_tools("anthropic")

        for tool in tool_ctx.provider_tools:
            if isinstance(tool, AnthropicTool):
                tool_schemas.append(tool.to_dict())
                if tool.beta_flag:
                    beta_flag = tool.beta_flag

        extra["tools"] = tool_schemas

        tool_choice = (
            cast(ToolChoice, tool_choice) if is_given(tool_choice) else self._opts.tool_choice
        )
        if is_given(tool_choice):
            anthropic_tool_choice: dict[str, Any] | None = {"type": "auto"}
            if isinstance(tool_choice, dict) and tool_choice.get("type") == "function":
                anthropic_tool_choice = {
                    "type": "tool",
                    "name": tool_choice["function"]["name"],
                }
            elif isinstance(tool_choice, str):
                if tool_choice == "required":
                    anthropic_tool_choice = {"type": "any"}
                elif tool_choice == "none":
                    extra["tools"] = []
                    anthropic_tool_choice = None
            if anthropic_tool_choice is not None:
                parallel_tool_calls = (
                    parallel_tool_calls
                    if is_given(parallel_tool_calls)
                    else self._opts.parallel_tool_calls
                )
                if is_given(parallel_tool_calls):
                    anthropic_tool_choice["disable_parallel_tool_use"] = not parallel_tool_calls
                extra["tool_choice"] = anthropic_tool_choice

    anthropic_ctx, extra_data = chat_ctx.to_provider_format(format="anthropic")
    messages = cast(list[anthropic.types.MessageParam], anthropic_ctx)
    if extra_data.system_messages:
        extra["system"] = [
            anthropic.types.TextBlockParam(text=content, type="text")
            for content in extra_data.system_messages
        ]

    # add cache control
    if self._opts.caching == "ephemeral":
        if extra.get("system"):
            extra["system"][-1]["cache_control"] = CACHE_CONTROL_EPHEMERAL

        if extra.get("tools"):
            extra["tools"][-1]["cache_control"] = CACHE_CONTROL_EPHEMERAL

        seen_assistant = False
        for msg in reversed(messages):
            if (
                msg["role"] == "assistant"
                and (content := msg["content"])
                and not seen_assistant
            ):
                content[-1]["cache_control"] = CACHE_CONTROL_EPHEMERAL  # type: ignore
                seen_assistant = True

            elif msg["role"] == "user" and (content := msg["content"]) and seen_assistant:
                content[-1]["cache_control"] = CACHE_CONTROL_EPHEMERAL  # type: ignore
                break

    if beta_flag:
        stream = self._client.beta.messages.create(
            betas=[beta_flag],
            messages=messages,  # type: ignore[arg-type]
            model=self._opts.model,
            stream=True,
            timeout=conn_options.timeout,
            **extra,
        )
    else:
        stream = self._client.messages.create(
            messages=messages,
            model=self._opts.model,
            stream=True,
            timeout=conn_options.timeout,
            **extra,
        )

    return LLMStream(
        self,
        anthropic_stream=stream,  # type: ignore[arg-type]
        chat_ctx=chat_ctx,
        tools=tools or [],
        conn_options=conn_options,
    )

Inherited members

• EventEmitter:
  • emit
  • off
  • on
  • once

class LLMStream (llm: LLM,
*,
anthropic_stream: Awaitable[anthropic.AsyncStream[anthropic.types.RawMessageStreamEvent]],
chat_ctx: llm.ChatContext,
tools: list[Tool],
conn_options: APIConnectOptions)

Expand source code

class LLMStream(llm.LLMStream):
    def __init__(
        self,
        llm: LLM,
        *,
        anthropic_stream: Awaitable[anthropic.AsyncStream[anthropic.types.RawMessageStreamEvent]],
        chat_ctx: llm.ChatContext,
        tools: list[Tool],
        conn_options: APIConnectOptions,
    ) -> None:
        super().__init__(llm, chat_ctx=chat_ctx, tools=tools, conn_options=conn_options)
        self._awaitable_anthropic_stream = anthropic_stream
        self._anthropic_stream: (
            anthropic.AsyncStream[anthropic.types.RawMessageStreamEvent] | None
        ) = None

        # current function call that we're waiting for full completion (args are streamed)
        self._tool_call_id: str | None = None
        self._fnc_name: str | None = None
        self._fnc_raw_arguments: str | None = None

        self._request_id: str = ""
        self._ignoring_cot = False  # ignore chain of thought
        self._input_tokens = 0
        self._cache_creation_tokens = 0
        self._cache_read_tokens = 0
        self._output_tokens = 0

    async def _run(self) -> None:
        retryable = True
        try:
            if not self._anthropic_stream:
                self._anthropic_stream = await self._awaitable_anthropic_stream

            async with self._anthropic_stream as stream:
                async for event in stream:
                    chat_chunk = self._parse_event(event)
                    if chat_chunk is not None:
                        self._event_ch.send_nowait(chat_chunk)
                        retryable = False

                # https://docs.anthropic.com/en/docs/build-with-claude/prompt-caching#tracking-cache-performance
                prompt_token = (
                    self._input_tokens + self._cache_creation_tokens + self._cache_read_tokens
                )
                self._event_ch.send_nowait(
                    llm.ChatChunk(
                        id=self._request_id,
                        usage=llm.CompletionUsage(
                            completion_tokens=self._output_tokens,
                            prompt_tokens=prompt_token,
                            total_tokens=prompt_token + self._output_tokens,
                            prompt_cached_tokens=self._cache_read_tokens,
                            cache_creation_tokens=self._cache_creation_tokens,
                            cache_read_tokens=self._cache_read_tokens,
                        ),
                    )
                )
        except anthropic.APITimeoutError as e:
            raise APITimeoutError(retryable=retryable) from e
        except anthropic.APIStatusError as e:
            raise APIStatusError(
                e.message,
                status_code=e.status_code,
                request_id=e.request_id,
                body=e.body,
            ) from e
        except Exception as e:
            raise APIConnectionError(retryable=retryable) from e

    def _parse_event(self, event: anthropic.types.RawMessageStreamEvent) -> llm.ChatChunk | None:
        if event.type == "message_start":
            self._request_id = event.message.id
            self._input_tokens = event.message.usage.input_tokens
            self._output_tokens = event.message.usage.output_tokens
            if event.message.usage.cache_creation_input_tokens:
                self._cache_creation_tokens = event.message.usage.cache_creation_input_tokens
            if event.message.usage.cache_read_input_tokens:
                self._cache_read_tokens = event.message.usage.cache_read_input_tokens
        elif event.type == "message_delta":
            self._output_tokens += event.usage.output_tokens
        elif event.type == "content_block_start":
            if event.content_block.type == "tool_use":
                self._tool_call_id = event.content_block.id
                self._fnc_name = event.content_block.name
                self._fnc_raw_arguments = ""
        elif event.type == "content_block_delta":
            delta = event.delta
            if delta.type == "text_delta":
                text = delta.text

                if self._tools is not None:
                    # anthropic may inject COC when using functions
                    if text.startswith("<thinking>"):
                        self._ignoring_cot = True
                    elif self._ignoring_cot and "</thinking>" in text:
                        text = text.split("</thinking>")[-1]
                        self._ignoring_cot = False

                if self._ignoring_cot:
                    return None

                return llm.ChatChunk(
                    id=self._request_id,
                    delta=llm.ChoiceDelta(content=text, role="assistant"),
                )
            elif delta.type == "input_json_delta":
                assert self._fnc_raw_arguments is not None
                self._fnc_raw_arguments += delta.partial_json

        elif event.type == "content_block_stop":
            if self._tool_call_id is not None:
                assert self._fnc_name is not None
                assert self._fnc_raw_arguments is not None

                chat_chunk = llm.ChatChunk(
                    id=self._request_id,
                    delta=llm.ChoiceDelta(
                        role="assistant",
                        tool_calls=[
                            llm.FunctionToolCall(
                                arguments=self._fnc_raw_arguments or "",
                                name=self._fnc_name or "",
                                call_id=self._tool_call_id or "",
                            )
                        ],
                    ),
                )
                self._tool_call_id = self._fnc_raw_arguments = self._fnc_name = None
                return chat_chunk

        return None

Helper class that provides a standard way to create an ABC using inheritance.

Ancestors

• livekit.agents.llm.llm.LLMStream
• abc.ABC