| Name | cozepy JSON |
| Version |
0.11.0
JSON |
| download |
| home_page | None |
| Summary | OpenAPI SDK for Coze(coze.com/coze.cn) |
| upload_time | 2025-01-23 23:37:22 |
| maintainer | None |
| docs_url | None |
| author | chyroc |
| requires_python | <4.0,>=3.7 |
| license | MIT |
| keywords |
|
| VCS |
|
| bugtrack_url |
|
| requirements |
No requirements were recorded.
|
| Travis-CI |
No Travis.
|
| coveralls test coverage |
No coveralls.
|
# Coze Python API SDK
[](https://pypi.org/project/cozepy/)
[](https://codecov.io/github/coze-dev/coze-py)
## Introduction
The Coze API SDK for Python is a versatile tool for integrating Coze's open APIs into
your projects.
- Supports all Coze open APIs and authentication APIs
- Supports both synchronous and asynchronous SDK calls
- Optimized for streaming apis, returning Stream and AsyncStream objects
- Optimized for list apis, returning Iterator Page objects
- Features a simple and user-friendly API design for ease of use
## Requirements
Python 3.7 or higher.
## Install
```shell
pip install cozepy
```
## Usage
### Examples
| Example | File |
|------------------------------|------------------------------------------------------------------------------------------|
| pat auth | [examples/auth_pat.py](examples/auth_pat.py) |
| oauth by web code | [examples/auth_oauth_web.py](examples/auth_oauth_web.py) |
| oauth by jwt flow | [examples/auth_oauth_jwt.py](examples/auth_oauth_jwt.py) |
| oauth by pkce flow | [examples/auth_oauth_pkce.py](examples/auth_oauth_pkce.py) |
| oauth by device flow | [examples/auth_oauth_device.py](examples/auth_oauth_device.py) |
| bot create, publish and chat | [examples/bot_publish.py](examples/bot_publish.py) |
| non-stream chat | [examples/chat_no_stream.py](examples/chat_no_stream.py) |
| steam chat | [examples/chat_stream.py](examples/chat_stream.py) |
| chat with conversation | [examples/chat_conversation_stream.py](examples/chat_conversation_stream.py) |
| chat with local plugin | [examples/chat_local_plugin.py](examples/chat_local_plugin.py) |
| chat with image | [examples/chat_multimodal_stream.py](examples/chat_multimode_stream.py) |
| chat with audio | [examples/chat_audio.py](examples/chat_audio.py) |
| workflow non-stream run | [examples/workflow_no_stream.py](examples/workflow_no_stream.py) |
| workflow stream run | [examples/workflow_stream.py](examples/workflow_stream.py) |
| async workflow run | [examples/workflow_async.py](examples/workflow_async.py) |
| workflow chat | [examples/workflow_chat_stream.py](examples/workflow_chat_stream.py) |
| workflow chat with image | [examples/workflow_chat_multimode_stream.py](examples/workflow_chat_multimode_stream.py) |
| conversation | [examples/conversation.py](examples/conversation.py) |
| workspace | [examples/workspace.py](examples/workspace.py) |
| timeout config | [examples/timeout.py](examples/timeout.py) |
| setup coze log config | [examples/log.py](examples/log.py) |
| how to handle exception | [examples/exception.py](examples/exception.py) |
### Initialize the Coze client
Firstly, you need to access https://www.coze.com/open/oauth/pats (for the cn environment,
visit https://www.coze.cn/open/oauth/pats).
Click to add a new token. After setting the
appropriate name, expiration time, and permissions, click OK to generate your personal
access token.
Please store it in a secure environment to prevent this personal access
token from being disclosed.
```python
import os
from cozepy import Coze, TokenAuth, COZE_COM_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_COM_BASE_URL
coze = Coze(auth=TokenAuth(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-app](https://github.com/coze-dev/coze-py?tab=readme-ov-file#web-oauth-app)
- [jwt-oauth-app](https://github.com/coze-dev/coze-py?tab=readme-ov-file#jwt-oauth-app)
- [pkce-oauth-app](https://github.com/coze-dev/coze-py?tab=readme-ov-file#pkce-oauth-app)
- [device-oauth-app](https://github.com/coze-dev/coze-py?tab=readme-ov-file#device-oauth-app)
### Chat
Create a bot instance in Coze, copy the last number from the web link as the bot's ID.
#### Non-stream Chat
To simplify the call, the SDK provides a wrapped function to complete non-streaming chat,
polling, and obtaining the messages of the chat. Developers can use create_and_poll to
simplify the process.
```python
import os
from cozepy import Coze, TokenAuth, Message, ChatStatus
coze = Coze(auth=TokenAuth(os.getenv("COZE_API_TOKEN")))
chat_poll = coze.chat.create_and_poll(
# 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?")]
)
for message in chat_poll.messages:
print(message.content, end="")
if chat_poll.chat.status == ChatStatus.COMPLETED:
print()
print("token usage:", chat_poll.chat.usage.token_count)
```
#### Stream Chat
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.
```python
import os
from cozepy import Coze, TokenAuth, Message, ChatEventType
coze = Coze(auth=TokenAuth(os.getenv("COZE_API_TOKEN")))
# 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?")]
):
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)
```
### Bots
You can create, update, publish and get the list of bots via api.
```python
import os
from cozepy import Coze, TokenAuth
coze = Coze(auth=TokenAuth(os.getenv("COZE_API_TOKEN")))
# retrieve bot info
bot = coze.bots.retrieve(bot_id='bot id')
# list bot list
# 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_num=1)
bots = bots_page.items
# create bot
bot = coze.bots.create(
# id of workspace, 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
space_id='workspace id',
# name of bot
name='bot name',
# description of bot
description='bot description',
)
# update bot info
coze.bots.update(
# id of workspace
bot_id='bot id',
# name of bot
name='bot name',
# description of bot
description='bot description',
)
# delete bot
bot = coze.bots.publish(bot_id='bot id')
```
### Conversations
Users can create conversations, and conduct conversations, inquire about messages,
etc. on conversations.
```python
import os
from cozepy import Coze, TokenAuth, Message, MessageContentType
coze = Coze(auth=TokenAuth(os.getenv("COZE_API_TOKEN")))
# If there is a need to build a context with user-assistant question-and-answer pairs,
# a conversation can be created through the create interface.
conversation = coze.conversations.create(
messages=[
# user question: how are you?
Message.build_user_question_text('who are you?'),
# assistant answer: I am Coze Bot.
Message.build_assistant_answer('I am Coze Bot.')
],
)
# retrieve conversation
conversation = coze.conversations.retrieve(conversation_id=conversation.id)
# append message to conversation
message = coze.conversations.messages.create(
# id of conversation
conversation_id=conversation.id,
content='how are you?',
content_type=MessageContentType.TEXT,
)
# retrieve message
message = coze.conversations.messages.retrieve(conversation_id=conversation.id, message_id=message.id)
# update message
coze.conversations.messages.update(
conversation_id=conversation.id,
message_id=message.id,
content='hey, how are you?',
content_type=MessageContentType.TEXT,
)
# delete message
coze.conversations.messages.delete(conversation_id=conversation.id, message_id=message.id)
# list messages
message_list = coze.conversations.messages.list(conversation_id=conversation.id)
```
### Files
Coze enables users to upload pictures and files. The uploaded pictures and files
can be utilized in the bot avatar and multimodal conversations.
```python
import os
from pathlib import Path
from cozepy import Coze, TokenAuth
coze = Coze(auth=TokenAuth(os.getenv("COZE_API_TOKEN")))
# upload file
file = coze.files.upload(file=Path('/filepath'))
# retrieve file info
coze.files.retrieve(file_id=file.id)
```
### Workflows
Coze also enables users to directly invoke the workflow.
#### Non-stream workflow run
```python
import os
from cozepy import Coze, TokenAuth, Stream, WorkflowEvent, WorkflowEventType
coze = Coze(auth=TokenAuth(os.getenv("COZE_API_TOKEN")))
result = coze.workflows.runs.create(
# id of workflow
workflow_id='workflow id',
# params
parameters={
'input_key': 'input value',
}
)
```
#### Stream workflow run
The returned result of the streaming interface is an iterator and can be directly iterated.
When the workflow incorporates question-and-answer nodes, the streaming interface will
return the INTERRUPT event.
Users should call the resume interface to submit the results of the question-and-answer.
The return value of resume remains an iterator, so recursive processing might be necessary here.
```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:
# Users should call the resume interface to submit the results of the question-and-answer
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(
# id of workflow
workflow_id='workflow id',
# params
parameters={
'input_key': 'input value',
}
))
```
### Knowledge
```python
import os
from cozepy import Coze, TokenAuth, DocumentBase, DocumentSourceInfo, DocumentChunkStrategy, DocumentUpdateRule
coze = Coze(auth=TokenAuth(os.getenv("COZE_API_TOKEN")))
# create knowledge documents by local_file
documents = coze.knowledge.documents.create(
# id of dataset
dataset_id='dataset id',
# document data
document_bases=[
DocumentBase(
name='document name',
source_info=DocumentSourceInfo.build_local_file('local file content')
)
],
# chunk strategy, needed when first create
chunk_strategy=DocumentChunkStrategy.build_auto()
)
# create knowledge documents by web_page
documents = coze.knowledge.documents.create(
# id of dataset
dataset_id='dataset id',
# document data
document_bases=[
DocumentBase(
name='document name',
source_info=DocumentSourceInfo.build_web_page('https://example.com')
)
],
# chunk strategy, needed when first create
chunk_strategy=DocumentChunkStrategy.build_auto()
)
# update knowledge document
coze.knowledge.documents.update(
# id of document
document_id='document id',
# name of document
document_name='name',
# update rule, current set to no auto-update
update_rule=DocumentUpdateRule.build_no_auto_update()
)
# delete knowledge document
coze.knowledge.documents.delete(document_ids=['document id'])
# list knowledge documents
paged_documents = coze.knowledge.documents.list(
# id of dataset
dataset_id='dataset id',
page_num=1,
page_size=10
)
```
### OAuth App
#### Web OAuth App
Firstly, users need to access https://www.coze.com/open/oauth/apps. For the cn environment,
users need to access https://www.coze.cn/open/oauth/apps to create an OAuth App of the type
of Web application.
The specific creation process can be referred to in the document:
https://www.coze.com/docs/developer_guides/oauth_code. For the cn environment, it can be
accessed at https://www.coze.cn/docs/developer_guides/oauth_code.
After the creation is completed, three parameters, namely the client ID, client secret,
and redirect link, can be obtained. For the client secret, users need to keep it securely
to avoid leakage.
```python
import os
from cozepy import Coze, TokenAuth, WebOAuthApp
# client ID
web_oauth_client_id = os.getenv("COZE_WEB_OAUTH_CLIENT_ID")
# client secret
web_oauth_client_secret = os.getenv("COZE_WEB_OAUTH_CLIENT_SECRET")
web_oauth_app = WebOAuthApp(
client_id=web_oauth_client_id,
client_secret=web_oauth_client_secret,
)
```
The WebOAuth authorization process is to first generate a coze authorization link and
send it to the coze user requiring authorization.
Once the coze user opens the link, they can see the authorization consent button.
```python
# redirect link
web_oauth_redirect_uri = os.getenv("COZE_WEB_OAUTH_REDIRECT_URI")
# Generate the authorization link and direct the user to open it.
url = web_oauth_app.get_oauth_url(redirect_uri=web_oauth_redirect_uri)
```
After the user clicks the authorization consent button, the coze web page will redirect
to the redirect address configured in the authorization link and carry the authorization
code and state parameters in the address via the query string.
```python
# Open the authorization link in your browser and authorize this OAuth App
# After authorization, you will be redirected to the redirect_uri with a code and state
# You can use the code to get the access token
code = 'mock code'
# After obtaining the code after redirection, the interface to exchange the code for a
# token can be invoked to generate the coze access_token of the authorized user.
oauth_token = web_oauth_app.get_access_token(redirect_uri=web_oauth_redirect_uri, code=code)
# use the access token to init Coze client
coze = Coze(auth=TokenAuth(oauth_token.access_token))
# When the token expires, you can also refresh and re-obtain the token
oauth_token = web_oauth_app.refresh_access_token(oauth_token.refresh_token)
```
#### JWT OAuth App
Firstly, users need to access https://www.coze.com/open/oauth/apps. For the cn environment,
users need to access https://www.coze.cn/open/oauth/apps to create an OAuth App of the type
of Service application.
The specific creation process can be referred to in the document:
https://www.coze.com/docs/developer_guides/oauth_jwt. For the cn environment, it can be
accessed at https://www.coze.cn/docs/developer_guides/oauth_jwt.
After the creation is completed, three parameters, namely the client ID, private key,
and public key id, can be obtained. For the client secret and public key id, users need to
keep it securely to avoid leakage.
```python
import os
from cozepy import Coze, TokenAuth, JWTOAuthApp
# client ID
jwt_oauth_client_id = os.getenv("COZE_JWT_OAUTH_CLIENT_ID")
# private key
jwt_oauth_private_key = os.getenv("COZE_JWT_OAUTH_PRIVATE_KEY")
# public key id
jwt_oauth_public_key_id = os.getenv("COZE_JWT_OAUTH_PUBLIC_KEY_ID")
jwt_oauth_app = JWTOAuthApp(
client_id=jwt_oauth_client_id,
private_key=jwt_oauth_private_key,
public_key_id=jwt_oauth_public_key_id,
)
```
The jwt oauth type requires using private to be able to issue a jwt token, and through
the jwt token, apply for an access_token from the coze service.
The sdk encapsulates this procedure, and only needs to use get_access_token to obtain
the access_token under the jwt oauth process.
```python
# The jwt process does not require any other operations, you can directly apply for a token
oauth_token = jwt_oauth_app.get_access_token(ttl=3600)
# use the access token to init Coze client
coze = Coze(auth=TokenAuth(oauth_token.access_token))
# The jwt oauth process does not support refreshing tokens. When the token expires,
# just directly call get_access_token to generate a new token.
```
#### PKCE OAuth App
PKCE stands for Proof Key for Code Exchange, and it's an extension to the OAuth 2.0 authorization
code flow designed to enhance security for public clients, such as mobile and single-page
applications.
Firstly, users need to access https://www.coze.com/open/oauth/apps. For the cn environment,
users need to access https://www.coze.cn/open/oauth/apps to create an OAuth App of the type
of Mobile/PC/Single-page application.
The specific creation process can be referred to in the document:
https://www.coze.com/docs/developer_guides/oauth_pkce. For the cn environment, it can be
accessed at https://www.coze.cn/docs/developer_guides/oauth_pkce.
After the creation is completed, three parameters, namely the client ID can be obtained.
```python
import os
from cozepy import PKCEOAuthApp
# client ID
pkce_oauth_client_id = os.getenv("COZE_PKCE_OAUTH_CLIENT_ID")
# redirect link
web_oauth_redirect_uri = os.getenv("COZE_WEB_OAUTH_REDIRECT_URI")
pkce_oauth_app = PKCEOAuthApp(client_id=pkce_oauth_client_id)
```
In the pkce oauth process, first, need to select a suitable code_challenge_method.
Coze supports two types: plain and s256.
Then, based on the selected code_challenge_method type, hash the code_verifier into
the code_challenge. Finally, based on the callback address,
code_challenge, and code_challenge_method, an authorization link can be generated.
```python
# In the SDK, we have wrapped up the code_challenge process of PKCE. Developers only need
# to select the code_challenge_method.
code_verifier = "random code verifier"
url = pkce_oauth_app.get_oauth_url(
redirect_uri=web_oauth_redirect_uri,
code_verifier=code_verifier,
code_challenge_method="S256"
)
```
Developers should lead users to open up this authorization link.
When the user consents to the authorization, Coze will redirect with the code to the
callback address configured by the developer, and the developer can obtain this code.
```python
# Open the authorization link in your browser and authorize this OAuth App
# After authorization, you can exchange code_verifier for access token
code = 'mock code'
# After obtaining the code after redirection, the interface to exchange the code for a
# token can be invoked to generate the coze access_token of the authorized user.
oauth_token = pkce_oauth_app.get_access_token(
redirect_uri=web_oauth_redirect_uri, code=code, code_verifier=code_verifier
)
# use the access token to init Coze client
coze = Coze(auth=TokenAuth(oauth_token.access_token))
# When the token expires, you can also refresh and re-obtain the token
oauth_token = pkce_oauth_app.refresh_access_token(oauth_token.refresh_token)
```
#### Device OAuth App
The OAuth 2.0 device flow, also known as the device authorization grant, is an extension to t
e OAuth 2.0 protocol designed for devices that have limited input capabilities or lack a suitable
browser.
Firstly, users need to access https://www.coze.com/open/oauth/apps. For the cn environment,
users need to access https://www.coze.cn/open/oauth/apps to create an OAuth App of the type
of TVs/Limited Input devices/Command line programs.
The specific creation process can be referred to in the document:
https://www.coze.com/docs/developer_guides/oauth_device_code. For the cn environment, it can be
accessed at https://www.coze.cn/docs/developer_guides/oauth_device_code.
After the creation is completed, three parameters, namely the client ID can be obtained.
```python
import os
from cozepy import Coze, TokenAuth, DeviceOAuthApp
# client ID
device_oauth_client_id = os.getenv("COZE_DEVICE_OAUTH_CLIENT_ID")
device_oauth_app = DeviceOAuthApp(client_id=device_oauth_client_id)
```
In the device oauth authorization process, developers need to first call the interface
of Coze to generate the device code to obtain the user_code and device_code.
Then generate the authorization link through the user_code, guide the user to open the
link, fill in the user_code, and consent to the authorization.
Developers need to call the interface of Coze to generate the token through the device_code.
When the user has not authorized or rejected the authorization, the interface will throw an
error and return a specific error code.
After the user consents to the authorization, the interface will succeed and return the
access_token.
```python
# First, you need to request the server to obtain the device code required in the device auth flow
device_code = device_oauth_app.get_device_code()
# The returned device_code contains an authorization link. Developers need to guide users
# to open up this link.
# open device_code.verification_url
```
The developers then need to use the device_code to poll Coze's interface to obtain the token.
The SDK has encapsulated this part of the code in and handled the different returned error
codes. The developers only need to invoke get_access_token.
```python
try:
oauth_token = device_oauth_app.get_access_token(
device_code=device_code.device_code,
poll=True,
)
except CozePKCEAuthError as e:
if e.error == CozePKCEAuthErrorType.ACCESS_DENIED:
# The user rejected the authorization.
# Developers need to guide the user to open the authorization link again.
pass
elif e.error == CozePKCEAuthErrorType.EXPIRED_TOKEN:
# The token has expired. Developers need to guide the user to open
# the authorization link again.
pass
else:
# Other errors
pass
raise # for example, re-raise the error
# use the access token to init Coze client
coze = Coze(auth=TokenAuth(oauth_token.access_token))
# When the token expires, you can also refresh and re-obtain the token
oauth_token = device_oauth_app.refresh_access_token(oauth_token.refresh_token)
```
### Debug
The SDK support returning the logid of the request, which can be used to debug the request.
You can get the logid from the response of the request and submit it to the coze support team for further assistance.
```python
import os
from cozepy import Coze, AsyncCoze, 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
cozepy supports asynchronous calls through `httpx.AsyncClient`.
Just replace the `Coze` client with the `AsyncCoze` client to use all the asynchronous calls of the Coze OpenAPI.
```python
import os
import asyncio
from cozepy import TokenAuth, Message, AsyncCoze
coze = AsyncCoze(auth=TokenAuth(os.getenv("COZE_API_TOKEN")))
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 usage
Bot chat and workflow run support running in streaming mode.
chat streaming example:
```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)
```
workflow streaming example:
```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',
}
))
```
Asynchronous calls also support streaming mode:
```python
import os
import asyncio
from cozepy import TokenAuth, ChatEventType, Message, AsyncCoze
coze = AsyncCoze(auth=TokenAuth(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())
```
### Paginator Iterator
The result returned by all list interfaces (both synchronous and asynchronous) is a paginator, which supports iteration.
Take the example of listing the bots in a space to explain the three ways to use the paginator iterator:
#### 1. Not using iterators
```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
```
#### 2. Iterate over the paginator, getting T
```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)
```
Asynchronous methods also support:
```python
import os
import asyncio
from cozepy import TokenAuth, AsyncCoze
coze = AsyncCoze(auth=TokenAuth(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
```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 page in bots_page.iter_pages():
print('got page:', page.page_num)
for bot in page.items:
print('got bot:', bot)
```
Asynchronous methods also support:
```python
import asyncio
import os
from cozepy import TokenAuth, AsyncCoze
coze = AsyncCoze(auth=TokenAuth(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())
```
### Config
#### Log Config
coze support config logging level
```python
import logging
from cozepy import setup_logging
# open debug logging, default is warning
setup_logging(level=logging.DEBUG)
```
#### Timeout Config
Coze client is built on httpx, and supports passing a custom httpx.Client when initializing
Coze, and setting a timeout on the httpx.Client
```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
)
```
Raw data
{
"_id": null,
"home_page": null,
"name": "cozepy",
"maintainer": null,
"docs_url": null,
"requires_python": "<4.0,>=3.7",
"maintainer_email": null,
"keywords": null,
"author": "chyroc",
"author_email": "chyroc@bytedance.com",
"download_url": "https://files.pythonhosted.org/packages/80/1f/88ac3e421f5ff31376866320af8497038d1a995a9ddd0275b108fda44982/cozepy-0.11.0.tar.gz",
"platform": null,
"description": "# Coze Python API SDK\n\n[](https://pypi.org/project/cozepy/)\n\n[](https://codecov.io/github/coze-dev/coze-py)\n\n## Introduction\n\nThe Coze API SDK for Python is a versatile tool for integrating Coze's open APIs into\nyour projects.\n\n- Supports all Coze open APIs and authentication APIs\n- Supports both synchronous and asynchronous SDK calls\n- Optimized for streaming apis, returning Stream and AsyncStream objects\n- Optimized for list apis, returning Iterator Page objects\n- Features a simple and user-friendly API design for ease of use\n\n## Requirements\n\nPython 3.7 or higher.\n\n## Install\n\n```shell\npip install cozepy\n```\n\n## Usage\n\n### Examples\n\n| Example | File |\n|------------------------------|------------------------------------------------------------------------------------------|\n| pat auth | [examples/auth_pat.py](examples/auth_pat.py) |\n| oauth by web code | [examples/auth_oauth_web.py](examples/auth_oauth_web.py) |\n| oauth by jwt flow | [examples/auth_oauth_jwt.py](examples/auth_oauth_jwt.py) |\n| oauth by pkce flow | [examples/auth_oauth_pkce.py](examples/auth_oauth_pkce.py) |\n| oauth by device flow | [examples/auth_oauth_device.py](examples/auth_oauth_device.py) |\n| bot create, publish and chat | [examples/bot_publish.py](examples/bot_publish.py) |\n| non-stream chat | [examples/chat_no_stream.py](examples/chat_no_stream.py) |\n| steam chat | [examples/chat_stream.py](examples/chat_stream.py) |\n| chat with conversation | [examples/chat_conversation_stream.py](examples/chat_conversation_stream.py) |\n| chat with local plugin | [examples/chat_local_plugin.py](examples/chat_local_plugin.py) |\n| chat with image | [examples/chat_multimodal_stream.py](examples/chat_multimode_stream.py) |\n| chat with audio | [examples/chat_audio.py](examples/chat_audio.py) |\n| workflow non-stream run | [examples/workflow_no_stream.py](examples/workflow_no_stream.py) |\n| workflow stream run | [examples/workflow_stream.py](examples/workflow_stream.py) |\n| async workflow run | [examples/workflow_async.py](examples/workflow_async.py) |\n| workflow chat | [examples/workflow_chat_stream.py](examples/workflow_chat_stream.py) |\n| workflow chat with image | [examples/workflow_chat_multimode_stream.py](examples/workflow_chat_multimode_stream.py) |\n| conversation | [examples/conversation.py](examples/conversation.py) |\n| workspace | [examples/workspace.py](examples/workspace.py) |\n| timeout config | [examples/timeout.py](examples/timeout.py) |\n| setup coze log config | [examples/log.py](examples/log.py) |\n| how to handle exception | [examples/exception.py](examples/exception.py) |\n\n### Initialize the Coze client\n\nFirstly, you need to access https://www.coze.com/open/oauth/pats (for the cn environment,\nvisit https://www.coze.cn/open/oauth/pats).\n\nClick to add a new token. After setting the\nappropriate name, expiration time, and permissions, click OK to generate your personal\naccess token.\n\nPlease store it in a secure environment to prevent this personal access\ntoken from being disclosed.\n\n```python\nimport os\n\nfrom cozepy import Coze, TokenAuth, COZE_COM_BASE_URL\n\n# Get an access_token through personal access token or oauth.\ncoze_api_token = os.getenv(\"COZE_API_TOKEN\")\n# The default access is api.coze.com, but if you need to access api.coze.cn,\n# please use base_url to configure the api endpoint to access\ncoze_api_base = os.getenv(\"COZE_API_BASE\") or COZE_COM_BASE_URL\n\ncoze = Coze(auth=TokenAuth(coze_api_token), base_url=coze_api_base)\n```\n\ncoze api access_token can also be generated via the OAuth App. For details, refer to:\n\n- [web-oauth-app](https://github.com/coze-dev/coze-py?tab=readme-ov-file#web-oauth-app)\n- [jwt-oauth-app](https://github.com/coze-dev/coze-py?tab=readme-ov-file#jwt-oauth-app)\n- [pkce-oauth-app](https://github.com/coze-dev/coze-py?tab=readme-ov-file#pkce-oauth-app)\n- [device-oauth-app](https://github.com/coze-dev/coze-py?tab=readme-ov-file#device-oauth-app)\n\n### Chat\n\nCreate a bot instance in Coze, copy the last number from the web link as the bot's ID.\n\n#### Non-stream Chat\n\nTo simplify the call, the SDK provides a wrapped function to complete non-streaming chat,\npolling, and obtaining the messages of the chat. Developers can use create_and_poll to\nsimplify the process.\n\n```python\nimport os\n\nfrom cozepy import Coze, TokenAuth, Message, ChatStatus\n\ncoze = Coze(auth=TokenAuth(os.getenv(\"COZE_API_TOKEN\")))\n\nchat_poll = coze.chat.create_and_poll(\n # id of bot\n bot_id='bot_id',\n # id of user, Note: The user_id here is specified by the developer, for example, it can be the\n # business id in the developer system, and does not include the internal attributes of coze.\n user_id='user_id',\n # user input\n additional_messages=[Message.build_user_question_text(\"How are you?\")]\n)\nfor message in chat_poll.messages:\n print(message.content, end=\"\")\n\nif chat_poll.chat.status == ChatStatus.COMPLETED:\n print()\n print(\"token usage:\", chat_poll.chat.usage.token_count)\n```\n\n#### Stream Chat\n\nCall the coze.chat.stream method to create a chat. The create method is a streaming\nchat and will return a Chat Iterator. Developers should iterate the iterator to get\nchat event and handle them.\n\n```python\nimport os\n\nfrom cozepy import Coze, TokenAuth, Message, ChatEventType\n\ncoze = Coze(auth=TokenAuth(os.getenv(\"COZE_API_TOKEN\")))\n\n# The return values of the streaming interface can be iterated immediately.\nfor event in coze.chat.stream(\n # id of bot\n bot_id='bot_id',\n # id of user, Note: The user_id here is specified by the developer, for example, it can be the\n # business id in the developer system, and does not include the internal attributes of coze.\n user_id='user_id',\n # user input\n additional_messages=[Message.build_user_question_text(\"How are you?\")]\n):\n if event.event == ChatEventType.CONVERSATION_MESSAGE_DELTA:\n print(event.message.content, end=\"\")\n\n if event.event == ChatEventType.CONVERSATION_CHAT_COMPLETED:\n print()\n print(\"token usage:\", event.chat.usage.token_count)\n```\n\n### Bots\n\nYou can create, update, publish and get the list of bots via api.\n\n```python\nimport os\nfrom cozepy import Coze, TokenAuth\n\ncoze = Coze(auth=TokenAuth(os.getenv(\"COZE_API_TOKEN\")))\n\n# retrieve bot info\nbot = coze.bots.retrieve(bot_id='bot id')\n\n# list bot list\n# open your workspace, browser url will be https://www.coze.com/space/<this is workspace id>/develop\n# copy <this is workspace id> as workspace id\nbots_page = coze.bots.list(space_id='workspace id', page_num=1)\nbots = bots_page.items\n\n# create bot\nbot = coze.bots.create(\n # id of workspace, open your workspace, browser url will be https://www.coze.com/space/<this is workspace id>/develop\n # copy <this is workspace id> as workspace id\n space_id='workspace id',\n # name of bot\n name='bot name',\n # description of bot\n description='bot description',\n)\n\n# update bot info\ncoze.bots.update(\n # id of workspace\n bot_id='bot id',\n # name of bot\n name='bot name',\n # description of bot\n description='bot description',\n)\n\n# delete bot\nbot = coze.bots.publish(bot_id='bot id')\n```\n\n### Conversations\n\nUsers can create conversations, and conduct conversations, inquire about messages,\netc. on conversations.\n\n```python\nimport os\nfrom cozepy import Coze, TokenAuth, Message, MessageContentType\n\ncoze = Coze(auth=TokenAuth(os.getenv(\"COZE_API_TOKEN\")))\n\n# If there is a need to build a context with user-assistant question-and-answer pairs,\n# a conversation can be created through the create interface.\nconversation = coze.conversations.create(\n messages=[\n # user question: how are you?\n Message.build_user_question_text('who are you?'),\n # assistant answer: I am Coze Bot.\n Message.build_assistant_answer('I am Coze Bot.')\n ],\n)\n\n# retrieve conversation\nconversation = coze.conversations.retrieve(conversation_id=conversation.id)\n\n# append message to conversation\nmessage = coze.conversations.messages.create(\n # id of conversation\n conversation_id=conversation.id,\n content='how are you?',\n content_type=MessageContentType.TEXT,\n)\n\n# retrieve message\nmessage = coze.conversations.messages.retrieve(conversation_id=conversation.id, message_id=message.id)\n\n# update message\ncoze.conversations.messages.update(\n conversation_id=conversation.id,\n message_id=message.id,\n content='hey, how are you?',\n content_type=MessageContentType.TEXT,\n)\n\n# delete message\ncoze.conversations.messages.delete(conversation_id=conversation.id, message_id=message.id)\n\n# list messages\nmessage_list = coze.conversations.messages.list(conversation_id=conversation.id)\n```\n\n### Files\n\nCoze enables users to upload pictures and files. The uploaded pictures and files\ncan be utilized in the bot avatar and multimodal conversations.\n\n```python\nimport os\nfrom pathlib import Path\nfrom cozepy import Coze, TokenAuth\n\ncoze = Coze(auth=TokenAuth(os.getenv(\"COZE_API_TOKEN\")))\n\n# upload file\nfile = coze.files.upload(file=Path('/filepath'))\n\n# retrieve file info\ncoze.files.retrieve(file_id=file.id)\n```\n\n### Workflows\n\nCoze also enables users to directly invoke the workflow.\n\n#### Non-stream workflow run\n\n```python\nimport os\nfrom cozepy import Coze, TokenAuth, Stream, WorkflowEvent, WorkflowEventType\n\ncoze = Coze(auth=TokenAuth(os.getenv(\"COZE_API_TOKEN\")))\n\nresult = coze.workflows.runs.create(\n # id of workflow\n workflow_id='workflow id',\n # params\n parameters={\n 'input_key': 'input value',\n }\n)\n```\n\n#### Stream workflow run\n\nThe returned result of the streaming interface is an iterator and can be directly iterated.\n\nWhen the workflow incorporates question-and-answer nodes, the streaming interface will\nreturn the INTERRUPT event.\n\nUsers should call the resume interface to submit the results of the question-and-answer.\n\nThe return value of resume remains an iterator, so recursive processing might be necessary here.\n\n```python\nimport os\n\nfrom cozepy import Coze, TokenAuth, Stream, WorkflowEvent, WorkflowEventType\n\ncoze = Coze(auth=TokenAuth(os.getenv(\"COZE_API_TOKEN\")))\n\n\ndef handle_workflow_iterator(stream: Stream[WorkflowEvent]):\n for event in stream:\n if event.event == WorkflowEventType.MESSAGE:\n print('got message', event.message)\n elif event.event == WorkflowEventType.ERROR:\n print('got error', event.error)\n elif event.event == WorkflowEventType.INTERRUPT:\n # Users should call the resume interface to submit the results of the question-and-answer\n handle_workflow_iterator(coze.workflows.runs.resume(\n workflow_id='workflow id',\n event_id=event.interrupt.interrupt_data.event_id,\n resume_data='hey',\n interrupt_type=event.interrupt.interrupt_data.type,\n ))\n\n\nhandle_workflow_iterator(coze.workflows.runs.stream(\n # id of workflow\n workflow_id='workflow id',\n # params\n parameters={\n 'input_key': 'input value',\n }\n))\n```\n\n### Knowledge\n\n```python\nimport os\nfrom cozepy import Coze, TokenAuth, DocumentBase, DocumentSourceInfo, DocumentChunkStrategy, DocumentUpdateRule\n\ncoze = Coze(auth=TokenAuth(os.getenv(\"COZE_API_TOKEN\")))\n\n# create knowledge documents by local_file\ndocuments = coze.knowledge.documents.create(\n # id of dataset\n dataset_id='dataset id',\n # document data\n document_bases=[\n DocumentBase(\n name='document name',\n source_info=DocumentSourceInfo.build_local_file('local file content')\n )\n ],\n # chunk strategy, needed when first create\n chunk_strategy=DocumentChunkStrategy.build_auto()\n)\n\n# create knowledge documents by web_page\ndocuments = coze.knowledge.documents.create(\n # id of dataset\n dataset_id='dataset id',\n # document data\n document_bases=[\n DocumentBase(\n name='document name',\n source_info=DocumentSourceInfo.build_web_page('https://example.com')\n )\n ],\n # chunk strategy, needed when first create\n chunk_strategy=DocumentChunkStrategy.build_auto()\n)\n\n# update knowledge document\ncoze.knowledge.documents.update(\n # id of document\n document_id='document id',\n # name of document\n document_name='name',\n # update rule, current set to no auto-update\n update_rule=DocumentUpdateRule.build_no_auto_update()\n)\n\n# delete knowledge document\ncoze.knowledge.documents.delete(document_ids=['document id'])\n\n# list knowledge documents\npaged_documents = coze.knowledge.documents.list(\n # id of dataset\n dataset_id='dataset id',\n page_num=1,\n page_size=10\n)\n```\n\n### OAuth App\n\n#### Web OAuth App\n\nFirstly, users need to access https://www.coze.com/open/oauth/apps. For the cn environment,\nusers need to access https://www.coze.cn/open/oauth/apps to create an OAuth App of the type\nof Web application.\n\nThe specific creation process can be referred to in the document:\nhttps://www.coze.com/docs/developer_guides/oauth_code. For the cn environment, it can be\naccessed at https://www.coze.cn/docs/developer_guides/oauth_code.\n\nAfter the creation is completed, three parameters, namely the client ID, client secret,\nand redirect link, can be obtained. For the client secret, users need to keep it securely\nto avoid leakage.\n\n```python\nimport os\nfrom cozepy import Coze, TokenAuth, WebOAuthApp\n\n# client ID\nweb_oauth_client_id = os.getenv(\"COZE_WEB_OAUTH_CLIENT_ID\")\n# client secret\nweb_oauth_client_secret = os.getenv(\"COZE_WEB_OAUTH_CLIENT_SECRET\")\n\nweb_oauth_app = WebOAuthApp(\n client_id=web_oauth_client_id,\n client_secret=web_oauth_client_secret,\n)\n```\n\nThe WebOAuth authorization process is to first generate a coze authorization link and\nsend it to the coze user requiring authorization.\n\nOnce the coze user opens the link, they can see the authorization consent button.\n\n```python\n# redirect link\nweb_oauth_redirect_uri = os.getenv(\"COZE_WEB_OAUTH_REDIRECT_URI\")\n\n# Generate the authorization link and direct the user to open it.\nurl = web_oauth_app.get_oauth_url(redirect_uri=web_oauth_redirect_uri)\n```\n\nAfter the user clicks the authorization consent button, the coze web page will redirect\nto the redirect address configured in the authorization link and carry the authorization\ncode and state parameters in the address via the query string.\n\n```python\n# Open the authorization link in your browser and authorize this OAuth App\n# After authorization, you will be redirected to the redirect_uri with a code and state\n# You can use the code to get the access token\ncode = 'mock code'\n\n# After obtaining the code after redirection, the interface to exchange the code for a\n# token can be invoked to generate the coze access_token of the authorized user.\noauth_token = web_oauth_app.get_access_token(redirect_uri=web_oauth_redirect_uri, code=code)\n\n# use the access token to init Coze client\ncoze = Coze(auth=TokenAuth(oauth_token.access_token))\n\n# When the token expires, you can also refresh and re-obtain the token\noauth_token = web_oauth_app.refresh_access_token(oauth_token.refresh_token)\n```\n\n#### JWT OAuth App\n\nFirstly, users need to access https://www.coze.com/open/oauth/apps. For the cn environment,\nusers need to access https://www.coze.cn/open/oauth/apps to create an OAuth App of the type\nof Service application.\n\nThe specific creation process can be referred to in the document:\nhttps://www.coze.com/docs/developer_guides/oauth_jwt. For the cn environment, it can be\naccessed at https://www.coze.cn/docs/developer_guides/oauth_jwt.\n\nAfter the creation is completed, three parameters, namely the client ID, private key,\nand public key id, can be obtained. For the client secret and public key id, users need to\nkeep it securely to avoid leakage.\n\n```python\nimport os\nfrom cozepy import Coze, TokenAuth, JWTOAuthApp\n\n# client ID\njwt_oauth_client_id = os.getenv(\"COZE_JWT_OAUTH_CLIENT_ID\")\n# private key\njwt_oauth_private_key = os.getenv(\"COZE_JWT_OAUTH_PRIVATE_KEY\")\n# public key id\njwt_oauth_public_key_id = os.getenv(\"COZE_JWT_OAUTH_PUBLIC_KEY_ID\")\n\njwt_oauth_app = JWTOAuthApp(\n client_id=jwt_oauth_client_id,\n private_key=jwt_oauth_private_key,\n public_key_id=jwt_oauth_public_key_id,\n)\n```\n\nThe jwt oauth type requires using private to be able to issue a jwt token, and through\nthe jwt token, apply for an access_token from the coze service.\n\nThe sdk encapsulates this procedure, and only needs to use get_access_token to obtain\nthe access_token under the jwt oauth process.\n\n```python\n# The jwt process does not require any other operations, you can directly apply for a token\noauth_token = jwt_oauth_app.get_access_token(ttl=3600)\n\n# use the access token to init Coze client\ncoze = Coze(auth=TokenAuth(oauth_token.access_token))\n\n# The jwt oauth process does not support refreshing tokens. When the token expires,\n# just directly call get_access_token to generate a new token.\n```\n\n#### PKCE OAuth App\n\nPKCE stands for Proof Key for Code Exchange, and it's an extension to the OAuth 2.0 authorization\ncode flow designed to enhance security for public clients, such as mobile and single-page\napplications.\n\nFirstly, users need to access https://www.coze.com/open/oauth/apps. For the cn environment,\nusers need to access https://www.coze.cn/open/oauth/apps to create an OAuth App of the type\nof Mobile/PC/Single-page application.\n\nThe specific creation process can be referred to in the document:\nhttps://www.coze.com/docs/developer_guides/oauth_pkce. For the cn environment, it can be\naccessed at https://www.coze.cn/docs/developer_guides/oauth_pkce.\n\nAfter the creation is completed, three parameters, namely the client ID can be obtained.\n\n```python\nimport os\n\nfrom cozepy import PKCEOAuthApp\n\n# client ID\npkce_oauth_client_id = os.getenv(\"COZE_PKCE_OAUTH_CLIENT_ID\")\n# redirect link\nweb_oauth_redirect_uri = os.getenv(\"COZE_WEB_OAUTH_REDIRECT_URI\")\n\npkce_oauth_app = PKCEOAuthApp(client_id=pkce_oauth_client_id)\n```\n\nIn the pkce oauth process, first, need to select a suitable code_challenge_method.\nCoze supports two types: plain and s256.\n\nThen, based on the selected code_challenge_method type, hash the code_verifier into\nthe code_challenge. Finally, based on the callback address,\ncode_challenge, and code_challenge_method, an authorization link can be generated.\n\n```python\n# In the SDK, we have wrapped up the code_challenge process of PKCE. Developers only need\n# to select the code_challenge_method.\ncode_verifier = \"random code verifier\"\nurl = pkce_oauth_app.get_oauth_url(\n redirect_uri=web_oauth_redirect_uri,\n code_verifier=code_verifier,\n code_challenge_method=\"S256\"\n)\n```\n\nDevelopers should lead users to open up this authorization link.\n\nWhen the user consents to the authorization, Coze will redirect with the code to the\ncallback address configured by the developer, and the developer can obtain this code.\n\n```python\n# Open the authorization link in your browser and authorize this OAuth App\n# After authorization, you can exchange code_verifier for access token\ncode = 'mock code'\n\n# After obtaining the code after redirection, the interface to exchange the code for a\n# token can be invoked to generate the coze access_token of the authorized user.\noauth_token = pkce_oauth_app.get_access_token(\n redirect_uri=web_oauth_redirect_uri, code=code, code_verifier=code_verifier\n)\n\n# use the access token to init Coze client\ncoze = Coze(auth=TokenAuth(oauth_token.access_token))\n\n# When the token expires, you can also refresh and re-obtain the token\noauth_token = pkce_oauth_app.refresh_access_token(oauth_token.refresh_token)\n```\n\n#### Device OAuth App\n\nThe OAuth 2.0 device flow, also known as the device authorization grant, is an extension to t\ne OAuth 2.0 protocol designed for devices that have limited input capabilities or lack a suitable\nbrowser.\n\nFirstly, users need to access https://www.coze.com/open/oauth/apps. For the cn environment,\nusers need to access https://www.coze.cn/open/oauth/apps to create an OAuth App of the type\nof TVs/Limited Input devices/Command line programs.\n\nThe specific creation process can be referred to in the document:\nhttps://www.coze.com/docs/developer_guides/oauth_device_code. For the cn environment, it can be\naccessed at https://www.coze.cn/docs/developer_guides/oauth_device_code.\n\nAfter the creation is completed, three parameters, namely the client ID can be obtained.\n\n```python\nimport os\nfrom cozepy import Coze, TokenAuth, DeviceOAuthApp\n\n# client ID\ndevice_oauth_client_id = os.getenv(\"COZE_DEVICE_OAUTH_CLIENT_ID\")\n\ndevice_oauth_app = DeviceOAuthApp(client_id=device_oauth_client_id)\n```\n\nIn the device oauth authorization process, developers need to first call the interface\nof Coze to generate the device code to obtain the user_code and device_code.\n\nThen generate the authorization link through the user_code, guide the user to open the\nlink, fill in the user_code, and consent to the authorization.\n\nDevelopers need to call the interface of Coze to generate the token through the device_code.\n\nWhen the user has not authorized or rejected the authorization, the interface will throw an\nerror and return a specific error code.\n\nAfter the user consents to the authorization, the interface will succeed and return the\naccess_token.\n\n```python\n# First, you need to request the server to obtain the device code required in the device auth flow\ndevice_code = device_oauth_app.get_device_code()\n\n# The returned device_code contains an authorization link. Developers need to guide users\n# to open up this link.\n# open device_code.verification_url\n```\n\nThe developers then need to use the device_code to poll Coze's interface to obtain the token.\n\nThe SDK has encapsulated this part of the code in and handled the different returned error\ncodes. The developers only need to invoke get_access_token.\n\n```python\ntry:\n oauth_token = device_oauth_app.get_access_token(\n device_code=device_code.device_code,\n poll=True,\n )\nexcept CozePKCEAuthError as e:\n if e.error == CozePKCEAuthErrorType.ACCESS_DENIED:\n # The user rejected the authorization.\n # Developers need to guide the user to open the authorization link again.\n pass\n elif e.error == CozePKCEAuthErrorType.EXPIRED_TOKEN:\n # The token has expired. Developers need to guide the user to open\n # the authorization link again.\n pass\n else:\n # Other errors\n pass\n\n raise # for example, re-raise the error\n\n# use the access token to init Coze client\ncoze = Coze(auth=TokenAuth(oauth_token.access_token))\n\n# When the token expires, you can also refresh and re-obtain the token\noauth_token = device_oauth_app.refresh_access_token(oauth_token.refresh_token)\n```\n\n### Debug\n\nThe SDK support returning the logid of the request, which can be used to debug the request.\nYou can get the logid from the response of the request and submit it to the coze support team for further assistance.\n\n```python\nimport os\nfrom cozepy import Coze, AsyncCoze, TokenAuth\n\ncoze = Coze(auth=TokenAuth(os.getenv(\"COZE_API_TOKEN\")))\n\nbot = coze.bots.retrieve(bot_id='bot id')\nprint(bot.response.logid) # support for CozeModel\n\nstream = coze.chat.stream(bot_id='bot id', user_id='user id')\nprint(stream.response.logid) # support for stream\n\nworkspaces = coze.workspaces.list()\nprint(workspaces.response.logid) # support for paged\n\nmessages = coze.chat.messages.list(conversation_id='conversation id', chat_id='chat id')\nprint(messages.response.logid) # support for list(simple list, not paged)\n```\n\n\n### Async usage\n\ncozepy supports asynchronous calls through `httpx.AsyncClient`.\n\nJust replace the `Coze` client with the `AsyncCoze` client to use all the asynchronous calls of the Coze OpenAPI.\n\n```python\nimport os\nimport asyncio\n\nfrom cozepy import TokenAuth, Message, AsyncCoze\n\ncoze = AsyncCoze(auth=TokenAuth(os.getenv(\"COZE_API_TOKEN\")))\n\n\nasync def main() -> None:\n chat = await coze.chat.create(\n bot_id='bot id',\n # 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.\n user_id='user id',\n additional_messages=[\n Message.build_user_question_text('how are you?'),\n Message.build_assistant_answer('I am fine, thank you.')\n ],\n )\n print('chat', chat)\n\n\nasyncio.run(main())\n```\n\n### Streaming usage\n\nBot chat and workflow run support running in streaming mode.\n\nchat streaming example:\n\n```python\nimport os\nfrom cozepy import Coze, TokenAuth, ChatEventType, Message\n\ncoze = Coze(auth=TokenAuth(os.getenv(\"COZE_API_TOKEN\")))\n\nstream = coze.chat.stream(\n bot_id='bot id',\n # id of user, Note: The user_id here is specified by the developer, for example, it can be the\n # business id in the developer system, and does not include the internal attributes of coze.\n user_id='user id',\n additional_messages=[\n Message.build_user_question_text('how are you?'),\n Message.build_assistant_answer('I am fine, thank you.')\n ],\n)\nfor event in stream:\n if event.event == ChatEventType.CONVERSATION_MESSAGE_DELTA:\n print('got message delta:', event.message.content)\n```\n\nworkflow streaming example:\n\n```python\nimport os\nfrom cozepy import Coze, TokenAuth, Stream, WorkflowEvent, WorkflowEventType\n\ncoze = Coze(auth=TokenAuth(os.getenv(\"COZE_API_TOKEN\")))\n\n\ndef handle_workflow_iterator(stream: Stream[WorkflowEvent]):\n for event in stream:\n if event.event == WorkflowEventType.MESSAGE:\n print('got message', event.message)\n elif event.event == WorkflowEventType.ERROR:\n print('got error', event.error)\n elif event.event == WorkflowEventType.INTERRUPT:\n handle_workflow_iterator(coze.workflows.runs.resume(\n workflow_id='workflow id',\n event_id=event.interrupt.interrupt_data.event_id,\n resume_data='hey',\n interrupt_type=event.interrupt.interrupt_data.type,\n ))\n\n\nhandle_workflow_iterator(coze.workflows.runs.stream(\n workflow_id='workflow id',\n parameters={\n 'input_key': 'input value',\n }\n))\n```\n\nAsynchronous calls also support streaming mode:\n\n```python\nimport os\nimport asyncio\n\nfrom cozepy import TokenAuth, ChatEventType, Message, AsyncCoze\n\ncoze = AsyncCoze(auth=TokenAuth(os.getenv(\"COZE_API_TOKEN\")))\n\n\nasync def main():\n stream = coze.chat.stream(\n bot_id='bot id',\n # id of user, Note: The user_id here is specified by the developer, for example, it can be the\n # business id in the developer system, and does not include the internal attributes of coze.\n user_id='user id',\n additional_messages=[\n Message.build_user_question_text('how are you?'),\n Message.build_assistant_answer('I am fine, thank you.')\n ],\n )\n async for event in stream:\n if event.event == ChatEventType.CONVERSATION_MESSAGE_DELTA:\n print('got message delta:', event.message.content)\n\n\nasyncio.run(main())\n```\n\n### Paginator Iterator\n\nThe result returned by all list interfaces (both synchronous and asynchronous) is a paginator, which supports iteration.\n\nTake the example of listing the bots in a space to explain the three ways to use the paginator iterator:\n\n#### 1. Not using iterators\n\n```python\nimport os\nfrom cozepy import Coze, TokenAuth\n\ncoze = Coze(auth=TokenAuth(os.getenv(\"COZE_API_TOKEN\")))\n\n# open your workspace, browser url will be https://www.coze.com/space/<this is workspace id>/develop\n# copy <this is workspace id> as workspace id\nbots_page = coze.bots.list(space_id='workspace id', page_size=10)\nbots = bots_page.items\ntotal = bots_page.total\nhas_more = bots_page.has_more\n```\n\n#### 2. Iterate over the paginator, getting T\n\n```python\nimport os\nfrom cozepy import Coze, TokenAuth\n\ncoze = Coze(auth=TokenAuth(os.getenv(\"COZE_API_TOKEN\")))\n\n# open your workspace, browser url will be https://www.coze.com/space/<this is workspace id>/develop\n# copy <this is workspace id> as workspace id\nbots_page = coze.bots.list(space_id='workspace id', page_size=10)\nfor bot in bots_page:\n print('got bot:', bot)\n```\n\nAsynchronous methods also support:\n\n```python\nimport os\nimport asyncio\n\nfrom cozepy import TokenAuth, AsyncCoze\n\ncoze = AsyncCoze(auth=TokenAuth(os.getenv(\"COZE_API_TOKEN\")))\n\n\nasync def main():\n # open your workspace, browser url will be https://www.coze.com/space/<this is workspace id>/develop\n # copy <this is workspace id> as workspace id\n bots_page = await coze.bots.list(space_id='workspace id', page_size=10)\n async for bot in bots_page:\n print('got bot:', bot)\n\n\nasyncio.run(main())\n```\n\n#### 3. Iterate over the paginator iter_pages to get the next page paginator\n\n```python\nimport os\nfrom cozepy import Coze, TokenAuth\n\ncoze = Coze(auth=TokenAuth(os.getenv(\"COZE_API_TOKEN\")))\n\n# open your workspace, browser url will be https://www.coze.com/space/<this is workspace id>/develop\n# copy <this is workspace id> as workspace id\nbots_page = coze.bots.list(space_id='workspace id', page_size=10)\nfor page in bots_page.iter_pages():\n print('got page:', page.page_num)\n for bot in page.items:\n print('got bot:', bot)\n```\n\nAsynchronous methods also support:\n\n```python\nimport asyncio\nimport os\n\nfrom cozepy import TokenAuth, AsyncCoze\n\ncoze = AsyncCoze(auth=TokenAuth(os.getenv(\"COZE_API_TOKEN\")))\n\n\nasync def main():\n # open your workspace, browser url will be https://www.coze.com/space/<this is workspace id>/develop\n # copy <this is workspace id> as workspace id\n bots_page = await coze.bots.list(space_id='workspace id', page_size=10)\n async for page in bots_page.iter_pages():\n print('got page:', page.page_num)\n for bot in page.items:\n print('got bot:', bot)\n\n\nasyncio.run(main())\n```\n\n### Config\n\n#### Log Config\n\ncoze support config logging level\n\n```python\nimport logging\n\nfrom cozepy import setup_logging\n\n# open debug logging, default is warning\nsetup_logging(level=logging.DEBUG)\n```\n\n#### Timeout Config\n\nCoze client is built on httpx, and supports passing a custom httpx.Client when initializing\nCoze, and setting a timeout on the httpx.Client\n\n```python\nimport os\n\nimport httpx\n\nfrom cozepy import COZE_COM_BASE_URL, Coze, TokenAuth, SyncHTTPClient\n\n# Coze client is built on httpx, and supports passing a custom httpx.Client when initializing\n# Coze, and setting a timeout on the httpx.Client\nhttp_client = SyncHTTPClient(timeout=httpx.Timeout(\n # 600s timeout on elsewhere\n timeout=600.0,\n # 5s timeout on connect\n connect=5.0\n))\n\n# Init the Coze client through the access_token and custom timeout http client.\ncoze = Coze(auth=TokenAuth(token=os.getenv(\"COZE_API_TOKEN\")),\n base_url=COZE_COM_BASE_URL,\n http_client=http_client\n )\n```\n",
"bugtrack_url": null,
"license": "MIT",
"summary": "OpenAPI SDK for Coze(coze.com/coze.cn)",
"version": "0.11.0",
"project_urls": null,
"split_keywords": [],
"urls": [
{
"comment_text": "",
"digests": {
"blake2b_256": "77f6ff0cc36d6ddf6f35dbfab205bd9d42913ec7de364ffa29cdf0932bbe98a1",
"md5": "7e77c94d5079f9437e9548743120f396",
"sha256": "a1c80645c1b1985b576877a675650fd288196a9425fee3f5f2ffb9f5e2ea1a95"
},
"downloads": -1,
"filename": "cozepy-0.11.0-py3-none-any.whl",
"has_sig": false,
"md5_digest": "7e77c94d5079f9437e9548743120f396",
"packagetype": "bdist_wheel",
"python_version": "py3",
"requires_python": "<4.0,>=3.7",
"size": 80067,
"upload_time": "2025-01-23T23:37:20",
"upload_time_iso_8601": "2025-01-23T23:37:20.073390Z",
"url": "https://files.pythonhosted.org/packages/77/f6/ff0cc36d6ddf6f35dbfab205bd9d42913ec7de364ffa29cdf0932bbe98a1/cozepy-0.11.0-py3-none-any.whl",
"yanked": false,
"yanked_reason": null
},
{
"comment_text": "",
"digests": {
"blake2b_256": "801f88ac3e421f5ff31376866320af8497038d1a995a9ddd0275b108fda44982",
"md5": "34d0097bcccd4044f7bf3936ddb5b094",
"sha256": "334be8f1ca34f38209045cbf19a39defc4d67a5ef50f47275f9a742aef6aac18"
},
"downloads": -1,
"filename": "cozepy-0.11.0.tar.gz",
"has_sig": false,
"md5_digest": "34d0097bcccd4044f7bf3936ddb5b094",
"packagetype": "sdist",
"python_version": "source",
"requires_python": "<4.0,>=3.7",
"size": 64516,
"upload_time": "2025-01-23T23:37:22",
"upload_time_iso_8601": "2025-01-23T23:37:22.515880Z",
"url": "https://files.pythonhosted.org/packages/80/1f/88ac3e421f5ff31376866320af8497038d1a995a9ddd0275b108fda44982/cozepy-0.11.0.tar.gz",
"yanked": false,
"yanked_reason": null
}
],
"upload_time": "2025-01-23 23:37:22",
"github": false,
"gitlab": false,
"bitbucket": false,
"codeberg": false,
"lcname": "cozepy"
}
