DingTalk Stream SDK for Python

Warnings

• breaking The `dingtalk-stream` library is built entirely on `asyncio`. All client initialization, connection, and message handling must be performed within an asynchronous context using `async` and `await` keywords. Synchronous usage is not supported for core streaming operations.
  Fix: Ensure your application's entry point and all interactions with `DingTalkStreamClient` are designed to be asynchronous (e.g., `asyncio.run(main())`, `await client.start()`, `async def handler(...)`).
• gotcha Prior to version `0.22.1`, the `DINGTALK_OPENAPI_ENDPOINT` environment variable might not have been consistently applied across all internal API calls made by the SDK, potentially leading to some requests bypassing the custom endpoint. This could cause connectivity issues or incorrect routing.
  Fix: Upgrade to `dingtalk-stream` version `0.22.1` or newer to ensure consistent application of the `DINGTALK_OPENAPI_ENDPOINT` environment variable for all relevant API operations.
• gotcha Authentication is primarily done via `client_id` (AppKey) and `client_secret` (AppSecret). While these can be passed directly, common practice and examples often use `DINGTALK_APP_KEY` and `DINGTALK_APP_SECRET` environment variables. Incorrect or missing credentials will prevent the client from establishing a connection.
  Fix: Ensure `DINGTALK_APP_KEY` and `DINGTALK_APP_SECRET` environment variables are correctly set, or pass `client_id` and `client_secret` directly to the `DingTalkStreamClient` constructor.
• gotcha This SDK is specifically for the DingTalk Streaming API, which handles real-time events and messages. While it might include an `OpenApiClient` for general API calls, its primary purpose is stream processing. If your use case is solely for making one-off DingTalk OpenAPI requests without streaming, a different SDK or direct HTTP requests might be more appropriate.
  Fix: Understand the distinction between the Streaming API and the general OpenAPI. Use this SDK for real-time event processing; for other OpenAPI needs, evaluate if `OpenApiClient` within this SDK meets your requirements or if a different approach is better.

Install

• pip install dingtalk-stream Install stable version

Imports

• DingTalkStreamClient

  from dingtalk_stream import DingTalkStreamClient

• DingTalkMessageType

  from dingtalk_stream.message import DingTalkMessageType

• OpenApiClient

  from dingtalk_stream.client import OpenApiClient

  OpenApiClient is located within the `client` submodule, not directly under `dingtalk_stream`.

Quickstart

This quickstart demonstrates how to initialize the `DingTalkStreamClient`, register a default handler for incoming messages, and start the asynchronous client. Ensure `DINGTALK_APP_KEY` and `DINGTALK_APP_SECRET` environment variables are set for authentication.

import os
import asyncio
from dingtalk_stream import DingTalkStreamClient
from dingtalk_stream.message import DingTalkMessageType

async def quickstart_handler(msg):
    """A simple message handler that prints received message info."""
    print(f"\nReceived message type: {msg.type}\nContent: {msg.data}")
    if msg.type == DingTalkMessageType.INTERACTION_MESSAGE:
        # For interactive messages, you might want to reply to a card.
        # Example (requires card_instance_id and conversation_id from msg):
        # await msg.reply_to_card(card_data={'content': 'Acknowledged!'})
        print("Received an interactive message.")

async def main():
    app_key = os.environ.get("DINGTALK_APP_KEY", "YOUR_APP_KEY")
    app_secret = os.environ.get("DINGTALK_APP_SECRET", "YOUR_APP_SECRET")

    if app_key == "YOUR_APP_KEY" or app_secret == "YOUR_APP_SECRET":
        print("Please set DINGTALK_APP_KEY and DINGTALK_APP_SECRET environment variables.")
        print("Example: export DINGTALK_APP_KEY='...' DINGTALK_APP_SECRET='...' ")
        return

    client = DingTalkStreamClient(
        client_id=app_key,
        client_secret=app_secret
    )

    client.set_default_handler(quickstart_handler)

    print("DingTalk Stream Client starting... (Press Ctrl+C to stop)")
    try:
        await client.start()
    except asyncio.CancelledError:
        print("Client start cancelled.")
    finally:
        await client.stop()
        print("Client stopped.")

if __name__ == "__main__":
    try:
        asyncio.run(main())
    except KeyboardInterrupt:
        print("Application terminated by user.")