跳到主要内容
极客日志极客日志面向AI+效率的开发者社区
首页博客GitHub 精选镜像AI 生图工具UI配色美学隐私政策关于联系
搜索内容 / 工具 / 仓库 / 镜像...⌘K搜索
注册
博客列表
PythonAI

使用 Python SDK 调用扣子(Coze)工作流

使用 Python SDK 调用扣子(Coze)工作流。需先登录平台创建工作流并获取 ID,创建个人访问令牌(PAT)。安装 cozepy 后初始化客户端,支持同步非流式执行、流式实时输出及文件上传功能。代码示例涵盖参数传递、错误处理及环境变量配置。注意 Token 有效期与调用限额,国内国际 API 地址不同。官方文档提供完整接口参考。

观心发布于 2026/3/21更新于 2026/7/2349 浏览

使用 Python 官方 SDK cozepy 调用扣子(Coze)工作流(Workflow),这是字节跳动官方维护的包,支持完整的 Coze Open API,包括直接执行工作流(非流式/流式)、带文件上传、恢复中断等功能。

1. 准备工作

  1. 登录扣子平台:访问 https://www.coze.cn 或 https://www.coze.com(国际版)
  2. 创建并发布工作流:
    • 在工作流画布中搭建好逻辑(支持输入参数、LLM、代码节点、插件等)
    • 发布后,复制工作流 ID(通常在 URL 最后一段数字)
  3. 创建个人访问令牌(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,根据工作流输出节点决定
(, result.output.get(, ))


 (result, ):
     key, value  result.outputs.items():
        ()
print
"最终消息:"
"content"
"无输出"
# 如果工作流有多个输出节点,可遍历
if
hasattr
'outputs'
for
in
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 默认都有)
  • 限额:免费空间有调用次数/天限制,专业版/企业版更高
  • 异步调用: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

目录

  1. 1. 准备工作
  2. 2. 基本调用方式(非流式 / 同步)
  3. 初始化客户端(推荐从环境变量读取 Token,安全)
  4. 或直接写 TokenAuth("pat_xxxxxxxx")
  5. 工作流 ID(从 Coze 平台复制)
  6. 输入参数(根据你工作流定义的输入变量)
  7. 执行工作流(同步,非流式)
  8. 打印最终输出
  9. 通常是 dict,根据工作流输出节点决定
  10. 如果工作流有多个输出节点,可遍历
  11. 3. 流式调用(推荐用于长任务,实时获取进度)
  12. 流式执行
  13. 4. 支持文件上传(常见场景:OCR、文档分析等)
  14. 先上传文件
  15. 然后把 file_id 传给工作流
  16. 5. 完整推荐实践写法(带错误处理 + 环境变量)
  17. 6. 常见问题 & 注意事项
  18. 7. 官方文档参考
  • 免费图片AI生成工具免费生成了解详情
  • Magick API 一键接入全球大模型注册送1000万token查看
  • 免费图片视频在线生成30秒,将你的创意变成现实开始设计
  • X/Twitter免费视频下载器免登陆无限额度免费视频解析下载了解详情
  • 100+免费在线小游戏爽一把
极客日志微信公众号二维码

微信扫一扫,关注极客日志

微信公众号「极客日志V2」,在微信中扫描左侧二维码关注。展示文案:极客日志V2 zeeklog

更多推荐文章

查看全部
  • OpenClaw 自动化 AI 智能体跨平台部署与日常使用教程
  • Linux 五种 IO 模型
  • Stable Diffusion WebUI Windows 部署流程与常见报错解决方案
  • 机器人操作 VLA 模型强化学习综述
  • OpenClaw:低成本自适应机器人抓手技术解析
  • C++ 异常机制详解:从原理到工程实践
  • C++ 类与对象详解:封装、this 指针与默认成员函数
  • A*算法在网格路径规划中的三种优化策略对比与实战
  • Stable Diffusion 秋叶整合包:环境配置与使用指南
  • Mitmproxy 启动正常却抓不到包?端口冲突排查实录
  • 前端常用可视化图表组件选型指南
  • 实战开发 AI Skill:网页内容抓取工具实现
  • arXiv 论文:Reasoning Models Generate Societies of Thought
  • Mac Big Sur 使用 OpenClaw OpenCode OpenSpec 实现 AI 自动化开发流程
  • FPGA 实现 MIPI 协议全解析与时序规范
  • RaNER 模型中文命名实体识别服务 WebUI 与 API 部署实战
  • 用 Rust 构建 Git 提交历史可视化工具
  • Android Studio 结合 Trae 使用 Kotlin 开发 WebView 应用
  • 线性表、顺序表与链表详解(C 语言实现)
  • 前端国际化实战:i18n 选型、架构设计与 RTL 布局避坑

相关免费在线工具

  • RSA密钥对生成器

    生成新的随机RSA私钥和公钥pem证书。 在线工具,RSA密钥对生成器在线工具,online

  • Mermaid 预览与可视化编辑

    基于 Mermaid.js 实时预览流程图、时序图等图表,支持源码编辑与即时渲染。 在线工具,Mermaid 预览与可视化编辑在线工具,online

  • 随机西班牙地址生成器

    随机生成西班牙地址(支持马德里、加泰罗尼亚、安达卢西亚、瓦伦西亚筛选),支持数量快捷选择、显示全部与下载。 在线工具,随机西班牙地址生成器在线工具,online

  • curl 转代码

    解析常见 curl 参数并生成 fetch、axios、PHP curl 或 Python requests 示例代码。 在线工具,curl 转代码在线工具,online

  • Base64 字符串编码/解码

    将字符串编码和解码为其 Base64 格式表示形式即可。 在线工具,Base64 字符串编码/解码在线工具,online

  • Base64 文件转换器

    将字符串、文件或图像转换为其 Base64 表示形式。 在线工具,Base64 文件转换器在线工具,online