DOCSEARCHbyAlgolia
Submit Docs

DingTalk Stream SDK for Python

repository·main·Indexed 41 days ago

https://github.com/open-dingtalk/dingtalk-stream-sdk-python

A Python SDK for the DingTalk Stream Mode API designed to simplify the integration of event pushes, chatbot message receiving, and card callbacks. It provides a client for managing WebSocket connections and a handler system for processing CallbackMessages and responding via ChatbotMessages.

Tokens
1.2K
Snippets
4
Records
5
Agent score
2%

What's inside DingTalk Stream SDK for Python

  2. Create a DingTalk Stream Chatbot

    main

    To build a chatbot, extend dingtalk_stream.ChatbotHandler and implement the process method. Inside process, you receive a CallbackMessage, which you can convert to a ChatbotMessage using dingtalk_stream.ChatbotMessage.from_dict(callback.data). You can respond to users using self.reply_text().

    To run the bot, initialize a dingtalk_stream.Credential with your client_id and client_secret, create a dingtalk_stream.DingTalkStreamClient, register your handler using client.register_callback_handler(), and call client.start_forever().

    import logging
from dingtalk_stream import AckMessage
import dingtalk_stream

class CalcBotHandler(dingtalk_stream.ChatbotHandler):
    def __init__(self, logger: logging.Logger = None):
        super(dingtalk_stream.ChatbotHandler, self).__init__()
        if logger:
            self.logger = logger

    async def process(self, callback: dingtalk_stream.CallbackMessage):
        # Convert callback data to a ChatbotMessage
        incoming_message = dingtalk_stream.ChatbotMessage.from_dict(callback.data)
        expression = incoming_message.text.content.strip()
        
        # Logic for processing the message
        result = '...' 
        
        # Reply to the user
        response = 'Q: %s\nA: %s' % (expression, result)
        self.reply_text(response, incoming_message)

        # Return status to acknowledge receipt
        return AckMessage.STATUS_OK, 'OK'

def main():
    credential = dingtalk_stream.Credential('YOUR_CLIENT_ID', 'YOUR_CLIENT_SECRET')
    client = dingtalk_stream.DingTalkStreamClient(credential)
    
    # Register the handler for ChatbotMessage topics
    client.register_callback_handler(dingtalk_stream.chatbot.ChatbotMessage.TOPIC, CalcBotHandler())
    
    # Start the client loop
    client.start_forever()

if __name__ == '__main__':
    main()

  3. Configure WebSocket connection options

    main

    You can pass custom parameters to websockets.connect() via the websocket_connect_options argument when initializing DingTalkStreamClient. This is useful for adjusting handshake or heartbeat timeouts.

    client = dingtalk_stream.DingTalkStreamClient(
    credential,
    websocket_connect_options={
        'open_timeout': 10,
        'ping_interval': 20,
        'ping_timeout': 20,
    },
)

  4. Upgrade notice: register_callback_handler

    main
    When upgrading from versions prior to v0.13.0, ensure you use the correct spelling for the registration method: register_callback_handler (previously contained a typo register_callback_hanlder).

  5. Manage the client lifecycle with start() and stop()

    main

    If you are running the SDK within an existing asyncio event loop and do not want to use start_forever(), use client.start(). This method automatically handles reconnections after network errors. To gracefully shut down the client and stop reconnection attempts, call client.stop().

    client_task = asyncio.create_task(client.start())
try:
    await run_application()
finally:
    await client.stop()
    await client_task