通俗版 Claude Code 文档

结构照原 docs,内容改成真能照着做的人话版。

查看原始文档 目录顺序与官方保持一致

Input and output

Stream responses in real-time

Stream responses in real-time 这一页讲的,就是 Stream responses in real-time 这件事在 Claude Code 里到底怎么用。

页面信息

对应原页

Stream responses in real-time

页面性质

第三方中文解释页

使用建议

先看人话解释,再对照原页命令和代码

这页不是官方原文,而是顺着官方文档结构做的中文解释版。命令、参数、配置名这些硬东西尽量保留,解释部分则尽量讲成人能照着做的话。

如果你碰到特别敏感的配置、权限或企业环境差异,最好顺手点上面的“查看原始文档”再核一遍。

这一页先讲明白

这页主要讲 Stream responses in real-time:Get real-time responses from the Agent SDK as text and tool calls stream in

你可以把它当成"Input and output"这块里专门管这一摊事的说明书。

你可以把"Stream responses in real-time"理解成 Input and output 这一栏里的一把专门工具。这页不是让你背书,而是教你什么时候该把这把工具拿出来。

原文这页大多会按 Enable streaming output、StreamEvent reference、Message flow、Stream text responses 这些环节往下讲。

翻成人话,大概就是:Enable streaming output

第一,先别一上来全开全配。先按最小一步试通,确认没跑偏,再继续往下加。

第二,命令、配置名、参数名这些硬东西尽量保留原样。人话解释是帮你听懂,不是帮你改关键字。

第三,照着原文这几个环节挨个过:Enable streaming output -> StreamEvent reference -> Message flow -> Stream text responses。像下地先看水路、再试机器、再正式开干,一步一步最稳。

关键片段

原页关键片段:Enable streaming output

下面这块是这一段最值钱的原文样板,先对着看一眼。

from claude_agent_sdk import query, ClaudeAgentOptions
from claude_agent_sdk.types import StreamEvent
import asyncio


async def stream_response():
    options = ClaudeAgentOptions(
        include_partial_messages=True,
        allowed_tools=["Bash", "Read"],
    )

    async for message in query(prompt="List the files in my project", options=options):
        if isinstance(message, StreamEvent):
            event = message.event
            if event.get("type") == "content_block_delta":
                delta = event.get("delta", {})
                if delta.get("type") == "text_delta":
                    print(delta.get("text", ""), end="", flush=True)


asyncio.run(stream_response())
关键片段

原页关键片段:StreamEvent reference

"StreamEvent reference"这一段里最要紧的原始写法在下面,先看它怎么落地。

@dataclass
class StreamEvent:
    uuid: str  # Unique identifier for this event
    session_id: str  # Session identifier
    event: dict[str, Any]  # The raw Claude API stream event
    parent_tool_use_id: str | None  # Parent tool ID if from a subagent
关键片段

原页关键片段:Message flow

"Message flow"这一段里最要紧的原始写法在下面,先看它怎么落地。

StreamEvent (message_start)
StreamEvent (content_block_start) - text block
StreamEvent (content_block_delta) - text chunks...
StreamEvent (content_block_stop)
StreamEvent (content_block_start) - tool_use block
StreamEvent (content_block_delta) - tool input chunks...
StreamEvent (content_block_stop)
StreamEvent (message_delta)
StreamEvent (message_stop)
AssistantMessage - complete message with all content
... tool executes ...
... more streaming events for next turn ...
ResultMessage - final result
关键片段

原页关键片段:Stream text responses

先看下面这块原始片段,等会儿再回头看解释会顺得多。

from claude_agent_sdk import query, ClaudeAgentOptions
from claude_agent_sdk.types import StreamEvent
import asyncio


async def stream_text():
    options = ClaudeAgentOptions(include_partial_messages=True)

    async for message in query(prompt="Explain how databases work", options=options):
        if isinstance(message, StreamEvent):
            event = message.event
            if event.get("type") == "content_block_delta":
                delta = event.get("delta", {})
                if delta.get("type") == "text_delta":
                    # Print each text chunk as it arrives
                    print(delta.get("text", ""), end="", flush=True)

    print()  # Final newline


asyncio.run(stream_text())
关键片段

原页关键片段:Stream tool calls

"Stream tool calls"这一段里最要紧的原始写法在下面,先看它怎么落地。

from claude_agent_sdk import query, ClaudeAgentOptions
from claude_agent_sdk.types import StreamEvent
import asyncio


async def stream_tool_calls():
    options = ClaudeAgentOptions(
        include_partial_messages=True,
        allowed_tools=["Read", "Bash"],
    )

    # Track the current tool and accumulate its input JSON
    current_tool = None
    tool_input = ""

    async for message in query(prompt="Read the README.md file", options=options):
        if isinstance(message, StreamEvent):
            event = message.event
            event_type = event.get("type")

            if event_type == "content_block_start":
                # New tool call is starting
                content_block = event.get("content_block", {})
                if content_block.get("type") == "tool_use":
                    current_tool = content_block.get("name")
                    tool_input = ""
                    print(f"Starting tool: {current_tool}")

            elif event_type == "content_block_delta":
                delta = event.get("delta", {})
                if delta.get("type") == "input_json_delta":
                    # Accumulate JSON input as it streams in
                    chunk = delta.get("partial_json", "")
                    tool_input += chunk
                    print(f"  Input chunk: {chunk}")

            elif event_type == "content_block_stop":
                # Tool call complete - show final input
                if current_tool:
                    print(f"Tool {current_tool} called with: {tool_input}")
                    current_tool = None


asyncio.run(stream_tool_calls())
关键片段

原页关键片段:Build a streaming UI

"Build a streaming UI"这一段里最要紧的原始写法在下面,先看它怎么落地。

from claude_agent_sdk import query, ClaudeAgentOptions, ResultMessage
from claude_agent_sdk.types import StreamEvent
import asyncio
import sys


async def streaming_ui():
    options = ClaudeAgentOptions(
        include_partial_messages=True,
        allowed_tools=["Read", "Bash", "Grep"],
    )

    # Track whether we're currently in a tool call
    in_tool = False

    async for message in query(
        prompt="Find all TODO comments in the codebase", options=options
    ):
        if isinstance(message, StreamEvent):
            event = message.event
            event_type = event.get("type")

            if event_type == "content_block_start":
                content_block = event.get("content_block", {})
                if content_block.get("type") == "tool_use":
                    # Tool call is starting - show status indicator
                    tool_name = content_block.get("name")
                    print(f"\n[Using {tool_name}...]", end="", flush=True)
                    in_tool = True

            elif event_type == "content_block_delta":
                delta = event.get("delta", {})
                # Only stream text when not executing a tool
                if delta.get("type") == "text_delta" and not in_tool:
                    sys.stdout.write(delta.get("text", ""))
                    sys.stdout.flush()

            elif event_type == "content_block_stop":
                if in_tool:
                    # Tool call finished
                    print(" done", flush=True)
                    in_tool = False

        elif isinstance(message, ResultMessage):
            # Agent finished all work
            print(f"\n\n--- Complete ---")


asyncio.run(streaming_ui())

预留广告位

正文中段响应式广告 等你后面真接 AdSense,这里再放正式广告。

Documentation Index

这里不是让你背"Documentation Index"这个词,而是让你看它真干活时怎么使。

看这段时要特别盯工具和权限边界,别为了省事一把全开。

Enable streaming output

这一段主要是在拧开关,地方和顺序都别搞错。

关键片段

Enable streaming output

下面这块是这一段最值钱的原文样板,先对着看一眼。

from claude_agent_sdk import query, ClaudeAgentOptions
from claude_agent_sdk.types import StreamEvent
import asyncio


async def stream_response():
    options = ClaudeAgentOptions(
        include_partial_messages=True,
        allowed_tools=["Bash", "Read"],
    )

    async for message in query(prompt="List the files in my project", options=options):
        if isinstance(message, StreamEvent):
            event = message.event
            if event.get("type") == "content_block_delta":
                delta = event.get("delta", {})
                if delta.get("type") == "text_delta":
                    print(delta.get("text", ""), end="", flush=True)


asyncio.run(stream_response())

StreamEvent reference

这种内容最怕想当然,老老实实照它列的格式走最稳。

关键片段

StreamEvent reference

"StreamEvent reference"这一段里最要紧的原始写法在下面,先看它怎么落地。

@dataclass
class StreamEvent:
    uuid: str  # Unique identifier for this event
    session_id: str  # Session identifier
    event: dict[str, Any]  # The raw Claude API stream event
    parent_tool_use_id: str | None  # Parent tool ID if from a subagent

Message flow

这一块主要是在说"Message flow"真到手上该怎么用,哪里最容易踩坑。

这里还牵扯作用域,意思就是这条规则到底管当前项目、你个人,还是只管这一趟会话。

关键片段

Message flow

"Message flow"这一段里最要紧的原始写法在下面,先看它怎么落地。

StreamEvent (message_start)
StreamEvent (content_block_start) - text block
StreamEvent (content_block_delta) - text chunks...
StreamEvent (content_block_stop)
StreamEvent (content_block_start) - tool_use block
StreamEvent (content_block_delta) - tool input chunks...
StreamEvent (content_block_stop)
StreamEvent (message_delta)
StreamEvent (message_stop)
AssistantMessage - complete message with all content
... tool executes ...
... more streaming events for next turn ...
ResultMessage - final result

Stream text responses

看到这里,就把"Stream text responses"当成一件真要上手的活来看。

关键片段

Stream text responses

先看下面这块原始片段,等会儿再回头看解释会顺得多。

from claude_agent_sdk import query, ClaudeAgentOptions
from claude_agent_sdk.types import StreamEvent
import asyncio


async def stream_text():
    options = ClaudeAgentOptions(include_partial_messages=True)

    async for message in query(prompt="Explain how databases work", options=options):
        if isinstance(message, StreamEvent):
            event = message.event
            if event.get("type") == "content_block_delta":
                delta = event.get("delta", {})
                if delta.get("type") == "text_delta":
                    # Print each text chunk as it arrives
                    print(delta.get("text", ""), end="", flush=True)

    print()  # Final newline


asyncio.run(stream_text())

Stream tool calls

这一块主要是在说"Stream tool calls"真到手上该怎么用,哪里最容易踩坑。

看这段时要特别盯工具和权限边界,别为了省事一把全开。

关键片段

Stream tool calls

"Stream tool calls"这一段里最要紧的原始写法在下面,先看它怎么落地。

from claude_agent_sdk import query, ClaudeAgentOptions
from claude_agent_sdk.types import StreamEvent
import asyncio


async def stream_tool_calls():
    options = ClaudeAgentOptions(
        include_partial_messages=True,
        allowed_tools=["Read", "Bash"],
    )

    # Track the current tool and accumulate its input JSON
    current_tool = None
    tool_input = ""

    async for message in query(prompt="Read the README.md file", options=options):
        if isinstance(message, StreamEvent):
            event = message.event
            event_type = event.get("type")

            if event_type == "content_block_start":
                # New tool call is starting
                content_block = event.get("content_block", {})
                if content_block.get("type") == "tool_use":
                    current_tool = content_block.get("name")
                    tool_input = ""
                    print(f"Starting tool: {current_tool}")

            elif event_type == "content_block_delta":
                delta = event.get("delta", {})
                if delta.get("type") == "input_json_delta":
                    # Accumulate JSON input as it streams in
                    chunk = delta.get("partial_json", "")
                    tool_input += chunk
                    print(f"  Input chunk: {chunk}")

            elif event_type == "content_block_stop":
                # Tool call complete - show final input
                if current_tool:
                    print(f"Tool {current_tool} called with: {tool_input}")
                    current_tool = None


asyncio.run(stream_tool_calls())

Build a streaming UI

这一段不只是挂个标题,它是在说明"Build a streaming UI"这一块到底负责什么。

看这段时要特别盯工具和权限边界,别为了省事一把全开。

关键片段

Build a streaming UI

"Build a streaming UI"这一段里最要紧的原始写法在下面,先看它怎么落地。

from claude_agent_sdk import query, ClaudeAgentOptions, ResultMessage
from claude_agent_sdk.types import StreamEvent
import asyncio
import sys


async def streaming_ui():
    options = ClaudeAgentOptions(
        include_partial_messages=True,
        allowed_tools=["Read", "Bash", "Grep"],
    )

    # Track whether we're currently in a tool call
    in_tool = False

    async for message in query(
        prompt="Find all TODO comments in the codebase", options=options
    ):
        if isinstance(message, StreamEvent):
            event = message.event
            event_type = event.get("type")

            if event_type == "content_block_start":
                content_block = event.get("content_block", {})
                if content_block.get("type") == "tool_use":
                    # Tool call is starting - show status indicator
                    tool_name = content_block.get("name")
                    print(f"\n[Using {tool_name}...]", end="", flush=True)
                    in_tool = True

            elif event_type == "content_block_delta":
                delta = event.get("delta", {})
                # Only stream text when not executing a tool
                if delta.get("type") == "text_delta" and not in_tool:
                    sys.stdout.write(delta.get("text", ""))
                    sys.stdout.flush()

            elif event_type == "content_block_stop":
                if in_tool:
                    # Tool call finished
                    print(" done", flush=True)
                    in_tool = False

        elif isinstance(message, ResultMessage):
            # Agent finished all work
            print(f"\n\n--- Complete ---")


asyncio.run(streaming_ui())

Known limitations

这里不是让你背"Known limitations"这个词,而是让你看它真干活时怎么使。

Next steps

这里不是让你背"Next steps"这个词,而是让你看它真干活时怎么使。

看这段时要特别盯工具和权限边界,别为了省事一把全开。

照着做一遍

如果你不想来回翻,就先照这几步顺着做。

每做完一步就看一下结果,再决定要不要继续往下。

关键片段

第 1 步:Enable streaming output

下面这块是这一段最值钱的原文样板,先对着看一眼。

from claude_agent_sdk import query, ClaudeAgentOptions
from claude_agent_sdk.types import StreamEvent
import asyncio


async def stream_response():
    options = ClaudeAgentOptions(
        include_partial_messages=True,
        allowed_tools=["Bash", "Read"],
    )

    async for message in query(prompt="List the files in my project", options=options):
        if isinstance(message, StreamEvent):
            event = message.event
            if event.get("type") == "content_block_delta":
                delta = event.get("delta", {})
                if delta.get("type") == "text_delta":
                    print(delta.get("text", ""), end="", flush=True)


asyncio.run(stream_response())
关键片段

第 2 步:StreamEvent reference

"StreamEvent reference"这一段里最要紧的原始写法在下面,先看它怎么落地。

@dataclass
class StreamEvent:
    uuid: str  # Unique identifier for this event
    session_id: str  # Session identifier
    event: dict[str, Any]  # The raw Claude API stream event
    parent_tool_use_id: str | None  # Parent tool ID if from a subagent
关键片段

第 3 步:Message flow

"Message flow"这一段里最要紧的原始写法在下面,先看它怎么落地。

StreamEvent (message_start)
StreamEvent (content_block_start) - text block
StreamEvent (content_block_delta) - text chunks...
StreamEvent (content_block_stop)
StreamEvent (content_block_start) - tool_use block
StreamEvent (content_block_delta) - tool input chunks...
StreamEvent (content_block_stop)
StreamEvent (message_delta)
StreamEvent (message_stop)
AssistantMessage - complete message with all content
... tool executes ...
... more streaming events for next turn ...
ResultMessage - final result
关键片段

第 4 步:Stream text responses

先看下面这块原始片段,等会儿再回头看解释会顺得多。

from claude_agent_sdk import query, ClaudeAgentOptions
from claude_agent_sdk.types import StreamEvent
import asyncio


async def stream_text():
    options = ClaudeAgentOptions(include_partial_messages=True)

    async for message in query(prompt="Explain how databases work", options=options):
        if isinstance(message, StreamEvent):
            event = message.event
            if event.get("type") == "content_block_delta":
                delta = event.get("delta", {})
                if delta.get("type") == "text_delta":
                    # Print each text chunk as it arrives
                    print(delta.get("text", ""), end="", flush=True)

    print()  # Final newline


asyncio.run(stream_text())

一眼看懂这一页

先把这页到底在讲什么看明白,再去碰具体命令和配置,最不容易绕晕。

Stream responses in real-time
   |
   v
这是 Input and output 里的一摊要紧活
   |
   v
先弄懂,再下手

文末提醒

这站会按官方 docs 的导航和内容变化继续重生成,原站加页、删页、改页时,这里会跟着更新。

人话解释会尽量顺着原页往下讲,但命令、参数名、配置名这些硬东西还是保留原样,免得你抄过去跑不起来。