详解如何使用 Coze Workflow API 快速实现流程自动化

在现代应用开发中，流程自动化是提升效率的关键环节。Coze 提供的 Workflow API 允许开发者轻松调用已发布的工作流，实现复杂业务逻辑的自动化执行。本文将从基础到进阶，带你掌握 Coze Workflow API 的使用方法，包括环境准备、代码实现、高级特性及常见问题解决。

什么是 Coze Workflow API？

Coze Workflow API 是 Coze 平台提供的一套接口，用于管理和执行工作流。通过该 API，开发者可以：

• 执行已发布的工作流（同步/异步模式）
• 传递自定义参数给工作流
• 关联智能体（Bot）或应用（App）
• 获取执行结果及调试信息
• 查询工作流执行历史

其核心优势在于：无需关心工作流内部的节点编排细节，只需通过简单的 API 调用即可触发复杂流程，大幅降低跨系统集成的开发成本。

准备工作：调用 API 前的必要配置

在开始编码前，需要完成以下准备步骤：

1. 获取个人访问令牌（PAT）

PAT 是调用 Coze API 的身份凭证，获取方式：

• 登录 Coze 平台
• 进入「个人设置」-「访问令牌」
• 生成新的 PAT，记录令牌值（注意：令牌仅显示一次）

2. 准备已发布的工作流

API 仅能调用已发布的工作流，需确保：

• 在 Coze 工作台中创建并配置好工作流
• 完成工作流的发布（发布后会生成可调用的版本）
• 记录工作流 ID（可在工作流详情页获取）

3. 环境要求

• 开发语言：Python 3.6+
• 依赖库：requests（用于 HTTP 请求）
• 网络环境：能访问 Coze API 服务器（默认域名：www.coze.com）

快速开始：首个工作流调用示例

下面通过一个完整示例，演示如何调用 Coze Workflow API 执行工作流。

步骤 1：安装依赖

首先安装 requests 库（用于发送 HTTP 请求）：

pip install requests

步骤 2：编写 API 客户端代码

创建 coze_workflow_client.py 文件，实现调用逻辑：

import requests
import json

class CozeWorkflowAPI:
    def __init__(self, host, access_token):
        """
        初始化Coze工作流API客户端
        :param host: API主机地址，如"www.coze.com"
        :param access_token: 个人访问令牌(PAT)
        """
        self.base_url = f"https://{host}/v1"
        self.headers = {
            "Authorization": f"Bearer {access_token}",
            "Content-Type": "application/json"
        }

    def run_workflow(self, workflow_id, parameters=None, bot_id=None, app_id=None, is_async=False):
        """
        执行已发布的工作流
        :param workflow_id: 工作流ID（必选）
        :param parameters: 工作流输入参数（可选，字典类型）
        :param bot_id: 关联的智能体ID（可选）
        :param app_id: 关联的应用ID（可选）
        :param is_async: 是否异步执行（默认同步）
        :return: 执行结果字典
        """
        url = f"{self.base_url}/workflow/run"
        
        # 构建请求体（基础参数）
        payload = {
            "workflow_id": workflow_id,  # 必选参数，工作流唯一标识
            "is_async": is_async         # 可选，默认同步执行
        }
        
        # 添加可选参数
        if parameters is not None:
            # 参数需序列化为JSON字符串（根据API要求）
            payload["parameters"] = json.dumps(parameters)
        if bot_id is not None:
            payload["bot_id"] = bot_id  # 关联智能体（如有）
        if app_id is not None:
            payload["app_id"] = app_id  # 关联应用（如有）

        try:
            # 发送POST请求
            response = requests.post(
                url,
                headers=self.headers,
                data=json.dumps(payload),
                timeout=600  # 同步执行超时时间（10分钟）
            )
            response.raise_for_status()  # 抛出HTTP错误（如401、404）
            return response.json()
        except requests.exceptions.RequestException as e:
            error_msg = f"工作流执行失败: {str(e)}"
            print(error_msg)
            # 处理错误响应
            if 'response' in locals():
                try:
                    error_details = response.json()  # 解析JSON错误信息
                except json.JSONDecodeError:
                    error_details = response.text  # 非JSON响应直接返回文本
                return {
                    "error": True,
                    "status_code": response.status_code,
                    "message": error_msg,
                    "details": error_details
                }
            return {"error": True, "message": error_msg}

if __name__ == "__main__":
    # 配置参数（替换为你的实际信息）
    HOST = "www.coze.com"
    ACCESS_TOKEN = "pat_your_personal_access_token"  # 你的PAT
    WORKFLOW_ID = "73664689170551*****"  # 你的工作流ID
    
    # 初始化客户端
    coze_api = CozeWorkflowAPI(
        host=HOST,
        access_token=ACCESS_TOKEN
    )
    
    # 执行工作流（传递参数）
    result = coze_api.run_workflow(
        workflow_id=WORKFLOW_ID,
        parameters={  # 根据工作流定义的输入参数调整
            "user_id": "12345",
            "user_name": "George",
            "query": "如何使用Coze Workflow API？"
        },
        # bot_id="73428668*****",  # 如需关联智能体，取消注释并填写
        # app_id="749081945898306****",  # 如需关联应用，取消注释并填写
        # is_async=True  # 如需异步执行，取消注释
    )
    
    # 打印结果
    print(json.dumps(result, indent=2, ensure_ascii=False))
    
    # 输出调试链接（便于排查问题）
    if not result.get("error") and result.get("code") == 0:
        print(f"\n调试链接: {result.get('debug_url')}")

步骤 3：运行代码并验证结果

替换代码中的 ACCESS_TOKEN 和 WORKFLOW_ID 为实际值，执行脚本：

python coze_workflow_client.py

成功执行后，会返回类似以下结果：

{
  "code": 0,
  "msg": "success",
  "data": "{\"result\": \"Hello George, 以下是Coze Workflow API的使用指南...\"}",
  "execute_id": "1688849230123",
  "debug_url": "https://www.coze.com/debug?execute_id=1688849230123&space_id=123"
}

• code=0 表示执行成功
• data 为工作流输出结果（JSON字符串）
• debug_url 可用于在Coze平台查看执行详情

代码解析：核心逻辑详解

上面的示例代码封装了调用 Workflow API 的核心逻辑，关键部分解析如下：

1. 初始化客户端

def __init__(self, host, access_token):
    self.base_url = f"https://{host}/v1"
    self.headers = {
        "Authorization": f"Bearer {access_token}",
        "Content-Type": "application/json"
    }

• base_url：API 基础路径，基于 v1 版本（稳定版）
• headers：请求头必须包含身份凭证（Authorization）和数据格式（Content-Type: application/json）

2. 构建请求参数

根据 Coze Workflow API 定义，核心参数包括：

• workflow_id：必选，工作流的唯一标识
• parameters：可选，工作流输入参数（需序列化为JSON字符串）
• is_async：可选，是否异步执行（默认 false 同步执行）
• bot_id/app_id：可选，关联的智能体或应用（二选一，不可同时设置）

注意：parameters 需序列化为 JSON 字符串，因为 API 接收的是字符串格式（后端会通过 sonic.UnmarshalString 解析，见 workflow.go 源码）。

3. 处理响应与错误

• 成功响应：返回 JSON 格式的结果，包含 code、data、execute_id 等字段
• 错误处理：捕获 HTTP 异常（如网络错误、401 权限不足），并解析错误详情

高级用法：解锁更多功能

除了基础的同步执行，Coze Workflow API 还支持更多高级特性：

1. 异步执行工作流

当工作流执行时间较长（如超过10分钟），建议使用异步模式：

# 异步执行（立即返回执行ID，后续可查询结果）
result = coze_api.run_workflow(
    workflow_id=WORKFLOW_ID,
    parameters={"task": "long_running_job"},
    is_async=True
)
# 异步执行的结果中，execute_id 用于后续查询状态
execute_id = result.get("execute_id")

2. 流式获取执行结果

对于需要实时反馈的场景（如聊天机器人流程），可使用流式接口：

def stream_run_workflow(self, workflow_id, parameters=None):
    url = f"{self.base_url}/workflow/stream_run"
    payload = {"workflow_id": workflow_id, "parameters": json.dumps(parameters)}
    response = requests.post(
        url,
        headers=self.headers,
        data=json.dumps(payload),
        stream=True  # 启用流式响应
    )
    for line in response.iter_lines():
        if line:
            yield json.loads(line.decode("utf-8"))

# 使用示例
for chunk in coze_api.stream_run_workflow(WORKFLOW_ID, {"query": "流式返回示例"}):
    print(chunk.get("data"))  # 实时打印每一段结果

3. 查询工作流执行历史

通过 OpenAPIGetWorkflowRunHistory 接口查询历史执行记录：

def get_run_history(self, workflow_id, page=1, size=10):
    url = f"{self.base_url}/workflow/get_run_history"
    params = {
        "workflow_id": workflow_id,
        "page": page,
        "size": size
    }
    response = requests.get(
        url,
        headers=self.headers,
        params=params
    )
    return response.json()

# 查询最近10条执行记录
history = coze_api.get_run_history(WORKFLOW_ID)

常见问题与解决方案

在使用过程中，可能会遇到以下问题，这里提供对应的解决方法：

1. 错误：工作流未发布

现象：响应提示 ErrWorkflowNotPublished
原因：API 仅支持调用已发布的工作流，草稿状态的工作流无法被调用
解决：在 Coze 工作台中发布工作流（「发布」按钮），确保 LatestPublishedVersion 存在（见 workflow.go 源码校验逻辑）

2. 错误：参数格式错误

现象：响应提示 invalid parameter 或解析失败
原因：parameters 格式不正确（未序列化为 JSON 字符串）
解决：确保参数通过 json.dumps(parameters) 处理后再传入

3. 超时问题

现象：同步执行时超时（超过10分钟）
原因：同步执行有超时限制，长耗时任务不适合
解决：改用异步执行（is_async=True），通过 execute_id 轮询结果

4. 权限不足

现象：401 错误或 无权限访问
原因：PAT 无效、过期或没有工作流的调用权限
解决：重新生成 PAT，或检查工作流的权限设置（确保当前用户有「调用」权限）

总结

通过 Coze Workflow API，开发者可以轻松集成工作流能力到自己的应用中，无需关注复杂的流程编排细节。本文从基础示例出发，介绍了 API 的核心用法、参数配置及高级特性，同时提供了常见问题的解决方案。

如需进一步探索，可以参考：

• Coze API 官方文档
• 更多 API 接口（如创建工作流、保存工作流等）的使用方法

立即动手尝试，让流程自动化为你的应用赋能吧！