SuperAPI AI Directory
Partner中文AI newsFavorites
Skillscoze-api

coze-api

Coding & development

View repository

Updated 10-16 01:34

About this skill

Coze API Integration Skill

Coze is an AI agent development platform launched by ByteDance. This Skill provides a complete guide to calling the Coze API.

Core Features

  1. Chat API - Converse with the agent
  2. Workflow API - Execute workflows
  3. Message Management - Query conversation status and message lists

Quick Start

1. Preparation

Before using the Coze API, complete the following preparations:

Obtain an Access Token (Personal Access Token)

  1. Log in to the Coze platform: https://www.coze.cn
  2. Go to Personal Center → API Management
  3. Create a Personal Access Token (PAT)
  4. Save the token (displayed only once)

Obtain the Bot ID

  1. Go to the Bot edit page
  2. Get the Bot ID from the URL
    • For example: https://www.coze.cn/space/123/bot/7348293334
    • The Bot ID is: 7348293334

Publish the Bot as an API service

  1. Click "Publish" on the Bot page
  2. Select "Bot as API"
  3. Wait for approval

2. API Basic Information

API Base URL

  • Domestic version: https://api.coze.cn
  • API version: v3 (recommended)

Authentication Method

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

Usage Limits

  • Basic edition: 100 API calls per account (one-time)
  • Professional edition: unlimited, billed by Token consumption

Chat API

Non-streaming Conversation

Initiate a complete conversation and return after waiting for the complete result.

API Endpoint

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

Python Example 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))

Streaming Conversation

Receive AI replies in real time, similar to a typewriter effect.

Python Example 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("写一首关于春天的诗")

Important Parameter Descriptions

Request Parameters

  • bot_id (required): The unique identifier of the Bot
  • user_id (required): User identifier, used to distinguish different users
  • stream (required): Whether to use streaming output
    • true: streaming output,auto_save_history must be false
    • false: non-streaming output,auto_save_history must be true
  • auto_save_history: Whether to automatically save history
  • additional_messages: Message array
    • role: Role, usually "user"
    • content: Message content
    • content_type: Content type, usually "text"
  • conversation_id (optional): Conversation ID, used to continue a previous conversation

Response Fields

  • conversation_id: Conversation ID
  • chat_id: The ID of this conversation
  • status: Conversation status
    • in_progress: Processing
    • completed: Completed
    • failed: Failed

Workflow API

Execute a published workflow.

API Endpoint

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

Python Example 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))

Complete Example: Poll for Conversation Results

Non-streaming conversations require polling to query the status until the conversation is complete.

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("介绍一下人工智能的发展历史")

Error Handling

Common Error Codes

  • 400 - Request parameter error

    • Check whether bot_id correct?
    • Check stream and auto_save_history combination is correct

  • 401 - Authentication failed

    • Check whether the PAT Token is correct
    • Check whether the Authorization header format is correct

  • 403 - Insufficient permissions

    • Check whether the Bot has been published as an API service
    • Check whether the account has permission to access the Bot

  • 429 - Request rate limit

    • Basic plan has reached the limit of 100 requests
    • Reduce the request rate

  • 500 - Server error

    • Retry later

Error Handling Example

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

Best Practices

1. Manage conversation context

Use conversation_id to maintain multi-turn conversations:

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. Choose the appropriate response mode

  • Non-streaming (stream=False): suitable for scenarios that require complete results

    • Data analysis
    • Batch processing
    • API integration

  • Streaming (stream=True): suitable for interactive scenarios

    • Chat applications
    • Real-time feedback
    • User experience first

3. Set timeouts reasonably

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

4. Use custom variables

Pass context information during the conversation:

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

Reference Materials

Notes

  1. Token security: Do not hardcode the PAT Token in code; use environment variables or a configuration file
  2. Cost management: The Professional plan is billed by Token consumption; monitor usage carefully
  3. Version updates: The API may be updated; it is recommended to check the official documentation regularly
  4. Streaming limitation: In streaming mode, auto_save_history must be false
  5. User identifier: Use a unique user_id to distinguish different users, making tracking and management easier