適用于 Python 的 Azure 通訊聊天套件用戶端程式庫 - 1.2.0 版

此套件包含適用于聊天Azure 通訊服務的 Python SDK。 在這裡深入瞭解Azure 通訊服務

| 原始程式碼套件 (Pypi) | 封裝 (Conda) | API 參考檔 | 產品檔

免責聲明

Python 2.7 的 Azure SDK Python 套件支援已于 2022 年 1 月 1 日結束。 如需詳細資訊和問題，請參閱 https://github.com/Azure/azure-sdk-for-python/issues/20691

開始使用

Prerequisites

• 需要 Python 3.7 或更新版本才能使用此套件。
• 已部署通訊服務資源。 您可以使用Azure 入口網站或Azure PowerShell來設定。

安裝套件

安裝 Azure 通訊服務聊天 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-Read-Delete) 作業

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

初始化 ChatThreadClient 類別之後，您可以執行下列聊天作業：

更新執行緒

線上程主題上執行更新作業

update_topic(topic, **kwargs)

取得聊天對話屬性

get_properties(**kwargs)

傳送、取得、更新和刪除訊息

對訊息執行 CRUD (Create-Read-Update-Delete) 作業

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

取得、新增和移除參與者

線上程參與者上執行 CRD (Create-Read-Delete) 作業

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，必要，這是 CommunicationUserIdentifier 您從使用者存取權杖建立的 CommunicationIdentityClient.create_user ()
  • display_name選擇性的 是執行緒參與者的顯示名稱。
  • share_history_time、選擇性、與參與者共用聊天歷程記錄的時間。
• 使用 idempotency_token ，選擇性地指定要求的唯一識別碼。

CreateChatThreadResult 是建立對話所傳回的結果，您可以使用它來擷取 id 所建立之聊天對話的 。 id然後，您可以使用 方法擷取 ChatThreadClient 物件 get_chat_thread_client 。 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)

取得執行緒

Use get_properties 方法會 ChatThreadProperties 從服務擷取 ， thread_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 是執行緒的唯一識別碼。

• 使用 thread_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 是傳送訊息時的傳回回應，它包含的識別碼就是訊息的唯一識別碼。

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)

取得訊息

Use get_message 方法會從服務擷取訊息; message_id 這是訊息的唯一識別碼。

• 使用 message_id ，必要，若要指定現有訊息 ChatMessage 的訊息識別碼是從取得訊息傳回的回應，其中包含識別碼，這是訊息的唯一識別碼，其他欄位請參閱 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 ，必要，是訊息的唯一識別碼。
• 使用 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 ，必要專案是訊息的唯一識別碼。

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，必要，這是 CommunicationUserIdentifier 您從使用者存取權杖CommunicationIdentityClient.create_user () 所建立的
  • 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 所識別的執行緒中移除執行緒參與者。 identifierCommunicationUserIdentifier是您從 CommunicationIdentityClient.create_user () 建立的azure-communication-identity

和 已新增至此聊天對話。

• 用來 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 指定要傳送讀取回條之訊息的識別碼

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 通訊聊天用戶端程式庫的常見案例作業。 範例的非同步版本 (附加 _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

疑難排解

發生問題嗎？ 本節應包含詳細資料，以瞭解該處的用途。

下一步

更多範例程式碼應該 移至這裡，以及適當範例測試的連結。

參與

如果您遇到任何錯誤或有建議，請在專案的 [ 問題 ] 區段中提出問題。