Coze Python API SDK

Introduction

Requirements

Install

pip install cozepy

Usage

Examples

Initialize client

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)

Bot Chat

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

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

import os
from cozepy import Coze, TokenAuth
coze = Coze(auth=TokenAuth(os.getenv("COZE_API_TOKEN")))
bot = coze.bots.retrieve(bot_id='bot id')
print(bot.response.logid) # support for CozeModel
stream = coze.chat.stream(bot_id='bot id', user_id='user id')
print(stream.response.logid) # support for stream
workspaces = coze.workspaces.list()
print(workspaces.response.logid) # support for paged
messages = coze.chat.messages.list(conversation_id='conversation id', chat_id='chat id')
print(messages.response.logid) # support for list(simple list, not paged)

Async Usage

import os
import asyncio
from cozepy import AsyncTokenAuth, Message, AsyncCoze, COZE_CN_BASE_URL
# 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 = AsyncCoze(auth=AsyncTokenAuth(coze_api_token), base_url=coze_api_base)
async def main() -> None:
 chat = await coze.chat.create(
 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',
 additional_messages=[
 Message.build_user_question_text('how are you?'),
 Message.build_assistant_answer('I am fine, thank you.')
 ],
 )
 print('chat', chat)
asyncio.run(main())

Streaming Support

import os
from cozepy import Coze, TokenAuth, ChatEventType, Message
coze = Coze(auth=TokenAuth(os.getenv("COZE_API_TOKEN")))
stream = coze.chat.stream(
 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',
 additional_messages=[
 Message.build_user_question_text('how are you?'),
 Message.build_assistant_answer('I am fine, thank you.')
 ],
)
for event in stream:
 if event.event == ChatEventType.CONVERSATION_MESSAGE_DELTA:
 print('got message delta:', event.message.content)

import os
from cozepy import Coze, TokenAuth, Stream, WorkflowEvent, WorkflowEventType
coze = Coze(auth=TokenAuth(os.getenv("COZE_API_TOKEN")))
def handle_workflow_iterator(stream: Stream[WorkflowEvent]):
 for event in stream:
 if event.event == WorkflowEventType.MESSAGE:
 print('got message', event.message)
 elif event.event == WorkflowEventType.ERROR:
 print('got error', event.error)
 elif event.event == WorkflowEventType.INTERRUPT:
 handle_workflow_iterator(coze.workflows.runs.resume(
 workflow_id='workflow id',
 event_id=event.interrupt.interrupt_data.event_id,
 resume_data='hey',
 interrupt_type=event.interrupt.interrupt_data.type,
 ))
handle_workflow_iterator(coze.workflows.runs.stream(
 workflow_id='workflow id',
 parameters={
 'input_key': 'input value',
 }
))

import os
import asyncio
from cozepy import AsyncTokenAuth, ChatEventType, Message, AsyncCoze
coze = AsyncCoze(auth=AsyncTokenAuth(os.getenv("COZE_API_TOKEN")))
async def main():
 stream = coze.chat.stream(
 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',
 additional_messages=[
 Message.build_user_question_text('how are you?'),
 Message.build_assistant_answer('I am fine, thank you.')
 ],
 )
 async for event in stream:
 if event.event == ChatEventType.CONVERSATION_MESSAGE_DELTA:
 print('got message delta:', event.message.content)
asyncio.run(main())

Smart Pagination

1. Not using iterators

import os
from cozepy import Coze, TokenAuth
coze = Coze(auth=TokenAuth(os.getenv("COZE_API_TOKEN")))
# open your workspace, browser url will be https://www.coze.com/space/<this is workspace id>/develop
# copy <this is workspace id> as workspace id
bots_page = coze.bots.list(space_id='workspace id', page_size=10)
bots = bots_page.items
total = bots_page.total
has_more = bots_page.has_more

2. Iterate over the paginator, getting T

import os
from cozepy import Coze, TokenAuth
coze = Coze(auth=TokenAuth(os.getenv("COZE_API_TOKEN")))
# open your workspace, browser url will be https://www.coze.com/space/<this is workspace id>/develop
# copy <this is workspace id> as workspace id
bots_page = coze.bots.list(space_id='workspace id', page_size=10)
for bot in bots_page:
 print('got bot:', bot)

import os
import asyncio
from cozepy import AsyncTokenAuth, AsyncCoze
coze = AsyncCoze(auth=AsyncTokenAuth(os.getenv("COZE_API_TOKEN")))
async def main():
 # open your workspace, browser url will be https://www.coze.com/space/<this is workspace id>/develop
 # copy <this is workspace id> as workspace id
 bots_page = await coze.bots.list(space_id='workspace id', page_size=10)
 async for bot in bots_page:
 print('got bot:', bot)
asyncio.run(main())

3. Iterate over the paginator iter_pages to get the next page paginator

import os
from cozepy import Coze, TokenAuth
coze = Coze(auth=TokenAuth(os.getenv("COZE_API_TOKEN")))
# open your workspace, browser url will be https://www.coze.com/space/<this is workspace id>/develop
# copy <this is workspace id> as workspace id
bots_page = coze.bots.list(space_id='workspace id', page_size=10)
for page in bots_page.iter_pages():
 print('got page:', page.page_num)
 for bot in page.items:
 print('got bot:', bot)

import asyncio
import os
from cozepy import AsyncTokenAuth, AsyncCoze
coze = AsyncCoze(auth=AsyncTokenAuth(os.getenv("COZE_API_TOKEN")))
async def main():
 # open your workspace, browser url will be https://www.coze.com/space/<this is workspace id>/develop
 # copy <this is workspace id> as workspace id
 bots_page = await coze.bots.list(space_id='workspace id', page_size=10)
 async for page in bots_page.iter_pages():
 print('got page:', page.page_num)
 for bot in page.items:
 print('got bot:', bot)
asyncio.run(main())

Configuration Options

Logging Configuration

import logging
from cozepy import setup_logging
# Enable debug logging (default: warning)
setup_logging(level=logging.DEBUG)

Timeout Configuration

import os
import httpx
from cozepy import COZE_COM_BASE_URL, Coze, TokenAuth, SyncHTTPClient
# Coze client is built on httpx, and supports passing a custom httpx.Client when initializing
# Coze, and setting a timeout on the httpx.Client
http_client = SyncHTTPClient(timeout=httpx.Timeout(
 # 600s timeout on elsewhere
 timeout=600.0,
 # 5s timeout on connect
 connect=5.0
))
# Init the Coze client through the access_token and custom timeout http client.
coze = Coze(auth=TokenAuth(token=os.getenv("COZE_API_TOKEN")),
 base_url=COZE_COM_BASE_URL,
 http_client=http_client
 )