通过python API来调用扣子coze的工作流
通过 Python 调用**扣子（Coze）**的工作流（Workflow），目前最推荐的方式是使用官方提供的 cozepy SDK（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：
pip install cozepy 2. 基本调用方式（非流式 / 同步）
import os from cozepy import Coze, 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'):for key, value in result.outputs.items():print(f"{key}: {value}")3. 流式调用（推荐用于长任务，实时获取进度）
from cozepy import Stream, WorkflowEvent, WorkflowEventType defhandle_stream(stream: Stream[WorkflowEvent]):for event in stream:if event.event == WorkflowEventType.MESSAGE:# 收到消息增量（类似聊天流式输出）print(event.message.content, end="", flush=True)elif event.event == WorkflowEventType.ERROR:print("\n错误：", event.error)elif event.event == WorkflowEventType.INTERRUPT:# 中断（需要用户补充信息）print("\n中断，需要补充：", event.interrupt)# 可调用 resume 接口继续# coze.workflows.runs.resume(workflow_id=workflow_id, event_id=..., resume_data="补充内容")elif event.event == WorkflowEventType.DONE:print("\n执行完成")# 流式执行 stream = coze.workflows.runs.create_stream( workflow_id=workflow_id, parameters=parameters ) handle_stream(stream)4. 支持文件上传（常见场景：OCR、文档分析等）
from pathlib import Path # 先上传文件 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. 完整推荐实践写法（带错误处理 + 环境变量）
import os import sys from cozepy import Coze, TokenAuth, WorkflowEventType defmain(): token = os.getenv("COZE_API_TOKEN")ifnot token: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)except Exception as e:print("调用失败：", e)if __name__ =="__main__": main()6. 常见问题 & 注意事项
- Token 权限：确保 Token 有「工作流执行」权限（一般个人 Token 默认都有）
- 限额：免费空间有调用次数/天限制，专业版/企业版更高
- 异步调用：SDK 也支持 AsyncCoze（asyncio），适合高并发场景
- 官方示例仓库：https://github.com/coze-dev/coze-py （examples 文件夹有 workflow_stream.py 等）
地域：国内用 https://api.coze.cn，国际用 https://api.coze.com（SDK 可通过 base_url 指定）
coze = Coze(auth=TokenAuth(token), base_url="https://api.coze.com")7. 官方文档参考（强烈建议看）
- 工作流运行接口：https://www.coze.cn/open/docs/developer_guides/workflow_run
- Python SDK 完整文档：https://github.com/coze-dev/coze-py
- API 总览：https://www.coze.cn/open/docs
如果你的工作流涉及文件上传、复杂参数、流式中断恢复等特殊场景，告诉我具体需求，我可以帮你写更针对性的代码示例！
