Pywayne Lark Bot Listener

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, chatid, username, 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

pip install pywayne lark-oapi

Quick Start

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

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

# 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(...).

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.

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.

@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:

@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:

@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.

@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:

@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:

@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:

@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.

@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:

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:

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:

@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.

@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:

@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:

@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).

@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:

@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).

@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:

@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.

@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:

@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.

@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:

@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:

@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.

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.