### Initialize AsyncCoze Client and Get User Info

Source: https://context7.com/coze-dev/coze-py/llms.txt

Set up an asynchronous Coze client using `AsyncCoze` and `AsyncTokenAuth`. This example demonstrates fetching the current user's information asynchronously.

```python
import asyncio
from cozepy import AsyncCoze, AsyncTokenAuth, COZE_COM_BASE_URL

async def main():
    coze = AsyncCoze(auth=AsyncTokenAuth("pat_xxx"), base_url=COZE_COM_BASE_URL)
    user = await coze.users.me()
    print(user.name)

asyncio.run(main())
```

--------------------------------

### Install Project Dependencies

Source: https://github.com/coze-dev/coze-py/blob/main/CONTRIBUTING.md

Install all project dependencies managed by Poetry.

```shell
poetry install
```

--------------------------------

### Install Pre-Commit Hooks

Source: https://github.com/coze-dev/coze-py/blob/main/CONTRIBUTING.md

Install pre-commit hooks to automatically format and lint code before commits.

```shell
pre-commit install
```

--------------------------------

### Install Coze Python SDK

Source: https://github.com/coze-dev/coze-py/blob/main/README.md

Install the Coze Python SDK using pip. Ensure you have Python 3.7 or higher.

```shell
pip install cozepy
```

--------------------------------

### Install Poetry

Source: https://github.com/coze-dev/coze-py/blob/main/CONTRIBUTING.md

Install Poetry, a dependency management tool for Python, using pip.

```shell
python -m pip install poetry
```

--------------------------------

### Create and Manage Datasets (Knowledge Bases) with Python

Source: https://context7.com/coze-dev/coze-py/llms.txt

Create knowledge bases (datasets) to store text, image, or table documents that bots can query. Documents can be ingested from local files or URLs. This example shows creating a text-based dataset.

```python
from cozepy import Coze, TokenAuth, DocumentFormatType, DocumentSourceInfo, DocumentSourceType

coze = Coze(auth=TokenAuth("pat_xxx"))

# Create a text knowledge base
resp = coze.datasets.create(
    name="Product FAQ",
    space_id="space_id_here",
    format_type=DocumentFormatType.TEXT,
    description="Frequently asked questions about our product",
)
dataset_id = resp.dataset_id
print("Dataset ID:", dataset_id)
```

--------------------------------

### Asynchronous Iteration Over Bot Paginator

Source: https://github.com/coze-dev/coze-py/blob/main/README.md

This example shows how to asynchronously iterate over the bot paginator to fetch bots. It requires an asyncio event loop.

```python
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())
```

--------------------------------

### List Bots Asynchronously

Source: https://github.com/coze-dev/coze-py/blob/main/README.md

This example shows how to list bots asynchronously using the `AsyncCoze` client. It requires an API token set in the environment variable `COZE_API_TOKEN` and an actual workspace ID.

```python
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())
```

--------------------------------

### Submit Tool Outputs for Function Calls

Source: https://context7.com/coze-dev/coze-py/llms.txt

When a bot requires tool outputs (e.g., for a plugin), use `submit_tool_outputs` to provide the results and continue the conversation. This example shows how to break from a stream to submit outputs and then resume streaming.

```python
from cozepy import Coze, TokenAuth, Message, ChatEventType, ToolOutput

coze = Coze(auth=TokenAuth("pat_xxx"))

stream = coze.chat.stream(
    bot_id="bot_id_here",
    user_id="user_123",
    additional_messages=[Message.build_user_question_text("What is today's weather in Paris?")],
)

conversation_id = chat_id = None
for event in stream:
    if event.event == ChatEventType.CONVERSATION_CHAT_REQUIRES_ACTION:
        chat = event.chat
        conversation_id = chat.conversation_id
        chat_id = chat.id
        tool_calls = chat.required_action.submit_tool_outputs.tool_calls
        break

# Execute local function and submit result
result_stream = coze.chat.submit_tool_outputs(
    conversation_id=conversation_id,
    chat_id=chat_id,
    tool_outputs=[ToolOutput(tool_call_id=tool_calls[0].id, output='{"temperature": "18°C", "condition": "Cloudy"}')],
    stream=True,
)
for event in result_stream:
    if event.event == ChatEventType.CONVERSATION_MESSAGE_DELTA:
        print(event.message.content, end="")
```

--------------------------------

### Stream Workflow Executions with Python

Source: https://github.com/coze-dev/coze-py/blob/main/README.md

This example demonstrates how to stream events from a Coze workflow execution. It includes handling messages, errors, and interrupts, allowing for dynamic workflow progression.

```python
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',
    }
))
```

--------------------------------

### List Bots without Iterators

Source: https://github.com/coze-dev/coze-py/blob/main/README.md

This example shows how to fetch a list of bots using the Coze SDK without directly iterating over the paginator. It retrieves the first page of results and provides access to total count and has_more flag.

```python
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
```

--------------------------------

### AsyncCoze - Asynchronous Entry Point

Source: https://context7.com/coze-dev/coze-py/llms.txt

Demonstrates how to initialize and use the AsyncCoze client for asynchronous operations, such as retrieving user information.

```APIDOC
## AsyncCoze — Asynchronous Entry Point

`AsyncCoze` mirrors `Coze` with `async/await` interfaces on every sub-client method.

```python
import asyncio
from cozepy import AsyncCoze, AsyncTokenAuth, COZE_COM_BASE_URL

async def main():
    coze = AsyncCoze(auth=AsyncTokenAuth("pat_xxx"), base_url=COZE_COM_BASE_URL)
    user = await coze.users.me()
    print(user.name)

asyncio.run(main())
```
```

--------------------------------

### Real-time Audio Chat with WebSocket

Source: https://github.com/coze-dev/coze-py/blob/main/README.md

Demonstrates setting up a real-time audio chat using WebSockets. This involves handling audio deltas, sending audio buffers, and completing the chat. Ensure you have the necessary environment variables set for API token, base URL, and bot ID.

```python
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())

```

--------------------------------

### Initialize Coze Client (Sync)

Source: https://context7.com/coze-dev/coze-py/llms.txt

The `Coze` class is the primary synchronous client. Pass any `Auth` implementation and optionally a custom `base_url`.

```python
from cozepy import Coze, TokenAuth, COZE_CN_BASE_URL, COZE_COM_BASE_URL
```

--------------------------------

### Initialize Coze Client with Custom HTTP Client

Source: https://context7.com/coze-dev/coze-py/llms.txt

Configure the Coze client with a custom httpx client to manage settings like timeouts or proxies. This allows for more control over network requests.

```python
import httpx
from cozepy import Coze, TokenAuth, SyncHTTPClient

http = httpx.Client(timeout=30)
coze_custom = Coze(
    auth=TokenAuth("pat_xxx"),
    http_client=SyncHTTPClient(http_client=http),
)
```

--------------------------------

### List Bots (Asynchronous)

Source: https://github.com/coze-dev/coze-py/blob/main/README.md

Demonstrates how to list bots within a workspace using the asynchronous Coze SDK. It shows how to iterate through paginated results using async for.

```APIDOC
## List Bots (Asynchronous)

### Description
Retrieves a paginated list of bots available in a specified workspace asynchronously.

### Method
`coze.bots.list`

### Parameters
#### Path Parameters
None

#### Query Parameters
- **space_id** (string) - Required - The ID of the workspace to list bots from.
- **page_size** (integer) - Optional - The number of bots to retrieve per page.

### Request Example
```python
import asyncio
import os

from cozepy import AsyncTokenAuth, AsyncCoze

# Ensure COZE_API_TOKEN is set in your environment variables
coze = AsyncCoze(auth=AsyncTokenAuth(os.getenv("COZE_API_TOKEN")))

async def main():
    # Replace 'workspace id' with your actual workspace ID
    space_id = 'workspace id'
    bots_page = await coze.bots.list(space_id=space_id, page_size=10)
    async for page in bots_page.iter_pages():
        print(f'got page: {page.page_num}')
        for bot in page.items:
            print(f'got bot: {bot}')

asyncio.run(main())
```

### Response
#### Success Response (200)
- **page_num** (integer) - The current page number.
- **items** (list) - A list of bot objects.
  - Each bot object contains details about a specific bot.
```

--------------------------------

### Iterate Through Bots with Pagination

Source: https://context7.com/coze-dev/coze-py/llms.txt

Demonstrates two pagination patterns: flat iteration over all items and page-by-page iteration to access metadata like total count and individual page items.

```python
from cozepy import Coze, TokenAuth

coze = Coze(auth=TokenAuth("pat_xxx"))

# Pattern 1: flat iteration (all items across all pages)
for bot in coze.bots.list(space_id="space_id_here", page_size=20):
    print(bot.name)

# Pattern 2: page-by-page (access metadata like total count)
paged = coze.bots.list(space_id="space_id_here", page_size=20)
for page in paged.iter_pages():
    print(f"Page {page.page_num} of {page.total} total items")
    for bot in page.items:
        print(" -", bot.name)
```

--------------------------------

### Async Usage with AsyncCoze

Source: https://github.com/coze-dev/coze-py/blob/main/README.md

Demonstrates how to use the Coze SDK with full async/await support by utilizing `httpx.AsyncClient`. Replace `Coze` with `AsyncCoze` for non-blocking API calls. Ensure COZE_API_TOKEN is set and configure COZE_API_BASE if needed.

```python
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())

```

--------------------------------

### List Bots (Synchronous)

Source: https://github.com/coze-dev/coze-py/blob/main/README.md

Demonstrates how to list bots within a workspace using the synchronous Coze SDK. It shows how to iterate through paginated results.

```APIDOC
## List Bots (Synchronous)

### Description
Retrieves a paginated list of bots available in a specified workspace.

### Method
`coze.bots.list`

### Parameters
#### Path Parameters
None

#### Query Parameters
- **space_id** (string) - Required - The ID of the workspace to list bots from.
- **page_size** (integer) - Optional - The number of bots to retrieve per page.

### Request Example
```python
import coze

# Replace 'workspace id' with your actual workspace ID
space_id = 'workspace id'
bots_page = coze.bots.list(space_id=space_id, page_size=10)

for page in bots_page.iter_pages():
    print(f'got page: {page.page_num}')
    for bot in page.items:
        print(f'got bot: {bot}')
```

### Response
#### Success Response (200)
- **page_num** (integer) - The current page number.
- **items** (list) - A list of bot objects.
  - Each bot object contains details about a specific bot.
```

--------------------------------

### Async Usage Pattern - Paginated Bots List

Source: https://context7.com/coze-dev/coze-py/llms.txt

Demonstrates asynchronous retrieval of a paginated list of bots within a specified space.

```APIDOC
## Async Usage Pattern - Paginated Bots List

### Description
Asynchronously retrieves a paginated list of bots.

### Method Signature
```python
await coze.bots.list(
    space_id: str
)
```

### Parameters
- **space_id** (str) - Required - The ID of the space to list bots from.

### Request Example
```python
import asyncio
from cozepy import AsyncCoze, AsyncTokenAuth

async def main():
    coze = AsyncCoze(auth=AsyncTokenAuth("pat_xxx"))

    # Paginated bots list
    paged = await coze.bots.list(space_id="space_id_here")
    async for bot in paged:
        print(bot.name)

asyncio.run(main())
```

### Response
- Returns an async iterable of bot objects.
```

--------------------------------

### Initialize Coze Client with TokenAuth (Sync)

Source: https://context7.com/coze-dev/coze-py/llms.txt

Use TokenAuth for server-side scripts with a Personal Access Token. This method injects the token as a Bearer header.

```python
from cozepy import Coze, TokenAuth, COZE_COM_BASE_URL

coze = Coze(auth=TokenAuth(token="pat_your_personal_access_token"))

# Verify the token works by fetching the current user
user = coze.users.me()
print(user.name, user.user_id)
```

--------------------------------

### Initialize Coze Client

Source: https://github.com/coze-dev/coze-py/blob/main/README.md

Initialize the Coze client with your API token and optionally specify the base URL for Coze China. Ensure your API token is stored securely and not exposed in code.

```python
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)
```

--------------------------------

### Manage Coze Bots with Python SDK

Source: https://context7.com/coze-dev/coze-py/llms.txt

Create, update, publish, and list bots using the Coze Python SDK. Requires bot configuration including name, description, prompt, and onboarding information. Bots can be updated with new prompts and model configurations, then published to channels.

```python
from cozepy import Coze, TokenAuth, BotPromptInfo, BotOnboardingInfo, BotModelInfo

coze = Coze(auth=TokenAuth("pat_xxx"))

# Create a new bot (starts as unpublished draft)
bot = coze.bots.create(
    space_id="space_id_here",
    name="My Assistant",
    description="A helpful coding assistant",
    prompt_info=BotPromptInfo(prompt="You are a Python expert. Always use type hints."),
    onboarding_info=BotOnboardingInfo(
        prologue="Hello! Ask me anything about Python.",
        suggested_questions=["How do I use decorators?", "Explain list comprehensions"],
    ),
)
print("Created bot:", bot.bot_id)

# Update prompt
coze.bots.update(
    bot_id=bot.bot_id,
    prompt_info=BotPromptInfo(prompt="You are a senior Python and TypeScript engineer."),
    model_info_config=BotModelInfo(model_id="gpt-4o", temperature=0.7, max_tokens=2048),
)

# Publish to API channel (connector_id "1024")
published = coze.bots.publish(bot_id=bot.bot_id, connector_ids=["1024"])
print("Published version:", published.version)

# List bots in a space (paginated)
paged = coze.bots.list(space_id="space_id_here", page_size=10)
for bot_info in paged:
    print(bot_info.name, bot_info.bot_id, "published:", bot_info.is_published)
```

--------------------------------

### Initialize Coze Client for International and China Endpoints

Source: https://context7.com/coze-dev/coze-py/llms.txt

Instantiate the Coze client for either the international or China endpoint using token authentication. Ensure you have the correct base URLs and authentication tokens.

```python
from cozepy import Coze, TokenAuth

COZE_COM_BASE_URL = "https://api.coze.com"
COZE_CN_BASE_URL = "https://api.coze.cn"

# International endpoint
coze = Coze(auth=TokenAuth("pat_xxx"), base_url=COZE_COM_BASE_URL)

# China endpoint
coze_cn = Coze(auth=TokenAuth("pat_yyy"), base_url=COZE_CN_BASE_URL)
```

--------------------------------

### Coze Client Initialization

Source: https://context7.com/coze-dev/coze-py/llms.txt

Initializes the synchronous `Coze` client. Accepts an `Auth` implementation and an optional custom `base_url`.

```APIDOC
## Coze / AsyncCoze Client Initialization

### Coze — Synchronous Entry Point

The `Coze` class is the primary synchronous client. Pass any `Auth` implementation and optionally a custom `base_url` (default `https://api.coze.com`; use `COZE_CN_BASE_URL` for the China endpoint).

```python
from cozepy import Coze, TokenAuth, COZE_CN_BASE_URL, COZE_COM_BASE_URL


```

--------------------------------

### Async Chat Streaming and Bot Listing

Source: https://context7.com/coze-dev/coze-py/llms.txt

Demonstrates asynchronous usage of the Coze SDK for streaming chat responses and paginated bot listing. Requires an event loop to run.

```python
import asyncio
from cozepy import AsyncCoze, AsyncTokenAuth, Message, ChatEventType

async def main():
    coze = AsyncCoze(auth=AsyncTokenAuth("pat_xxx"))

    # Streaming chat
    async for event in coze.chat.stream(
        bot_id="bot_id_here",
        user_id="user_123",
        additional_messages=[Message.build_user_question_text("Tell me a joke")],
    ):
        if event.event == ChatEventType.CONVERSATION_MESSAGE_DELTA:
            print(event.message.content, end="", flush=True)
    print()

    # Paginated bots list
    paged = await coze.bots.list(space_id="space_id_here")
    async for bot in paged:
        print(bot.name)

asyncio.run(main())
```

--------------------------------

### Create Virtual Environment

Source: https://github.com/coze-dev/coze-py/blob/main/CONTRIBUTING.md

Use this command to create a Python virtual environment named ./.venv in your project directory.

```shell
python -m venv ./.venv
```

--------------------------------

### Logging Configuration

Source: https://github.com/coze-dev/coze-py/blob/main/README.md

Shows how to configure the logging level for the Coze SDK to aid in debugging.

```APIDOC
## Logging Configuration

### Description
Fine-tune SDK logging to match your debugging needs.

### Method
`setup_logging`

### Parameters
#### Query Parameters
- **level** (logging.Level) - Optional - The logging level to set. Defaults to `logging.WARNING`. Use `logging.DEBUG` for detailed logs.

### Request Example
```python
import logging
from cozepy import setup_logging

# Enable debug logging
setup_logging(level=logging.DEBUG)
```
```

--------------------------------

### List Workspaces

Source: https://context7.com/coze-dev/coze-py/llms.txt

Retrieves a paginated list of workspaces accessible with the current authentication token. The page size can be specified.

```python
from cozepy import Coze, TokenAuth

coze = Coze(auth=TokenAuth("pat_xxx"))

for workspace in coze.workspaces.list(page_size=20):
    print(workspace.id, workspace.name, workspace.workspace_type)
```

--------------------------------

### Activate Virtual Environment

Source: https://github.com/coze-dev/coze-py/blob/main/CONTRIBUTING.md

Activate the created virtual environment. This command is for Unix-based systems (Linux, macOS).

```shell
source ./.venv/bin/activate
```

--------------------------------

### Initialize Coze Client with JWTAuth (Sync)

Source: https://context7.com/coze-dev/coze-py/llms.txt

JWTAuth automatically generates short-lived access tokens using a service-account private key. Tokens are cached and renewed transparently. Ensure the private key is loaded and environment variables for client ID and public key ID are set.

```python
import os
from cozepy import Coze, JWTAuth, Scope, COZE_COM_BASE_URL

private_key = open("private_key.pem").read()

auth = JWTAuth(
    client_id=os.environ["COZE_CLIENT_ID"],
    private_key=private_key,
    public_key_id=os.environ["COZE_PUBLIC_KEY_ID"],
    ttl=7200,  # token TTL in seconds (max 86399)
)
coze = Coze(auth=auth, base_url=COZE_COM_BASE_URL)

# Optionally scope the token to specific bots
scoped_auth = JWTAuth(
    client_id=os.environ["COZE_CLIENT_ID"],
    private_key=private_key,
    public_key_id=os.environ["COZE_PUBLIC_KEY_ID"],
    oauth_app=auth._oauth_cli,  # reuse the JWTOAuthApp
)
```

--------------------------------

### Bot Management (Create, Update, Publish, List)

Source: https://context7.com/coze-dev/coze-py/llms.txt

Manage bots programmatically, including creating new bots, updating their prompts and models, publishing them to channels, and listing existing bots within a space.

```APIDOC
## coze.bots.create / update / publish / list

Create and manage bots programmatically: set prompts, bind knowledge bases, configure models, and publish to channels.

```python
from cozepy import Coze, TokenAuth, BotPromptInfo, BotOnboardingInfo, BotModelInfo

coze = Coze(auth=TokenAuth("pat_xxx"))

# Create a new bot (starts as unpublished draft)
bot = coze.bots.create(
    space_id="space_id_here",
    name="My Assistant",
    description="A helpful coding assistant",
    prompt_info=BotPromptInfo(prompt="You are a Python expert. Always use type hints."),
    onboarding_info=BotOnboardingInfo(
        prologue="Hello! Ask me anything about Python.",
        suggested_questions=["How do I use decorators?", "Explain list comprehensions"],
    ),
)
print("Created bot:", bot.bot_id)

# Update prompt
coze.bots.update(
    bot_id=bot.bot_id,
    prompt_info=BotPromptInfo(prompt="You are a senior Python and TypeScript engineer."),
    model_info_config=BotModelInfo(model_id="gpt-4o", temperature=0.7, max_tokens=2048),
)

# Publish to API channel (connector_id "1024")
published = coze.bots.publish(bot_id=bot.bot_id, connector_ids=["1024"])
print("Published version:", published.version)

# List bots in a space (paginated)
paged = coze.bots.list(space_id="space_id_here", page_size=10)
for bot_info in paged:
    print(bot_info.name, bot_info.bot_id, "published:", bot_info.is_published)
```
```

--------------------------------

### Dataset Management

Source: https://context7.com/coze-dev/coze-py/llms.txt

Create, list, and manage datasets (knowledge bases) which store text, image, or table documents that bots can query. Documents can be ingested from local files or URLs.

```APIDOC
## Datasets API

### coze.datasets — Create, List, and Manage Knowledge Bases

Datasets (knowledge bases) store text, image, or table documents that bots can query. Documents can be ingested from local files or URLs.

```python
from cozepy import Coze, TokenAuth, DocumentFormatType, DocumentSourceInfo, DocumentSourceType

coze = Coze(auth=TokenAuth("pat_xxx"))

# Create a text knowledge base
resp = coze.datasets.create(
    name="Product FAQ",
    space_id="space_id_here",
    format_type=DocumentFormatType.TEXT,
    description="Frequently asked questions about our product",
)
dataset_id = resp.dataset_id
print("Dataset ID:", dataset_id)
```
```

--------------------------------

### Upload and Retrieve Files with Python SDK

Source: https://context7.com/coze-dev/coze-py/llms.txt

Upload local files to the Coze platform and retrieve their metadata. The returned `file_id` can be used in multimodal messages, dataset ingestion, or bot icon configuration. Demonstrates uploading a PDF and using its ID in a chat message.

```python
from pathlib import Path
from cozepy import Coze, TokenAuth, Message, MessageObjectString

coze = Coze(auth=TokenAuth("pat_xxx"))

# Upload a file
uploaded = coze.files.upload(file=Path("report.pdf"))
print("File ID:", uploaded.id, "Size:", uploaded.bytes)

# Retrieve metadata
info = coze.files.retrieve(file_id=uploaded.id)
print("Name:", info.file_name, "Uploaded at:", info.created_at)

# Use file_id in a multimodal message
msg = Message.build_user_question_objects([
    MessageObjectString.build_text("Please summarize this document:"),
    MessageObjectString.build_file(file_id=uploaded.id),
])

for event in coze.chat.stream(bot_id="bot_id_here", user_id="u1", additional_messages=[msg]):
    from cozepy import ChatEventType
    if event.event == ChatEventType.CONVERSATION_MESSAGE_DELTA:
        print(event.message.content, end="")
```

--------------------------------

### Configure SDK Logging

Source: https://github.com/coze-dev/coze-py/blob/main/README.md

Adjust the SDK's logging level to aid in debugging. The default level is 'warning', but it can be set to 'DEBUG' for more verbose output.

```python
import logging

from cozepy import setup_logging

# Enable debug logging (default: warning)
setup_logging(level=logging.DEBUG)
```

--------------------------------

### Configure Custom HTTP Client with Timeouts

Source: https://github.com/coze-dev/coze-py/blob/main/README.md

Customize the HTTP client's timeout settings for both general requests and connection attempts. This is useful for managing performance and handling slow responses.

```python
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
            )
```

--------------------------------

### Async Usage Pattern - Streaming Chat

Source: https://context7.com/coze-dev/coze-py/llms.txt

Demonstrates asynchronous streaming chat functionality, allowing for real-time interaction with bots and receiving incremental message deltas.

```APIDOC
## Async Usage Pattern - Streaming Chat

### Description
All sub-client methods on `AsyncCoze` are coroutines. The streaming methods are async generators.

### Method Signature
```python
async for event in coze.chat.stream(
    bot_id: str,
    user_id: str,
    additional_messages: list[Message] = None,
):
    ...
```

### Parameters
- **bot_id** (str) - Required - The ID of the bot to interact with.
- **user_id** (str) - Required - The ID of the user.
- **additional_messages** (list[Message]) - Optional - A list of additional messages to include in the conversation.

### Request Example
```python
import asyncio
from cozepy import AsyncCoze, AsyncTokenAuth, Message, ChatEventType

async def main():
    coze = AsyncCoze(auth=AsyncTokenAuth("pat_xxx"))

    # Streaming chat
    async for event in coze.chat.stream(
        bot_id="bot_id_here",
        user_id="user_123",
        additional_messages=[Message.build_user_question_text("Tell me a joke")],
    ):
        if event.event == ChatEventType.CONVERSATION_MESSAGE_DELTA:
            print(event.message.content, end="", flush=True)
    print()

asyncio.run(main())
```

### Response Events
- `ChatEventType.CONVERSATION_MESSAGE_DELTA`: Contains incremental message content.
```

--------------------------------

### Upload Document from URL

Source: https://context7.com/coze-dev/coze-py/llms.txt

Uploads a document from a given web URL to a specified dataset. It then initiates a processing job for the uploaded documents.

```python
docs = coze.datasets.documents.create(
    dataset_id=dataset_id,
    document_bases=[
        {
            "name": "FAQ Page",
            "source_info": DocumentSourceInfo(
                document_source=DocumentSourceType.WEB_PAGE,
                web_url="https://example.com/faq",
            ).model_dump(),
        }
    ],
)

progress = coze.datasets.process(
    dataset_id=dataset_id,
    document_ids=[d.document_id for d in docs],
)
for p in progress:
    print(p.document_name, p.progress, "%", p.status)
```

--------------------------------

### Timeout Configuration

Source: https://github.com/coze-dev/coze-py/blob/main/README.md

Illustrates how to customize HTTP request timeouts by providing a custom httpx client to the Coze SDK.

```APIDOC
## Timeout Configuration

### Description
Customize HTTP timeouts using the underlying httpx client for optimal performance in your environment.

### Method
`Coze` constructor with `http_client` parameter.

### Parameters
#### Path Parameters
None

#### Query Parameters
None

#### Request Body
None

### Request Example
```python
import os
import httpx

from cozepy import COZE_COM_BASE_URL, Coze, TokenAuth, SyncHTTPClient

# Configure custom timeouts for the HTTP client
http_client = SyncHTTPClient(
    timeout=httpx.Timeout(
        timeout=600.0,  # 600s timeout for general requests
        connect=5.0     # 5s timeout for establishing a connection
    )
)

# Initialize the Coze client with the custom HTTP client
# Ensure COZE_API_TOKEN is set in your environment variables
coze = Coze(
    auth=TokenAuth(token=os.getenv("COZE_API_TOKEN")),
    base_url=COZE_COM_BASE_URL,
    http_client=http_client
)

# Now, any requests made through this 'coze' client will use the configured timeouts.
```

### Response
N/A - This section describes client initialization, not a specific API response.
```

--------------------------------

### Iterate Over Bot Paginator

Source: https://github.com/coze-dev/coze-py/blob/main/README.md

This snippet demonstrates iterating directly over the bot paginator to retrieve all bots. Each item yielded is a bot object.

```python
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)
```

--------------------------------

### coze.workspaces.list — List Workspaces

Source: https://context7.com/coze-dev/coze-py/llms.txt

Retrieves a paginated list of workspaces accessible by the current authentication token.

```APIDOC
## coze.workspaces.list — List Workspaces

### Description
Retrieve paginated workspaces accessible by the current auth token.

### Method Signature
```python
coze.workspaces.list(
    page_size: int = 20
)
```

### Parameters
- **page_size** (int) - Optional - The number of workspaces to return per page. Defaults to 20.

### Request Example
```python
from cozepy import Coze, TokenAuth

coze = Coze(auth=TokenAuth("pat_xxx"))

for workspace in coze.workspaces.list(page_size=20):
    print(workspace.id, workspace.name, workspace.workspace_type)
```

### Response
- Returns an iterable of workspace objects, each containing `id`, `name`, and `workspace_type`.
```

--------------------------------

### Real-Time WebSocket Chat

Source: https://context7.com/coze-dev/coze-py/llms.txt

Establishes a real-time bidirectional chat connection over WebSockets. Handles chat events, including message deltas, audio deltas, and completion. Supports sending audio chunks from a file.

```python
import threading
import wave
from cozepy import (
    Coze, TokenAuth,
    WebsocketsChatEventHandler, WebsocketsChatClient,
    ChatCreatedEvent, ConversationMessageDeltaEvent,
    ConversationAudioDeltaEvent, ConversationChatCompletedEvent,
    InputAudioBufferAppendEvent, AudioFormat,
)

coze = Coze(auth=TokenAuth("pat_xxx"))

class MyHandler(WebsocketsChatEventHandler):
    def on_chat_created(self, cli, event: ChatCreatedEvent):
        print("Connected!")

    def on_conversation_message_delta(self, cli, event: ConversationMessageDeltaEvent):
        print(event.data.content, end="", flush=True)

    def on_conversation_audio_delta(self, cli, event: ConversationAudioDeltaEvent):
        audio_bytes = event.data.get_audio()
        with open("response.pcm", "ab") as f:
            f.write(audio_bytes)

    def on_conversation_chat_completed(self, cli, event: ConversationChatCompletedEvent):
        print("\nChat complete. Tokens:", event.data.usage.token_count if event.data.usage else "N/A")

ws_client = coze.websockets.chat.create(
    bot_id="bot_id_here",
    on_event=MyHandler(),
)

# Read PCM audio from a file and send it chunk by chunk
with ws_client:
    with open("input.pcm", "rb") as audio_file:
        while chunk := audio_file.read(3200):  # 100ms chunks at 16kHz 16-bit mono
            ws_client.input_audio_buffer_append(
                InputAudioBufferAppendEvent.Data(audio=chunk.hex())
            )
    ws_client.input_audio_buffer_complete()
    # wait_events=[CONVERSATION_CHAT_COMPLETED] — client auto-waits
```

--------------------------------

### Execute Workflows Synchronously with Python

Source: https://context7.com/coze-dev/coze-py/llms.txt

Execute published workflows synchronously and retrieve the complete result. Supports asynchronous execution for long-running workflows by setting `is_async=True`. Requires a `workflow_id` and parameters for execution.

```python
import json
from cozepy import Coze, TokenAuth

coze = Coze(auth=TokenAuth("pat_xxx"))

result = coze.workflows.runs.create(
    workflow_id="workflow_id_here",
    parameters={
        "user_query": "Analyze the sentiment of: 'I love this product!'",
        "language": "en",
    },
)
print("Debug URL:", result.debug_url)
data = json.loads(result.data or "{}")
print("Output:", data)
```

--------------------------------

### List Bots Synchronously

Source: https://github.com/coze-dev/coze-py/blob/main/README.md

Use this snippet to list bots within a specified workspace using synchronous methods. Ensure you replace 'workspace id' with your actual workspace ID.

```python
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)
```

--------------------------------

### Text-to-Speech Conversion

Source: https://context7.com/coze-dev/coze-py/llms.txt

Converts input text into an audio file (MP3 format). Requires listing available voices first to select a voice ID. Supports adjustable speed and sample rate.

```python
from cozepy import Coze, TokenAuth, AudioFormat

coze = Coze(auth=TokenAuth("pat_xxx"))

# List available voices first
voices = list(coze.audio.voices.list())
voice_id = voices[0].voice_id if voices else "default_voice_id"

audio_response = coze.audio.speech.create(
    input="Hello! This is a demonstration of Coze text-to-speech.",
    voice_id=voice_id,
    response_format=AudioFormat.MP3,
    speed=1.0,
    sample_rate=24000,
)

with open("output.mp3", "wb") as f:
    f.write(audio_response.read())
print("Audio saved to output.mp3")
```

--------------------------------

### coze.chat.submit_tool_outputs - Local Plugin / Function Call

Source: https://context7.com/coze-dev/coze-py/llms.txt

Submits tool outputs to continue a conversation when the bot requires action, typically after a function call.

```APIDOC
### coze.chat.submit_tool_outputs — Local Plugin / Function Call

When a bot's chat enters `REQUIRES_ACTION` status (i.e., it needs the results of a local function), submit the outputs to continue the conversation.

```python
from cozepy import Coze, TokenAuth, Message, ChatEventType, ToolOutput

coze = Coze(auth=TokenAuth("pat_xxx"))

stream = coze.chat.stream(
    bot_id="bot_id_here",
    user_id="user_123",
    additional_messages=[Message.build_user_question_text("What is today's weather in Paris?")],
)

conversation_id = chat_id = None
for event in stream:
    if event.event == ChatEventType.CONVERSATION_CHAT_REQUIRES_ACTION:
        chat = event.chat
        conversation_id = chat.conversation_id
        chat_id = chat.id
        tool_calls = chat.required_action.submit_tool_outputs.tool_calls
        break

# Execute local function and submit result
result_stream = coze.chat.submit_tool_outputs(
    conversation_id=conversation_id,
    chat_id=chat_id,
    tool_outputs=[ToolOutput(tool_call_id=tool_calls[0].id, output='{"temperature": "18°C", "condition": "Cloudy"}')],
    stream=True,
)
for event in result_stream:
    if event.event == ChatEventType.CONVERSATION_MESSAGE_DELTA:
        print(event.message.content, end="")
```
```

--------------------------------

### TokenAuth - Personal Access Token Authentication

Source: https://context7.com/coze-dev/coze-py/llms.txt

Uses a static Personal Access Token (PAT) to authenticate requests by injecting it as a Bearer header. Suitable for server-side scripts.

```APIDOC
## TokenAuth — Personal Access Token

`TokenAuth` (sync) and `AsyncTokenAuth` (async) wrap a static Personal Access Token (PAT) and inject it as a `Bearer` header on every request. This is the simplest auth method for server-side scripts.

```python
from cozepy import Coze, TokenAuth, COZE_COM_BASE_URL

coze = Coze(auth=TokenAuth(token="pat_your_personal_access_token"))

# Verify the token works by fetching the current user
user = coze.users.me()
print(user.name, user.user_id)
```
```

--------------------------------

### coze.audio.speech.create — Text-to-Speech

Source: https://context7.com/coze-dev/coze-py/llms.txt

Converts text into natural-sounding audio. Supports multiple languages, formats (MP3, WAV, PCM, OGG), and adjustable speed and sample rate.

```APIDOC
## coze.audio.speech.create — Text-to-Speech

### Description
Convert text to a natural-sounding audio file. Supports multiple languages, formats (MP3, WAV, PCM, OGG), and adjustable speed/sample-rate.

### Method Signature
```python
coze.audio.speech.create(
    input: str,
    voice_id: str,
    response_format: AudioFormat = AudioFormat.MP3,
    speed: float = 1.0,
    sample_rate: int = 24000,
)
```

### Parameters
- **input** (str) - Required - The text to convert to speech.
- **voice_id** (str) - Required - The ID of the voice to use.
- **response_format** (AudioFormat) - Optional - The format of the audio file (e.g., MP3, WAV). Defaults to MP3.
- **speed** (float) - Optional - The speaking speed. Defaults to 1.0.
- **sample_rate** (int) - Optional - The sample rate of the audio. Defaults to 24000.

### Request Example
```python
from cozepy import Coze, TokenAuth, AudioFormat

coze = Coze(auth=TokenAuth("pat_xxx"))

# List available voices first
voices = list(coze.audio.voices.list())
voice_id = voices[0].voice_id if voices else "default_voice_id"

audio_response = coze.audio.speech.create(
    input="Hello! This is a demonstration of Coze text-to-speech.",
    voice_id=voice_id,
    response_format=AudioFormat.MP3,
    speed=1.0,
    sample_rate=24000,
)

with open("output.mp3", "wb") as f:
    f.write(audio_response.read())
print("Audio saved to output.mp3")
```

### Response
- Returns a file-like object containing the audio data.
```

--------------------------------

### PKCEOAuthApp - PKCE OAuth (Public Clients)

Source: https://context7.com/coze-dev/coze-py/llms.txt

Suitable for desktop and mobile applications where a client secret cannot be stored securely, using the PKCE flow.

```APIDOC
### PKCEOAuthApp — PKCE OAuth (Public Clients)

PKCE is suitable for desktop / mobile apps where a client secret cannot be stored securely.

```python
import secrets
from cozepy import PKCEOAuthApp, TokenAuth, Coze

app = PKCEOAuthApp(client_id="your_client_id")
code_verifier = secrets.token_urlsafe(64)

auth_url = app.get_oauth_url(
    redirect_uri="myapp://callback",
    code_verifier=code_verifier,
    code_challenge_method="S256",
    state="state123",
)
print("Open:", auth_url)

# After user authorizes and you receive ?code=...
token = app.get_access_token(
    redirect_uri="myapp://callback",
    code="received_code",
    code_verifier=code_verifier,
)
coze = Coze(auth=TokenAuth(token.access_token))
```
```

--------------------------------

### Stream Bot Conversations with Python

Source: https://github.com/coze-dev/coze-py/blob/main/README.md

Use this snippet to stream responses from a Coze bot in real-time. Ensure your COZE_API_TOKEN is set in the environment variables.

```python
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)
```