在当下的智能应用开发中，与第三方 API 进行高效集成是关键环节之一。本文将详细介绍如何使用 Python 实现与扣子 (Coze) API 的回调功能集成，包括 OAuth 授权流程、会话管理及消息交互的完整实现方案。

---

🔑 OAuth 授权流程

Coze API 采用 OAuth 2.0 授权框架，通过 JWT (JSON Web Token) 实现安全的身份验证。整个授权流程分为以下步骤：

步骤一：进入应用创建入口

在左侧导航栏点击「API 管理」，切换到「授权」标签页，点击右上角「创建新应用」，开始 OAuth 应用的创建流程。

步骤二：配置应用基本信息与权限

进入「创建新应用」页面后：

• 密钥生成：点击「+ 创建 Key」，生成用于 JWT 签名的公钥和私钥（私钥会自动下载，需妥善保管）。

• 权限配置：在「权限」区域勾选「全选」，确保应用获得 Bot 管理、会话管理、回调应用等所有必要权限，为后续 API 调用提供完整的权限支持。

🛠️ 环境准备与依赖安装

首先需要安装必要的 Python 依赖库：

pip install requests pyjwt cryptography

• requests：用于发送 HTTP 请求与 Coze API 交互

• pyjwt：用于生成和解析 JWT 令牌

• cryptography：提供加密算法支持，用于 JWT 签名

⚙️ 核心配置参数

在开始编码前，需要准备以下配置参数（从 Coze 开放平台获取）：

# 用户 ID
USER_ID = "..."
# 应用 ID
APP_ID = "..."
# 空间 ID
BOT_ID = "..."
# 公钥
PUBLIC_KEY = "..."
# 私钥
PRIVATE_KEY = """
-----BEGIN PRIVATE KEY-----
......
-----END PRIVATE KEY-----
"""

🔐 JWT 令牌生成与 Token 获取

实现获取访问令牌的核心函数：

def get_access_token() -> str:
    # 生成 JWT
    payload = {
        "iat": int(time.time()),
        "exp": int(time.time()) + 600,  # 10分钟有效期
        "jti": str(uuid.uuid4()),  # 防重放攻击
        "aud": "api.coze.cn",
        "iss": APP_ID,
        "session_name": USER_ID  # 用户隔离标识
    }
    headers = {"kid": PUBLIC_KEY}
    new_jwt = jwt.encode(payload, PRIVATE_KEY, algorithm="RS256", headers=headers)
    print("【调试】生成 JWT 前 50 字符：", new_jwt[:50])
​
    # 获取 Token
    url = "https://api.coze.cn/api/permission/oauth2/token"
    body = {
        "grant_type": "urn:ietf:params:oauth:grant-type:jwt-bearer",
        "duration_seconds": 86399
    }
    response = requests.post(url, json=body, headers={"Authorization": f"Bearer {new_jwt}"})
    new_token = response.json()["access_token"]
    print("【调试】获取 token 前 50 字符：", new_token[:50])
    return new_token

🤖 Coze 机器人交互类实现

封装一个完整的 Coze 机器人交互类，包含会话管理和消息交互功能：

初始化与会话创建

class CozeBot:
    def __init__(self, bot_id: str, user_id: str):
        self.bot_id = bot_id
        self.user_id = user_id
        self.token = get_access_token()
        self.sess_id = self.create_session()
    
    def create_session(self) -> str:
        # 创建会话
        url = "https://api.coze.cn/v1/conversation/create"
        body = {
            "bot_id": self.bot_id,
            "meta_data": {"user_id": self.user_id}  # 用户隔离
        }
        # 重试机制
        for i in range(3):
            response = requests.post(url, json=body, headers={"Authorization": f"Bearer {self.token}"})
            print(f"【调试】第{i + 1}次创建会话返回：", response.status_code, response.text)
            if response.status_code == 200 and response.json().get("code") == 0:
                return response.json()["data"]["id"]
            time.sleep(2)
        raise RuntimeError("创建会话连续失败")

消息发送与接收

def chat(self, query: str) -> str:
    # 插入用户消息
    url = f"https://api.coze.cn/v1/conversation/message/create?conversation_id={self.sess_id}"
    body = {
        "role": "user",
        "content_type": "text",
        "content": query
    }
    response = requests.post(url, json=body, headers={"Authorization": f"Bearer {self.token}"})
    print("【调试】插入消息返回：", response.status_code, response.text)
    
    # 发起对话请求
    url = f"https://api.coze.cn/v3/chat?conversation_id={self.sess_id}"
    body = {
        "bot_id": self.bot_id,
        "user_id": self.user_id,
        "stream": False,  # 流式输出开关
        "auto_save_history": True,
        "additional_messages": [{
            "role": "user",
            "type": "question",
            "content_type": "text",
            "content": query
        }]
    }
    response = requests.post(url, json=body, headers={"Authorization": f"Bearer {self.token}"})
    return response.json()["data"]["messages"][-1]["content"]

对话历史查询

def get_round_messages(self):
    # 获取本轮对话的完整消息记录
    url = f"https://api.coze.cn/v1/conversation/message/list?conversation_id={self.sess_id}"
    response = requests.post(url, headers={"Authorization": f"Bearer {self.token}"})
    if response.status_code != 200:
        raise RuntimeError("获取对话消息失败: " + response.text)
    messages_data = response.json()["data"]
    return [(m["role"], m["content"]) for m in messages_data]

🚀 运行示例

if __name__ == "__main__":
    # 初始化智能体
    bot = CozeBot(BOT_ID, USER_ID)
    
    # 发起对话
    answer = bot.chat("你好，请介绍下你自己")
    print("Bot 回答：", answer)
    
    # 查看完整对话历史
    print("\n=== 对话历史 ===")
    for role, content in bot.get_round_messages():
        print(f"{role}: {content}")