Azure Communication Chat Package客户端库，用于Python

此包包含用于Azure Communication Services for Chat的Python SDK。了解更多关于Azure Communication Services的信息，请参阅此处

源代码 | 包（PyPI） | 包（Conda） | API参考文档 | 产品文档

免责声明

Azure SDK Python包对Python 2.7的支持已于2022年1月1日结束。有关更多信息及疑问，请参阅https://github.com/Azure/azure-sdk-for-python/issues/20691

入门

先决条件

• 要使用此包，需要Python 3.7或更高版本。
• 已部署的通信服务资源。您可以使用Azure 门户或Azure PowerShell来设置。

安装包

安装 Azure Communication Service Chat SDK。

pip install --pre azure-communication-chat

用户访问令牌

用户访问令牌允许您构建直接对 Azure 通信服务进行身份验证的客户端应用程序。您可以使用 azure.communication.identity 模块生成这些令牌，然后使用它们来初始化通信服务 SDK。使用 azure.communication.identity 的示例

pip install azure-communication-identity

from azure.communication.identity import CommunicationIdentityClient
identity_client = CommunicationIdentityClient.from_connection_string("<connection string of your Communication service>")
user = identity_client.create_user()
tokenresponse = identity_client.get_token(user, scopes=["chat"])
token = tokenresponse.token

上面创建的 user 将在以后使用，因为当您使用此令牌创建它时，该用户应被添加为新聊天线程的参与者。这是因为创建请求的发起者必须在聊天线程的参与者列表中。

创建聊天客户端

这将允许您创建、获取、列出或删除聊天线程。

from azure.communication.chat import ChatClient, CommunicationTokenCredential

# Your unique Azure Communication service endpoint
endpoint = "https://<RESOURCE_NAME>.communcationservices.azure.com"
chat_client = ChatClient(endpoint, CommunicationTokenCredential(token))

创建聊天线程客户端

ChatThreadClient 允许您执行特定于聊天线程的操作，例如发送消息、获取消息、更新聊天线程主题、向聊天线程添加参与者等。

您可以通过使用 ChatClient 创建新的聊天线程来获取它

create_chat_thread_result = chat_client.create_chat_thread(topic)
chat_thread_client = chat_client.get_chat_thread_client(create_chat_thread_result.chat_thread.id)

此外，客户端还可以直接指定请求的可重复性；也就是说，如果客户端多次使用相同的 Idempotency-Token 发出请求，它将获得适当的响应，而服务器不会多次执行请求。Idempotency-Token 的值是一个不透明的字符串，表示客户端生成的、在所有时间都全球唯一的请求标识符。

create_chat_thread_result = chat_client.create_chat_thread(
    topic,
    thread_participants=thread_participants,
    idempotency_token=idempotency_token
)
chat_thread_client = chat_client.get_chat_thread_client(create_chat_thread_result.chat_thread.id)

或者，如果您之前已创建聊天线程并且有其 thread_id，您可以通过以下方式创建它

chat_thread_client = chat_client.get_chat_thread_client(thread_id) # thread_id is the id of an existing chat thread

关键概念

聊天对话由聊天线程表示。线程中的每个用户都称为线程参与者。线程参与者可以彼此进行私密聊天，在 1:1 聊天或 1:N 群聊中聚在一起。用户还可以获得其他人正在输入和阅读消息的近乎实时的更新。

初始化 ChatClient 类后，您可以执行以下聊天操作

创建、获取、更新和删除线程

在线程上执行 CRD（创建-读取-删除）操作

create_chat_thread(topic, **kwargs)
list_chat_threads(**kwargs)
delete_chat_thread(thread_id, **kwargs)

初始化 ChatThreadClient 类后，您可以执行以下聊天操作

更新线程

在线程主题上执行更新操作

update_topic(topic, **kwargs)

获取聊天线程属性

get_properties(**kwargs)

发送、获取、更新和删除消息

在消息上执行 CRUD（创建-读取-更新-删除）操作

send_message(content, **kwargs)
get_message(message_id, **kwargs)
list_messages(**kwargs)
update_message(message_id, content, **kwargs)
delete_message(message_id, **kwargs)

获取、添加和删除参与者

在线程参与者上执行 CRD（创建-读取-删除）操作

list_participants(**kwargs)
add_participants(thread_participants, **kwargs)
remove_participant(participant_identifier, **kwargs)

发送打字通知

通知服务打字通知

send_typing_notification(**kwargs)

发送和获取已读回执

通知服务消息已读并获取已读消息列表。

send_read_receipt(message_id, **kwargs)
list_read_receipts(**kwargs)

示例

以下部分提供了一些代码片段，涵盖了最常见的一些任务，包括

• 线程操作
• 消息操作
• 线程参与者操作
• 事件操作

线程操作

创建线程

使用 create_chat_thread 方法创建聊天线程。

• 使用必需的 topic 给定线程主题；
• 使用可选的 thread_participants 提供要添加到线程中的 ChatParticipant 列表；
  • user 是必需的，它是通过 CommunicationIdentityClient.create_user() 从用户访问令牌创建的 CommunicationUserIdentifier；
  • display_name 是可选的，它是线程参与者的显示名称。
  • share_history_time 是可选的，是从与参与者共享聊天历史的时间。
• 使用 idempotency_token（可选），指定请求的唯一标识符。

CreateChatThreadResult 是创建线程时返回的结果，您可以使用它来获取创建的聊天线程的 id。然后，可以使用 get_chat_thread_client 方法使用此 id 获取 ChatThreadClient 对象。ChatThreadClient 可以用来执行其他针对此聊天线程的聊天操作。

# Without idempotency_token and thread_participants
topic = "test topic"
create_chat_thread_result = chat_client.create_chat_thread(topic)
chat_thread_client = chat_client.get_chat_thread_client(create_chat_thread_result.chat_thread.id)

# With idempotency_token and thread_participants
from azure.communication.identity import CommunicationIdentityClient
from azure.communication.chat import ChatParticipant, ChatClient, CommunicationTokenCredential
import uuid
from datetime import datetime

# create an user
identity_client = CommunicationIdentityClient.from_connection_string('<connection_string>')
user = identity_client.create_user()

# user access tokens
tokenresponse = identity_client.get_token(user, scopes=["chat"])
token = tokenresponse.token

## OR pass existing user
# from azure.communication.chat import CommunicationUserIdentifier
# user_id = 'some_user_id'
# user = CommunicationUserIdentifier(user_id)

# create the chat_client
endpoint = "https://<RESOURCE_NAME>.communcationservices.azure.com"
chat_client = ChatClient(endpoint, CommunicationTokenCredential(token))

# modify function to implement customer logic
def get_unique_identifier_for_request(**kwargs):
    res = uuid.uuid4()
    return res

topic = "test topic"
thread_participants = [ChatParticipant(
    identifier=user,
    display_name='name',
    share_history_time=datetime.utcnow()
)]

# obtains idempotency_token using some customer logic
idempotency_token = get_unique_identifier_for_request()

create_chat_thread_result = chat_client.create_chat_thread(
    topic,
    thread_participants=thread_participants,
    idempotency_token=idempotency_token)
thread_id = create_chat_thread_result.chat_thread.id

# fetch ChatThreadClient
chat_thread_client = chat_client.get_chat_thread_client(create_chat_thread_result.chat_thread.id)

# Additionally, you can also check if all participants were successfully added or not
# and subsequently retry adding the failed participants again
def decide_to_retry(error, **kwargs):
    """
    Insert some custom logic to decide if retry is applicable based on error
    """
    return True

retry = [thread_participant for thread_participant, error in create_chat_thread_result.errors if decide_to_retry(error)]
if retry:
    chat_thread_client.add_participants(retry)

获取线程

使用 get_properties 方法从服务中检索 ChatThreadProperties；thread_id 是线程的唯一 ID。

chat_thread_properties = chat_thread_client.get_properties()

列出聊天线程

使用 list_chat_threads 方法检索已创建的聊天线程列表

• 使用 results_per_page（可选），每页最多返回的消息数。
• 使用 start_time（可选），范围查询的起始时间。

列出线程时的响应是一个 [ChatThreadItem] 迭代器。

from azure.communication.chat import ChatClient, CommunicationTokenCredential
from datetime import datetime, timedelta

token = "<token>"
endpoint = "https://<RESOURCE_NAME>.communcationservices.azure.com"
chat_client = ChatClient(endpoint, CommunicationTokenCredential(token))
start_time = datetime.utcnow() - timedelta(days=2)

chat_threads = chat_client.list_chat_threads(results_per_page=5, start_time=start_time)
for chat_thread_item_page in chat_threads.by_page():
    for chat_thread_item in chat_thread_item_page:
        print("thread id:", chat_thread_item.id)

更新线程主题

使用 update_topic 方法更新线程的属性。topic 用于描述线程主题的变化

• 使用 topic 为线程分配一个新的主题；

topic = "new topic"
chat_thread_client.update_topic(topic=topic)

chat_thread = chat_thread_client.get_properties(thread_id)

assert chat_thread.topic == topic

删除线程

使用 delete_chat_thread 方法删除线程；thread_id 是线程的唯一 ID。

• 使用 thread_id（必需），指定线程的唯一 ID。

chat_client.delete_chat_thread(thread_id=thread_id)

消息操作

发送消息

使用 send_message 方法向通过 thread_id 标识的线程发送消息。

• 使用 content（必需），提供聊天消息内容。
• 使用 chat_message_type（可选），提供聊天消息类型。可能的值包括：ChatMessageType.TEXT、ChatMessageType.HTML、'text'、'html'；如果未指定，则默认设置为 ChatMessageType.TEXT。
• 使用 sender_display_name（可选），指定发送者的显示名称，如果未指定，则设置为空名称。

SendChatMessageResult 是发送消息时返回的响应，其中包含一个 id，即消息的唯一 ID。

from azure.communication.chat import ChatMessageType

topic = "test topic"
create_chat_thread_result = chat_client.create_chat_thread(topic)
thread_id = create_chat_thread_result.chat_thread.id
chat_thread_client = chat_client.get_chat_thread_client(create_chat_thread_result.chat_thread.id)

content='hello world'
sender_display_name='sender name'
chat_message_type = ChatMessageType.TEXT

# without specifying sender_display_name and chat_message_type
send_message_result = chat_thread_client.send_message(content)
send_message_result_id = send_message_result.id
print("Message sent: id: ", send_message_result_id)

# specifying sender_display_name and chat_message_type
send_message_result_w_type = chat_thread_client.send_message(
            content,
            sender_display_name=sender_display_name,
            chat_message_type=chat_message_type # equivalent to chat_message_type = 'text'
)
send_message_result_w_type_id = send_message_result_w_type.id
print("Message sent: id: ", send_message_result_w_type_id)

获取消息

使用 get_message 方法从服务中检索消息；message_id 是消息的唯一 ID。

• 使用 message_id（必需），指定现有消息的 id。《ChatMessage》是从获取消息时返回的响应，它包含一个 id，即消息的唯一 ID，以及其他字段请参阅 azure.communication.chat.ChatMessage。

chat_message = chat_thread_client.get_message(message_id=send_message_result_id)
print("get_chat_message succeeded, message id:", chat_message.id, "content: ", chat_message.content)

列出消息

使用 list_messages 方法从服务中检索消息。

• 使用 results_per_page（可选），每页最多返回的消息数。
• 使用 start_time（可选），范围查询的起始时间。

列出消息时返回的响应是一个 [ChatMessage] 迭代器。

from datetime import datetime, timedelta

start_time = datetime.utcnow() - timedelta(days=1)

chat_messages = chat_thread_client.list_messages(results_per_page=1, start_time=start_time)
for chat_message_page in chat_messages.by_page():
    for chat_message in chat_message_page:
        print("ChatMessage: Id=", chat_message.id, "; Content=", chat_message.content.message)

更新消息

使用 update_message 更新通过 threadId 和 messageId 标识的消息。

• 使用 message_id（必需），是消息的唯一 ID。
• 使用 content（可选），是要更新的消息内容；如果未指定，则将其设置为空。

content = "updated message content"
chat_thread_client.update_message(send_message_result_id, content=content)

chat_message = chat_thread_client.get_message(message_id=send_message_result_id)

assert chat_message.content.message == content

删除消息

使用 delete_message 删除消息。

• 使用 message_id（必需），是消息的唯一 ID。

chat_thread_client.delete_message(message_id=send_message_result_id)

线程参与者操作

列出线程参与者

使用 list_participants 检索线程的参与者。

• 使用 results_per_page（可选），每页最多返回的参与者数。
• 使用 skip（可选），在响应中跳过指定位置的参与者。

列出参与者时返回的响应是一个 [ChatParticipant] 迭代器。

chat_participants = chat_thread_client.list_participants(results_per_page=5, skip=5)
for chat_participant_page in chat_participants.by_page():
    for chat_participant in chat_participant_page:
        print("ChatParticipant: ", chat_participant)

添加线程参与者

使用 add_participants 方法向线程添加线程参与者。

• 使用 thread_participants（必需），列出要添加到线程的 ChatParticipant。
  • user 是必需的，它是通过 CommunicationIdentityClient.create_user() 从用户访问令牌创建的 CommunicationUserIdentifier；
  • display_name 是可选的，它是线程参与者的显示名称。
  • share_history_time 是可选的，是从与参与者共享聊天历史的时间。

返回一个 list(tuple(ChatParticipant, ChatError))。当参与者成功添加时，期望得到一个空列表。如果在添加参与者时遇到错误，则列表中将包含失败的参与者以及遇到的错误。

from azure.communication.identity import CommunicationIdentityClient
from azure.communication.chat import ChatParticipant
from datetime import datetime

# create 2 users
identity_client = CommunicationIdentityClient.from_connection_string('<connection_string>')
new_users = [identity_client.create_user() for i in range(2)]

# # conversely, you can also add an existing user to a chat thread; provided the user_id is known
# from azure.communication.chat import CommunicationUserIdentifier
#
# user_id = 'some user id'
# user_display_name = "Wilma Flinstone"
# new_user = CommunicationUserIdentifier(user_id)
# participant = ChatParticipant(
#     identifier=new_user,
#     display_name=user_display_name,
#     share_history_time=datetime.utcnow())

participants = []
for _user in new_users:
  chat_participant = ChatParticipant(
    identifier=_user,
    display_name='Fred Flinstone',
    share_history_time=datetime.utcnow()
  )
  participants.append(chat_participant)

response = chat_thread_client.add_participants(thread_participants=participants)

def decide_to_retry(error, **kwargs):
    """
    Insert some custom logic to decide if retry is applicable based on error
    """
    return True

# verify if all users has been successfully added or not
# in case of partial failures, you can retry to add all the failed participants
retry = [p for p, e in response if decide_to_retry(e)]
if retry:
    chat_thread_client.add_participants(retry)

删除线程参与者

使用 remove_participant 方法从通过 threadId 标识的线程中移除线程参与者。identifier 是您通过 CommunicationIdentityClient.create_user() 在 azure-communication-identity 中创建的 CommunicationUserIdentifier。

并已添加到此聊天线程。

• 使用 identifier 指定您创建的 CommunicationUserIdentifier

chat_thread_client.remove_participant(identifier=new_user)

# # conversely you can also do the following; provided the user_id is known
# from azure.communication.chat import CommunicationUserIdentifier
#
# user_id = 'some user id'
# chat_thread_client.remove_participant(identifier=CommunicationUserIdentifier(new_user))

事件操作

发送打字通知

使用 send_typing_notification 方法代表用户向线程发布打字通知事件。

chat_thread_client.send_typing_notification()

发送已读回执

使用 send_read_receipt 方法代表用户向线程发布已读回执事件。

• 使用 message_id 指定要发送已读回执的消息的 ID

content='hello world'
send_message_result = chat_thread_client.send_message(content)
send_message_result_id = send_message_result.id
chat_thread_client.send_read_receipt(message_id=send_message_result_id)

列出已读回执

使用 list_read_receipts 方法检索线程的已读回执。

• 使用 results_per_page（可选），每页返回的最大已读回执数量。
• 使用 skip（可选），在响应中跳过指定位置的已读回执。

从列出已读回执返回的 [ChatMessageReadReceipt] 迭代器是响应。

read_receipts = chat_thread_client.list_read_receipts(results_per_page=5, skip=5)

for read_receipt_page in read_receipts.by_page():
    for read_receipt in read_receipt_page:
        print(read_receipt)
        print(read_receipt.sender)
        print(read_receipt.chat_message_id)
        print(read_receipt.read_on)

示例代码

这些是 Azure Communication Chat 客户端库中常见场景操作的代码示例。示例的异步版本（以 _async 结尾的 Python 示例文件）显示异步操作。在运行示例代码之前，请参阅先决条件

创建资源，然后设置一些环境变量

set AZURE_COMMUNICATION_SERVICE_ENDPOINT="https://<RESOURCE_NAME>.communcationservices.azure.com"
set COMMUNICATION_SAMPLES_CONNECTION_STRING="<connection string of your Communication service>"

pip install azure-communication-identity

python samples\chat_client_sample.py
python samples\chat_client_sample_async.py
python samples\chat_thread_client_sample.py
python samples\chat_thread_client_sample_async.py

故障排除

遇到问题？本节应包含有关如何处理这些问题的详细信息。

下一步

更多示例代码请参阅此处，以及链接到适当的示例测试。

贡献

如果您遇到任何错误或有所建议，请在项目的问题部分提交问题。