Install & Compatibility
Where this runs
tested against v0.24.3 · pip install
no network on importno background threads
Install × environment matrix
Each cell = how many times install + import succeeded across repeated harness runs. Partial = flaky.
glibc = Debian/Ubuntu slim · musl = Alpine Linux
muslpy 3.10–3.95 runs
installs and imports cleanly · install 0.0s · import 0.878s · 31.3MB
glibcpy 3.10–3.95 runs
installs and imports cleanly · install 5.0s · import 0.792s · 34MB
31MB installed
● package 31MB
Code
Verified usage
Verified import paths — ran on the pinned version, not inferred.
DingTalkStreamClient
✓ from dingtalk_stream import DingTalkStreamClient
DingTalkMessageType
✓ from dingtalk_stream.message import DingTalkMessageType
OpenApiClient
✓ from dingtalk_stream.client import OpenApiClient
✗ from dingtalk_stream import OpenApiClient
OpenApiClient is located within the `client` submodule, not directly under `dingtalk_stream`.
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.")
Debug
Known issues
breakingThe `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.fixEnsure 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(...)`).
affects: All versions from v0.21.0 onwards.
gotchaPrior 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.fixUpgrade 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.
affects: < 0.22.1
gotchaAuthentication 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.fixEnsure `DINGTALK_APP_KEY` and `DINGTALK_APP_SECRET` environment variables are correctly set, or pass `client_id` and `client_secret` directly to the `DingTalkStreamClient` constructor.
affects: All versions
gotchaThis 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.fixUnderstand 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.
affects: All versions
Errors
Common errors & fixes
ModuleNotFoundError: No module named 'dingtalk_stream'
The `dingtalk-stream` library is not installed in your Python environment or is not accessible.
fixpip install dingtalk-stream
coroutine 'DingTalkStreamClient.start' was never awaited
The `DingTalkStreamClient.start()` method is an asynchronous coroutine and must be `await`ed, but it was called without `await`.
fixEnsure you are calling `await client.start()` within an `async` function and that your application's entry point is set up to run an asyncio event loop.
error processing message: object tuple can't be used in 'await' expression
This error often occurs when the `ChatbotHandler.process()` method, or a similar message handling callback, is defined as a synchronous function but the `dingtalk-stream` SDK (especially versions >= 0.20) expects it to be an `async` function and attempts to `await` its return value.
fixRedefine your message processing methods (e.g., `process`) as `async def` functions. For example: `async def process(self, message): ...`
"DINGTALK_CLIENT_ID and DINGTALK_CLIENT_SECRET required"
The necessary authentication credentials (AppKey and AppSecret) for your DingTalk bot are not provided, either as environment variables or directly in the client configuration.
fixSet `DINGTALK_CLIENT_ID` (your AppKey) and `DINGTALK_CLIENT_SECRET` (your AppSecret) as environment variables, or pass them directly when initializing the `StreamClient`.
Upgrade
Version history
0.24.3latest on PyPI · released Oct 24, 2025
Audit
Dependencies
aiohttprequiredAsynchronous HTTP client for API communication.
pydanticrequiredData validation and settings management.
websocket-clientrequiredWebSocket client for maintaining the streaming connection.
httpxrequiredAnother HTTP client, likely used for specific OpenAPI calls.