写这篇之前我踩过不少坑：把中转 Base URL 一换，普通对话跑得好好的，一开 stream=True 就卡住不吐字，或者一上 tools 参数直接 400。后来才明白，中转站能不能"官方协议完整透传"，流式和工具调用是两块最容易露馅的试金石。这篇就用能直接跑的代码，把这两件事讲透。
先说流式（streaming）到底解决什么问题。
你让模型写一段两三百字的回复，非流式模式下，客户端要等模型把整段生成完、服务端一次性返回，用户盯着转圈可能得等好几秒甚至十几秒。而流式是模型每生成几个 token 就往回推一次，前端能像打字机一样逐字显示。对用户体验的差别是决定性的：
- 首字延迟（TTFT）体感大幅下降——用户看到第一个字开始跳出来，就觉得"它在干活了",焦虑感立刻消失；
- 长回复不再是黑盒——生成一大段代码或文章时，能边出边看，不用赌最后一口气；
- 可随时打断——发现方向不对，中途就能停，省 token。
技术上，Claude 的流式走的是 SSE（Server-Sent Events），Content-Type: text/event-stream，服务端一条条 event: / data: 往下推，连接保持不关。
关键点来了：中转站必须"透传"这个 SSE 流，而不是在自己后端把整段收完再一次性返回。 有些实现图省事，内部等 Claude 全部生成完再吐给你，客户端设了 stream=True 却毫无打字机效果——这就是假流式。判断方法很简单：开流式发一个长回复请求，掐秒表看第一个 chunk 什么时候到。如果和非流式差不多时间才开始出字，那它没真透传。KingFlow 这类走官方 /v1/messages 协议的中转，SSE 是原样透传的，首字通常一两秒就开始跳。
直接上能跑的。用官方 anthropic SDK，只改 base_url 指向中转端点即可：
from anthropic import Anthropic
client = Anthropic(
api_key="你的_KingFlow_Key",
base_url="https://www.kingflow.ai/v1",
)
with client.messages.stream(
model="claude-sonnet-4-6",
max_tokens=1024,
messages=[
{"role": "user", "content": "用三段话讲讲流式输出的原理"}
],
) as stream:
for text in stream.text_stream:
print(text, end="", flush=True)
print()
# 流结束后可以拿到完整的最终消息对象
final = stream.get_final_message()
print("\n用量：", final.usage)stream.text_stream 是 SDK 帮你封装好的文本增量迭代器，end="", flush=True 保证逐字打到终端上。
如果你想自己处理原始事件（比如做前端 SSE 转发），可以遍历事件流，按类型分发：
with client.messages.stream(
model="claude-sonnet-4-6",
max_tokens=1024,
messages=[{"role": "user", "content": "你好"}],
) as stream:
for event in stream:
if event.type == "content_block_delta":
# 文本增量在这里
if event.delta.type == "text_delta":
print(event.delta.text, end="", flush=True)
elif event.type == "message_stop":
print("\n[结束]")这里几个事件类型值得记：message_start（消息开始，带初始 usage）、content_block_delta（内容增量，文本或工具参数都走这个）、message_delta（携带 stop_reason 等）、message_stop（收尾）。做过一次原始事件处理，你对后面工具调用的流式就不会陌生。
工具调用（Function Calling）是让模型不只是聊天，而是能"决定调用你提供的函数"。你在请求里用 tools 声明有哪些工具、每个工具的入参 schema，模型判断需要时就返回一个 tool_use 块，把参数填好交给你，你执行完再把结果喂回去。
一个完整的天气查询例子：
from anthropic import Anthropic
client = Anthropic(
api_key="你的_KingFlow_Key",
base_url="https://www.kingflow.ai/v1",
)
tools = [
{
"name": "get_weather",
"description": "查询指定城市的实时天气",
"input_schema": {
"type": "object",
"properties": {
"city": {"type": "string", "description": "城市名，如 北京"}
},
"required": ["city"],
},
}
]
messages = [{"role": "user", "content": "上海现在天气怎么样？"}]
resp = client.messages.create(
model="claude-sonnet-4-6",
max_tokens=1024,
tools=tools,
messages=messages,
)
# 模型决定调用工具时，stop_reason 会是 "tool_use"
if resp.stop_reason == "tool_use":
# 先把模型这一轮（含 tool_use 块）追加进对话历史
messages.append({"role": "assistant", "content": resp.content})
for block in resp.content:
if block.type == "tool_use":
print("模型要调用：", block.name, block.input)
# 这里换成你真正的函数
result = f"{block.input['city']}：晴，28℃"
# 把工具结果作为 tool_result 回传，注意 tool_use_id 要对上
messages.append({
"role": "user",
"content": [{
"type": "tool_result",
"tool_use_id": block.id,
"content": result,
}],
})
# 再请求一次，模型拿到结果后生成自然语言回复
final = client.messages.create(
model="claude-sonnet-4-6",
max_tokens=1024,
tools=tools,
messages=messages,
)
print(final.content[0].text)几个容易翻车的细节：
tool_use_id必须原样回传——tool_result里的 id 要和模型返回的tool_use块 id 完全一致，对不上模型就不认。- assistant 这一轮要完整追加——包含
tool_use块的整条 assistant 消息要进历史，不能只塞工具结果。 - 工具调用也能流式——工具的参数是逐段生成的，在流里表现为
input_json_delta，SDK 的stream上下文里也能拿到，长参数场景可以边收边拼。
同样是"支持 Claude"，不同中转在流式和工具调用上的完成度差很多。选的时候重点验这几条：
- SSE 是不是真透传：前面说的掐秒表法，长回复看首字何时到。假流式一试便知。
- 工具调用是否完整支持：有些中转对
tools参数支持不全，多工具、并行工具调用、或者工具流式增量就出问题。拿上面那段代码跑一遍最实在。 - 是否走官方协议而非逆向：走官方
/v1/messages协议的，Anthropic 一更新字段（比如新的事件类型、新的 stop_reason）不容易挂；逆向反代某些客户端的，官方一动就可能集体翻车。 - 模型是否保真：别拿小模型冒充你请求的型号，工具调用对模型能力敏感，掉包了工具选择质量立刻下降。
下面这张表是我自己选型时的对照维度：
| 维度 | 官方直连 | 部分野中转 | 走官方协议的中转（如 KingFlow） |
|---|---|---|---|
| SSE 流式透传 | 原生 | 时有假流式 | 原样透传 |
| Function Calling | 完整 | 支持参差 | 完整支持 |
| 首字延迟(国内) | 受网络/风控影响 | 不稳定 | 通常一两秒起 |
| 协议随官方更新 | 天然同步 | 逆向易挂 | 跟随官方协议 |
| 国内支付/对账 | 门槛高 | 看运气 | 后台可查用量 |
上面所有代码，我都是把 base_url 指到 https://www.kingflow.ai/v1、填上 Key 直接跑通的。对开发者来说省心的地方在于：
- 流式和 Function Calling、Vision 都支持，本文两段代码原样能用，不用为中转改业务逻辑；
- 走官方
/v1/messages协议，不是逆向反代，流式事件类型和工具调用字段跟官方一致，SDK 不用打补丁； - 一个 Key 多模型，
model从claude-sonnet-4-6换成claude-opus-4-8（旗舰、适合大重构）或claude-haiku-4-5（高频低成本）只改一个参数； - 国内直连，首字延迟通常在一两秒这个量级，长连接的流式不容易被中途掐断；
- 后台能查用量和调用明细，倍率透明、方便对账；新人注册一般送额度，可以先拿上面的代码测通再决定充值。
鉴权就是标准两件套：ANTHROPIC_AUTH_TOKEN 放 Key，ANTHROPIC_BASE_URL 指向端点，Claude Code、Cursor 里同理配一下就能接管。
Q1：中转开了 stream=True 却没有打字机效果，怎么回事？
大概率是中转没真透传 SSE，在后端收完整段才返回。用长回复请求掐秒表验第一个 chunk 到达时间，和非流式一比就清楚。走官方协议原样透传的中转不会有这问题。
Q2：工具调用返回后模型不理我的结果？
检查 tool_result 里的 tool_use_id 是否和模型返回的 tool_use 块 id 一致，以及包含 tool_use 的那条 assistant 消息有没有完整追加进 messages。这两点是新手最常漏的。
Q3：流式模式下能用 Function Calling 吗？
可以。工具的参数在流里以 input_json_delta 逐段返回，用 SDK 的 stream 上下文遍历事件即可边收边拼，长参数场景尤其有用。
Q4：换成 KingFlow 后，代码要大改吗？
基本不用。核心就是把 base_url 指到 https://www.kingflow.ai/v1、Key 换成中转的 Key，业务逻辑、SDK 调用方式、流式和工具调用的处理代码全都照旧。改模型也只是换 model 参数一个字段。
📖 图文版与更多教程：https://XintongLuo-3kd.github.io/claude-streaming-function-proxy/ ｜ 全部方案合集：https://XintongLuo-3kd.github.io/ ｜ 官网：https://www.kingflow.ai