Coze Api

Coze API 集成 Skill

扣子(Coze)是字节跳动推出的 AI 智能体开发平台,本 Skill 提供完整的 Coze API 调用指南。

核心功能

1. 对话 API (Chat API) - 与智能体进行对话
2. 工作流 API (Workflow API) - 执行工作流
3. 消息管理 - 查询对话状态和消息列表

快速开始

1. 准备工作

在使用 Coze API 前需要完成以下准备:

获取访问令牌 (Personal Access Token)

1. 登录 Coze 平台: https://www.coze.cn
2. 进入个人中心 → API 管理
3. 创建个人访问令牌(PAT)
4. 保存令牌(仅显示一次)

获取 Bot ID

1. 进入 Bot 编辑页面
2. 从 URL 中获取 Bot ID
   • 例如: https://www.coze.cn/space/123/bot/7348293334
   • Bot ID 为: 7348293334

发布 Bot 为 API 服务

1. 在 Bot 页面点击"发布"
2. 选择"Bot as API"
3. 等待审核通过

2. API 基础信息

API 基础 URL

• 国内版: https://api.coze.cn
• API 版本: v3(推荐)

认证方式

Copy code
Authorization: Bearer {YOUR_PAT_TOKEN}
Content-Type: application/json

使用限制

• 基础版: 每账号 100 次 API 调用(一次性)
• 专业版: 无限制,按 Token 消耗计费

对话 API (Chat API)

非流式对话

发起一次完整对话,等待完整结果后返回。

API 端点

Copy code
POST https://api.coze.cn/v3/chat

Python 示例代码

pythonCopy code
import requests
import json

API_URL = "https://api.coze.cn/v3/chat"
RETRIEVE_URL = "https://api.coze.cn/v3/chat/retrieve"
MESSAGE_LIST_URL = "https://api.coze.cn/v3/chat/message/list"

# 配置参数
PAT_TOKEN = "YOUR_PAT_TOKEN"
BOT_ID = "YOUR_BOT_ID"
USER_ID = "unique_user_id"

def send_message(message):
    """发起对话"""
    headers = {
        "Authorization": f"Bearer {PAT_TOKEN}",
        "Content-Type": "application/json"
    }
    
    data = {
        "bot_id": BOT_ID,
        "user_id": USER_ID,
        "stream": False,
        "auto_save_history": True,
        "additional_messages": [
            {
                "role": "user",
                "content": message,
                "content_type": "text"
            }
        ]
    }
    
    response = requests.post(API_URL, headers=headers, json=data)
    return response.json()

def check_status(conversation_id, chat_id):
    """查询对话状态"""
    headers = {
        "Authorization": f"Bearer {PAT_TOKEN}",
        "Content-Type": "application/json"
    }
    
    params = {
        "conversation_id": conversation_id,
        "chat_id": chat_id
    }
    
    response = requests.get(RETRIEVE_URL, headers=headers, params=params)
    return response.json()

def get_messages(conversation_id, chat_id):
    """获取对话消息"""
    headers = {
        "Authorization": f"Bearer {PAT_TOKEN}",
        "Content-Type": "application/json"
    }
    
    params = {
        "conversation_id": conversation_id,
        "chat_id": chat_id
    }
    
    response = requests.get(MESSAGE_LIST_URL, headers=headers, params=params)
    return response.json()

# 使用示例
result = send_message("你好,请介绍一下自己")
print(json.dumps(result, ensure_ascii=False, indent=2))

流式对话

实时接收 AI 回复,类似打字机效果。

Python 示例代码

pythonCopy code
import requests
import json

def stream_chat(message):
    """流式对话"""
    headers = {
        "Authorization": f"Bearer {PAT_TOKEN}",
        "Content-Type": "application/json"
    }
    
    data = {
        "bot_id": BOT_ID,
        "user_id": USER_ID,
        "stream": True,
        "auto_save_history": False,  # 流式时必须为 False
        "additional_messages": [
            {
                "role": "user",
                "content": message,
                "content_type": "text"
            }
        ]
    }
    
    response = requests.post(API_URL, headers=headers, json=data, stream=True)
    
    # 处理流式响应
    for line in response.iter_lines():
        if line:
            line_str = line.decode('utf-8')
            
            # 跳过非 data 行
            if not line_str.startswith('data:'):
                continue
                
            # 提取 JSON 数据
            json_str = line_str.split('data:', 1)[1].strip()
            
            try:
                data = json.loads(json_str)
                
                # 处理消息事件
                if data.get('event') == 'conversation.message.delta':
                    content = data.get('data', {}).get('content', '')
                    print(content, end='', flush=True)
                    
                # 处理完成事件
                elif data.get('event') == 'conversation.message.completed':
                    print("\n[对话完成]")
                    
            except json.JSONDecodeError:
                continue

# 使用示例
stream_chat("写一首关于春天的诗")

重要参数说明

请求参数

• bot_id (必填): Bot 的唯一标识符
• user_id (必填): 用户标识符,用于区分不同用户
• stream (必填): 是否使用流式输出
  • true: 流式输出,auto_save_history 必须为 false
  • false: 非流式输出,auto_save_history 必须为 true
• auto_save_history: 是否自动保存历史记录
• additional_messages: 消息数组
  • role: 角色,通常为 "user"
  • content: 消息内容
  • content_type: 内容类型,通常为 "text"
• conversation_id (可选): 对话 ID,用于继续之前的对话

响应字段

• conversation_id: 对话 ID
• chat_id: 本次对话的 ID
• status: 对话状态
  • in_progress: 处理中
  • completed: 已完成
  • failed: 失败

工作流 API

执行已发布的工作流。

API 端点

Copy code
POST https://api.coze.cn/v3/workflows/run

Python 示例代码

pythonCopy code
import requests
import json

WORKFLOW_RUN_URL = "https://api.coze.cn/v3/workflows/run"

def run_workflow(workflow_id, parameters):
    """执行工作流"""
    headers = {
        "Authorization": f"Bearer {PAT_TOKEN}",
        "Content-Type": "application/json"
    }
    
    data = {
        "workflow_id": workflow_id,
        "parameters": parameters
    }
    
    response = requests.post(WORKFLOW_RUN_URL, headers=headers, json=data)
    return response.json()

# 使用示例
workflow_id = "73xxx47"
params = {
    "input_text": "需要处理的文本",
    "option": "选项A"
}

result = run_workflow(workflow_id, params)
print(json.dumps(result, ensure_ascii=False, indent=2))

完整示例: 轮询获取对话结果

非流式对话需要轮询查询状态,直到对话完成。

pythonCopy code
import requests
import time
import json

def chat_with_polling(message, max_retries=30, interval=2):
    """
    发起对话并轮询获取结果
    
    Args:
        message: 用户消息
        max_retries: 最大重试次数
        interval: 轮询间隔(秒)
    """
    headers = {
        "Authorization": f"Bearer {PAT_TOKEN}",
        "Content-Type": "application/json"
    }
    
    # 1. 发起对话
    data = {
        "bot_id": BOT_ID,
        "user_id": USER_ID,
        "stream": False,
        "auto_save_history": True,
        "additional_messages": [
            {
                "role": "user",
                "content": message,
                "content_type": "text"
            }
        ]
    }
    
    response = requests.post(API_URL, headers=headers, json=data)
    result = response.json()
    
    if response.status_code != 200:
        print(f"发起对话失败: {result}")
        return None
    
    conversation_id = result['data']['conversation_id']
    chat_id = result['data']['id']
    
    print(f"对话已创建: conversation_id={conversation_id}, chat_id={chat_id}")
    
    # 2. 轮询查询状态
    retrieve_url = f"https://api.coze.cn/v3/chat/retrieve"
    params = {
        "conversation_id": conversation_id,
        "chat_id": chat_id
    }
    
    for i in range(max_retries):
        time.sleep(interval)
        
        status_response = requests.get(retrieve_url, headers=headers, params=params)
        status_data = status_response.json()
        
        status = status_data['data']['status']
        print(f"查询状态 [{i+1}/{max_retries}]: {status}")
        
        if status == "completed":
            # 3. 获取消息列表
            message_url = f"https://api.coze.cn/v3/chat/message/list"
            message_response = requests.get(message_url, headers=headers, params=params)
            message_data = message_response.json()
            
            # 提取 AI 回复
            messages = message_data['data']
            for msg in messages:
                if msg['role'] == 'assistant' and msg['type'] == 'answer':
                    print("\n=== AI 回复 ===")
                    print(msg['content'])
                    return msg['content']
            
        elif status == "failed":
            print("对话失败")
            return None
    
    print("轮询超时")
    return None

# 使用示例
response = chat_with_polling("介绍一下人工智能的发展历史")

错误处理

常见错误码

• 400 - 请求参数错误

  • 检查 bot_id 是否正确
  • 检查 stream 和 auto_save_history 的组合是否正确

• 401 - 认证失败

  • 检查 PAT Token 是否正确
  • 检查 Authorization 头格式是否正确

• 403 - 权限不足

  • 检查 Bot 是否已发布为 API 服务
  • 检查账号是否有权限访问该 Bot

• 429 - 请求频率限制

  • 基础版达到 100 次限制
  • 降低请求频率

• 500 - 服务器错误

  • 稍后重试

错误处理示例

pythonCopy code
def safe_api_call(func, *args, **kwargs):
    """安全的 API 调用"""
    try:
        result = func(*args, **kwargs)
        return result
    except requests.exceptions.RequestException as e:
        print(f"请求错误: {e}")
        return None
    except json.JSONDecodeError as e:
        print(f"JSON 解析错误: {e}")
        return None
    except Exception as e:
        print(f"未知错误: {e}")
        return None

最佳实践

1. 管理对话上下文

使用 conversation_id 维持多轮对话:

pythonCopy code
def multi_turn_chat(conversation_id=None):
    """多轮对话"""
    data = {
        "bot_id": BOT_ID,
        "user_id": USER_ID,
        "stream": False,
        "auto_save_history": True,
        "additional_messages": [
            {
                "role": "user",
                "content": "继续我们的对话",
                "content_type": "text"
            }
        ]
    }
    
    # 如果有 conversation_id,继续之前的对话
    if conversation_id:
        data["conversation_id"] = conversation_id
    
    # ... 发送请求

2. 选择合适的响应模式

• 非流式 (stream=False): 适合需要完整结果的场景

  • 数据分析
  • 批量处理
  • API 集成

• 流式 (stream=True): 适合交互式场景

  • 聊天应用
  • 实时反馈
  • 用户体验优先

3. 合理设置超时

pythonCopy code
response = requests.post(
    API_URL, 
    headers=headers, 
    json=data,
    timeout=30  # 设置 30 秒超时
)

4. 使用自定义变量

在对话中传递上下文信息:

pythonCopy code
data = {
    "bot_id": BOT_ID,
    "user_id": USER_ID,
    "stream": False,
    "auto_save_history": True,
    "additional_messages": [...],
    "custom_variables": {
        "user_name": "张三",
        "company": "ABC公司",
        "department": "技术部"
    }
}

参考资料

• 官方文档: https://www.coze.cn/open/docs/developer_guides/coze_api_overview
• 开发者平台: https://www.coze.cn/open/playground
• API 参考: https://www.coze.cn/docs/developer_guides/chat_v3

注意事项

1. Token 安全: 不要在代码中硬编码 PAT Token,使用环境变量或配置文件
2. 费用管理: 专业版按 Token 消耗计费,注意监控使用量
3. 版本更新: API 可能更新,建议定期查看官方文档
4. 流式限制: 流式模式下 auto_save_history 必须为 false
5. 用户标识: 使用唯一的 user_id 区分不同用户,便于追踪和管理