# Pywayne Lark Bot Listener

> Feishu/Lark message listener for real-time event processing via WebSocket. Use when users need to listen for incoming Feishu messages (text, image, file, audio, media, sticker, post, interactive) and events (recall, read, reaction, bot added/removed, member changes, chat updates) with automatic deduplication, async handling, resource auto-download, and convenient decorators. Provides high-level h…

- **Type:** Skill
- **Install:** `agentstack add skill-wangyendt-wayne-skills-lark-bot-listener`
- **Verified:** Yes — security-reviewed for prompt injection and unsafe behavior
- **Seller:** [wangyendt](https://agentstack.voostack.com/s/wangyendt)
- **Installs:** 0
- **Category:** [Content & Media](https://agentstack.voostack.com/c/content-and-media)
- **Latest version:** 0.1.0
- **License:** MIT
- **Upstream author:** [wangyendt](https://github.com/wangyendt)
- **Source:** https://github.com/wangyendt/wayne-skills/tree/main/pywayne/lark-bot-listener

## Install

```sh
agentstack add skill-wangyendt-wayne-skills-lark-bot-listener
```

Requires the [AgentStack CLI](https://agentstack.voostack.com/docs/cli). Works with Claude Code, Cursor, and any MCP-compatible agent.

## About

# Pywayne Lark Bot Listener - Real-Time Event Processing

## Overview

`LarkBotListener` is a WebSocket-based event listener for Feishu (Lark) that enables **real-time message and event processing**. It provides decorator-based handlers for all message types and bot events, with automatic resource handling, message deduplication, and async/sync compatibility.

**Key Features**:
- **Message Handlers**: text, image, file, audio, media, sticker, post, interactive
- **Event Handlers**: recall, read, reaction, bot added/removed, member changes, chat updates
- **Auto Resource Handling**: Download images/files to temp dir, auto-cleanup after processing
- **Auto Upload/Send**: Return new file path from handler → automatically upload and send back
- **Flexible Parameters**: Handlers declare only parameters they need (text, chat_id, user_name, etc.)
- **Message Deduplication**: Per-handler deduplication with configurable expiry
- **Async/Sync Compatible**: Support both `async def` and `def` handler functions
- **Built-in LarkBot**: Access full `LarkBot` API via `listener.bot`
- **Streaming Reply Helpers**: Keep one card updated in place during long-running work
- **Card Action Callbacks**: HTTP handler for interactive card button clicks

**Companion**:
- Uses `LarkBot` internally for sending messages (see `pywayne-lark-bot` skill)

## Installation

```bash
pip install pywayne lark-oapi
```

## Quick Start

```python
from pywayne.lark_bot_listener import LarkBotListener

# Initialize listener
listener = LarkBotListener(
    app_id="cli_xxxxxxxxxxxx",
    app_secret="your_app_secret",
    message_expiry_time=60  # Deduplication expiry in seconds
)

# Handle text messages
@listener.text_handler()
async def on_text(text: str, chat_id: str):
    print(f"Received: {text}")
    listener.send_message(chat_id, f"Echo: {text}")

# Start listening
listener.run()
```

## LarkBotListener Class

### Constructor

```python
listener = LarkBotListener(
    app_id: str,
    app_secret: str,
    message_expiry_time: int = 60  # Deduplication cache expiry (seconds)
)
```

**Instance Attributes**:
- `bot`: Built-in `LarkBot` instance for sending messages and API calls
- `temp_dir`: Temporary directory for downloaded files (auto-created at system temp)

### Access Built-in Bot

```python
# Use bot for any LarkBot operations
listener.bot.send_text_to_chat("oc_xxx", "Message from listener")
listener.bot.reply_message("om_xxx", "text", {"text": "Reply"})
listener.bot.add_reaction("om_xxx", "THUMBSUP")
# ... any LarkBot method
```

**Tip**:
- `listener.bot.send_interactive_to_chat(...)` returns a response `Dict`.
- On success, that response includes `message_id`, which you can store and later pass to `listener.bot.update_interactive_card(...)`.

```python
from pywayne.lark_bot import CardContentV2

card = CardContentV2(title="Job Status", template="blue")
card.add_markdown("Job accepted. Waiting for worker...")

msg = listener.bot.send_interactive_to_chat("oc_xxx", card.get_card())
message_id = msg["message_id"]

# ... later ...

updated_card = CardContentV2(title="Job Status", template="green")
updated_card.add_markdown("Job finished successfully")

listener.bot.update_interactive_card(message_id, updated_card.get_card())
```

## MessageContext

All low-level `@listen()` handlers receive a `MessageContext` object.

```python
from pywayne.lark_bot_listener import MessageContext

@listener.listen()
async def handle_any(ctx: MessageContext):
    print(ctx.chat_id)
    print(ctx.message_id)
    print(ctx.message_type)
```

**MessageContext Fields**:
- `chat_id`: Chat/conversation ID
- `user_id`: Sender's open ID
- `message_type`: Message type string ("text", "image", "file", etc.)
- `content`: Message content (text string or JSON string)
- `is_group`: Boolean indicating if message is from group chat
- `chat_type`: Feishu chat type ("group" or "p2p")
- `message_id`: Unique message ID
- `thread_id`: Thread ID if message is in a thread
- `root_id`: Root message ID for thread
- `parent_id`: Parent message ID for thread
- `mentions`: List of @mentions in message
- `raw_event`: Original Feishu SDK event object

## Core Message Handlers

### listen - Universal Message Entry Point

Generic message handler for any message type. Provides full `MessageContext`.

```python
@listener.listen(
    message_type: Optional[str] = None,  # Specific type or None for all
    group_only: bool = False,            # Only group messages
    user_only: bool = False              # Only private messages
)
async def handler(ctx: MessageContext):
    # Handler implementation
    pass
```

**Parameters**:
- `message_type`: Filter by type: `"text"`, `"image"`, `"file"`, `"audio"`, `"media"`, `"sticker"`, `"post"`, `"interactive"`, or `None` for all
- `group_only`: Only process group messages
- `user_only`: Only process private (p2p) messages

**Use Cases**:
- Need full message context
- Need `message_id` for replies, reactions, etc.
- Routing different message types from single handler
- Don't need automatic file download/upload

**Example: Universal Router**:

```python
@listener.listen()  # All message types
async def router(ctx: MessageContext):
    print(f"Message type: {ctx.message_type}")
    print(f"From {'group' if ctx.is_group else 'user'}: {ctx.chat_id}")
    print(f"Content: {ctx.content}")
    
    # Reply to any message
    listener.bot.reply_message(
        ctx.message_id,
        "text",
        {"text": f"Received {ctx.message_type} message"}
    )
```

**Example: Type-Specific Handling**:

```python
@listener.listen(message_type="post", group_only=True)
async def handle_group_posts(ctx: MessageContext):
    import json
    post_content = json.loads(ctx.content)
    print(f"Rich text post: {post_content}")
    
    # Add reaction
    listener.bot.add_reaction(ctx.message_id, "THUMBSUP")
```

### text_handler - Text Message Handler

Simplified handler for text messages with automatic parameter extraction.

```python
@listener.text_handler(
    group_only: bool = False,
    user_only: bool = False
)
async def handler(**kwargs):
    # Handler declares which parameters it needs
    pass
```

**Available Parameters** (declare any subset):
- `text` (str): Text message content
- `chat_id` (str): Chat ID
- `is_group` (bool): Whether from group
- `group_name` (str): Group name (empty for private chat)
- `user_name` (str): Sender's display name
- `message_id` (str): Message ID (for replies, requires manual declaration)

**Example: Simple Text Echo**:

```python
@listener.text_handler()
async def echo(text: str, chat_id: str):
    listener.send_message(chat_id, f"You said: {text}")
```

**Example: Group-Only Command Bot**:

```python
@listener.text_handler(group_only=True)
async def commands(text: str, chat_id: str, user_name: str):
    if text == "/status":
        listener.send_message(chat_id, f"{user_name}, system is healthy ✅")
    elif text == "/help":
        listener.send_message(chat_id, "Commands: /status, /help")
```

**Example: Context-Aware Response**:

```python
@listener.text_handler()
async def smart_reply(text: str, chat_id: str, is_group: bool, group_name: str, user_name: str):
    context = f"in {group_name}" if is_group else "in private"
    listener.send_message(
        chat_id,
        f"Hi {user_name}, I received your message {context}: {text}"
    )
```

### image_handler - Image Message Handler

Auto-downloads image to temp file, passes `Path` to handler. Optionally auto-uploads and sends returned image.

```python
@listener.image_handler(
    group_only: bool = False,
    user_only: bool = False
)
async def handler(**kwargs) -> Optional[Path]:
    # Handler returns new image path or None
    return processed_image_path  # Auto-uploads and sends
    # or
    return None  # No image sent back
```

**Available Parameters**:
- `image_path` (Path): Temporary image file path (required parameter)
- `chat_id` (str): Chat ID
- `is_group` (bool): Whether from group
- `group_name` (str): Group name
- `user_name` (str): Sender's display name
- `message_id` (str): Message ID

**Return Value**:
- `Path`: Automatically upload and send this image back to chat
- `None`: Don't send any image

**Auto-Cleanup**: Original downloaded image and returned image (if different) are automatically deleted after processing.

**Example: Simple Image Echo**:

```python
from pathlib import Path

@listener.image_handler()
async def echo_image(image_path: Path) -> Path:
    print(f"Received image: {image_path}")
    return image_path  # Send same image back
```

**Example: OpenCV Image Processing**:

```python
import cv2
import tempfile
from pathlib import Path

@listener.image_handler()
async def add_watermark(image_path: Path, user_name: str) -> Path:
    # Read image
    img = cv2.imread(str(image_path))
    
    # Add watermark
    cv2.putText(
        img,
        f"Processed by {user_name}",
        (30, 60),
        cv2.FONT_HERSHEY_SIMPLEX,
        1,
        (0, 255, 0),
        2
    )
    
    # Save to new temp file
    with tempfile.NamedTemporaryFile(suffix=".png", delete=False) as f:
        result_path = Path(f.name)
    
    cv2.imwrite(str(result_path), img)
    
    return result_path  # Auto-upload and send
```

**Example: Conditional Processing**:

```python
@listener.image_handler(group_only=True)
async def process_group_images(image_path: Path, group_name: str) -> Optional[Path]:
    # Only process images from specific groups
    if group_name == "CV Project":
        # ... image processing ...
        return processed_path
    else:
        return None  # Don't send anything back
```

### file_handler - File Message Handler

Auto-downloads file to temp path. Optionally auto-uploads and sends returned file.

```python
@listener.file_handler(
    group_only: bool = False,
    user_only: bool = False
)
async def handler(**kwargs) -> Optional[Path]:
    # Handler returns new file path or None
    return processed_file_path  # Auto-uploads and sends
    # or
    return None  # No file sent back
```

**Available Parameters**:
- `file_path` (Path): Temporary file path
- `chat_id` (str): Chat ID
- `is_group` (bool): Whether from group
- `group_name` (str): Group name
- `user_name` (str): Sender's display name

**Return Value**: Same as `image_handler`

**Example: File Echo**:

```python
@listener.file_handler()
async def bounce_file(file_path: Path, user_name: str) -> Path:
    print(f"Received file from {user_name}: {file_path}")
    return file_path  # Send same file back
```

**Example: File Processing**:

```python
@listener.file_handler()
async def process_csv(file_path: Path, chat_id: str) -> Optional[Path]:
    # Check file extension
    if not str(file_path).endswith('.csv'):
        listener.send_message(chat_id, "Please send a CSV file")
        return None
    
    # Process CSV
    import pandas as pd
    df = pd.read_csv(file_path)
    
    # Create summary
    summary = df.describe()
    
    # Save summary to new file
    summary_path = file_path.with_suffix('.summary.csv')
    summary.to_csv(summary_path)
    
    return summary_path  # Send summary back
```

### audio_handler - Audio Message Handler

Auto-downloads audio file (typically `.opus` format).

```python
@listener.audio_handler(
    group_only: bool = False,
    user_only: bool = False
)
async def handler(**kwargs) -> Optional[Path]:
    # Return new audio path or None
    return processed_audio_path
```

**Available Parameters**:
- `audio_path` (Path): Temporary audio file path
- `chat_id` (str): Chat ID
- `is_group` (bool): Whether from group
- `group_name` (str): Group name
- `user_name` (str): Sender's display name
- `message_id` (str): Message ID
- `thread_id` (str): Thread ID

**Example: Audio Acknowledgment**:

```python
@listener.audio_handler()
async def on_audio(audio_path: Path, message_id: str):
    size = audio_path.stat().st_size
    listener.bot.reply_message(
        message_id,
        "text",
        {"text": f"Received audio: {size} bytes"}
    )
    # Return None - don't send audio back
    return None
```

### media_handler - Media/Video Message Handler

Auto-downloads media file (typically `.mp4` format).

```python
@listener.media_handler(
    group_only: bool = False,
    user_only: bool = False
)
async def handler(**kwargs) -> Optional[Path]:
    # Return new media path or None
    return processed_media_path
```

**Available Parameters**: Same as `audio_handler`, but parameter is `media_path`

**Example: Video Processing**:

```python
@listener.media_handler()
async def process_video(media_path: Path, chat_id: str) -> None:
    # Extract video metadata
    import cv2
    cap = cv2.VideoCapture(str(media_path))
    
    fps = cap.get(cv2.CAP_PROP_FPS)
    frame_count = int(cap.get(cv2.CAP_PROP_FRAME_COUNT))
    duration = frame_count / fps
    
    cap.release()
    
    listener.send_message(
        chat_id,
        f"Video: {duration:.2f}s, {fps:.1f} FPS, {frame_count} frames"
    )
    
    return None  # Don't send video back
```

### sticker_handler - Sticker Message Handler

Handle Feishu sticker messages.

```python
@listener.sticker_handler(
    group_only: bool = False,
    user_only: bool = False
)
async def handler(**kwargs):
    pass
```

**Available Parameters**:
- `sticker_content` (dict or str): Parsed JSON content or raw string
- `raw_content` (str): Original raw content string
- `chat_id` (str): Chat ID
- `is_group` (bool): Whether from group
- `group_name` (str): Group name
- `user_name` (str): Sender's display name
- `message_id` (str): Message ID
- `thread_id` (str): Thread ID

**Example: Sticker Response**:

```python
@listener.sticker_handler()
async def on_sticker(sticker_content: dict, chat_id: str):
    sticker_key = sticker_content.get("file_key", "unknown")
    listener.send_message(chat_id, f"Nice sticker! ({sticker_key})")
```

### mention_handler - @Mention Handler

Only triggers when bot is @mentioned in messages.

```python
@listener.mention_handler(
    group_only: bool = False,
    user_only: bool = False
)
async def handler(**kwargs):
    pass
```

**Available Parameters**:
- `text` (str): Message text content
- `mentions` (list): List of @mention objects
- `chat_id` (str): Chat ID
- `is_group` (bool): Whether from group
- `group_name` (str): Group name
- `user_name` (str): Sender's display name
- `message_id` (str): Message ID
- `thread_id` (str): Thread ID
- `root_id` (str): Root message ID
- `parent_id` (str): Parent message ID
- `raw_event`: Original event object

**Example: Respond to @Mentions Only**:

```python
@listener.mention_handler(group_only=True)
async def when_mentioned(text: str, user_name: str, message_id: str):
    listener.bot.reply_message(
        message_id,
        "text",
        {"text": f"{user_name}, how can I help you?"}
    )
```

**Example: Command Parsing from @Mention**:

```python
@listener.mention_handler()
async def handle_commands(text: str, chat_id: str):
    # Remove @bot mention from text
    command = text.strip().lower()
    
    if "status" in command:
        listener.send_message(chat_id, "System status: ✅ Healthy")
    elif "help" in command:
        listener.send_message(chat_id, "Commands: status, help, ping")
    else:
        listener.send_message(chat_id, "Unknown command")
```

### Streaming Card Reply Helpers

These wrappers let a listener handler stream output back into one reply card without dropping down to raw `listener.bot` calls.

```python
reply = listener.reply_streaming_card(
    target: Union[str, MessageContext],
    *,
    title: str = "Streaming Reply",
    template: str = "blue",
    initial_md: str = "",
    reply_in_thread: bool = False,
    uuid: str = "",
    status_text: str = "Generating...",
    max_chunk_bytes: int = 18_000
) -> Dict

response = listener.update_streaming_card(
    card_message_id: str,
    md_text: str,
    *,
    title: str = "Streaming Reply",
    template: str = "blue",
    done: bool = False,
    status_text: str = ""

…

## Source & license

This open-source skill is cataloged on AgentStack and links to its original source — we do not rehost the code.

- **Author:** [wangyendt](https://github.com/wangyendt)
- **Source:** [wangyendt/wayne-skills](https://github.com/wangyendt/wayne-skills)
- **License:** MIT

Install and usage instructions live in the source repository linked above.

## Pricing

- **Free** — Free

## Security capabilities

Automated source analysis of v0.1.0 — what this tool can access:

- **Network access:** yes
- **Filesystem access:** yes
- **Shell / process execution:** no
- **Environment & secrets:** no
- **Dynamic code execution:** no

*"Yes" means the capability is present in the source — more access means more to trust, not that it is unsafe.*


## Versions

- **0.1.0** — security scan: passed — Imported from the upstream source.

## Links

- Listing page: https://agentstack.voostack.com/l/skill-wangyendt-wayne-skills-lark-bot-listener
- Seller: https://agentstack.voostack.com/s/wangyendt
- Browse the marketplace: https://agentstack.voostack.com/browse

---
Listed on AgentStack — the marketplace for AI agent skills and MCP servers. Every listing is security-reviewed. Creators keep 70%.