当前位置：首页 > news >正文

通过python API来调用扣子coze的工作流

news 2026/3/26 17:15:13

通过 Python 调用**扣子（Coze）**的工作流（Workflow），目前最推荐的方式是使用官方提供的cozepySDK（Python SDK）。这是字节跳动官方维护的包，支持完整的 Coze Open API，包括直接执行工作流（非流式/流式）、带文件上传、恢复中断等功能。

1. 准备工作（必须先完成）

登录扣子平台：https://www.coze.cn 或 https://www.coze.com（国际版）
创建并发布工作流：
- 在工作流画布中搭建好逻辑（支持输入参数、LLM、代码节点、插件等）
- 发布后，复制工作流 ID（通常在 URL 最后一段数字）
创建个人访问令牌（Personal Access Token）：
- 进入「个人空间」→「设置」→「API 密钥」→「创建新密钥」
- 记录下生成的pat_xxx…（这就是 Token）
- 注意：Token 有有效期，过期需重新生成
安装 SDK：
```
pipinstallcozepy
```

2. 基本调用方式（非流式 / 同步）

importosfromcozepyimportCoze,TokenAuth,Message# 初始化客户端（推荐从环境变量读取 Token，安全）coze=Coze(auth=TokenAuth(os.getenv("COZE_API_TOKEN")))# 或直接写 TokenAuth("pat_xxxxxxxx")# 工作流 ID（从 Coze 平台复制）workflow_id="你的工作流ID，例如 7423xxxxxx"# 输入参数（根据你工作流定义的输入变量）parameters={"topic":"2025年AI发展趋势","length":"800字","style":"专业分析"# ... 其他你定义的输入键值对}# 执行工作流（同步，非流式）result=coze.workflows.runs.create(workflow_id=workflow_id,parameters=parameters,# 可选：user_id（自定义用户标识，用于追踪）# user_id="user_123")# 打印最终输出print("工作流执行结果：")print(result.output)# 通常是 dict，根据工作流输出节点决定print("最终消息：",result.output.get("content","无输出"))# 如果工作流有多个输出节点，可遍历ifhasattr(result,'outputs'):forkey,valueinresult.outputs.items():print(f"{key}:{value}")

3. 流式调用（推荐用于长任务，实时获取进度）

fromcozepyimportStream,WorkflowEvent,WorkflowEventTypedefhandle_stream(stream:Stream[WorkflowEvent]):foreventinstream:ifevent.event==WorkflowEventType.MESSAGE:# 收到消息增量（类似聊天流式输出）print(event.message.content,end="",flush=True)elifevent.event==WorkflowEventType.ERROR:print("\n错误：",event.error)elifevent.event==WorkflowEventType.INTERRUPT:# 中断（需要用户补充信息）print("\n中断，需要补充：",event.interrupt)# 可调用 resume 接口继续# coze.workflows.runs.resume(workflow_id=workflow_id, event_id=..., resume_data="补充内容")elifevent.event==WorkflowEventType.DONE:print("\n执行完成")# 流式执行stream=coze.workflows.runs.create_stream(workflow_id=workflow_id,parameters=parameters)handle_stream(stream)

4. 支持文件上传（常见场景：OCR、文档分析等）

frompathlibimportPath# 先上传文件file_obj=coze.files.upload(file=Path("/path/to/your/document.pdf"))# 然后把 file_id 传给工作流parameters={"file_id":file_obj.id,"question":"总结这份文档的主要观点"}result=coze.workflows.runs.create(workflow_id=workflow_id,parameters=parameters)print(result.output)

5. 完整推荐实践写法（带错误处理 + 环境变量）

importosimportsysfromcozepyimportCoze,TokenAuth,WorkflowEventTypedefmain():token=os.getenv("COZE_API_TOKEN")ifnottoken:print("请设置环境变量：export COZE_API_TOKEN=pat_xxx")sys.exit(1)coze=Coze(auth=TokenAuth(token))workflow_id="7423xxxxxxxxxxxx"# 替换成你的params={"input_text":"帮我写一篇关于Python异步编程的文章","max_length":1200}try:result=coze.workflows.runs.create(workflow_id=workflow_id,parameters=params)print("执行成功！")print(result.output)exceptExceptionase:print("调用失败：",e)if__name__=="__main__":main()

6. 常见问题 & 注意事项

Token 权限：确保 Token 有「工作流执行」权限（一般个人 Token 默认都有）
地域：国内用https://api.coze.cn，国际用https://api.coze.com（SDK 可通过 base_url 指定）
```
coze=Coze(auth=TokenAuth(token),base_url="https://api.coze.com")
```
限额：免费空间有调用次数/天限制，专业版/企业版更高
异步调用：SDK 也支持 AsyncCoze（asyncio），适合高并发场景
官方示例仓库：https://github.com/coze-dev/coze-py （examples 文件夹有 workflow_stream.py 等）