Coze Python API SDK

Introduction

The Coze Python SDK is your comprehensive solution for seamlessly integrating Coze's powerful open APIs into Python applications.

• Complete API coverage: All Coze open APIs and authentication methods supported
• Dual interface: Both synchronous and asynchronous SDK calls available
• Stream-optimized: Native Stream and AsyncStream objects for real-time data
• Pagination made easy: Iterator-based Page objects for efficient list operations
• Developer-first: Intuitive API design for rapid integration

Requirements

Python 3.7 or higher.

Install

pip install cozepy

Usage

Examples

Module	Example	File
Auth	JWT OAuth	examples/auth_oauth_jwt.py
Web OAuth	examples/auth_oauth_web.py
PKCE OAuth	examples/auth_oauth_pkce.py
Device OAuth	examples/auth_oauth_device.py
Personal Access Token	examples/auth_pat.py
Websockets	Audio Speech by Websocket	examples/websockets_audio_speech.py
Audio Transcription by Websocket	examples/websockets_audio_transcriptions.py
Audio Chat by Websocket	examples/websockets_chat.py
Audio Chat Realtime by Websocket	examples/websockets_chat_realtime_gui.py
Bot Chat	Bot Chat with Stream	examples/chat_stream.py
Bot Chat without Stream	examples/chat_no_stream.py
Bot Chat with Conversation	examples/chat_conversation_stream.py
Bot Chat with Image	examples/chat_multimode_stream.py
Bot Chat with Audio	examples/chat_simple_audio.py
Bot Chat with Audio One-One	examples/chat_oneonone_audio.py
Bot Chat With Local Plugin	examples/chat_local_plugin.py
Workflow Run & Chat	Workflow Run with Stream	examples/workflow_stream.py
Workflow Run without Stream	examples/workflow_no_stream.py
Workflow Async Run & Fetch	examples/workflow_async.py
Workflow Chat with Stream	examples/workflow_chat_stream.py
Workflow chat with Image and Stream	examples/workflow_chat_multimode_stream.py
List Workflow Versions	examples/workflow_version_list.py
Conversation	Conversation	examples/conversation.py
List Conversation	examples/conversation_list.py
Dataset	Create Dataset	examples/dataset_create.py
Audio	Audio Example by HTTP	examples/audio.py
Bot	Publish Bot	examples/bot_publish.py
File	Upload File	examples/files_upload.py
template	Duplicate Template	examples/template_duplicate.py
User	Get Current User	examples/users_me.py
Workspace	List Workspaces	examples/workspace.py
Variable	Retrieve Variable	examples/variable_retrieve.py
Update Variable	examples/variable_update.py
Other	Timeout Config	examples/timeout.py
Log Config	examples/log.py
Exception Usage	examples/exception.py

Initialize client

Start by obtaining your access token from the Coze platform:

• For coze.cn: Visit https://www.coze.cn/open/oauth/pats
• For coze.com: Visit https://www.coze.com/open/oauth/pats

Create a new personal access token by specifying a name, expiration period, and required permissions. Store this token securely—never expose it in code or version control.

import os

from cozepy import Coze, TokenAuth, COZE_CN_BASE_URL, AsyncCoze, AsyncTokenAuth

# Get an access_token through personal access token or oauth.
coze_api_token = os.getenv("COZE_API_TOKEN")
# 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_CN_BASE_URL

# init coze with token and base_url
coze = Coze(auth=TokenAuth(coze_api_token), base_url=coze_api_base)
async_coze = AsyncCoze(auth=AsyncTokenAuth(coze_api_token), base_url=coze_api_base)

coze api access_token can also be generated via the OAuth App. For details, refer to:

• Web OAuth
• JWT OAuth
• PKCE OAuth
• Device OAuth

Bot Chat

Create your bot in Coze and extract the bot ID from the URL (the final numeric segment).

Use coze.chat.stream for real-time streaming conversations. This method returns a Chat Iterator that yields events as they occur—simply iterate through the stream to process each event.

import os

from cozepy import Coze, TokenAuth, Message, ChatEventType, COZE_CN_BASE_URL

# initialize client
coze_api_token = os.getenv("COZE_API_TOKEN")
coze_api_base = os.getenv("COZE_API_BASE") or COZE_CN_BASE_URL
coze = Coze(auth=TokenAuth(coze_api_token), base_url=coze_api_base)

# The return values of the streaming interface can be iterated immediately.
for event in coze.chat.stream(
        # id of bot
        bot_id='bot_id',
        # id of user, Note: The user_id here is specified by the developer, for example, it can be the
        # business id in the developer system, and does not include the internal attributes of coze.
        user_id='user_id',
        # user input
        additional_messages=[Message.build_user_question_text("How are you?")]
        # conversation id, for Chaining conversation context
        # conversation_id='<conversation_id>',
):
    if event.event == ChatEventType.CONVERSATION_MESSAGE_DELTA:
        print(event.message.content, end="")

    if event.event == ChatEventType.CONVERSATION_CHAT_COMPLETED:
        print()
        print("token usage:", event.chat.usage.token_count)

Workflow Chat

Execute workflows directly through the SDK for powerful automation capabilities.

import os

from cozepy import TokenAuth, ChatEventType, Coze, COZE_CN_BASE_URL, Message

# Get the workflow id
workflow_id = os.getenv("COZE_WORKFLOW_ID") or "workflow id"
# Get the bot id
bot_id = os.getenv("COZE_BOT_ID")

# initialize client
coze_api_token = os.getenv("COZE_API_TOKEN")
coze_api_base = os.getenv("COZE_API_BASE") or COZE_CN_BASE_URL
coze = Coze(auth=TokenAuth(coze_api_token), base_url=coze_api_base)

conversation = coze.conversations.create()

# Call the coze.chat.stream method to create a chat. The create method is a streaming
# chat and will return a Chat Iterator. Developers should iterate the iterator to get
# chat event and handle them.
for event in coze.workflows.chat.stream(
        workflow_id=workflow_id,
        bot_id=bot_id,
        conversation_id=conversation.id,
        additional_messages=[
            Message.build_user_question_text("How are you?"),
        ],
):
    if event.event == ChatEventType.CONVERSATION_MESSAGE_DELTA:
        print(event.message.content, end="", flush=True)

    if event.event == ChatEventType.CONVERSATION_CHAT_COMPLETED:
        print()
        print("token usage:", event.chat.usage.token_count)

Real-time Audio Chat with WebSocket

import asyncio
import os

from cozepy import (
    AsyncTokenAuth,
    COZE_CN_BASE_URL,
    AsyncCoze,
    AsyncWebsocketsChatClient,
    AsyncWebsocketsChatEventHandler,
    AudioFormat,
    ConversationAudioDeltaEvent,
    ConversationChatCompletedEvent,
    ConversationChatCreatedEvent,
    ConversationMessageDeltaEvent,
    InputAudioBufferAppendEvent,
)
from cozepy.log import log_info
from cozepy.util import write_pcm_to_wav_file


class AsyncWebsocketsChatEventHandlerSub(AsyncWebsocketsChatEventHandler):
    delta = []

    async def on_conversation_chat_created(self, cli: AsyncWebsocketsChatClient, event: ConversationChatCreatedEvent):
        log_info("[examples] asr completed, logid=%s", event.detail.logid)

    async def on_conversation_message_delta(self, cli: AsyncWebsocketsChatClient, event: ConversationMessageDeltaEvent):
        print("Received:", event.data.content)

    async def on_conversation_audio_delta(self, cli: AsyncWebsocketsChatClient, event: ConversationAudioDeltaEvent):
        self.delta.append(event.data.get_audio())

    async def on_conversation_chat_completed(
        self, cli: "AsyncWebsocketsChatClient", event: ConversationChatCompletedEvent
    ):
        log_info("[examples] Saving audio data to output.wav")
        write_pcm_to_wav_file(b"".join(self.delta), "output.wav")


def wrap_coze_speech_to_iterator(coze: AsyncCoze, text: str):
    async def iterator():
        voices = await coze.audio.voices.list()
        content = await coze.audio.speech.create(
            input=text,
            voice_id=voices.items[0].voice_id,
            response_format=AudioFormat.WAV,
            sample_rate=24000,
        )
        for data in content._raw_response.iter_bytes(chunk_size=1024):
            yield data

    return iterator


async def main():
    coze_api_token = os.getenv("COZE_API_TOKEN")
    coze_api_base = os.getenv("COZE_API_BASE") or COZE_CN_BASE_URL
    coze = AsyncCoze(auth=AsyncTokenAuth(coze_api_token), base_url=coze_api_base)

    bot_id = os.getenv("COZE_BOT_ID")
    text = os.getenv("COZE_TEXT") or "How Are you?"

    # Initialize Audio
    speech_stream = wrap_coze_speech_to_iterator(coze, text)

    chat = coze.websockets.chat.create(
        bot_id=bot_id,
        on_event=AsyncWebsocketsChatEventHandlerSub(),
    )

    # Create and connect WebSocket client
    async with chat() as client:
        # Read and send audio data
        async for delta in speech_stream():
            await client.input_audio_buffer_append(
                InputAudioBufferAppendEvent.Data.model_validate(
                    {
                        "delta": delta,
                    }
                )
            )
        await client.input_audio_buffer_complete()
        await client.wait()


asyncio.run(main())

Request Debugging with LogID

Every SDK request includes a unique log ID for debugging purposes. Retrieve this ID from any response object to troubleshoot issues with Coze support.

```python import os from cozepy import Co