[2][Coze API][python] 通过 OAuth PKCE 实现扣子 BOT 对话

在授权页面创建 OAuth 应用

注意下面12345

picture.image

本地电脑测试，重定向URL ：需要跟代码里的设置匹配（红字1）

客户端ID ：以 .app.coze 结尾（红字2）

选中BOT管理 和 会话管理 ：我们需要让BOT能够创建一个会话并跟用户对话。（根据具体业务自己设定即可）

picture.image

安装 flask

pip install flask

确保你的BOT 获取了 API 权限

BOT ID

picture.image

发布并勾选 API权限

picture.image

获取 token 代码 (python)

需要设定：client_id = "你的 client ID ,参考 1 中的客户端ID ：以 .app.coze 结尾 "

import requests
import base64
import hashlib
import os
from flask import Flask, request
import threading
import webbrowser

# 生成随机的 code_verifier
def generate_code_verifier():
    return base64.urlsafe_b64encode(os.urandom(40)).rstrip(b'=').decode('utf-8')

# 根据 code_verifier 生成 code_challenge
def generate_code_challenge(code_verifier):
    code_challenge = hashlib.sha256(code_verifier.encode('utf-8')).digest()
    return base64.urlsafe_b64encode(code_challenge).rstrip(b'=').decode('utf-8')

# 发起授权请求
def initiate_authorization(client_id, redirect_uri):
    code_verifier = generate_code_verifier()
    code_challenge = generate_code_challenge(code_verifier)
    authorization_url = (
        f"https://www.coze.cn/api/permission/oauth2/authorize?"
        f"response_type=code&client_id={client_id}&redirect_uri={redirect_uri}"
        f"&state=random_state&code_challenge={code_challenge}&code_challenge_method=S256"
    )
    webbrowser.open(authorization_url)
    return code_verifier

# 获取授权码后请求访问令牌
def request_access_token(client_id, redirect_uri, code, code_verifier):
    token_url = "https://api.coze.cn/api/permission/oauth2/token"
    headers = {'Content-Type': 'application/json'}
    data = {
        "grant_type": "authorization_code",
        "code": code,
        "redirect_uri": redirect_uri,
        "client_id": client_id,
        "code_verifier": code_verifier
    }
    response = requests.post(token_url, headers=headers, json=data)
    return response.json()

# 刷新访问令牌
def refresh_access_token(client_id, refresh_token):
    token_url = "https://api.coze.cn/api/permission/oauth2/token"
    headers = {'Content-Type': 'application/json'}
    data = {
        "grant_type": "refresh_token",
        "refresh_token": refresh_token,
        "client_id": client_id
    }
    response = requests.post(token_url, headers=headers, json=data)
    return response.json()

# Flask 应用
app = Flask(__name__)
authorization_code = None

@app.route('/callback')
def callback():
    global authorization_code
    authorization_code = request.args.get('code')
    return "授权成功！请返回应用程序。"

# 主函数
def main():
    client_id = "你的 client ID ,参考 1 中的 客户端ID ：以 .app.coze 结尾 "
    redirect_uri = "http://localhost:8080/callback" 
    
    # 启动 Flask 服务器
    threading.Thread(target=lambda: app.run(port=8080)).start()
    
    # 发起授权请求
    code_verifier = initiate_authorization(client_id, redirect_uri)
    
    # 等待用户授权并获取授权码
    while authorization_code is None:
        pass
    
    # 请求访问令牌
    token_response = request_access_token(client_id, redirect_uri, authorization_code, code_verifier)
    print("访问令牌响应:", token_response)
    
    # 使用 refresh_token 刷新访问令牌
    refresh_token = token_response.get("refresh_token")
    if refresh_token:
        refreshed_token_response = refresh_access_token(client_id, refresh_token)
        print("刷新后的访问令牌响应:", refreshed_token_response)

if __name__ == "__main__":
    main()

运行上面函数后，函数会等待授权

    # 等待用户授权并获取授权码
    while authorization_code is None:
        pass

弹出的浏览器中点击授权

picture.image

授权成功后，跳转到我们自己的临时后端

# Flask 应用
app = Flask(__name__)
authorization_code = None

@app.route('/callback')
def callback():
    global authorization_code
    authorization_code = request.args.get('code')
    return "授权成功！请返回应用程序。"

可以看到，我们本地的后端获取了 code

picture.image

4 完整的授权 + BOT对话

效果

* Running on http://127.0.0.1:8080
Press CTRL+C to quit
127.0.0.1 - - [02/Dec/2024 09:30:47] "GET /callback?state=random_state&code=code_wzDngPmMFZP6KGZwYbT9kxv8libzoJGq5YVvvk4zBJsJvGDh HTTP/1.1" 200 -
你: 你好  
对话已创建: {'id': '7443620962052014117', 'conversation_id': '7443620962051997733', 'bot_id': '7443362942911496207', 'created_at': 1733103064, 'last_error': {'code': 0, 'msg': ''}, 'status': 'created', 'usage': {'token_count': 0, 'output_count': 0, 'input_count': 0}, 'section_id': '7443620962051997733'}
对话进行中: {'id': '7443620962052014117', 'conversation_id': '7443620962051997733', 'bot_id': '7443362942911496207', 'created_at': 1733103064, 'last_error': {'code': 0, 'msg': ''}, 'status': 'in_progress', 'usage': {'token_count': 0, 'output_count': 0, 'input_count': 0}, 'section_id': '7443620962051997733'}
助手消息: 您好
助手消息: 呀
助手消息: ！
助手消息: 请问
助手消息: 有
助手消息: 什么
助手消息: 我
助手消息: 能
助手消息: 帮忙
助手消息: 的
助手消息: ？
助手消息:
未知事件类型: conversation.message.completed
未知事件类型: conversation.message.completed
未知事件类型: conversation.message.completed
未知事件类型: conversation.message.completed
未知事件类型: conversation.message.completed
对话已完成: {'id': '7443620962052014117', 'conversation_id': '7443620962051997733', 'bot_id': '7443362942911496207', 'created_at': 1733103064, 'completed_at': 1733103066, 'last_error': {'code': 0, 'msg': ''}, 'status': 'completed', 'usage': {'token_count': 266, 'output_count': 12, 'input_count': 254}, 'section_id': '7443620962051997733'}
未知事件类型: done

代码

需要设置

OAuth 中的值 USER_CLIENT_ID = "你的client id : xxxxxx.app.coze" # 你的机器人ID USER_BOT_ID = "你的机器人ID"

import requests
import base64
import hashlib
import os
import json
import uuid
from flask import Flask, request
import threading
import webbrowser

# OAuth 中的值
USER_CLIENT_ID = "你的client id : xxxxxx.app.coze"

# 你的机器人ID
USER_BOT_ID = "你的机器人ID"

# 生成随机的 code_verifier
def generate_code_verifier():
    return base64.urlsafe_b64encode(os.urandom(40)).rstrip(b'=').decode('utf-8')

# 根据 code_verifier 生成 code_challenge
def generate_code_challenge(code_verifier):
    code_challenge = hashlib.sha256(code_verifier.encode('utf-8')).digest()
    return base64.urlsafe_b64encode(code_challenge).rstrip(b'=').decode('utf-8')

# 发起授权请求
def initiate_authorization(client_id, redirect_uri):
    code_verifier = generate_code_verifier()
    code_challenge = generate_code_challenge(code_verifier)
    authorization_url = (
        f"https://www.coze.cn/api/permission/oauth2/authorize?"
        f"response_type=code&client_id={client_id}&redirect_uri={redirect_uri}"
        f"&state=random_state&code_challenge={code_challenge}&code_challenge_method=S256"
    )
    webbrowser.open(authorization_url)
    return code_verifier

# 获取授权码后请求访问令牌
def request_access_token(client_id, redirect_uri, code, code_verifier):
    token_url = "https://api.coze.cn/api/permission/oauth2/token"
    headers = {'Content-Type': 'application/json'}
    data = {
        "grant_type": "authorization_code",
        "code": code,
        "redirect_uri": redirect_uri,
        "client_id": client_id,
        "code_verifier": code_verifier
    }
    response = requests.post(token_url, headers=headers, json=data)
    return response.json()

# Flask 应用
app = Flask(__name__)
authorization_code = None

@app.route('/callback')
def callback():
    global authorization_code
    authorization_code = request.args.get('code')
    return "授权成功！请返回应用程序。"

class ChatBot:
    def __init__(self, access_token):
        self.messages = []
        self.input_value = ''
        self.conversation_id = None
        self.bot_id = USER_BOT_ID
        self.user_id = self.generate_uuid()
        self.access_token = access_token

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

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

        if not self.conversation_id:
            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}')

def main():
    client_id = USER_CLIENT_ID
    redirect_uri = "http://localhost:8080/callback"
    
    threading.Thread(target=lambda: app.run(port=8080)).start()
    
    code_verifier = initiate_authorization(client_id, redirect_uri)
    
    # 等待用户授权
    while authorization_code is None:
        pass
    
    token_response = request_access_token(client_id, redirect_uri, authorization_code, code_verifier)
    access_token = token_response.get("access_token")
    
    if access_token:
        chat_bot = ChatBot(access_token)
        
        # 一问一答循环
        while True:
            user_input = input("你: ")
            if user_input.lower() in ['exit', 'quit']:
                print("结束对话。")
                break
            chat_bot.input_value = user_input
            chat_bot.send_message()

if __name__ == "__main__":
    main()

[2][Coze API][python] 通过 OAuth PKCE 实现 扣子 BOT 对话