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

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

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

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

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

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

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

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

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
    
    # ... 发送请求

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

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