扣子 Coze：扣子（Coze）智能体发布为 API 之后，可以通过 /v3/chat 流式对话

扣子（Coze）智能体发布为 API 之后，可以通过 /v3/chat 流式对话。这篇只用 Python 标准库写一个流式客户端，重点处理三件容易出错的事：

• 错误不一定是 HTTP 错误码：实测用错误的 Token 请求，即使设置了 stream=true，返回的也是 HTTP 200 + JSON {"code":4101,...}，而不是 SSE 流；
• 只拼接 type=answer 的增量：verbose、follow_up 等消息不能混进回答；
• 对话失败或需要调用端插件时（chat.failed、requires_action）要明确报错，不能静默返回半截内容。

先看清成品（box 上真实运行）

• 用无效 Token 请求真实的 api.coze.cn：输出 ✗ [4101] The token you entered is incorrect...（logid=20261021185426B9B6341F09390DC62FA2），退出码为 1。
• 本地替身上，回答、推荐问题、用量、失败、requires_action 各场景的测试全部通过；官方 SDK cozepy 也能正确解析替身返回的流。pytest 6 项通过。

合规提醒：扣子中国站（coze.cn）和国际站（coze.com）是两套独立的账号体系，个人访问令牌（PAT）也不能互用。请使用你所在地区对应的站点，不要用代理绕过地区限制，不要共享账号或 Token。Token 只放在环境变量里，不要写进代码或仓库。发给智能体和工作流的数据会由字节跳动的服务处理，涉及个人信息或公司数据时，先确认合规要求。本文不列价格和额度，以官网为准。

运行说明：box 上没有扣子账号，所以没有调用真实的智能体或工作流。真实运行过的部分是：用无效 Token 请求 api.coze.cn，记录真实返回的错误格式。“成功”路径用的是我自己写的本地替身（不是扣子官方程序），它的返回格式取自官方 Python SDK cozepy 0.20.0 的解析代码，并且用 cozepy 实际请求替身来校验格式。我也尝试过自部署开源版 Coze Studio v0.5.1，但 docker compose 拉不到它固定的 minio/minio:RELEASE.2025-06-13T11-33-47Z-cpuv1 镜像（上游 issue #2775），所以放弃了。

做完之后你会得到什么？

需要准备哪些环境？

关键事实

文件结构

59-coze-A-src/
├── coze_chat.py
├── fake_coze.py
└── test_chat.py

第 1 步：客户端

"""Coze v3 对话（流式）最小客户端，仅标准库。
要点（实测）：鉴权失败等错误会以 HTTP 200 + JSON {"code": 4101, ...} 返回，即使请求了 stream=true 也不是 SSE。
所以必须先看 Content-Type，再看 code，不能只看 HTTP 状态码。"""
import json, os, urllib.error, urllib.request

BASE = {"cn": "https://api.coze.cn", "com": "https://api.coze.com"}   # 两个站点账号和 Token 互不通用
UA = "binbho-tutorial/1.0"

class CozeError(RuntimeError):
    def __init__(self, code, msg, logid=None):
        super().__init__(f"[{code}] {msg} (logid={logid})")
        self.code, self.msg, self.logid = code, msg, logid

def _raise_json(body):
    js = json.loads(body)
    if js.get("code", 0) != 0:
        raise CozeError(js.get("code"), js.get("msg"), (js.get("detail") or {}).get("logid"))
    return js

def iter_sse(lines):
    """把 event:/data: 行组装成 (event, data_dict)。空行分隔一个事件。"""
    event, data = None, []
    for raw in lines:
        line = raw.decode("utf-8") if isinstance(raw, bytes) else raw
        line = line.rstrip("\r\n")
        if not line:
            if event:
                payload = "\n".join(data)
                yield event, (json.loads(payload) if payload.startswith(("{", "[")) else payload)
            event, data = None, []
        elif line.startswith("event:"):
            event = line[6:].strip()
        elif line.startswith("data:"):
            data.append(line[5:].lstrip())
    if event:
        payload = "\n".join(data)
        yield event, (json.loads(payload) if payload.startswith(("{", "[")) else payload)

def chat_stream(bot_id, text, user_id, token=None, site="cn", base=None, timeout=120):
    """返回 {"answer", "follow_up", "usage", "chat_id", "conversation_id"}；失败抛 CozeError。"""
    token = token or os.environ["COZE_API_TOKEN"]
    body = {"bot_id": bot_id, "user_id": user_id, "stream": True, "auto_save_history": True,
            "additional_messages": [{"role": "user", "content": text, "content_type": "text"}]}
    req = urllib.request.Request((base or BASE[site]) + "/v3/chat", data=json.dumps(body).encode(), method="POST",
                                 headers={"Authorization": f"Bearer {token}", "Content-Type": "application/json", "User-Agent": UA})
    try:
        resp = urllib.request.urlopen(req, timeout=timeout)
    except urllib.error.HTTPError as e:
        _raise_json(e.read()); raise
    with resp:
        if "text/event-stream" not in resp.headers.get("Content-Type", ""):
            _raise_json(resp.read())                 # HTTP 200 的错误 JSON 走这里
            raise CozeError(-1, "期望 SSE，却收到非流式成功响应")
        out = {"answer": "", "follow_up": [], "usage": None, "chat_id": None, "conversation_id": None}
        for ev, d in iter_sse(resp):
            if ev == "conversation.chat.created":
                out["chat_id"], out["conversation_id"] = d["id"], d["conversation_id"]
            elif ev == "conversation.message.delta" and d.get("type") == "answer":
                out["answer"] += d.get("content", "")
            elif ev == "conversation.message.completed" and d.get("type") == "follow_up":
                out["follow_up"].append(d["content"])
            elif ev == "conversation.chat.completed":
                out["usage"] = d.get("usage")
            elif ev == "conversation.chat.failed":
                err = d.get("last_error") or {}
                raise CozeError(err.get("code"), err.get("msg"))
            elif ev == "conversation.chat.requires_action":
                raise CozeError(-2, "智能体请求调用端插件（requires_action），本客户端未实现 submit_tool_outputs")
            elif ev == "error":
                raise CozeError((d or {}).get("code") if isinstance(d, dict) else -3, str(d))
            elif ev == "done":
                break
        return out

if __name__ == "__main__":
    import sys
    try:
        r = chat_stream(os.environ["COZE_BOT_ID"], " ".join(sys.argv[1:]) or "你好", user_id="demo-user",
                        site=os.environ.get("COZE_SITE", "cn"))
        print(r["answer"]); print("usage:", r["usage"])
    except CozeError as e:
        print("✗", e); sys.exit(1)

第 2 步：本地替身

"""本地替身服务器（我自己写的，不是扣子官方程序）。字段取自官方 cozepy 0.20.0 的数据模型；
错误场景（HTTP 200 + code 4101 的 JSON）照搬对 api.coze.cn 的真实调用结果。"""
import json, threading, urllib.parse
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer

GOOD = "pat_good"
SCENARIO = {"mode": "ok"}      # ok / failed / requires_action

def sse(ev, data):
    return f"event:{ev}\ndata:{json.dumps(data, ensure_ascii=False) if not isinstance(data, str) else data}\n\n".encode()

class H(BaseHTTPRequestHandler):
    def log_message(self, *a):
        pass

    def _json(self, code, obj):
        raw = json.dumps(obj).encode()
        self.send_response(code); self.send_header("Content-Type", "application/json; charset=utf-8")
        self.send_header("Content-Length", str(len(raw))); self.end_headers(); self.wfile.write(raw)

    def do_POST(self):
        body = json.loads(self.rfile.read(int(self.headers["Content-Length"])))
        if self.headers.get("Authorization") != f"Bearer {GOOD}":
            return self._json(200, {"code": 4101, "msg": "The token you entered is incorrect.", "detail": {"logid": "fake-logid"}})
        if urllib.parse.urlsplit(self.path).path != "/v3/chat":     # SDK 会带查询串，如 ?conversation_id=
            return self._json(404, {"code": 404, "msg": "not found"})
        self.send_response(200); self.send_header("Content-Type", "text/event-stream"); self.end_headers()
        chat = {"id": "c1", "conversation_id": "v1", "bot_id": body["bot_id"], "status": "created"}
        w = self.wfile.write
        w(sse("conversation.chat.created", chat))
        w(sse("conversation.chat.in_progress", {**chat, "status": "in_progress"}))
        if SCENARIO["mode"] == "failed":
            w(sse("conversation.chat.failed", {**chat, "status": "failed", "last_error": {"code": 4011, "msg": "余额不足"}}))
            w(sse("done", '"[DONE]"')); return
        if SCENARIO["mode"] == "requires_action":
            w(sse("conversation.chat.requires_action", {**chat, "status": "requires_action"})); return
        for part in ["您好，", "退款会在 ", "3 个工作日内原路退回。"]:
            w(sse("conversation.message.delta", {"id": "m1", "conversation_id": "v1", "chat_id": "c1", "bot_id": body["bot_id"],
                                                 "role": "assistant", "type": "answer", "content": part, "content_type": "text"}))
        w(sse("conversation.message.completed", {"id": "m1", "conversation_id": "v1", "chat_id": "c1", "bot_id": body["bot_id"],
              "role": "assistant", "type": "answer", "content": "您好，退款会在 3 个工作日内原路退回。", "content_type": "text"}))
        w(sse("conversation.message.completed", {"id": "m2", "conversation_id": "v1", "chat_id": "c1", "bot_id": body["bot_id"],
              "role": "assistant", "type": "verbose", "content": "{\"msg_type\":\"generate_answer_finish\"}", "content_type": "text"}))
        w(sse("conversation.message.completed", {"id": "m3", "conversation_id": "v1", "chat_id": "c1", "bot_id": body["bot_id"],
              "role": "assistant", "type": "follow_up", "content": "怎么查询退款进度？", "content_type": "text"}))
        w(sse("conversation.chat.completed", {**chat, "status": "completed",
              "usage": {"token_count": 120, "output_count": 30, "input_count": 90}}))
        w(sse("done", '"[DONE]"'))

def start():
    s = ThreadingHTTPServer(("127.0.0.1", 0), H)
    threading.Thread(target=s.serve_forever, daemon=True).start()
    return s, f"http://127.0.0.1:{s.server_address[1]}"

第 3 步：测试

import warnings
import pytest
import coze_chat, fake_coze

@pytest.fixture(scope="module")
def base():
    s, url = fake_coze.start(); yield url; s.shutdown()

@pytest.fixture(autouse=True)
def reset():
    fake_coze.SCENARIO["mode"] = "ok"

def test_answer_only_and_followup(base):
    r = coze_chat.chat_stream("b1", "怎么退款", "u1", token="pat_good", base=base)
    assert r["answer"] == "您好，退款会在 3 个工作日内原路退回。"            # verbose 消息没有混进答案
    assert r["follow_up"] == ["怎么查询退款进度？"] and r["usage"]["token_count"] == 120 and r["chat_id"] == "c1"

def test_bad_token_is_http200_json(base):
    with pytest.raises(coze_chat.CozeError) as e:
        coze_chat.chat_stream("b1", "hi", "u1", token="pat_bad", base=base)
    assert e.value.code == 4101 and e.value.logid == "fake-logid"

@pytest.mark.parametrize("mode,code", [("failed", 4011), ("requires_action", -2)])
def test_chat_failed(base, mode, code):
    fake_coze.SCENARIO["mode"] = mode
    with pytest.raises(coze_chat.CozeError) as e:
        coze_chat.chat_stream("b1", "hi", "u1", token="pat_good", base=base)
    assert e.value.code == code

def test_sse_parser_multiline():
    evs = list(coze_chat.iter_sse(["event:x\n", 'data:{"a":\n', "data:1}\n", "\n"]))
    assert evs == [("x", {"a": 1})]

def test_official_sdk_parses_fake(base):
    """用官方 cozepy SDK 解析替身服务器的流，确认替身的事件格式与 SDK 一致。"""
    with warnings.catch_warnings():
        warnings.simplefilter("ignore")
        from cozepy import Coze, TokenAuth, Message, ChatEventType, CozeAPIError
    c = Coze(auth=TokenAuth("pat_good"), base_url=base)
    text = "".join(e.message.content for e in c.chat.stream(bot_id="b1", user_id="u1",
                   additional_messages=[Message.build_user_question_text("hi")])
                   if e.event == ChatEventType.CONVERSATION_MESSAGE_DELTA)
    assert text == "您好，退款会在 3 个工作日内原路退回。"
    with pytest.raises(CozeAPIError) as e:
        list(Coze(auth=TokenAuth("pat_bad"), base_url=base).chat.stream(bot_id="b1", user_id="u1",
             additional_messages=[Message.build_user_question_text("hi")]))
    assert e.value.code == 4101

第 4 步：运行

$ export COZE_API_TOKEN=pat_invalid COZE_BOT_ID=7000000000000000000
$ python coze_chat.py 你好
✗ [4101] The token you entered is incorrect. Please double-check and try again. (logid=20261021185426B9B6341F09390DC62FA2)
$ python -m pytest -q
6 passed

换成真实的 Token 和 bot_id，就会输出智能体的回答和 usage（这一步我没有运行）。

调试示例

1. 错误 Token 却拿到了 HTTP 200。 我最初的写法是“HTTP 200 就按 SSE 解析”，结果用错误 Token 请求真实接口时，解析器没有读到任何事件，返回了空回答，看起来像是“成功了，只是没内容”。打印响应头后发现 Content-Type 是 JSON，响应体是 {"code":4101,...}。所以现在先判断 Content-Type：不是 text/event-stream 时，按 JSON 解析 code。

2. 官方 SDK 请求替身时报 404。 用 cozepy 请求替身的测试失败，报错 CozeAPIError: code: 404, msg: not found。我打印了 SDK 实际发出的请求，是 POST /v3/chat?conversation_id=：SDK 总会带上查询串，而替身里写的是 self.path != "/v3/chat"，把整条路径拿去比较了。改成只比较 urlsplit(self.path).path 之后，测试通过。这个问题也说明：替身要用真实客户端去请求，才能发现格式上的偏差。

常见问题

模板规则

• 判断成功要同时看 HTTP 状态码、Content-Type 和 code。
• 报错时保留 logid。
• 只拼接 answer。
• 替身要用官方 SDK 请求一遍，确认格式一致。

项目资源

扣子的错误经常“伪装”成 HTTP 200，客户端要把 code 和 logid 认真当回事。