[3][Coze API][python] 通过 OAuth JWT 实现扣子 BOT 对话

建立 OAuth JWT 应用

https://www.coze.cn/docs/developer_guides/oauth_jwt

选择服务类应用 2

应用ID 1

picture.image

创建 Key (红字2) ：获得一个 private_key.pem 文件，保存在本地

公钥： (红字1)

设置对应的权限 (红字3) ：例子中仅需要 BOT对话，所以选择这2个

picture.image

确保你的Coze Bot 发布并开通了 SDK 权限

BOT ID

picture.image

发布并勾选 API权限

picture.image

获取token 逻辑

将 private_key.pem 文件放到跟下面代码同级目录下

输入下面的内容：

KID = 'your_kid' # 公钥指纹，见步骤1 ISS = 'your_oauth_app_id' # OAuth应用ID ，见步骤1 AUD = 'api.coze.cn' # 扣子API的Endpoint，固定值

代码

import jwt
import requests
import time
import uuid
# pip install pyjwt

# 定义常量
PRIVATE_KEY_PATH = './private_key.pem'  # 私钥文件路径
KID = 'your_kid'  # 公钥指纹
ISS = 'your_oauth_app_id'  # OAuth应用ID
AUD = 'api.coze.cn'  # 扣子API的Endpoint



# 生成JWT
def generate_jwt():
    with open(PRIVATE_KEY_PATH, 'r') as key_file:
        private_key = key_file.read()

    current_time = int(time.time())
    payload = {
        "iss": ISS,
        "aud": AUD,
        "iat": current_time,
        "exp": current_time + 3600,  # 1小时后过期
        "jti": str(uuid.uuid4())  # 随机字符串
    }

    headers = {
        "alg": "RS256",
        "typ": "JWT",
        "kid": KID
    }

    token = jwt.encode(payload, private_key, algorithm='RS256', headers=headers)
    return token

# 获取访问令牌
def get_access_token(jwt_token):
    url = 'https://api.coze.cn/api/permission/oauth2/token'
    headers = {
        'Content-Type': 'application/json',
        'Authorization': f'Bearer {jwt_token}'
    }
    data = {
        "grant_type": "urn:ietf:params:oauth:grant-type:jwt-bearer",
        "duration_seconds": 86399  # 可选，设置为24小时
    }

    response = requests.post(url, headers=headers, json=data)
    if response.status_code == 200:
        return response.json().get('access_token')
    else:
        raise Exception(f"Failed to get access token: {response.text}")

# 主函数
def main():
    jwt_token = generate_jwt()
    access_token = get_access_token(jwt_token)
    print(f"Access Token: {access_token}")

if __name__ == "__main__":
    main()

完整授权 + 对话

import jwt
import requests
import json
import uuid
import time

# 定义常量
PRIVATE_KEY_PATH = './private_key.pem'  # 私钥文件路径
KID = 'your_kid'  # 公钥指纹
ISS = 'your_oauth_app_id'  # OAuth应用ID
AUD = 'api.coze.cn'  # 扣子API的Endpoint
USER_BOT_ID = '你的BOT ID '


class ChatBot:
    def __init__(self):
        self.messages = []
        self.input_value = ''
        self.conversation_id = None
        self.bot_id = USER_BOT_ID 
        self.user_id = self.generate_uuid()  # 初始化 user_id
        self.access_token = self.get_access_token()  # 获取访问令牌

    def generate_uuid(self):
        return str(uuid.uuid4())

    def generate_jwt(self):
        with open(PRIVATE_KEY_PATH, 'r') as key_file:
            private_key = key_file.read()

        current_time = int(time.time())
        payload = {
            "iss": ISS,
            "aud": AUD,
            "iat": current_time,
            "exp": current_time + 3600,  # 1小时后过期
            "jti": str(uuid.uuid4())  # 随机字符串
        }

        headers = {
            "alg": "RS256",
            "typ": "JWT",
            "kid": KID
        }

        token = jwt.encode(payload, private_key, algorithm='RS256', headers=headers)
        return token

    def get_access_token(self):
        jwt_token = self.generate_jwt()
        url = 'https://api.coze.cn/api/permission/oauth2/token'
        headers = {
            'Content-Type': 'application/json',
            'Authorization': f'Bearer {jwt_token}'
        }
        data = {
            "grant_type": "urn:ietf:params:oauth:grant-type:jwt-bearer",
            "duration_seconds": 86399  # 可选，设置为24小时
        }

        response = requests.post(url, headers=headers, json=data)
        response.raise_for_status()  # 检查请求是否成功
        return response.json().get('access_token')

    def send_message(self):
        message = self.input_value
        if not message:
            return

        # 检查 conversation_id 是否为 None
        if not self.conversation_id:
            # 如果是 None，先创建一个新的会话
            self.create_empty_conversation(self.send_message)
            return

        # 添加用户消息到消息列表
        self.messages.append({'role': 'user', 'content': message})
        self.input_value = ''

        headers = {
            'Authorization': f'Bearer {self.access_token}',
            'Content-Type': 'application/json'
        }

        data = {
            'conversation_id': self.conversation_id,  # 可选参数
            'bot_id': self.bot_id,
            'user_id': self.user_id,
            'stream': True,  # 启用流式返回
            'auto_save_history': True,
            'additional_messages': [
                {
                    'role': 'user',
                    'content': message,
                    'content_type': 'text'
                }
            ]
        }

        response = requests.post('https://api.coze.cn/v3/chat', headers=headers, json=data, stream=True)

        try:
            response.raise_for_status()  # 检查请求是否成功
            for line in response.iter_lines():
                if line:
                    decoded_line = line.decode('utf-8')
                    if decoded_line.startswith('event:'):
                        event_type = decoded_line.split(':', 1)[1].strip()
                    elif decoded_line.startswith('data:'):
                        event_data = decoded_line.split(':', 1)[1].strip()
                        self.handle_event(event_type, event_data)
        except requests.exceptions.HTTPError as http_err:
            print(f'HTTP error occurred: {http_err}')
        except json.JSONDecodeError:
            print('JSON decode error: 响应不是有效的 JSON 格式')
            print('响应内容:', response.text)
        except Exception as err:
            print(f'Other error occurred: {err}')

    def handle_event(self, event_type, event_data):
        # 处理不同类型的事件
        if event_type == 'conversation.chat.created':
            data = json.loads(event_data)
            print('对话已创建:', data)
        elif event_type == 'conversation.chat.in_progress':
            data = json.loads(event_data)
            print('对话进行中:', data)
        elif event_type == 'conversation.message.delta':
            data = json.loads(event_data)
            if data.get('role') == 'assistant':
                self.messages.append({'role': 'assistant', 'content': data.get('content')})
                print('助手消息:', data.get('content'))
        elif event_type == 'conversation.chat.completed':
            data = json.loads(event_data)
            print('对话已完成:', data)
        elif event_type == 'conversation.chat.failed':
            data = json.loads(event_data)
            print('对话失败:', data)
        else:
            print('未知事件类型:', event_type)

    def create_empty_conversation(self, callback):
        headers = {
            'Authorization': f'Bearer {self.access_token}',
            'Content-Type': 'application/json'
        }

        data = {
            'bot_id': self.bot_id,
            'user_id': self.user_id
        }

        response = requests.post('https://api.coze.cn/v1/conversation/create', headers=headers, json=data)

        try:
            response.raise_for_status()  # 检查请求是否成功
            res_data = response.json()
            if res_data.get('code') == 0:
                self.conversation_id = res_data['data']['id']
                self.messages.append({'role': 'system', 'content': f'空会话已创建，ID: {self.conversation_id}'})
                if callback:
                    callback()  # 执行回调函数
            else:
                print('创建空会话失败:', res_data.get('msg'))
        except requests.exceptions.HTTPError as http_err:
            print(f'HTTP error occurred: {http_err}')
        except json.JSONDecodeError:
            print('JSON decode error: 响应不是有效的 JSON 格式')
            print('响应内容:', response.text)
        except Exception as err:
            print(f'Other error occurred: {err}')

if __name__ == '__main__':
    chat_bot = ChatBot()
    while True:
        chat_bot.input_value = input("你: ")
        chat_bot.send_message()

[3][Coze API][python] 通过 OAuth JWT 实现 扣子 BOT 对话