Class ClaudeApiClient

mini_claude_code/api_client.py:143–215

源码: main.rs:2384-2500 (AnthropicRuntimeClient) emit_output: 是否实时打印文本到终端。 CC 的做法: streaming 时同步渲染 markdown 到 stdout (main.rs:2436-2460)。 收到 TextDelta 时立刻 print，不等整个响应结束。

# ============================================================

class ClaudeApiClient(ApiClient):
    """源码: main.rs:2384-2500 (AnthropicRuntimeClient)

    emit_output: 是否实时打印文本到终端。
    CC 的做法: streaming 时同步渲染 markdown 到 stdout (main.rs:2436-2460)。
    收到 TextDelta 时立刻 print，不等整个响应结束。
    """

    def __init__(
        self,
        api_key: str,
        model: str,
        tools: list[dict] | None = None,
        emit_output: bool = True,
    ):
        import anthropic
        self.client = anthropic.Anthropic(api_key=api_key)
        self.model = model
        self.tools = tools or []
        self.emit_output = emit_output

    def stream(self, system_prompt: list[str], messages: list[Message]) -> list[AssistantEvent]:
        import sys as _sys

        converted = _convert_messages(messages)
        system_prompt_str = '\n'.join(system_prompt)

        kwargs: dict = {
            "model": self.model,
            "max_tokens": 8096,
            "system": system_prompt_str,
            "messages": converted,
        }
        if self.tools:
            kwargs["tools"] = self.tools

        events: list[AssistantEvent] = []
        streaming_text = False  # 追踪是否正在流式输出文本

        with self.client.messages.stream(**kwargs) as s:
            for event in s:
                if event.type == 'content_block_delta':
                    if event.delta.type == 'text_delta':
                        events.append(TextDeltaEvent(text=event.delta.text))
                        # 实时流式输出 — CC 的 emit_output
                        if self.emit_output:
                            if not streaming_text:
                                _sys.stdout.write("\n")
                                streaming_text = True
                            _sys.stdout.write(event.delta.text)
                            _sys.stdout.flush()

                elif event.type == 'content_block_stop':
                    if event.content_block.type == 'tool_use':
                        # 工具调用前换行
                        if self.emit_output and streaming_text:
                            _sys.stdout.write("\n")
                            _sys.stdout.flush()