[ PROMPT_NODE_23838 ]
streaming
[ SKILL_DOCUMENTATION ]
# Streaming — Python
## 快速开始
python
with client.messages.stream(
model="claude-opus-4-7",
max_tokens=64000,
messages=[{"role": "user", "content": "Write a story"}]
) as stream:
for text in stream.text_stream:
print(text, end="", flush=True)
### 异步
python
async with async_client.messages.stream(
model="claude-opus-4-7",
max_tokens=64000,
messages=[{"role": "user", "content": "Write a story"}]
) as stream:
async for text in stream.text_stream:
print(text, end="", flush=True)
---
## 处理不同内容类型
Claude 可能会返回文本、思考块或工具使用。请相应地处理每种类型：
> **Opus 4.7 / Opus 4.6：** 使用 `thinking: {type: "adaptive"}`。在旧模型上，请改用 `thinking: {type: "enabled", budget_tokens: N}`。
python
with client.messages.stream(
model="claude-opus-4-7",
max_tokens=64000,
thinking={"type": "adaptive"},
messages=[{"role": "user", "content": "Analyze this problem"}]
) as stream:
for event in stream:
if event.type == "content_block_start":
if event.content_block.type == "thinking":
print("n[Thinking...]")
elif event.content_block.type == "text":
print("n[Response:]")
elif event.type == "content_block_delta":
if event.delta.type == "thinking_delta":
print(event.delta.thinking, end="", flush=True)
elif event.delta.type == "text_delta":
print(event.delta.text, end="", flush=True)
---
## 使用工具进行流式传输
Python 工具运行器目前返回完整消息。如果您需要针对每个 Token 进行工具流式传输，请在手动循环中使用流式传输：
python
with client.messages.stream(
model="claude-opus-4-7",
max_tokens=64000,
tools=tools,
messages=messages
) as stream:
for text in stream.text_stream:
print(text, end="", flush=True)
response = stream.get_final_message()
# 如果 response.stop_reason == "tool_use"，则继续执行工具
---
## 获取最终消息
python
with client.messages.stream(
model="claude-opus-4-7",
max_tokens=64000,
messages=[{"role": "user", "content": "Hello"}]
) as stream:
for text in stream.text_stream:
print(text, end="", flush=True)
# 流式传输后获取完整消息
final_message = stream.get_final_message()
print(f"nnTokens used: {final_mes
粤公网安备44030002003366号