扣子 Coze：扣子工作流可以通过 /v1/workflow/run 调用

扣子工作流可以通过 /v1/workflow/run 调用。这篇写一个批量运行器：从 CSV 读取多组参数，分别运行同一个工作流，把每一行的结果、execute_id 和 debug_url 写入结果 CSV。 设计要点：

• 返回体里的 data 是 JSON 字符串，需要再解析一次；
• 只对网络错误和 5xx 退避重试；鉴权错误、参数错误这类业务错误不重试，以免白白消耗额度；
• 并发限制为 2。

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

• 用无效 Token 请求真实接口：3 行都失败，错误是 [4100] authentication is invalid（真实接口返回的是 HTTP 401），退出码为 1。
• 本地替身上：2 行成功、1 行参数缺失（4000）；连续两次 502 之后第 3 次成功；鉴权错误只请求了 1 次。官方 SDK 也能正确解析替身的返回。pytest 4 项通过。

合规提醒：扣子中国站（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），所以放弃了。

做完之后你会得到什么？

需要准备哪些环境？

Python 3.9 以上；测试需要 pip install pytest cozepy。真实运行需要一个已发布的工作流 ID，以及有运行权限的 Token。

关键事实

文件结构

59-coze-B-src/
├── batch_run.py
├── fake_workflow.py
├── inputs.csv
└── test_batch.py

第 1 步：批量运行器

"""批量运行扣子工作流：/v1/workflow/run（同步），标准库实现。
- 返回体的 data 是 JSON **字符串**（官方 SDK 也是原样返回），需要再 json.loads 一次
- 业务错误（code > 0，包括 HTTP 200 的错误）不重试；只有网络错误和 HTTP 5xx 才退避重试
- 每行记录 execute_id 和 debug_url，便于在扣子后台追查
"""
import concurrent.futures as cf, csv, json, os, sys, time, urllib.error, urllib.request

BASE = os.environ.get("COZE_API_BASE", "https://api.coze.cn")
UA = "binbho-tutorial/1.0"
MAX_WORKERS, MAX_TRIES = 2, 3

class Retryable(Exception):
    pass

def run_once(workflow_id, params, token, base=BASE, timeout=120):
    req = urllib.request.Request(base + "/v1/workflow/run", method="POST",
        data=json.dumps({"workflow_id": workflow_id, "parameters": params}).encode(),
        headers={"Authorization": f"Bearer {token}", "Content-Type": "application/json", "User-Agent": UA})
    try:
        with urllib.request.urlopen(req, timeout=timeout) as r:
            body = json.load(r)
    except urllib.error.HTTPError as e:
        if e.code >= 500:
            raise Retryable(f"HTTP {e.code}")
        body = json.loads(e.read() or b"{}")
    except (urllib.error.URLError, TimeoutError) as e:
        raise Retryable(str(e))
    if body.get("code", 0) != 0:
        return {"status": "error", "code": body.get("code"), "msg": body.get("msg"),
                "logid": (body.get("detail") or {}).get("logid")}
    data = body.get("data")
    try:
        output = json.loads(data) if isinstance(data, str) else data
    except ValueError:
        output = data
    return {"status": "ok", "output": output, "execute_id": body.get("execute_id"),
            "debug_url": body.get("debug_url"), "usage": body.get("usage")}

def run_with_retry(*a, sleep=time.sleep, **k):
    for i in range(MAX_TRIES):
        try:
            return run_once(*a, **k)
        except Retryable as e:
            if i == MAX_TRIES - 1:
                return {"status": "error", "code": "network", "msg": str(e)}
            sleep(2 ** i)

def main(src, dst, workflow_id, token, base=BASE):
    rows = list(csv.DictReader(open(src, encoding="utf-8-sig")))
    with cf.ThreadPoolExecutor(MAX_WORKERS) as ex:
        results = list(ex.map(lambda r: run_with_retry(workflow_id, r, token, base=base), rows))
    with open(dst, "w", encoding="utf-8", newline="") as f:
        w = csv.writer(f); w.writerow(["row", "status", "output", "code", "msg", "execute_id", "debug_url"])
        for i, x in enumerate(results, 1):
            w.writerow([i, x["status"], json.dumps(x.get("output"), ensure_ascii=False) if "output" in x else "",
                        x.get("code", ""), x.get("msg", ""), x.get("execute_id", ""), x.get("debug_url", "")])
    ok = sum(x["status"] == "ok" for x in results)
    print(f"完成 {ok}/{len(results)}，结果写入 {dst}")
    for i, x in enumerate(results, 1):
        if x["status"] != "ok":
            print(f"  第 {i} 行失败：[{x.get('code')}] {x.get('msg')}")
    return 0 if ok == len(results) else 1

if __name__ == "__main__":
    sys.exit(main(sys.argv[1], sys.argv[2], os.environ["COZE_WORKFLOW_ID"], os.environ["COZE_API_TOKEN"]))

第 2 步：替身和输入

"""本地替身（我自己写的，不是扣子官方程序）。返回格式取自官方 cozepy 0.20.0 的解析代码：
{"code":0,"msg":"","data":"<JSON 字符串>","debug_url":...,"execute_id":...,"usage":{...}}；鉴权失败照搬真实 api.coze.cn 的 401/4100。"""
import json, threading
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer

STATE = {"fail_5xx": 0, "calls": 0}

class H(BaseHTTPRequestHandler):
    def log_message(self, *a):
        pass
    def _json(self, code, obj):
        raw = json.dumps(obj, ensure_ascii=False).encode()
        self.send_response(code); self.send_header("Content-Type", "application/json"); self.end_headers(); self.wfile.write(raw)
    def do_POST(self):
        STATE["calls"] += 1
        body = json.loads(self.rfile.read(int(self.headers["Content-Length"])))
        if self.headers.get("Authorization") != "Bearer pat_good":
            return self._json(401, {"code": 4100, "msg": "authentication is invalid", "detail": {"logid": "fake"}})
        if STATE["fail_5xx"] > 0:
            STATE["fail_5xx"] -= 1
            return self._json(502, {"code": 502, "msg": "bad gateway"})
        p = body.get("parameters") or {}
        if not p.get("city"):
            return self._json(200, {"code": 4000, "msg": "parameter city is required", "detail": {"logid": "fake"}})
        out = {"output": f"{p['city']}：晴，{len(p['city']) * 7 + 10}°C"}
        self._json(200, {"code": 0, "msg": "", "data": json.dumps(out, ensure_ascii=False),
                         "debug_url": f"https://www.coze.cn/work_flow?execute_id=e-{p['city']}", "execute_id": f"e-{p['city']}",
                         "usage": {"token_count": 50, "output_count": 10, "input_count": 40}})

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]}"

city
杭州
北京
""

第 3 步：测试

import csv, warnings
import pytest
import batch_run, fake_workflow

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

def test_batch(base, tmp_path):
    out = tmp_path / "out.csv"
    code = batch_run.main("inputs.csv", out, "wf1", "pat_good", base=base)
    rows = list(csv.DictReader(open(out, encoding="utf-8")))
    assert code == 1 and [r["status"] for r in rows] == ["ok", "ok", "error"]
    assert '"output": "杭州：晴，24°C"' in rows[0]["output"]          # data 字符串已二次解析
    assert rows[0]["execute_id"] == "e-杭州" and rows[2]["code"] == "4000"

def test_retry_5xx_then_ok(base):
    fake_workflow.STATE.update(fail_5xx=2, calls=0)
    r = batch_run.run_with_retry("wf1", {"city": "上海"}, "pat_good", base=base, sleep=lambda s: None)
    assert r["status"] == "ok" and fake_workflow.STATE["calls"] == 3

def test_no_retry_on_auth_error(base):
    fake_workflow.STATE.update(fail_5xx=0, calls=0)
    r = batch_run.run_with_retry("wf1", {"city": "上海"}, "pat_bad", base=base, sleep=lambda s: None)
    assert r["code"] == 4100 and fake_workflow.STATE["calls"] == 1

def test_official_sdk_parses_fake(base):
    with warnings.catch_warnings():
        warnings.simplefilter("ignore")
        from cozepy import Coze, TokenAuth
    r = Coze(auth=TokenAuth("pat_good"), base_url=base).workflows.runs.create(workflow_id="wf1", parameters={"city": "广州"})
    assert r.execute_id == "e-广州" and "广州" in r.data           # SDK 也把 data 当字符串返回

第 4 步：运行

$ COZE_API_TOKEN=pat_invalid COZE_WORKFLOW_ID=7000000000000000000 python batch_run.py inputs.csv out.csv
完成 0/3，结果写入 out.csv
  第 1 行失败：[4100] authentication is invalid
  第 2 行失败：[4100] authentication is invalid
  第 3 行失败：[4100] authentication is invalid
$ python -m pytest -q
4 passed

调试示例

1. CSV 里的空行被悄悄丢掉了。 我最初用空行表示“城市为空”的测试用例，写的是 city\n杭州\n\n北京。实际读取的结果是 [{'city': '杭州'}, {'city': '北京'}]：csv.DictReader 会跳过空行，这一行既不运行，也不会报错。想测试空值，必须写成 ""（示例输入的第 3 行就是这样写的），这样它才会进入结果表，得到 4000。

2. 鉴权错误不能重试。 真实接口对错误 Token 返回的是 HTTP 401（/v3/chat 返回的是 200），而 urllib 会把 401 当作异常抛出。我把所有 HTTPError 都归为非重试错误，只对 >= 500 的错误重试；test_no_retry_on_auth_error 断言替身只被请求了 1 次。

常见问题

模板规则

• 业务错误不重试，只对网络错误和 5xx 退避重试。
• 每行记录 execute_id 和 debug_url。
• 空值写成 ""。
• 并发数不要超过你账号允许的限额（以官网为准）。

项目资源

批量跑工作流时，最重要的是每一行都能追查：成功的有 execute_id，失败的有 code。