在当下的智能应用开发中，与第三方 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}")
