anthropic

feishu.agent.adapters.anthropic

AnthropicBackend

Anthropic Messages API 的 feishu.agent.llm.LlmBackend 实现。

将与厂商无关的 feishu.agent.llm.Message、feishu.agent.llm.ToolSpec 翻译为 Anthropic 请求格式， 并把其流式事件归一化为 [feishu.agent.llm.StreamChunk][]。anthropic SDK 仅在未注入 client 时按需 懒加载，核心模块本身不依赖该 SDK。

参数：

名称	类型	描述	默认
client	Any	已构造的 anthropic.AsyncAnthropic 客户端。为 None 时自动创建一个。	None
model	str	模型名称，例如 claude-sonnet-4-5。	必需
**defaults	Any	透传给 Anthropic API 的默认参数，例如 max_tokens、temperature。	{}

示例：

>>> backend = AnthropicBackend(model="claude-sonnet-4-5", max_tokens=1024)

源代码位于： feishu/agent/adapters/anthropic.py

class AnthropicBackend:
    r"""
    Anthropic Messages API 的 [feishu.agent.llm.LlmBackend][] 实现。

    将与厂商无关的 [feishu.agent.llm.Message][]、[feishu.agent.llm.ToolSpec][] 翻译为 Anthropic 请求格式，
    并把其流式事件归一化为 [feishu.agent.llm.StreamChunk][]。`anthropic` SDK 仅在未注入 `client` 时按需
    懒加载，核心模块本身不依赖该 SDK。

    Args:
        client: 已构造的 `anthropic.AsyncAnthropic` 客户端。为 `None` 时自动创建一个。
        model: 模型名称，例如 `claude-sonnet-4-5`。
        **defaults: 透传给 Anthropic API 的默认参数，例如 `max_tokens`、`temperature`。

    Examples:
        >>> backend = AnthropicBackend(model="claude-sonnet-4-5", max_tokens=1024)  # doctest:+SKIP
    """

    def __init__(self, client: Any = None, *, model: str, **defaults: Any):
        if client is None:
            import anthropic  # imported lazily; core never imports the SDK

            client = anthropic.AsyncAnthropic()
        self._client = client
        self._model = model
        self._defaults = defaults

    def stream(
        self,
        *,
        messages: Sequence[Message],
        tools: Sequence[ToolSpec] = (),
        system: str | None = None,
        **kwargs: Any,
    ) -> AsyncIterator[StreamChunk]:
        r"""
        调用 Anthropic Messages API 流式生成一轮响应。

        Args:
            messages: 截至当前轮次的对话历史。
            tools: 本轮可供模型调用的工具声明。
            system: 系统提示词，作为顶层 `system` 参数传入。
            **kwargs: 覆盖默认参数（如 `model`、`max_tokens`）或追加的额外参数。

        Returns:
            逐个产出归一化 [feishu.agent.llm.StreamChunk][] 的异步迭代器。

        飞书文档:
            [Anthropic Messages API](https://docs.anthropic.com/en/api/messages)

        Examples:
            >>> async for chunk in backend.stream(messages=history):  # doctest:+SKIP
            ...     print(chunk)
            TextDelta(text='你好')
            MessageStop(stop_reason=<StopReason.END_TURN: 'end_turn'>, usage={'output_tokens': 5})
        """
        params: dict[str, Any] = {
            "model": kwargs.pop("model", self._model),
            "max_tokens": kwargs.pop("max_tokens", self._defaults.get("max_tokens", 1024)),
            "messages": _to_anthropic_messages(messages),
            **{k: v for k, v in self._defaults.items() if k != "max_tokens"},
            **kwargs,
        }
        if tools:
            params["tools"] = _to_anthropic_tools(tools)
        if system is not None:
            params["system"] = system

        async def _gen() -> AsyncIterator[StreamChunk]:
            async with self._client.messages.stream(**params) as stream:
                async for chunk in _translate_events(stream):
                    yield chunk

        return _gen()

stream

Python

stream(*, messages: Sequence[Message], tools: Sequence[ToolSpec] = (), system: str | None = None, **kwargs: Any) -> AsyncIterator[StreamChunk]

调用 Anthropic Messages API 流式生成一轮响应。

参数：

名称	类型	描述	默认
messages	Sequence[Message]	截至当前轮次的对话历史。	必需
tools	Sequence[ToolSpec]	本轮可供模型调用的工具声明。	()
system	str | None	系统提示词，作为顶层 system 参数传入。	None
**kwargs	Any	覆盖默认参数（如 model、max_tokens）或追加的额外参数。	{}

返回：

类型	描述
AsyncIterator[StreamChunk]	逐个产出归一化 [feishu.agent.llm.StreamChunk][] 的异步迭代器。

飞书文档

Anthropic Messages API

示例：

>>> async for chunk in backend.stream(messages=history):
...     print(chunk)
TextDelta(text='你好')
MessageStop(stop_reason=<StopReason.END_TURN: 'end_turn'>, usage={'output_tokens': 5})

源代码位于： feishu/agent/adapters/anthropic.py

def stream(
    self,
    *,
    messages: Sequence[Message],
    tools: Sequence[ToolSpec] = (),
    system: str | None = None,
    **kwargs: Any,
) -> AsyncIterator[StreamChunk]:
    r"""
    调用 Anthropic Messages API 流式生成一轮响应。

    Args:
        messages: 截至当前轮次的对话历史。
        tools: 本轮可供模型调用的工具声明。
        system: 系统提示词，作为顶层 `system` 参数传入。
        **kwargs: 覆盖默认参数（如 `model`、`max_tokens`）或追加的额外参数。

    Returns:
        逐个产出归一化 [feishu.agent.llm.StreamChunk][] 的异步迭代器。

    飞书文档:
        [Anthropic Messages API](https://docs.anthropic.com/en/api/messages)

    Examples:
        >>> async for chunk in backend.stream(messages=history):  # doctest:+SKIP
        ...     print(chunk)
        TextDelta(text='你好')
        MessageStop(stop_reason=<StopReason.END_TURN: 'end_turn'>, usage={'output_tokens': 5})
    """
    params: dict[str, Any] = {
        "model": kwargs.pop("model", self._model),
        "max_tokens": kwargs.pop("max_tokens", self._defaults.get("max_tokens", 1024)),
        "messages": _to_anthropic_messages(messages),
        **{k: v for k, v in self._defaults.items() if k != "max_tokens"},
        **kwargs,
    }
    if tools:
        params["tools"] = _to_anthropic_tools(tools)
    if system is not None:
        params["system"] = system

    async def _gen() -> AsyncIterator[StreamChunk]:
        async with self._client.messages.stream(**params) as stream:
            async for chunk in _translate_events(stream):
                yield chunk

    return _gen()