Coze Python SDK

repository·main·Indexed 19 days ago

https://github.com/coze-dev/coze-py

An OpenAPI SDK for Coze (coze.com/coze.cn) that integrates Coze's open APIs into Python applications. Version 0.20.0 supports both synchronous and asynchronous workflows, featuring complete API coverage, stream-optimized objects for real-time data, and iterator-based pagination. It provides specialized clients for bots, workflows, chat, audio/speech, datasets, knowledge bases, and WebSockets for low-latency audio interactions. Requires Python 3.7 or higher.

Tokens
11.7K
Snippets
38
Records
47
Agent score
62%

What's inside cozepy

  1. Overview of Coze Python SDK features

    main

    The Coze Python SDK provides a comprehensive interface for Coze's open APIs with the following capabilities:

    • Complete API coverage: Supports all Coze open APIs and authentication methods.
    • Dual interface: Provides both synchronous and asynchronous SDK calls.
    • Stream-optimized: Includes native Stream and AsyncStream objects for real-time data handling.
    • Pagination made easy: Uses iterator-based Page objects for efficient list operations.
    • Requirements: Requires Python 3.7 or higher.
  2. Use Smart Pagination for list endpoints

    main

    All list endpoints return intelligent paginators. You can interact with them in three ways:

    1. Manual Access: Access .items, .total, and .has_more directly from the page object.
    2. Direct Iteration: Iterate over the paginator object to get individual items.
    3. Page Iteration: Use .iter_pages() to iterate over page objects, which is useful if you need metadata like page_num.
    # 1. Not using iterators
    import os
    from cozepy import Coze, TokenAuth
    coze = Coze(auth=TokenAuth(os.getenv("COZE_API_TOKEN")))
    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 items
    for bot in bots_page:
        print('got bot:', bot)
    
    # 3. Iterate over the paginator iter_pages to get the next page paginator
    for page in bots_page.iter_pages():
        print('got page:', page.page_num)
        for bot in page.items:
            print('got bot:', bot)
  3. Initialize the Coze client

    main

    To use the SDK, you must first obtain an access token from the Coze platform (coze.cn or coze.com). You can initialize either a synchronous Coze client or an asynchronous AsyncCoze client. By default, the SDK uses api.coze.com, but you can configure the base_url to COZE_CN_BASE_URL if you need to access api.coze.cn.

    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)
  4. Overview of Coze SDK Client Modules

    main

    The cozepy package is organized into specialized client modules. Most modules provide both a synchronous client (e.g., BotsClient) and an asynchronous client (e.g., AsyncBotsClient).

    Key functional areas include:

    • Bots & Workflows: Manage bots (BotsClient), workflows (WorkflowsClient), and their versions or collaborators.
    • Chat & Conversations: Handle real-time chat (ChatClient), message history (ChatMessagesClient), and conversation management (ConversationsClient).
    • Audio & Speech: Manage audio playback, speech synthesis (SpeechClient), transcriptions (TranscriptionsClient), and voiceprints (VoiceprintGroupsClient).
    • Datasets & Knowledge: Manage datasets (DatasetsClient), documents (DocumentsClient), and knowledge bases (KnowledgeClient).
    • Websockets: Real-time streaming via WebsocketsClient, including specialized clients for WebsocketsChatClient, WebsocketsAudioClient, and WebsocketsAudioTranscriptionsClient.
    • Auth: Various authentication methods including TokenAuth, JWTAuth, and OAuth flows (OAuthApp, WebOAuthApp, etc.).
    • Workspace & Enterprise: Manage organizational structures, workspaces (WorkspacesClient), and enterprise members (EnterprisesClient).
  5. Handle paginated API responses

    main

    The SDK provides several abstractions for handling paginated data. Depending on the API endpoint, you will encounter one of the following pagination patterns:

    1. Number-based Pagination (NumberPaged / AsyncNumberPaged)

    Used when the API uses page_num and page_size. You can iterate over all items directly or iterate page by page.

    2. Token-based Pagination (TokenPaged / AsyncTokenPaged)

    Used when the API returns a next_page_token. The SDK automatically fetches the next page when you iterate.

    3. Last-ID Pagination (LastIDPaged / AsyncLastIDPaged)

    Used when the API uses before_id and after_id for cursor-based navigation.

    Common Properties for all Paged types:

    • .items: A list of the objects in the current page.
    • .has_more: A boolean indicating if more pages are available.
    • .total: The total number of items (if provided by the API).
    • .response: The HTTPResponse for the current page.
    # Example: Iterating through all items in a NumberPaged response
    for item in paged_response:  # This iterates through all pages and all items
        print(item)
    
    # Example: Iterating page by page (Sync)
    for page in paged_response.iter_pages():
        print(f"Page items: {len(page.items)}")
        if not page.has_more:
            break
  6. Access Coze service clients

    main

    The Coze and AsyncCoze classes act as gateways to various specialized service clients via properties. Each property returns a client dedicated to a specific domain of the Coze API.

    Common service clients include:

    • bots: Manage bots
    • chat: Handle chat interactions
    • workflows: Manage workflows
    • datasets: Manage datasets (replaces the deprecated knowledge client)
    • files: Manage files
    • workspaces: Manage workspaces
    • conversations: Manage conversations
    • users: Manage users
    • variables: Manage variables
    • apps: Manage apps
    • connectors: Manage connectors
    • audio: Manage audio resources
    • templates: Manage templates
    • folders: Manage folders
    • benefits / benefit_limitations: Manage billing and benefits
    • bill_tasks: Manage billing tasks
    • enterprises: Manage enterprise features
    • api_apps: Manage API-specific apps

    Deprecation Warning: The knowledge property is deprecated. Please migrate to using datasets instead.

  7. Configure SyncHTTPClient and AsyncHTTPClient

    main

    The Requester uses SyncHTTPClient (inheriting from httpx.Client) and AsyncHTTPClient (inheriting from httpx.AsyncClient) to manage connections.

    By default, these clients are configured with:

    • timeout: Uses DEFAULT_TIMEOUT from cozepy.config.
    • limits: Uses DEFAULT_CONNECTION_LIMITS from cozepy.config.
    • follow_redirects: Set to True by default.

    You can provide your own instances of these clients to the Requester constructor to customize connection pooling, timeouts, or proxy settings.

  8. Handle streaming API responses

    main

    For APIs that stream data (like chat completions), use the Stream (synchronous) or AsyncStream (asynchronous) classes. These classes wrap the raw HTTP stream and yield parsed objects as they arrive.

    • Stream: Use for event in stream: to iterate.
    • AsyncStream: Use async for event in stream: to iterate.
    # Async streaming example
    async for event in async_stream:  # event is the parsed object
        print(event)
  9. Configure Logging and Timeouts

    main

    Logging

    Use setup_logging to adjust the SDK's logging level. The default is warning.

    Timeouts

    Since the Coze client is built on httpx, you can customize timeouts by passing a custom SyncHTTPClient (or AsyncHTTPClient) during initialization.

    # Logging
    import logging
    from cozepy import setup_logging
    setup_logging(level=logging.DEBUG)
    
    # Timeout
    import os
    import httpx
    from cozepy import COZE_COM_BASE_URL, Coze, TokenAuth, SyncHTTPClient
    
    http_client = SyncHTTPClient(timeout=httpx.Timeout(
        timeout=600.0,
        connect=5.0
    ))
    
    coze = Coze(auth=TokenAuth(token=os.getenv("COZE_API_TOKEN")),
                base_url=COZE_COM_BASE_URL,
                http_client=http_client
                )
  10. Debug requests using LogID

    main

    Every SDK request includes a unique logid. You can retrieve this from any response object (including single responses, streaming responses, paginated responses, and list responses) to troubleshoot issues with Coze support.

    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)