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

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

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

1. 登录扣子平台：https://www.coze.cn 或 https://www.coze.com（国际版）
2. 创建并发布工作流：

• 在工作流画布中搭建好逻辑（支持输入参数、LLM、代码节点、插件等）
• 发布后，复制工作流 ID（通常在 URL 最后一段数字）

1. 创建个人访问令牌（Personal Access Token）：

• 进入「个人空间」→「设置」→「API 密钥」→「创建新密钥」
• 记录下生成的 pat_xxx…（这就是 Token）
• 注意：Token 有有效期，过期需重新生成

1. 安装 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", "无输出"))

# 如果工作流有多个输出节点，可遍历
if hasattr(result, 'outputs'):
    for key, value in result.outputs.items():
        print(f"{key}: {value}")

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

from cozepy import Stream, WorkflowEvent, WorkflowEventType

def handle_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

def main():
    token = os.getenv("COZE_API_TOKEN")
    if not 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 默认都有）
• 地域：国内用 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 等）

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

如果你的工作流涉及文件上传、复杂参数、流式中断恢复等特殊场景，告诉我具体需求，我可以帮你写更针对性的代码示例！