### Complete AIMarkdownCardInstance Example

Source: https://github.com/open-dingtalk/dingtalk-stream-sdk-python/blob/main/_autodocs/api-reference/CardInstances.md

A comprehensive example demonstrating the lifecycle of an AIMarkdownCardInstance, from initialization and starting to streaming content, finishing with buttons, and handling exceptions by failing the card.

```python
async def process(self, callback):
    incoming = ChatbotMessage.from_dict(callback.data)
    card = AIMarkdownCardInstance(self.dingtalk_client, incoming)
    
    card.set_title_and_logo("AI Assistant", "@lALPDfJ6V_FPDmvNAfTNAfQ")
    card.ai_start()
    
    try:
        response = ""
        async for chunk in llm_stream():
            response += chunk
            card.ai_streaming(response, append=False)
        
        card.ai_finish(
            markdown=response,
            button_list=[{"text": "Copy", "url": "..."}]
        )
    except Exception as e:
        card.ai_fail()
    
    return AckMessage.STATUS_OK, 'OK'
```

--------------------------------

### Setup Default Logger

Source: https://github.com/open-dingtalk/dingtalk-stream-sdk-python/blob/main/_autodocs/api-reference/UtilityFunctions.md

Creates and configures a default logger instance. Use this to get a logger with a predefined format and handler.

```python
from dingtalk_stream.log import setup_default_logger

logger = setup_default_logger('my_module')
logger.info("Application started")
logger.error("An error occurred")
```

--------------------------------

### Example Card Creation and Update Workflow

Source: https://github.com/open-dingtalk/dingtalk-stream-sdk-python/blob/main/_autodocs/api-reference/CardReplier.md

Demonstrates a typical workflow for handling incoming messages, creating a new card, and later updating its content using the CardReplier. Ensure you have the necessary imports and client setup.

```python
from dingtalk_stream import CardReplier, DingTalkStreamClient, ChatbotMessage

# In your handler
async def process(self, callback):
    incoming = ChatbotMessage.from_dict(callback.data)
    replier = CardReplier(self.dingtalk_client, incoming)
    
    # Create card
    card_id = replier.create_and_send_card(
        card_template_id="589420e2-c1e2-46ef-a5ed-b8728e654da9.schema",
        card_data={
            "markdown": "# Initial Content",
            "title": "Card Title"
        }
    )
    
    # Later, update card
    replier.put_card_data(
        card_instance_id=card_id,
        card_data={"markdown": "# Updated Content"}
    )
    
    return AckMessage.STATUS_OK, 'OK'
```

--------------------------------

### DingTalk Stream Chatbot Registration and Start Example

Source: https://github.com/open-dingtalk/dingtalk-stream-sdk-python/blob/main/_autodocs/api-reference/ChatbotHandler.md

A complete example demonstrating how to register a custom ChatbotHandler with the DingTalkStreamClient and start the client to listen for messages.

```python
from dingtalk_stream import DingTalkStreamClient, Credential, ChatbotMessage

class MyChatbot(ChatbotHandler):
    async def process(self, callback):
        incoming = ChatbotMessage.from_dict(callback.data)
        reply = "Hello! You said: " + incoming.text.content
        self.reply_text(reply, incoming)
        return AckMessage.STATUS_OK, 'OK'

credential = Credential(client_id, client_secret)
client = DingTalkStreamClient(credential)
client.register_callback_handler(ChatbotMessage.TOPIC, MyChatbot())
client.start_forever()
```

--------------------------------

### Stream AI Content with Markdown Card

Source: https://github.com/open-dingtalk/dingtalk-stream-sdk-python/blob/main/_autodocs/README.md

This example demonstrates how to stream AI-generated content in real-time using a Markdown card. It starts by initializing an AI card, then iteratively updates it with chunks of text from an LLM stream, and finally finishes the card.

```python
async def process(self, callback):
    incoming = ChatbotMessage.from_dict(callback.data)
    card = self.ai_markdown_card_start(incoming, title="AI Response")
    
    # Stream content from LLM
    response = ""
    async for chunk in llm_stream():
        response += chunk
        card.ai_streaming(response, append=False)
    
    card.ai_finish(response)
    return AckMessage.STATUS_OK, 'OK'
```

--------------------------------

### Install DingTalk Stream SDK

Source: https://github.com/open-dingtalk/dingtalk-stream-sdk-python/blob/main/_autodocs/README.md

Install the DingTalk Stream SDK and its dependencies using pip.

```bash
pip install dingtalk-stream
```

--------------------------------

### start

Source: https://github.com/open-dingtalk/dingtalk-stream-sdk-python/blob/main/_autodocs/api-reference/AICardReplier.md

Creates an AI card in the PROCESSING state, initiating streaming content generation.

```APIDOC
## start(card_template_id, card_data, recipients=None, support_forward=True)

### Description
Creates an AI card in the PROCESSING state, initiating streaming content generation.

### Method Signature
```python
def start(
    self,
    card_template_id: str,
    card_data: dict,
    recipients: list = None,
    support_forward: bool = True
) -> str
```

### Parameters
#### Path Parameters
- **card_template_id** (str) - Required - Card template UUID
- **card_data** (dict) - Required - Initial card data
- **recipients** (list[dict]) - Optional - Specific recipients
- **support_forward** (bool) - Optional - Card can be forwarded

### Returns
- **str** - Card instance ID

### Example
```python
card_id = replier.start(
    card_template_id="382e4302-551d-4880-bf29-a30acfab2e71.schema",
    card_data={}
)
```
```

--------------------------------

### CarouselCardInstance Example

Source: https://github.com/open-dingtalk/dingtalk-stream-sdk-python/blob/main/_autodocs/api-reference/CardInstances.md

Demonstrates how to create and send a carousel card with product images and a selection button.

```python
images = [
    ("Product A", "@lADPDe7s2ySi18PNA6XNBXg"),
    ("Product B", "@lADPDf0i1beuNF3NAxTNBXg"),
]
card = CarouselCardInstance(client, incoming)
card.reply(
    "Select a product:",
    images,
    button_text="Select"
)
```

--------------------------------

### Start Client Synchronously - Python

Source: https://github.com/open-dingtalk/dingtalk-stream-sdk-python/blob/main/_autodocs/api-reference/DingTalkStreamClient.md

Synchronously starts the client and runs indefinitely, automatically restarting after exceptions. Use this for blocking execution and handle KeyboardInterrupt.

```python
if __name__ == '__main__':
    client = DingTalkStreamClient(credential)
    client.register_callback_handler(ChatbotMessage.TOPIC, MyBotHandler())
    client.start_forever()  # Blocks indefinitely
```

--------------------------------

### pre_start()

Source: https://github.com/open-dingtalk/dingtalk-stream-sdk-python/blob/main/_autodocs/api-reference/EventHandlers.md

Method called before the client starts listening for events. Useful for performing initialization tasks.

```APIDOC
## pre_start()

### Description
Called before the client starts listening for events. Override this method to perform any necessary initialization tasks before the event loop begins.

### Signature
```python
def pre_start(self) -> None
```

### Example
```python
class MyEventHandler(EventHandler):
    def pre_start(self):
        print("Event handler initialized")
```
```

--------------------------------

### Custom SystemHandler Example

Source: https://github.com/open-dingtalk/dingtalk-stream-sdk-python/blob/main/_autodocs/api-reference/EventHandlers.md

Example of extending SystemHandler to process specific system messages, like disconnect events.

```python
class MySystemHandler(SystemHandler):
    async def process(self, message):
        if message.headers.topic == SystemMessage.TOPIC_DISCONNECT:
            print("Server initiated disconnect")
        return AckMessage.STATUS_OK, 'OK'
```

--------------------------------

### Token Cache Example

Source: https://github.com/open-dingtalk/dingtalk-stream-sdk-python/blob/main/_autodocs/configuration.md

An example of how to structure a token cache for access tokens, including the token string and its expiration time.

```python
token_cache = {
    'accessToken': 'token_string',
    'expireTime': 1234567890
}
```

--------------------------------

### Example: Using get_success_response

Source: https://github.com/open-dingtalk/dingtalk-stream-sdk-python/blob/main/_autodocs/api-reference/EventHandlers.md

Shows how to create a successful GraphResponse with specific data using the get_success_response method.

```python
response = handler.get_success_response({"key": "value"})
```

--------------------------------

### async_start

Source: https://github.com/open-dingtalk/dingtalk-stream-sdk-python/blob/main/_autodocs/api-reference/AICardReplier.md

Asynchronous version of start(). Creates an AI card in the PROCESSING state, initiating streaming content generation.

```APIDOC
## async_start(card_template_id, card_data, recipients=None, support_forward=True)

### Description
Asynchronous version of `start()`. Creates an AI card in the PROCESSING state, initiating streaming content generation.

### Method Signature
```python
async def async_start(
    self,
    card_template_id: str,
    card_data: dict,
    recipients: list = None,
    support_forward: bool = True
) -> str
```

### Parameters
#### Path Parameters
- **card_template_id** (str) - Required - Card template UUID
- **card_data** (dict) - Required - Initial card data
- **recipients** (list[dict]) - Optional - Specific recipients
- **support_forward** (bool) - Optional - Card can be forwarded

### Returns
- **str** - Card instance ID

### Example
```python
card_id = await replier.async_start(
    card_template_id="382e4302-551d-4880-bf29-a30acfab2e71.schema",
    card_data={}
)
```
```

--------------------------------

### Implement Pre-Start Initialization

Source: https://github.com/open-dingtalk/dingtalk-stream-sdk-python/blob/main/_autodocs/api-reference/EventHandlers.md

Override the pre_start method to perform initialization tasks before the client begins listening for events. This example prints a message indicating initialization.

```python
class MyEventHandler(EventHandler):
    def pre_start(self):
        print("Event handler initialized")
```

--------------------------------

### Register Chatbot and Event Handlers

Source: https://github.com/open-dingtalk/dingtalk-stream-sdk-python/blob/main/_autodocs/api-reference/EventHandlers.md

Demonstrates how to set up a DingTalk Stream client, register custom handlers for chatbot messages and general events, and start the client. Ensure you have your client ID and secret for credential setup.

```python
from dingtalk_stream import (
    DingTalkStreamClient, Credential,
    EventHandler, ChatbotHandler, SystemHandler,
    ChatbotMessage, AckMessage
)

class MyBotHandler(ChatbotHandler):
    async def process(self, callback):
        incoming = ChatbotMessage.from_dict(callback.data)
        self.reply_text(f"Echo: {incoming.text.content}", incoming)
        return AckMessage.STATUS_OK, 'OK'

class MyEventHandler(EventHandler):
    async def process(self, event):
        print(f"Event type: {event.headers.event_type}")
        return AckMessage.STATUS_OK, 'OK'

# Setup
credential = Credential(client_id, client_secret)
client = DingTalkStreamClient(credential)

# Register handlers
client.register_callback_handler(ChatbotMessage.TOPIC, MyBotHandler())
client.register_all_event_handler(MyEventHandler())

# Start client
client.start_forever()
```

--------------------------------

### Streaming Response Example

Source: https://github.com/open-dingtalk/dingtalk-stream-sdk-python/blob/main/_autodocs/api-reference/AICardReplier.md

An example demonstrating how to use AICardReplier to stream content from an LLM and update an AI card in real-time.

```APIDOC
## Example: Streaming Response

```python
from dingtalk_stream import AICardReplier, AICardStatus

async def process(self, callback):
    incoming = ChatbotMessage.from_dict(callback.data)
    replier = AICardReplier(self.dingtalk_client, incoming)
    
    # Create AI card
    card_id = await replier.async_start(
        card_template_id="382e4302-551d-4880-bf29-a30acfab2e71.schema",
        card_data={}
    )
    
    try:
        # Stream content from LLM
        response_text = ""
        for chunk in llm_stream():
            response_text += chunk
            await replier.async_streaming(
                card_instance_id=card_id,
                content_key="msgContent",
                content_value=response_text,
                append=False,
                finished=False,
                failed=False
            )
        
        # Mark as complete
        await replier.async_finish(
            card_instance_id=card_id,
            card_data={"msgContent": response_text}
        )
    except Exception as e:
        # Mark as failed
        await replier.async_fail(
            card_instance_id=card_id,
            card_data={}
        )
    
    return AckMessage.STATUS_OK, 'OK'
```
```

--------------------------------

### Example: Sending a Markdown Card

Source: https://github.com/open-dingtalk/dingtalk-stream-sdk-python/blob/main/_autodocs/api-reference/CardInstances.md

Demonstrates how to construct a list of buttons and send a markdown card with a title, buttons, and tips.

```python
buttons = [
    {"text": "Accept", "url": "https://example.com/accept", "color": "blue"},
    {"text": "Reject", "url": "https://example.com/reject", "color": "gray"}
]
card.reply(
    "# Action Required\nPlease review and respond",
    buttons,
    tips="Click a button to respond"
)
```

--------------------------------

### CallbackHandler Process Method Example

Source: https://github.com/open-dingtalk/dingtalk-stream-sdk-python/blob/main/_autodocs/api-reference/EventHandlers.md

An example implementation of the `process` method within a subclass of `CallbackHandler`. This demonstrates how to access callback data and return a status code and response message.

```python
class MyHandler(CallbackHandler):
    async def process(self, callback):
        data = callback.data
        # Handle callback...
        return AckMessage.STATUS_OK, 'Callback processed'
```

--------------------------------

### Start Client Asynchronously - Python

Source: https://github.com/open-dingtalk/dingtalk-stream-sdk-python/blob/main/_autodocs/api-reference/DingTalkStreamClient.md

Asynchronously starts the client, initiating message listening and automatic reconnection. Handles potential cancellation or connection errors.

```python
import asyncio

try:
    await client.start()
except asyncio.CancelledError:
    print("Client stopped")
except Exception as e:
    print(f"Error: {e}")
```

--------------------------------

### DingTalk Stream Client Lifecycle

Source: https://github.com/open-dingtalk/dingtalk-stream-sdk-python/blob/main/_autodocs/api-reference/DingTalkStreamClient.md

Demonstrates the typical usage pattern for initializing and starting the DingTalk Stream Client. This includes creating credentials, instantiating the client, registering handlers for specific events and all events, and starting the client to listen for messages.

```python
# 1. Create client
credential = Credential(client_id, client_secret)
client = DingTalkStreamClient(credential)

# 2. Register handlers
client.register_callback_handler(ChatbotMessage.TOPIC, MyBotHandler())
client.register_all_event_handler(MyEventHandler())

# 3. Start listening (blocks indefinitely)
client.start_forever()
```

--------------------------------

### AsyncChatbotHandler Example with Basic Processing

Source: https://github.com/open-dingtalk/dingtalk-stream-sdk-python/blob/main/_autodocs/api-reference/EventHandlers.md

A basic example of an AsyncChatbotHandler that processes incoming messages and replies with text. The process method is synchronous.

```python
from dingtalk_stream import AsyncChatbotHandler, AckMessage

class MyAsyncBot(AsyncChatbotHandler):
    def process(self, message):  # NOTE: NOT async
        # Long-running operations
        incoming = ChatbotMessage.from_dict(message.data)
        self.reply_text("Processing...", incoming)
        # This runs in thread pool
        return AckMessage.STATUS_OK, 'OK'

handler = MyAsyncBot(max_workers=4)
```

--------------------------------

### start_forever

Source: https://github.com/open-dingtalk/dingtalk-stream-sdk-python/blob/main/_autodocs/api-reference/DingTalkStreamClient.md

Synchronously starts the client and runs indefinitely, ensuring continuous operation by automatically restarting after exceptions.

```APIDOC
## start_forever()

### Description
Synchronously starts the client and runs indefinitely, automatically restarting after exceptions.

### Method Signature
```python
def start_forever(self) -> None
```

### Raises
`KeyboardInterrupt` - When user presses Ctrl+C

### Returns
`None` (never returns normally)

### Example
```python
if __name__ == '__main__':
    client = DingTalkStreamClient(credential)
    client.register_callback_handler(ChatbotMessage.TOPIC, MyBotHandler())
    client.start_forever()  # Blocks indefinitely
```
```

--------------------------------

### CallbackHandler.pre_start Method

Source: https://github.com/open-dingtalk/dingtalk-stream-sdk-python/blob/main/_autodocs/api-reference/EventHandlers.md

Method called before the client starts listening for messages. It can be overridden for any necessary initialization tasks.

```APIDOC
## CallbackHandler.pre_start()

### Description
Called before the client starts listening. Override this method for any initialization logic required before message processing begins.

### Signature
```python
def pre_start(self) -> None
```

### Returns
- `None`
```

--------------------------------

### start

Source: https://github.com/open-dingtalk/dingtalk-stream-sdk-python/blob/main/_autodocs/api-reference/DingTalkStreamClient.md

Asynchronously starts the DingTalk stream client, initiating message listening and handling automatic reconnections.

```APIDOC
## start()

### Description
Asynchronously starts the client and begins listening for messages. Implements automatic reconnection on connection loss.

### Method Signature
```python
async def start(self) -> None
```

### Raises
- `asyncio.exceptions.CancelledError` - When the async task is cancelled
- `websockets.exceptions.ConnectionClosedError` - When WebSocket connection closes

### Returns
`None` (never returns; runs indefinitely until exception or cancellation)

### Example
```python
import asyncio

try:
    await client.start()
except asyncio.CancelledError:
    print("Client stopped")
except Exception as e:
    print(f"Error: {e}")
```
```

--------------------------------

### Initialize and Start DingTalk Stream Client

Source: https://github.com/open-dingtalk/dingtalk-stream-sdk-python/blob/main/_autodocs/README.md

Set up authentication with client ID and secret, then initialize the DingTalkStreamClient. Register a callback handler for chatbot messages and start the client to listen for incoming events.

```python
credential = Credential(client_id, client_secret)
client = DingTalkStreamClient(credential)
client.register_callback_handler(ChatbotMessage.TOPIC, MyHandler())
client.start_forever()
```

--------------------------------

### AsyncChatbotHandler Example with Database Query

Source: https://github.com/open-dingtalk/dingtalk-stream-sdk-python/blob/main/_autodocs/api-reference/EventHandlers.md

Demonstrates using AsyncChatbotHandler for synchronous, long-running operations like database queries within its process method.

```python
class DatabaseBot(AsyncChatbotHandler):
    def process(self, message):
        incoming = ChatbotMessage.from_dict(message.data)
        
        # Synchronous database query (runs in thread)
        result = database.query(incoming.text.content)
        self.reply_text(result, incoming)
        
        return AckMessage.STATUS_OK, 'OK'
```

--------------------------------

### CallbackHandler Example Usage

Source: https://github.com/open-dingtalk/dingtalk-stream-sdk-python/blob/main/_autodocs/api-reference/EventHandlers.md

An example demonstrating how to create a custom handler by subclassing `CallbackHandler` and implementing the `process` method. This shows the basic structure for handling DingTalk callbacks.

```python
from dingtalk_stream import CallbackHandler, AckMessage

class MyHandler(CallbackHandler):
    async def process(self, callback):
        # Handle callback
        return AckMessage.STATUS_OK, 'OK'

handler = MyHandler()
```

--------------------------------

### Asynchronous AI Card Creation

Source: https://github.com/open-dingtalk/dingtalk-stream-sdk-python/blob/main/_autodocs/api-reference/AICardReplier.md

Asynchronous version of the `start` method. Use this in async contexts to initiate AI card creation without blocking.

```python
async def async_start(
    self,
    card_template_id: str,
    card_data: dict,
    recipients: list = None,
    support_forward: bool = True
) -> str:
    pass
```

```python
card_id = await replier.async_start(
    card_template_id="382e4302-551d-4880-bf29-a30acfab2e71.schema",
    card_data={}
)
```

--------------------------------

### Custom ChatbotHandler Implementation Example

Source: https://github.com/open-dingtalk/dingtalk-stream-sdk-python/blob/main/_autodocs/api-reference/ChatbotHandler.md

An example of how to implement a custom ChatbotHandler by overriding the process method to handle incoming messages and return an acknowledgment.

```python
class MyHandler(ChatbotHandler):
    async def process(self, callback):
        msg = ChatbotMessage.from_dict(callback.data)
        # Custom logic
        return AckMessage.STATUS_OK, 'Processed'
```

--------------------------------

### Basic Bot Setup

Source: https://github.com/open-dingtalk/dingtalk-stream-sdk-python/blob/main/_autodocs/api-reference/UtilityFunctions.md

Sets up a basic chatbot that echoes received text messages. Requires client ID and secret for authentication. The bot processes incoming messages and sends an acknowledgment.

```python
from dingtalk_stream import (
    DingTalkStreamClient,
    Credential,
    ChatbotHandler,
    ChatbotMessage,
    AckMessage
)

class MyBot(ChatbotHandler):
    async def process(self, callback):
        msg = ChatbotMessage.from_dict(callback.data)
        self.reply_text(f"Echo: {msg.text.content}", msg)
        return AckMessage.STATUS_OK, 'OK'

if __name__ == '__main__':
    credential = Credential(
        client_id="your_client_id",
        client_secret="your_client_secret"
    )
    client = DingTalkStreamClient(credential)
    client.register_callback_handler(
        ChatbotMessage.TOPIC,
        MyBot()
    )
    client.start_forever()
```

--------------------------------

### API Request Headers Example

Source: https://github.com/open-dingtalk/dingtalk-stream-sdk-python/blob/main/_autodocs/configuration.md

Example of auto-generated request headers for API calls. Ensure the 'x-acs-dingtalk-access-token' is correctly set.

```json
{
    'Content-Type': 'application/json',
    'Accept': 'application/json',
    'User-Agent': 'DingTalkStream/1.0 SDK/0.24.3 Python/3.9.0 ...',
    'x-acs-dingtalk-access-token': '<token>'
}
```

--------------------------------

### ai_markdown_card_start

Source: https://github.com/open-dingtalk/dingtalk-stream-sdk-python/blob/main/_autodocs/api-reference/ChatbotHandler.md

Initiates an AI card for streaming content generation. This is the starting point for AI-driven content updates.

```APIDOC
## ai_markdown_card_start

### Description
Initiates an AI card for streaming content generation.

### Method Signature
`ai_markdown_card_start(incoming_message: ChatbotMessage, title: str = "", logo: str = "", recipients: list = None) -> AIMarkdownCardInstance`

### Parameters
#### Path Parameters
* `incoming_message` (ChatbotMessage) - Required - Original incoming message
* `title` (str) - Optional - Card title
* `logo` (str) - Optional - Card logo
* `recipients` (list) - Optional - List of recipients

### Returns
`AIMarkdownCardInstance` - Use for streaming updates
```

--------------------------------

### Start AI Markdown Card

Source: https://github.com/open-dingtalk/dingtalk-stream-sdk-python/blob/main/_autodocs/api-reference/ChatbotHandler.md

Initiates an AI card for streaming content generation. Use this to begin a session where content will be updated incrementally.

```python
def ai_markdown_card_start(
    self,
    incoming_message: ChatbotMessage,
    title: str = "",
    logo: str = "",
    recipients: list = None
) -> AIMarkdownCardInstance:
    # ... implementation details ...
    pass
```

--------------------------------

### Instantiate EventHandler

Source: https://github.com/open-dingtalk/dingtalk-stream-sdk-python/blob/main/_autodocs/api-reference/EventHandlers.md

Example of how to create an instance of a custom EventHandler by subclassing the base class and implementing the process method.

```python
from dingtalk_stream import EventHandler, AckMessage, EventMessage

class MyEventHandler(EventHandler):
    async def process(self, event):
        # Handle event
        return AckMessage.STATUS_OK, 'OK'

handler = MyEventHandler()
```

--------------------------------

### Accept Self-Signed Certificates (Development)

Source: https://github.com/open-dingtalk/dingtalk-stream-sdk-python/blob/main/_autodocs/configuration.md

Example of creating a custom SSL context to accept self-signed certificates for development purposes. This is not recommended for production environments.

```python
import ssl
import websockets

# Create custom SSL context
ssl_context = ssl.create_default_context()
ssl_context.check_hostname = False
ssl_context.verify_mode = ssl.CERT_NONE

# This requires modifying websockets.connect() call
# Not directly exposed in current SDK version
```

--------------------------------

### Card Button Color Example

Source: https://github.com/open-dingtalk/dingtalk-stream-sdk-python/blob/main/_autodocs/types.md

Example of how to specify a button color for a card. Supported colors include 'blue', 'gray', 'red', and 'green'.

```python
{
    "text": "Label",
    "url": "https://...",
    "color": "blue"      # or "gray", etc.
}
```

--------------------------------

### DingTalkStreamClient Usage

Source: https://github.com/open-dingtalk/dingtalk-stream-sdk-python/blob/main/_autodocs/api-reference/DingTalkStreamClient.md

Demonstrates the typical usage pattern for the DingTalkStreamClient, including client creation, handler registration, and starting the client.

```APIDOC
## DingTalkStreamClient Initialization and Usage

### Description
This section outlines the standard procedure for initializing and utilizing the DingTalkStreamClient. It covers creating a client instance with credentials, registering specific event handlers, and starting the client to listen for incoming events.

### Usage Pattern

```python
# 1. Create client
credential = Credential(client_id, client_secret)
client = DingTalkStreamClient(credential)

# 2. Register handlers
client.register_callback_handler(ChatbotMessage.TOPIC, MyBotHandler())
client.register_all_event_handler(MyEventHandler())

# 3. Start listening (blocks indefinitely)
client.start_forever()
```

### Error Handling and Graceful Shutdown

The client includes built-in mechanisms for handling common issues such as WebSocket disconnections (with exponential backoff), token expiration and refresh, and network errors. For asynchronous contexts, a graceful shutdown can be managed using `try-except` blocks.

```python
try:
    await client.start()
except KeyboardInterrupt:
    # Clean shutdown
    pass
except asyncio.CancelledError:
    # Task cancelled
    pass
```
```

--------------------------------

### Example: Using reply_markdown in GraphHandler

Source: https://github.com/open-dingtalk/dingtalk-stream-sdk-python/blob/main/_autodocs/api-reference/EventHandlers.md

Demonstrates how to use the reply_markdown method within a custom GraphHandler to send a markdown response to a callback.

```python
class MyGraphHandler(GraphHandler):
    async def process(self, callback):
        request = GraphRequest.from_dict(callback.data.get('request'))
        response_content = "# Response"
        
        # Reply via webhook
        status, response = await self.reply_markdown(
            webhook=callback.data.get('webhook'),
            content=response_content
        )
        return AckMessage.STATUS_OK, 'OK'
```

--------------------------------

### CallbackHandler pre_start Method Signature

Source: https://github.com/open-dingtalk/dingtalk-stream-sdk-python/blob/main/_autodocs/api-reference/EventHandlers.md

The `pre_start` method signature. This method is called before the client starts listening for messages and can be overridden for custom initialization tasks.

```python
def pre_start(self) -> None
```

--------------------------------

### Start AI Card Creation

Source: https://github.com/open-dingtalk/dingtalk-stream-sdk-python/blob/main/_autodocs/api-reference/AICardReplier.md

Initiates the creation of an AI card in the PROCESSING state. Use this to begin streaming content generation for a new card.

```python
def start(
    self,
    card_template_id: str,
    card_data: dict,
    recipients: list = None,
    support_forward: bool = True
) -> str:
    pass
```

```python
card_id = replier.start(
    card_template_id="382e4302-551d-4880-bf29-a30acfab2e71.schema",
    card_data={}
)
```

--------------------------------

### Setup Default Logger

Source: https://github.com/open-dingtalk/dingtalk-stream-sdk-python/blob/main/_autodocs/configuration.md

Sets up the default logger for the DingTalk Stream SDK. Use this if you don't need custom logging configurations.

```python
import logging
from dingtalk_stream.log import setup_default_logger

logger = setup_default_logger('dingtalk_stream.client')
```

--------------------------------

### Start AI Card Processing

Source: https://github.com/open-dingtalk/dingtalk-stream-sdk-python/blob/main/_autodocs/api-reference/CardInstances.md

Initiates the AI card, setting its state to PROCESSING. Supports optional recipients and forward functionality.

```python
card.ai_start()
```

--------------------------------

### Initialize and Run a Chatbot

Source: https://github.com/open-dingtalk/dingtalk-stream-sdk-python/blob/main/_autodocs/README.md

This snippet shows how to initialize the DingTalkStreamClient with credentials, register a custom chatbot handler, and start the client to listen for incoming messages. Ensure your credentials are set up correctly in the DingTalk console.

```python
from dingtalk_stream import (
    DingTalkStreamClient,
    Credential,
    ChatbotHandler,
    ChatbotMessage,
    AckMessage
)

class MyChatbot(ChatbotHandler):
    async def process(self, callback):
        # Parse incoming message
        incoming = ChatbotMessage.from_dict(callback.data)
        
        # Simple text reply
        self.reply_text(f"You said: {incoming.text.content}", incoming)
        
        # Or send a card
        card = self.reply_markdown_card(
            markdown="# Response\nMarkdown content here",
            incoming_message=incoming,
            title="Bot",
            at_sender=True
        )
        
        # Acknowledge the message
        return AckMessage.STATUS_OK, 'OK'

if __name__ == '__main__':
    # Create credentials from DingTalk console
    credential = Credential(
        client_id="your_app_key",
        client_secret="your_app_secret"
    )
    
    # Create client
    client = DingTalkStreamClient(credential)
    
    # Register handler
    client.register_callback_handler(ChatbotMessage.TOPIC, MyChatbot())
    
    # Start listening (blocks indefinitely)
    client.start_forever()
```

--------------------------------

### Generate Card ID Example Usage

Source: https://github.com/open-dingtalk/dingtalk-stream-sdk-python/blob/main/_autodocs/api-reference/CardReplier.md

Shows how to call the `gen_card_id` static method with an incoming message object to obtain a card ID.

```python
card_id = CardReplier.gen_card_id(incoming_message)
```

--------------------------------

### Create and Send an AckMessage

Source: https://github.com/open-dingtalk/dingtalk-stream-sdk-python/blob/main/_autodocs/api-reference/Messages.md

Example of creating an acknowledgment message with a success status and a message. This is typically returned after processing an incoming message.

```python
async def process(self, callback):
    # Process message...
    ack = AckMessage()
    ack.code = AckMessage.STATUS_OK
    ack.message = 'Success'
    return ack
```

--------------------------------

### AsyncChatbotHandler for Blocking I/O

Source: https://github.com/open-dingtalk/dingtalk-stream-sdk-python/blob/main/_autodocs/README.md

Example of using AsyncChatbotHandler for blocking I/O operations like database queries. The `process` method should not be async, and blocking calls will be executed in a thread pool.

```python
# For blocking database queries
class DBHandler(AsyncChatbotHandler):
    def process(self, message):  # NO async!
        result = database.query(...)  # Blocking I/O runs in thread pool
        return AckMessage.STATUS_OK, 'OK'
```

--------------------------------

### Handler Status Code Example

Source: https://github.com/open-dingtalk/dingtalk-stream-sdk-python/blob/main/_autodocs/README.md

Illustrates returning appropriate status codes from a handler based on processing outcomes. Use STATUS_OK for success, STATUS_BAD_REQUEST for invalid input, and STATUS_SYSTEM_EXCEPTION for unexpected errors.

```python
async def process(self, callback):
    try:
        # ... process message
        return AckMessage.STATUS_OK, 'Success'
    except ValueError:
        return AckMessage.STATUS_BAD_REQUEST, 'Invalid input'
    except Exception as e:
        self.logger.exception(f"Error: {e}")
        return AckMessage.STATUS_SYSTEM_EXCEPTION, 'Error'
```

--------------------------------

### Graceful Shutdown in Async Contexts

Source: https://github.com/open-dingtalk/dingtalk-stream-sdk-python/blob/main/_autodocs/api-reference/DingTalkStreamClient.md

Provides an example of how to handle graceful shutdown of the DingTalk Stream Client in asynchronous contexts using a try-except block to catch KeyboardInterrupt and asyncio.CancelledError.

```python
try:
    await client.start()
except KeyboardInterrupt:
    # Clean shutdown
    pass
except asyncio.CancelledError:
    # Task cancelled
    pass
```

--------------------------------

### Initialize AIMarkdownCardInstance

Source: https://github.com/open-dingtalk/dingtalk-stream-sdk-python/blob/main/_autodocs/api-reference/CardInstances.md

Instantiates an AIMarkdownCardInstance. Requires a DingTalkStreamClient and an incoming ChatbotMessage.

```python
card = AIMarkdownCardInstance(client, incoming)
```

--------------------------------

### Initialize DingTalkStreamClient

Source: https://github.com/open-dingtalk/dingtalk-stream-sdk-python/blob/main/_autodocs/api-reference/DingTalkStreamClient.md

Instantiate the client with your application's credentials. Ensure you have imported the necessary classes.

```python
from dingtalk_stream import DingTalkStreamClient, Credential

credential = Credential(
    client_id="your_app_key",
    client_secret="your_app_secret"
)
client = DingTalkStreamClient(credential)
```

--------------------------------

### Creating Card Instances with ChatbotHandler

Source: https://github.com/open-dingtalk/dingtalk-stream-sdk-python/blob/main/_autodocs/api-reference/CardInstances.md

Demonstrates how to use ChatbotHandler methods to reply with markdown cards or instantiate card classes directly. This is useful for building chatbot responses.

```python
class MyHandler(ChatbotHandler):
    async def process(self, callback):
        incoming = ChatbotMessage.from_dict(callback.data)
        
        # Use high-level methods
        card = self.reply_markdown_card(
            "# Content",
            incoming,
            title="Bot Response"
        )
        
        # Or instantiate directly
        card2 = MarkdownButtonCardInstance(self.dingtalk_client, incoming)
        
        return AckMessage.STATUS_OK, 'OK'
```

--------------------------------

### Get Access Token

Source: https://github.com/open-dingtalk/dingtalk-stream-sdk-python/blob/main/_autodocs/README.md

Retrieves an access token for authentication with DingTalk APIs.

```APIDOC
## POST /v1.0/oauth2/accessToken

### Description
Obtains an access token required for authenticating API requests.

### Method
POST

### Endpoint
/v1.0/oauth2/accessToken

### Request Body
(No specific request body details provided in the source)

### Response
(No specific response details provided in the source)
```

--------------------------------

### Create and Send Card (Sync)

Source: https://github.com/open-dingtalk/dingtalk-stream-sdk-python/blob/main/_autodocs/api-reference/CardReplier.md

Use this method to create a card instance and send it. It supports various options like mentioning users and enabling forwarding. Returns the card instance ID.

```python
card_id = replier.create_and_send_card(
    card_template_id="589420e2-c1e2-46ef-a5ed-b8728e654da9.schema",
    card_data={"markdown": "# Hello", "title": "Greeting"},
    at_sender=True,
    support_forward=True
)
```

--------------------------------

### Create and Deliver Card Instance

Source: https://github.com/open-dingtalk/dingtalk-stream-sdk-python/blob/main/_autodocs/README.md

Creates and delivers a card instance in a single step.

```APIDOC
## POST /v1.0/card/instances/createAndDeliver

### Description
Creates and delivers a card instance in one step.

### Method
POST

### Endpoint
/v1.0/card/instances/createAndDeliver

### Request Body
(No specific request body details provided in the source)

### Response
(No specific response details provided in the source)
```

--------------------------------

### open_connection

Source: https://github.com/open-dingtalk/dingtalk-stream-sdk-python/blob/main/_autodocs/api-reference/DingTalkStreamClient.md

Manually opens a WebSocket connection to DingTalk for real-time communication. Note: This is typically handled automatically by `start()`.

```APIDOC
## open_connection()

### Description
Opens a WebSocket connection by exchanging credentials with the DingTalk server.

### Method Signature
```python
def open_connection(self) -> dict | None
```

### Returns
`dict` with keys:
- `endpoint`: WebSocket URI
- `ticket`: Authentication ticket

Returns `None` if connection fails

### Note
This is called automatically by `start()`. Manual usage is not recommended.
```

--------------------------------

### Graceful Shutdown with Exception Handling

Source: https://github.com/open-dingtalk/dingtalk-stream-sdk-python/blob/main/_autodocs/errors.md

Implement a try-except block to handle KeyboardInterrupt, asyncio.CancelledError, and other unexpected exceptions during client startup for a graceful shutdown.

```python
import asyncio

async def safe_main():
    client = DingTalkStreamClient(credential)
    client.register_callback_handler(ChatbotMessage.TOPIC, MyHandler())
    
    try:
        await client.start()
    except KeyboardInterrupt:
        print("Shutting down...")
    except asyncio.CancelledError:
        print("Task cancelled")
    except Exception as e:
        print(f"Unexpected error: {e}")
```

--------------------------------

### Initialize MarkdownButtonCardInstance

Source: https://github.com/open-dingtalk/dingtalk-stream-sdk-python/blob/main/_autodocs/api-reference/CardInstances.md

Constructor for the MarkdownButtonCardInstance. Requires a DingTalkStreamClient and an incoming ChatbotMessage.

```python
def __init__(self, dingtalk_client: "DingTalkStreamClient", incoming_message: "ChatbotMessage")
```

--------------------------------

### setup_default_logger

Source: https://github.com/open-dingtalk/dingtalk-stream-sdk-python/blob/main/_autodocs/api-reference/UtilityFunctions.md

Creates and configures a default logger instance. It adds a StreamHandler if the logger has no handlers, sets the level to INFO, and uses a specific format.

```APIDOC
## setup_default_logger(name=None)

### Description
Creates and configures a default logger instance.

### Method Signature
```python
def setup_default_logger(name: str = None) -> logging.Logger
```

### Parameters
#### Path Parameters
- **name** (`str`) - Optional - Logger name (typically module name)

### Returns
- `logging.Logger` - Configured logger instance

### Behavior
- Creates logger with specified name
- Adds StreamHandler if logger has no handlers
- Sets level to INFO
- Uses format: `%(asctime)s %(name)-8s %(levelname)-8s %(message)s [%(filename)s:%(lineno)d]`

### Example
```python
from dingtalk_stream.log import setup_default_logger

logger = setup_default_logger('my_module')
logger.info("Application started")
logger.error("An error occurred")
```
```

--------------------------------

### Initialize MarkdownCardInstance

Source: https://github.com/open-dingtalk/dingtalk-stream-sdk-python/blob/main/_autodocs/api-reference/CardInstances.md

Instantiate a MarkdownCardInstance with the DingTalk client and incoming message. This is the first step before setting card content or sending it.

```python
card = MarkdownCardInstance(client, incoming)
```

--------------------------------

### Get SDK Version String

Source: https://github.com/open-dingtalk/dingtalk-stream-sdk-python/blob/main/_autodocs/api-reference/UtilityFunctions.md

Retrieves the current version of the DingTalk Stream SDK. Useful for constructing user agents or logging.

```python
from dingtalk_stream.version import VERSION_STRING

print(VERSION_STRING)  # Output: "0.24.3"
```

```python
import platform
from dingtalk_stream.version import VERSION_STRING

user_agent = f"MyBot/1.0 SDK/{VERSION_STRING} Python/{platform.python_version()}"
```

--------------------------------

### Configure HTTP/HTTPS Proxies

Source: https://github.com/open-dingtalk/dingtalk-stream-sdk-python/blob/main/_autodocs/configuration.md

Set environment variables to configure HTTP and HTTPS proxies for SDK network requests. Use NO_PROXY to exclude specific hosts.

```bash
# HTTP proxy
export HTTP_PROXY="http://proxy.example.com:8080"
export HTTPS_PROXY="http://proxy.example.com:8080"

# No proxy for certain hosts
export NO_PROXY="localhost,127.0.0.1"
```

--------------------------------

### Initialize DingTalk Stream Client with Credentials

Source: https://github.com/open-dingtalk/dingtalk-stream-sdk-python/blob/main/_autodocs/README.md

Use this snippet to set up your DingTalk Stream client by providing your application's App Key and App Secret obtained from the DingTalk developer console. This is the first step to interacting with the DingTalk Stream API.

```python
from dingtalk_stream import Credential

credential = Credential(
    client_id="app_key_from_console",
    client_secret="app_secret_from_console"
)

client = DingTalkStreamClient(credential)
```

--------------------------------

### Create Card Instance

Source: https://github.com/open-dingtalk/dingtalk-stream-sdk-python/blob/main/_autodocs/README.md

Creates a new instance of a pre-built card template.

```APIDOC
## POST /v1.0/card/instances

### Description
Creates a new instance of a card.

### Method
POST

### Endpoint
/v1.0/card/instances

### Request Body
(No specific request body details provided in the source)

### Response
(No specific response details provided in the source)
```

--------------------------------

### Get Request Header

Source: https://github.com/open-dingtalk/dingtalk-stream-sdk-python/blob/main/_autodocs/api-reference/CardReplier.md

Constructs HTTP headers required for making API requests to DingTalk. It includes authentication, content-type, and user-agent information.

```python
@staticmethod
def get_request_header(access_token: str) -> dict
```

--------------------------------

### AIMarkdownCardInstance Constructor

Source: https://github.com/open-dingtalk/dingtalk-stream-sdk-python/blob/main/_autodocs/api-reference/CardInstances.md

Initializes an AIMarkdownCardInstance with a DingTalk client and an incoming message.

```APIDOC
## AIMarkdownCardInstance Constructor

Initializes an AI card instance.

### Signature
```python
def __init__(self, dingtalk_client: "DingTalkStreamClient", incoming_message: "ChatbotMessage")
```
```

--------------------------------

### Get Host IP Address - Python

Source: https://github.com/open-dingtalk/dingtalk-stream-sdk-python/blob/main/_autodocs/api-reference/DingTalkStreamClient.md

Determines the local machine's IPv4 address. This method is used internally for connection initialization.

```python
client.get_host_ip()
```

--------------------------------

### Registering Event and Callback Handlers

Source: https://github.com/open-dingtalk/dingtalk-stream-sdk-python/blob/main/_autodocs/configuration.md

Shows how to instantiate the DingTalkStreamClient and register different types of handlers, including event handlers and specific callback handlers for topics like ChatbotMessage and Card_Callback_Router_Topic.

```python
client = DingTalkStreamClient(credential)

# Register event handler (optional)
client.register_all_event_handler(MyEventHandler())

# Register callback handlers (required)
client.register_callback_handler(ChatbotMessage.TOPIC, MyChatbotHandler())
client.register_callback_handler(Card_Callback_Router_Topic, MyCardHandler())
```

--------------------------------

### Get Image Download URL

Source: https://github.com/open-dingtalk/dingtalk-stream-sdk-python/blob/main/_autodocs/errors.md

Retrieve the image download URL using the provided download code. If the URL cannot be obtained, check the logs for specific reasons.

```python
url = handler.get_image_download_url(download_code)
if not url:
    print("Cannot download image - check logs")
```

--------------------------------

### Get Access Token - Python

Source: https://github.com/open-dingtalk/dingtalk-stream-sdk-python/blob/main/_autodocs/api-reference/DingTalkStreamClient.md

Retrieves the current OAuth access token. Returns None if token retrieval fails. Use the token for subsequent API calls.

```python
token = client.get_access_token()
if token:
    # Use token for API calls
    pass
```

--------------------------------

### Open Connection - Python

Source: https://github.com/open-dingtalk/dingtalk-stream-sdk-python/blob/main/_autodocs/api-reference/DingTalkStreamClient.md

Opens a WebSocket connection by exchanging credentials with the DingTalk server. Returns connection details or None on failure. This method is typically called automatically by start().

```python
client.open_connection()
```

--------------------------------

### Process Incoming Chatbot Messages

Source: https://github.com/open-dingtalk/dingtalk-stream-sdk-python/blob/main/_autodocs/api-reference/Messages.md

Implement a handler to process incoming chatbot messages, extract text content, and reply with a processed message. This example demonstrates acknowledging receipt.

```python
class MyHandler(ChatbotHandler):
    async def process(self, callback: CallbackMessage):
        # Callback contains raw message data
        incoming = ChatbotMessage.from_dict(callback.data)
        
        # Access message properties
        if incoming.message_type == 'text':
            text = incoming.text.content
        
        # Send response
        self.reply_text("Got: " + text, incoming)
        
        # Acknowledge receipt
        return AckMessage.STATUS_OK, 'OK'
```

--------------------------------

### ai_start

Source: https://github.com/open-dingtalk/dingtalk-stream-sdk-python/blob/main/_autodocs/api-reference/CardInstances.md

Initiates the AI card in PROCESSING state, optionally specifying recipients and forward support.

```APIDOC
## ai_start(recipients=None, support_forward=True)

Initiates the AI card in PROCESSING state.

### Method
`ai_start`

### Parameters
#### Query Parameters
- **recipients** (list) - Optional - List of recipients.
- **support_forward** (bool) - Optional - Defaults to `True`. Whether to support forwarding.
```

--------------------------------

### Initialize Credential

Source: https://github.com/open-dingtalk/dingtalk-stream-sdk-python/blob/main/_autodocs/configuration.md

Initializes the Credential object with your App Key and App Secret. These are required for authentication.

```python
credential = Credential(client_id, client_secret)
```

--------------------------------

### Streaming AI Card Response

Source: https://github.com/open-dingtalk/dingtalk-stream-sdk-python/blob/main/_autodocs/api-reference/AICardReplier.md

Example of how to create, stream content to, and finish an AI card using AICardReplier. This pattern is useful for LLM-generated responses where content arrives in chunks.

```python
from dingtalk_stream import AICardReplier, AICardStatus

async def process(self, callback):
    incoming = ChatbotMessage.from_dict(callback.data)
    replier = AICardReplier(self.dingtalk_client, incoming)
    
    # Create AI card
    card_id = await replier.async_start(
        card_template_id="382e4302-551d-4880-bf29-a30acfab2e71.schema",
        card_data={}
    )
    
    try:
        # Stream content from LLM
        response_text = ""
        for chunk in llm_stream():
            response_text += chunk
            await replier.async_streaming(
                card_instance_id=card_id,
                content_key="msgContent",
                content_value=response_text,
                append=False,
                finished=False,
                failed=False
            )
        
        # Mark as complete
        await replier.async_finish(
            card_instance_id=card_id,
            card_data={"msgContent": response_text}
        )
    except Exception as e:
        # Mark as failed
        await replier.async_fail(
            card_instance_id=card_id,
            card_data={}
        )
    
    return AckMessage.STATUS_OK, 'OK'
```

--------------------------------

### Create and Deliver Card with Custom Options

Source: https://github.com/open-dingtalk/dingtalk-stream-sdk-python/blob/main/_autodocs/api-reference/CardReplier.md

This method combines card creation and delivery into a single call, allowing for additional custom delivery parameters like `openSpaceId`.

```python
card_id = replier.create_and_deliver_card(
    card_template_id="589420e2-c1e2-46ef-a5ed-b8728e654da9.schema",
    card_data={"markdown": "# Card"},
    openSpaceId="dtv1.card//IM_GROUP.xxxxx"  # Custom space
)
```

--------------------------------

### Implement Event Processing Logic

Source: https://github.com/open-dingtalk/dingtalk-stream-sdk-python/blob/main/_autodocs/api-reference/EventHandlers.md

Override the process method to define custom logic for handling incoming EventMessage objects. This example checks the event type and logs user login details.

```python
class MyEventHandler(EventHandler):
    async def process(self, event: EventMessage):
        if event.headers.event_type == 'user.login':
            print(f"User logged in: {event.data}")
        return AckMessage.STATUS_OK, 'Processed'
```

--------------------------------

### RPAPluginCardInstance Set Goal Method

Source: https://github.com/open-dingtalk/dingtalk-stream-sdk-python/blob/main/_autodocs/api-reference/CardInstances.md

Sets the execution goal description for the RPA plugin card.

```python
def set_goal(self, goal: str) -> None
```

--------------------------------

### HostingContext Class Definition

Source: https://github.com/open-dingtalk/dingtalk-stream-sdk-python/blob/main/_autodocs/types.md

Defines the context for delegated bot execution, including user ID and nickname.

```python
class HostingContext(object):
    def __init__(self)
```

--------------------------------

### Initialize Credential

Source: https://github.com/open-dingtalk/dingtalk-stream-sdk-python/blob/main/_autodocs/types.md

Use this to authenticate with the DingTalk API. Ensure you have your App Key and App Secret from the DingTalk console.

```python
class Credential(object):
    def __init__(self, client_id: str, client_secret: str)
```

```python
from dingtalk_stream import Credential

credential = Credential(
    client_id="your_app_key",
    client_secret="your_app_secret"
)
```

--------------------------------

### ImageContent from_dict Method

Source: https://github.com/open-dingtalk/dingtalk-stream-sdk-python/blob/main/_autodocs/api-reference/Messages.md

Constructs an ImageContent object from a dictionary. The dictionary should include the 'download_code' for the image.

```python
def from_dict(cls, d: dict)

```

--------------------------------

### Reply with Markdown Card and Buttons

Source: https://github.com/open-dingtalk/dingtalk-stream-sdk-python/blob/main/_autodocs/README.md

Create a card that includes interactive buttons. This is useful for prompting user actions, such as approvals or rejections, by providing direct links.

```python
async def process(self, callback):
    incoming = ChatbotMessage.from_dict(callback.data)
    buttons = [
        {"text": "Accept", "url": "https://example.com/accept"},
        {"text": "Reject", "url": "https://example.com/reject"}
    ]
    card = self.reply_markdown_button(
        incoming_message=incoming,
        markdown="# Approval Required",
        button_list=buttons
    )
    return AckMessage.STATUS_OK, 'OK'
```

--------------------------------

### RPAPluginCardInstance Reply Method

Source: https://github.com/open-dingtalk/dingtalk-stream-sdk-python/blob/main/_autodocs/api-reference/CardInstances.md

Sends an RPA plugin execution card. Requires plugin details, ability name, and arguments. Supports custom recipients and forwarding.

```python
def reply(
    self,
    plugin_id: str,
    plugin_version: str,
    plugin_name: str,
    ability_name: str,
    plugin_args: dict,
    recipients: list = None,
    support_forward: bool = True
) -> None
```

--------------------------------

### HostingContext

Source: https://github.com/open-dingtalk/dingtalk-stream-sdk-python/blob/main/_autodocs/types.md

Represents the delegated bot execution context, providing information about the user operating the bot.

```APIDOC
## Class: HostingContext

### Description
Delegated bot execution context.

### Attributes
- **user_id** (str) - ID of user operating the bot
- **nick** (str) - Display name of operator

### Methods
- **to_dict()** -> dict - Serialize to JSON dict
```

--------------------------------

### async_create_and_send_card

Source: https://github.com/open-dingtalk/dingtalk-stream-sdk-python/blob/main/_autodocs/api-reference/CardReplier.md

Asynchronous version of `create_and_send_card()`. Use this for non-blocking card creation and delivery.

```APIDOC
## async_create_and_send_card

### Description
Asynchronous version of `create_and_send_card()`.

### Method
`async_create_and_send_card`

### Parameters
#### Path Parameters
- **card_template_id** (str) - Required - Template ID (UUID) for the card design
- **card_data** (dict) - Required - Card parameters/variables
- **callback_type** (str) - Optional - Callback mode: "STREAM" or "HTTP". Defaults to "STREAM".
- **callback_route_key** (str) - Optional - Route key for HTTP callbacks. Defaults to "".
- **at_sender** (bool) - Optional - @ the message sender. Defaults to `False`.
- **at_all** (bool) - Optional - @ all conversation participants. Defaults to `False`.
- **recipients** (list[dict]) - Optional - Specific recipient list. Defaults to `None`.
- **support_forward** (bool) - Optional - Whether card can be forwarded. Defaults to `True`.

### Returns
`str` - Card instance ID

### Example
```python
card_id = await replier.async_create_and_send_card(
    card_template_id="589420e2-c1e2-46ef-a5ed-b8728e654da9.schema",
    card_data={"markdown": "# Async", "title": "Async Card"}
)
```
```

--------------------------------

### SystemHandler Process Method

Source: https://github.com/open-dingtalk/dingtalk-stream-sdk-python/blob/main/_autodocs/api-reference/EventHandlers.md

Default implementation for processing system messages. It accepts all system messages and returns a success status.

```python
async def process(self, message: SystemMessage) -> tuple[int, str]
```

--------------------------------

### Reply with RPA Plugin Card

Source: https://github.com/open-dingtalk/dingtalk-stream-sdk-python/blob/main/_autodocs/api-reference/ChatbotHandler.md

Initiates an RPA plugin execution card. All plugin configuration parameters can be passed directly.

```python
def reply_rpa_plugin_card(
    self,
    incoming_message: ChatbotMessage,
    plugin_id: str = "",
    plugin_version: str = "",
    plugin_name: str = "",
    ability_name: str = "",
    plugin_args: dict = {},
    goal: str = "",
    corp_id: str = "",
    recipients: list = None
) -> RPAPluginCardInstance:
    # ... implementation details ...
    pass
```

--------------------------------

### create_and_send_card

Source: https://github.com/open-dingtalk/dingtalk-stream-sdk-python/blob/main/_autodocs/api-reference/CardReplier.md

Creates a card instance and delivers it in two steps. This method is synchronous.

```APIDOC
## create_and_send_card

### Description
Creates a card instance and delivers it in two steps.

### Method
`create_and_send_card`

### Parameters
#### Path Parameters
- **card_template_id** (str) - Required - Template ID (UUID) for the card design
- **card_data** (dict) - Required - Card parameters/variables
- **callback_type** (str) - Optional - Callback mode: "STREAM" or "HTTP". Defaults to "STREAM".
- **callback_route_key** (str) - Optional - Route key for HTTP callbacks. Defaults to "".
- **at_sender** (bool) - Optional - @ the message sender. Defaults to `False`.
- **at_all** (bool) - Optional - @ all conversation participants. Defaults to `False`.
- **recipients** (list[dict]) - Optional - Specific recipient list. Defaults to `None`.
- **support_forward** (bool) - Optional - Whether card can be forwarded. Defaults to `True`.

### Returns
`str` - Card instance ID, or empty string on failure

### Example
```python
card_id = replier.create_and_send_card(
    card_template_id="589420e2-c1e2-46ef-a5ed-b8728e654da9.schema",
    card_data={"markdown": "# Hello", "title": "Greeting"},
    at_sender=True,
    support_forward=True
)
```
```