Python 实现 Coze API 集成：OAuth 授权与消息交互

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