feat: Enhance OAuth configuration and improve example scripts (#162)

- Add load_oauth_app_from_config function for simplified OAuth app setup
- Improve API base URL and token retrieval mechanisms in example scripts

cozepy/__init__.py

@@ -18,6 +18,7 @@
     Scope,
     TokenAuth,
     WebOAuthApp,
+    load_oauth_app_from_config,
 )
 from .bots import (
     Bot,
@@ -159,6 +160,7 @@
     # audio.transcriptions
     "CreateTranscriptionsResp",
     # auth
+    "load_oauth_app_from_config",
     "AsyncDeviceOAuthApp",
     "AsyncJWTOAuthApp",
     "AsyncPKCEOAuthApp",

cozepy/auth/__init__.py

@@ -1,6 +1,7 @@
 import abc
+import json
 import time
-from typing import List, Optional
+from typing import List, Optional, Union
 from urllib.parse import quote_plus, urlparse
 
 from authlib.jose import jwt  # type: ignore
@@ -660,6 +661,28 @@ async def refresh_access_token(self, refresh_token: str) -> OAuthToken:
         return await self._arefresh_access_token(refresh_token)
 
 
+def load_oauth_app_from_config(conf: str) -> Union[PKCEOAuthApp, JWTOAuthApp, DeviceOAuthApp, WebOAuthApp]:
+    config = json.loads(conf)
+    client_id = config.get("client_id", "")
+    client_type = config.get("client_type", "")
+    coze_api_base = config.get("coze_api_base", "")
+    coze_www_base = config.get("coze_www_base", "")
+
+    if client_type == "single_page":
+        return PKCEOAuthApp(client_id, coze_api_base, coze_www_base)
+    elif client_type == "server":
+        private_key = config.get("private_key", "")
+        public_key_id = config.get("public_key_id", "")
+        return JWTOAuthApp(client_id, private_key, public_key_id, coze_api_base)
+    elif client_type == "device":
+        return DeviceOAuthApp(client_id, coze_api_base, coze_www_base)
+    elif client_type == "web":
+        client_secret = config.get("client_secret", "")
+        return WebOAuthApp(client_id, client_secret, coze_api_base, coze_www_base)
+    else:
+        raise ValueError(f"Invalid OAuth client_type: {client_type}")
+
+
 class Auth(abc.ABC):
     """
     This class is the base class for all authorization types.

examples/chat_stream.py

@@ -4,18 +4,37 @@
 """
 
 import os
+from typing import Optional
 
-from cozepy import COZE_COM_BASE_URL
-from examples.utils import get_coze_api_token
+from cozepy import COZE_CN_BASE_URL, ChatEventType, Coze, DeviceOAuthApp, Message, TokenAuth  # noqa
 
-# The default access is api.coze.com, but if you need to access api.coze.cn,
-# please use base_url to configure the api endpoint to access
-coze_api_base = os.getenv("COZE_API_BASE") or COZE_COM_BASE_URL
 
-from cozepy import Coze, TokenAuth, Message, ChatStatus, MessageContentType, ChatEventType  # noqa
+def get_coze_api_base() -> str:
+    # The default access is api.coze.com, but if you need to access api.coze.cn,
+    # please use base_url to configure the api endpoint to access
+    coze_api_base = os.getenv("COZE_API_BASE")
+    if coze_api_base:
+        return coze_api_base
+
+    return COZE_CN_BASE_URL  # default
+
+
+def get_coze_api_token(workspace_id: Optional[str] = None) -> str:
+    # Get an access_token through personal access token or oauth.
+    coze_api_token = os.getenv("COZE_API_TOKEN")
+    if coze_api_token:
+        return coze_api_token
+
+    coze_api_base = get_coze_api_base()
+
+    device_oauth_app = DeviceOAuthApp(client_id="57294420732781205987760324720643.app.coze", base_url=coze_api_base)
+    device_code = device_oauth_app.get_device_code(workspace_id)
+    print(f"Please Open: {device_code.verification_url} to get the access token")
+    return device_oauth_app.get_access_token(device_code=device_code.device_code, poll=True).access_token
+
 
 # Init the Coze client through the access_token.
-coze = Coze(auth=TokenAuth(token=get_coze_api_token()), base_url=coze_api_base)
+coze = Coze(auth=TokenAuth(token=get_coze_api_token()), base_url=get_coze_api_base())
 
 # Create a bot instance in Coze, copy the last number from the web link as the bot's ID.
 bot_id = os.getenv("COZE_BOT_ID") or "bot id"

examples/websockets_audio_speech.py

@@ -1,27 +1,61 @@
 import asyncio
 import json
+import logging
 import os
+from typing import Optional
 
 from cozepy import (
+    COZE_CN_BASE_URL,
     AsyncCoze,
     AsyncWebsocketsAudioSpeechClient,
     AsyncWebsocketsAudioSpeechEventHandler,
+    DeviceOAuthApp,
     InputTextBufferAppendEvent,
     InputTextBufferCompletedEvent,
     SpeechAudioCompletedEvent,
     SpeechAudioUpdateEvent,
     TokenAuth,
+    setup_logging,
 )
 from cozepy.log import log_info
 from cozepy.util import write_pcm_to_wav_file
-from examples.utils import get_coze_api_base, get_coze_api_token, setup_examples_logger
+
+
+def get_coze_api_base() -> str:
+    # The default access is api.coze.com, but if you need to access api.coze.cn,
+    # please use base_url to configure the api endpoint to access
+    coze_api_base = os.getenv("COZE_API_BASE")
+    if coze_api_base:
+        return coze_api_base
+
+    return COZE_CN_BASE_URL  # default
+
+
+def get_coze_api_token(workspace_id: Optional[str] = None) -> str:
+    # Get an access_token through personal access token or oauth.
+    coze_api_token = os.getenv("COZE_API_TOKEN")
+    if coze_api_token:
+        return coze_api_token
+
+    coze_api_base = get_coze_api_base()
+
+    device_oauth_app = DeviceOAuthApp(client_id="57294420732781205987760324720643.app.coze", base_url=coze_api_base)
+    device_code = device_oauth_app.get_device_code(workspace_id)
+    print(f"Please Open: {device_code.verification_url} to get the access token")
+    return device_oauth_app.get_access_token(device_code=device_code.device_code, poll=True).access_token
+
+
+def setup_examples_logger():
+    coze_log = os.getenv("COZE_LOG")
+    if coze_log:
+        setup_logging(logging.getLevelNamesMapping().get(coze_log.upper(), logging.INFO))
+
 
 setup_examples_logger()
 
 kwargs = json.loads(os.getenv("COZE_KWARGS") or "{}")
 
 
-# todo review
 class AsyncWebsocketsAudioSpeechEventHandlerSub(AsyncWebsocketsAudioSpeechEventHandler):
     """
     Class is not required, you can also use Dict to set callback

examples/websockets_audio_transcriptions.py

@@ -1,19 +1,54 @@
 import asyncio
 import json
+import logging
 import os
+from typing import Optional
 
 from cozepy import (
+    COZE_CN_BASE_URL,
     AsyncCoze,
     AsyncWebsocketsAudioTranscriptionsClient,
     AsyncWebsocketsAudioTranscriptionsEventHandler,
     AudioFormat,
+    DeviceOAuthApp,
     InputAudioBufferAppendEvent,
     InputAudioBufferCompletedEvent,
     TokenAuth,
     TranscriptionsMessageUpdateEvent,
+    setup_logging,
 )
 from cozepy.log import log_info
-from examples.utils import get_coze_api_base, get_coze_api_token, setup_examples_logger
+
+
+def get_coze_api_base() -> str:
+    # The default access is api.coze.com, but if you need to access api.coze.cn,
+    # please use base_url to configure the api endpoint to access
+    coze_api_base = os.getenv("COZE_API_BASE")
+    if coze_api_base:
+        return coze_api_base
+
+    return COZE_CN_BASE_URL  # default
+
+
+def get_coze_api_token(workspace_id: Optional[str] = None) -> str:
+    # Get an access_token through personal access token or oauth.
+    coze_api_token = os.getenv("COZE_API_TOKEN")
+    if coze_api_token:
+        return coze_api_token
+
+    coze_api_base = get_coze_api_base()
+
+    device_oauth_app = DeviceOAuthApp(client_id="57294420732781205987760324720643.app.coze", base_url=coze_api_base)
+    device_code = device_oauth_app.get_device_code(workspace_id)
+    print(f"Please Open: {device_code.verification_url} to get the access token")
+    return device_oauth_app.get_access_token(device_code=device_code.device_code, poll=True).access_token
+
+
+def setup_examples_logger():
+    coze_log = os.getenv("COZE_LOG")
+    if coze_log:
+        setup_logging(logging.getLevelNamesMapping().get(coze_log.upper(), logging.INFO))
+
 
 setup_examples_logger()
 

examples/websockets_chat.py

@@ -1,8 +1,11 @@
 import asyncio
 import json
+import logging
 import os
+from typing import Optional
 
 from cozepy import (
+    COZE_CN_BASE_URL,
     AsyncCoze,
     AsyncWebsocketsChatClient,
     AsyncWebsocketsChatEventHandler,
@@ -13,13 +16,45 @@
     ConversationChatRequiresActionEvent,
     ConversationChatSubmitToolOutputsEvent,
     ConversationMessageDeltaEvent,
+    DeviceOAuthApp,
     InputAudioBufferAppendEvent,
     TokenAuth,
     ToolOutput,
+    setup_logging,
 )
 from cozepy.log import log_info
 from cozepy.util import write_pcm_to_wav_file
-from examples.utils import get_coze_api_base, get_coze_api_token, setup_examples_logger
+
+
+def get_coze_api_base() -> str:
+    # The default access is api.coze.com, but if you need to access api.coze.cn,
+    # please use base_url to configure the api endpoint to access
+    coze_api_base = os.getenv("COZE_API_BASE")
+    if coze_api_base:
+        return coze_api_base
+
+    return COZE_CN_BASE_URL  # default
+
+
+def get_coze_api_token(workspace_id: Optional[str] = None) -> str:
+    # Get an access_token through personal access token or oauth.
+    coze_api_token = os.getenv("COZE_API_TOKEN")
+    if coze_api_token:
+        return coze_api_token
+
+    coze_api_base = get_coze_api_base()
+
+    device_oauth_app = DeviceOAuthApp(client_id="57294420732781205987760324720643.app.coze", base_url=coze_api_base)
+    device_code = device_oauth_app.get_device_code(workspace_id)
+    print(f"Please Open: {device_code.verification_url} to get the access token")
+    return device_oauth_app.get_access_token(device_code=device_code.device_code, poll=True).access_token
+
+
+def setup_examples_logger():
+    coze_log = os.getenv("COZE_LOG")
+    if coze_log:
+        setup_logging(logging.getLevelNamesMapping().get(coze_log.upper(), logging.INFO))
+
 
 setup_examples_logger()
 

examples/websockets_chat_realtime_gui.py

@@ -1,5 +1,6 @@
 import asyncio
 import json
+import logging
 import os
 import queue
 import threading
@@ -18,11 +19,11 @@
     ChatUpdateEvent,
     ConversationAudioDeltaEvent,
     ConversationChatCompletedEvent,
+    InputAudio,
     InputAudioBufferAppendEvent,
     TokenAuth,
+    setup_logging,
 )
-from cozepy.websockets.ws import InputAudio
-from examples.utils import setup_examples_logger
 
 # 音频参数设置
 CHUNK = 1024
@@ -31,6 +32,13 @@
 RATE = 24000
 INPUT_BLOCK_TIME = 0.05  # 50ms per block
 
+
+def setup_examples_logger():
+    coze_log = os.getenv("COZE_LOG")
+    if coze_log:
+        setup_logging(logging.getLevelNamesMapping().get(coze_log.upper(), logging.INFO))
+
+
 setup_examples_logger()