{"id":5181,"library":"dingtalk-stream","title":"DingTalk Stream SDK for Python","description":"The `dingtalk-stream` library is the official Python SDK for interacting with the DingTalk Streaming API. It enables developers to build chatbots, AI assistants, and other applications that receive real-time messages and events from DingTalk, and send messages or interact with cards. The current version is 0.24.3, with an active release cadence addressing features and bug fixes.","status":"active","version":"0.24.3","language":"en","source_language":"en","source_url":"https://github.com/open-dingtalk/dingtalk-stream-sdk-python","tags":["dingtalk","chatbot","stream","messaging","ai","asyncio"],"install":[{"cmd":"pip install dingtalk-stream","lang":"bash","label":"Install stable version"}],"dependencies":[{"reason":"Asynchronous HTTP client for API communication.","package":"aiohttp"},{"reason":"Data validation and settings management.","package":"pydantic"},{"reason":"WebSocket client for maintaining the streaming connection.","package":"websocket-client"},{"reason":"Another HTTP client, likely used for specific OpenAPI calls.","package":"httpx"}],"imports":[{"symbol":"DingTalkStreamClient","correct":"from dingtalk_stream import DingTalkStreamClient"},{"symbol":"DingTalkMessageType","correct":"from dingtalk_stream.message import DingTalkMessageType"},{"note":"OpenApiClient is located within the `client` submodule, not directly under `dingtalk_stream`.","wrong":"from dingtalk_stream import OpenApiClient","symbol":"OpenApiClient","correct":"from dingtalk_stream.client import OpenApiClient"}],"quickstart":{"code":"import os\nimport asyncio\nfrom dingtalk_stream import DingTalkStreamClient\nfrom dingtalk_stream.message import DingTalkMessageType\n\nasync def quickstart_handler(msg):\n    \"\"\"A simple message handler that prints received message info.\"\"\"\n    print(f\"\\nReceived message type: {msg.type}\\nContent: {msg.data}\")\n    if msg.type == DingTalkMessageType.INTERACTION_MESSAGE:\n        # For interactive messages, you might want to reply to a card.\n        # Example (requires card_instance_id and conversation_id from msg):\n        # await msg.reply_to_card(card_data={'content': 'Acknowledged!'})\n        print(\"Received an interactive message.\")\n\nasync def main():\n    app_key = os.environ.get(\"DINGTALK_APP_KEY\", \"YOUR_APP_KEY\")\n    app_secret = os.environ.get(\"DINGTALK_APP_SECRET\", \"YOUR_APP_SECRET\")\n\n    if app_key == \"YOUR_APP_KEY\" or app_secret == \"YOUR_APP_SECRET\":\n        print(\"Please set DINGTALK_APP_KEY and DINGTALK_APP_SECRET environment variables.\")\n        print(\"Example: export DINGTALK_APP_KEY='...' DINGTALK_APP_SECRET='...' \")\n        return\n\n    client = DingTalkStreamClient(\n        client_id=app_key,\n        client_secret=app_secret\n    )\n\n    client.set_default_handler(quickstart_handler)\n\n    print(\"DingTalk Stream Client starting... (Press Ctrl+C to stop)\")\n    try:\n        await client.start()\n    except asyncio.CancelledError:\n        print(\"Client start cancelled.\")\n    finally:\n        await client.stop()\n        print(\"Client stopped.\")\n\nif __name__ == \"__main__\":\n    try:\n        asyncio.run(main())\n    except KeyboardInterrupt:\n        print(\"Application terminated by user.\")\n","lang":"python","description":"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."},"warnings":[{"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(...)`).","message":"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.","severity":"breaking","affected_versions":"All versions from v0.21.0 onwards."},{"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.","message":"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.","severity":"gotcha","affected_versions":"< 0.22.1"},{"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.","message":"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.","severity":"gotcha","affected_versions":"All versions"},{"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.","message":"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.","severity":"gotcha","affected_versions":"All versions"}],"env_vars":null,"last_verified":"2026-04-13T00:00:00.000Z","next_check":"2026-07-12T00:00:00.000Z"}