| name | pywayne-lark-bot-listener |
| description | 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 handlers (text_handler, image_handler, file_handler, audio_handler, media_handler, sticker_handler, mention_handler) with automatic download/cleanup and optional file回传, plus event handlers (recall_handler, message_read_handler, reaction_handler, bot_added/removed_handler, member_changed_handler, chat_updated/disbanded_handler), card action callback support, and listener-level streaming card helpers such as reply_streaming_card, update_streaming_card, recolor_streaming_card, stream_reply_card, and astream_reply_card. |
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
pip install pywayne lark-oapi
Quick Start
from pywayne.lark_bot_listener import LarkBotListener
listener = LarkBotListener(
app_id="cli_xxxxxxxxxxxx",
app_secret="your_app_secret",
message_expiry_time=60
)
@listener.text_handler()
async def on_text(text: str, chat_id: str):
print(f"Received: {text}")
listener.send_message(chat_id, f"Echo: {text}")
listener.run()
LarkBotListener Class
Constructor
listener = LarkBotListener(
app_id: str,
app_secret: str,
message_expiry_time: int = 60
)
Instance Attributes:
bot: Built-in LarkBot instance for sending messages and API callstemp_dir: Temporary directory for downloaded files (auto-created at system temp)
Access Built-in Bot
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")
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"]
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 IDuser_id: Sender's open IDmessage_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 chatchat_type: Feishu chat type ("group" or "p2p")message_id: Unique message IDthread_id: Thread ID if message is in a threadroot_id: Root message ID for threadparent_id: Parent message ID for threadmentions: List of @mentions in messageraw_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,
group_only: bool = False,
user_only: bool = False
)
async def handler(ctx: MessageContext):
pass
Parameters:
message_type: Filter by type: "text", "image", "file", "audio", "media", "sticker", "post", "interactive", or None for allgroup_only: Only process group messagesuser_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()
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}")
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}")
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):
pass
Available Parameters (declare any subset):
text (str): Text message contentchat_id (str): Chat IDis_group (bool): Whether from groupgroup_name (str): Group name (empty for private chat)user_name (str): Sender's display namemessage_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]:
return processed_image_path
return None
Available Parameters:
image_path (Path): Temporary image file path (required parameter)chat_id (str): Chat IDis_group (bool): Whether from groupgroup_name (str): Group nameuser_name (str): Sender's display namemessage_id (str): Message ID
Return Value:
Path: Automatically upload and send this image back to chatNone: 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
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:
img = cv2.imread(str(image_path))
cv2.putText(
img,
f"Processed by {user_name}",
(30, 60),
cv2.FONT_HERSHEY_SIMPLEX,
1,
(0, 255, 0),
2
)
with tempfile.NamedTemporaryFile(suffix=".png", delete=False) as f:
result_path = Path(f.name)
cv2.imwrite(str(result_path), img)
return result_path
Example: Conditional Processing:
@listener.image_handler(group_only=True)
async def process_group_images(image_path: Path, group_name: str) -> Optional[Path]:
if group_name == "CV Project":
return processed_path
else:
return None
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]:
return processed_file_path
return None
Available Parameters:
file_path (Path): Temporary file pathchat_id (str): Chat IDis_group (bool): Whether from groupgroup_name (str): Group nameuser_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
Example: File Processing:
@listener.file_handler()
async def process_csv(file_path: Path, chat_id: str) -> Optional[Path]:
if not str(file_path).endswith('.csv'):
listener.send_message(chat_id, "Please send a CSV file")
return None
import pandas as pd
df = pd.read_csv(file_path)
summary = df.describe()
summary_path = file_path.with_suffix('.summary.csv')
summary.to_csv(summary_path)
return summary_path
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 processed_audio_path
Available Parameters:
audio_path (Path): Temporary audio file pathchat_id (str): Chat IDis_group (bool): Whether from groupgroup_name (str): Group nameuser_name (str): Sender's display namemessage_id (str): Message IDthread_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
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 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:
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
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 stringraw_content (str): Original raw content stringchat_id (str): Chat IDis_group (bool): Whether from groupgroup_name (str): Group nameuser_name (str): Sender's display namemessage_id (str): Message IDthread_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 contentmentions (list): List of @mention objectschat_id (str): Chat IDis_group (bool): Whether from groupgroup_name (str): Group nameuser_name (str): Sender's display namemessage_id (str): Message IDthread_id (str): Thread IDroot_id (str): Root message IDparent_id (str): Parent message IDraw_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):
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 = "",
max_chunk_bytes: int = 18_000
) -> Dict
response = listener.recolor_streaming_card(
card_message_id: str,
md_text: str,
*,
title: str = "Streaming Reply",
template: str = "green",
status_text: str = "Done",
done: bool = True,
max_chunk_bytes: int = 18_000
) -> Dict
result = listener.stream_reply_card(
target: Union[str, MessageContext],
text_stream: Iterable[Any],
*,
title: str = "Streaming Reply",
template: str = "blue",
initial_md: str = "",
reply_in_thread: bool = False,
uuid: str = "",
update_interval: float = 0.25,
status_text: str = "Generating...",
final_status_text: str = "",
final_template: Optional[str] = "green",
max_chunk_bytes: int = 18_000
) -> Dict[str, Any]
result = await listener.astream_reply_card(
target: Union[str, MessageContext],
text_stream: AsyncIterable[Any],
*,
title: str = "Streaming Reply",
template: str = "blue",
initial_md: str = "",
reply_in_thread: bool = False,
uuid: str = "",
update_interval: float = 0.25,
status_text: str = "Generating...",
final_status_text: str = "",
final_template: Optional[str] = "green",
max_chunk_bytes: int = 18_000
) -> Dict[str, Any]
Key Rules:
target can be either a raw message_id or the whole MessageContext.update_streaming_card() updates the card message returned by reply_streaming_card(), not the original user message ID.- Streaming updates expect the full accumulated Markdown, not the latest delta only.
- Use
final_template="green" for success and recolor_streaming_card(..., template="red") for failure.
- If you need token-level refresh, decrease
update_interval, but remember Feishu message updates are rate-limited.
Example 1: Mention -> Async LLM Stream -> Auto Turn Green:
@listener.mention_handler(group_only=True)
async def answer_when_mentioned(text: str, user_name: str, message_id: str):
async def fake_llm_stream():
yield f"Hello {user_name}.\n\n"
yield "## Draft answer\n"
yield f"- You asked: `{text}`\n"
yield "- Suggested next step: review the checklist.\n"
await listener.astream_reply_card(
message_id,
fake_llm_stream(),
title="Assistant Reply",
template="blue",
status_text="Thinking...",
final_status_text="Answer complete",
final_template="green",
update_interval=0.4
)
Example 2: Universal Handler with Manual Progress Updates:
@listener.listen(message_type="text")
async def run_job(ctx: MessageContext):
reply = listener.reply_streaming_card(
ctx,
title="Data Pipeline",
template="wathet",
initial_md="Queued job...",
status_text="Starting"
)
card_message_id = reply["message_id"]
current_text = "Queued job..."
current_text += "\n- Loaded source files"
listener.update_streaming_card(
card_message_id,
current_text,
title="Data Pipeline",
template="wathet",
status_text="Transforming"
)
current_text += "\n- Applied transformations"
listener.update_streaming_card(
card_message_id,
current_text,
title="Data Pipeline",
template="wathet",
status_text="Uploading output"
)
listener.recolor_streaming_card(
card_message_id,
current_text + "\n- Upload completed",
title="Data Pipeline",
template="green",
status_text="Finished"
)
Example 3: Add Reaction While Working, Remove It When Done:
@listener.listen(message_type="text")
async def process_with_progress(ctx: MessageContext):
reaction = listener.bot.add_reaction(ctx.message_id, "OK")
reaction_id = reaction.get("reaction_id", "")
reply = listener.reply_streaming_card(
ctx,
title="Risk Review",
template="blue",
initial_md="Parsing request..."
)
card_message_id = reply["message_id"]
current_text = "Parsing request..."
try:
for step in [
"Loaded policy rules",
"Matched request fields",
"Generated recommendation",
]:
current_text += f"\n- {step}"
listener.update_streaming_card(
card_message_id,
current_text,
title="Risk Review",
template="blue",
status_text="Running"
)
listener.recolor_streaming_card(
card_message_id,
current_text + "\n\n**Decision**: approved",
title="Risk Review",
template="green",
status_text="Completed"
)
except Exception as exc:
listener.recolor_streaming_card(
card_message_id,
current_text + f"\n\n**Error**: {exc}",
title="Risk Review",
template="red",
status_text="Failed"
)
raise
finally:
if reaction_id:
listener.bot.delete_reaction(ctx.message_id, reaction_id)
Event Handlers
recall_handler - Message Recall Event
Triggered when a message is recalled.
@listener.recall_handler()
async def on_recall(**kwargs):
pass
Available Parameters:
message_id (str): ID of recalled messagechat_id (str): Chat IDrecall_time (str): Recall timestamprecall_type (str): Recall typeraw_event: Original event object
Example: Log Recalls:
@listener.recall_handler()
async def log_recalls(message_id: str, chat_id: str, recall_time: str):
print(f"Message {message_id} recalled in {chat_id} at {recall_time}")
message_read_handler - Message Read Event
Triggered when message(s) are marked as read.
@listener.message_read_handler()
async def on_read(**kwargs):
pass
Available Parameters:
reader: Reader information objectmessage_id_list (list): List of read message IDsraw_event: Original event object
Example: Track Read Status:
@listener.message_read_handler()
async def track_reads(message_id_list: list, reader):
for msg_id in message_id_list:
print(f"Message {msg_id} read by {reader}")
reaction_handler - Reaction Event
Triggered when reaction is added or removed.
@listener.reaction_handler()
async def on_reaction(**kwargs):
pass
Available Parameters:
action (str): "created" or "deleted"message_id (str): Message IDemoji_type (str): Feishu emoji code (e.g., "THUMBSUP")operator_type (str): Operator typeuser_id (str): User who reactedapp_id (str): App IDaction_time (str): Timestampraw_event: Original event object
Example: Handle Reactions:
@listener.reaction_handler()
async def on_reaction(action: str, message_id: str, emoji_type: str, user_id: str):
if action == "created":
if emoji_type == "THUMBSUP":
print(f"User {user_id} liked message {message_id}")
elif emoji_type == "HEART":
print(f"User {user_id} loved message {message_id}")
Example: Reaction Event Triggers Batch Follow-Up:
@listener.reaction_handler()
async def escalate_on_reaction(action: str, emoji_type: str, user_id: str, message_id: str):
if action != "created" or emoji_type != "OK":
return
listener.bot.batch_send_message(
"text",
content=f"Message {message_id} was acknowledged by {user_id}",
user_open_ids=["ou_manager_a", "ou_manager_b"]
)
bot_added_handler - Bot Added to Group
Triggered when bot is added to a group chat.
@listener.bot_added_handler()
async def on_added(**kwargs):
pass
Available Parameters:
chat_id (str): Chat IDoperator_id (str): User who added the botexternal (bool): Whether external groupoperator_tenant_key (str): Operator tenant keyname (str): Group nameraw_event: Original event object
Example: Welcome Message:
@listener.bot_added_handler()
async def welcome(chat_id: str, name: str):
listener.bot.send_markdown_to_chat(
chat_id,
md_text="""
# Bot Activated! 👋
Welcome to using our bot!
**Capabilities**:
- Auto-reply to messages
- Image processing
- File analysis
- @mention for help
Type `/help` to get started.
""",
title="Welcome"
)
bot_removed_handler - Bot Removed from Group
Triggered when bot is removed from a group chat.
@listener.bot_removed_handler()
async def on_removed(**kwargs):
pass
Available Parameters: Same as bot_added_handler
Example: Log Removal:
@listener.bot_removed_handler()
async def log_removal(chat_id: str, operator_id: str, name: str):
print(f"Bot removed from {name} ({chat_id}) by {operator_id}")
bot_p2p_chat_entered_handler - Bot Enters Private Chat
Triggered when bot enters a private chat with a user.
@listener.bot_p2p_chat_entered_handler()
async def on_p2p_entered(**kwargs):
pass
Available Parameters:
chat_id (str): Chat IDoperator_id (str): User's open IDlast_message_id (str): Last message ID in chatlast_message_create_time (str): Last message timestampraw_event: Original event object
Example: First-Time Welcome:
@listener.bot_p2p_chat_entered_handler()
async def first_contact(chat_id: str, operator_id: str):
listener.bot.send_text_to_chat(
chat_id,
"Hello! I'm here to help. Ask me anything!"
)
member_changed_handler - Group Member Changes
Unified handler for member added/removed/withdrawn events.
@listener.member_changed_handler()
async def on_member_change(**kwargs):
pass
Available Parameters:
action (str): "added", "deleted", or "withdrawn"chat_id (str): Chat IDoperator_id (str): User who performed the actionusers (list): List of affected usersexternal (bool): Whether external groupoperator_tenant_key (str): Operator tenant keyname (str): Group nameraw_event: Original event object
Example: Member Change Notifications:
@listener.member_changed_handler()
async def notify_member_changes(action: str, chat_id: str, users: list, name: str):
user_count = len(users)
if action == "added":
listener.bot.send_text_to_chat(
chat_id,
f"👋 Welcome {user_count} new member(s) to {name}!"
)
elif action == "deleted":
listener.bot.send_text_to_chat(
chat_id,
f"{user_count} member(s) removed from {name}"
)
elif action == "withdrawn":
listener.bot.send_text_to_chat(
chat_id,
f"{user_count} member(s) left {name}"
)
chat_updated_handler - Group Info Updated
Triggered when chat information changes (name, description, etc.).
@listener.chat_updated_handler()
async def on_chat_updated(**kwargs):
pass
Available Parameters:
chat_id (str): Chat IDoperator_id (str): User who made changesexternal (bool): Whether external groupoperator_tenant_key (str): Operator tenant keybefore_change: Info before changeafter_change: Info after changemoderator_list (list): List of moderatorsraw_event: Original event object
Example: Track Changes:
@listener.chat_updated_handler()
async def log_changes(chat_id: str, before_change, after_change):
print(f"Chat {chat_id} updated:")
print(f" Before: {before_change}")
print(f" After: {after_change}")
chat_disbanded_handler - Group Disbanded
Triggered when a group is disbanded/deleted.
@listener.chat_disbanded_handler()
async def on_disbanded(**kwargs):
pass
Available Parameters:
chat_id (str): Chat IDoperator_id (str): User who disbanded the groupexternal (bool): Whether external groupoperator_tenant_key (str): Operator tenant keyname (str): Group nameraw_event: Original event object
Example: Log Disbandment:
@listener.chat_disbanded_handler()
async def log_disbandment(chat_id: str, name: str, operator_id: str):
print(f"Group '{name}' ({chat_id}) disbanded by {operator_id}")
Interactive Card Callbacks
card_action_handler - Card Button Click Handler
Register handler for interactive card action callbacks (button clicks, form submissions).
@listener.card_action_handler(
verification_token: str,
encrypt_key: str = ""
)
def on_card_action(card_event):
return {"toast": {"type": "success", "content": "Action completed"}}
Returns:
- Dict with toast/modal response
- Can update card content in place using
listener.bot.update_interactive_card()
Example: Approval Workflow:
@listener.card_action_handler(verification_token="your_token")
def handle_approval(card_event):
action_value = card_event.action.value
message_id = card_event.event.context.open_message_id
if action_value == "approve":
from pywayne.lark_bot import CardContentV2
approved_card = CardContentV2(title="Approved", template="green")
approved_card.add_markdown("✅ Request approved")
listener.bot.update_interactive_card(message_id, approved_card.get_card())
return {"toast": {"type": "success", "content": "Approved!"}}
elif action_value == "reject":
from pywayne.lark_bot import CardContentV2
rejected_card = CardContentV2(title="Rejected", template="red")
rejected_card.add_markdown("❌ Request rejected")
listener.bot.update_interactive_card(message_id, rejected_card.get_card())
return {"toast": {"type": "warning", "content": "Rejected"}}
Example: Button Click Recolors an Existing Streaming Card:
@listener.card_action_handler(verification_token="your_token")
def handle_review_action(card_event):
message_id = card_event.event.context.open_message_id
action = card_event.action.value.get("action")
summary = card_event.action.value.get("summary", "No summary provided")
if action == "approve":
listener.bot.recolor_streaming_card(
message_id,
f"## Review Result\n\n{summary}",
title="Manual Review",
template="green",
status_text="Approved"
)
return {"toast": {"type": "success", "content": "Approved"}}
if action == "reject":
listener.bot.recolor_streaming_card(
message_id,
f"## Review Result\n\n{summary}",
title="Manual Review",
template="red",
status_text="Rejected"
)
return {"toast": {"type": "warning", "content": "Rejected"}}
return {"toast": {"type": "info", "content": "No action taken"}}
get_card_action_handler - Get HTTP Handler
Retrieve the HTTP handler for mounting in web frameworks.
handler = listener.get_card_action_handler() -> CardActionHandler
Usage with Flask:
from flask import Flask, request
app = Flask(__name__)
@app.route('/feishu/card', methods=['POST'])
def card_callback():
handler = listener.get_card_action_handler()
return handler(request)
app.run()
Usage with FastAPI:
from fastapi import FastAPI, Request
app = FastAPI()
app.post("/feishu/card")(listener.get_card_action_handler())
Utility Methods
send_message - Quick Markdown Send
Lightweight method for sending Markdown-formatted messages.
listener.send_message(chat_id: str, content: str) -> None
Note: Internally uses PostContent with Markdown. For more control, use listener.bot.send_markdown_to_chat().
Example:
listener.send_message(
"oc_xxx",
"**Bold** text and [link](https://example.com)"
)
run - Start Listening
Start the WebSocket listener service.
listener.run() -> None
Important: This is a blocking call. The listener will run until interrupted.
Example:
@listener.text_handler()
async def handle_text(text: str):
print(text)
listener.run()
Complete Usage Examples
Example 1: AI Chat Bot
from pywayne.lark_bot_listener import LarkBotListener
from pywayne.llm.chat_bot import ChatManager
import asyncio
listener = LarkBotListener(app_id="cli_xxx", app_secret="sec_xxx")
chat_manager = ChatManager(
base_url="https://api.deepseek.com/v1",
api_key="your_key"
)
@listener.text_handler()
async def ai_reply(text: str, chat_id: str, user_name: str):
chat_bot = chat_manager.get_chat(chat_id)
loop = asyncio.get_event_loop()
response = await loop.run_in_executor(
None,
lambda: ''.join(chat_bot.chat(text, stream=True))
)
listener.send_message(chat_id, f"Hi {user_name}!\n\n{response}")
listener.run()
Example 2: AprilTag Detection Bot
from pywayne.lark_bot_listener import LarkBotListener
from pywayne.cv.apriltag_detector import ApriltagCornerDetector
from pathlib import Path
import cv2
import tempfile
listener = LarkBotListener(app_id="cli_xxx", app_secret="sec_xxx")
detector = ApriltagCornerDetector()
@listener.image_handler()
async def detect_apriltags(image_path: Path, user_name: str) -> Path:
detected_img = detector.detect_and_draw(str(image_path))
with tempfile.NamedTemporaryFile(suffix=".png", delete=False) as f:
result_path = Path(f.name)
cv2.imwrite(str(result_path), detected_img)
return result_path
listener.run()
Example 3: Multi-Handler Router
from pywayne.lark_bot_listener import LarkBotListener, MessageContext
listener = LarkBotListener(app_id="cli_xxx", app_secret="sec_xxx")
@listener.text_handler(group_only=True)
async def commands(text: str, chat_id: str, user_name: str):
if text.startswith("/"):
command = text[1:].lower()
if command == "status":
listener.send_message(chat_id, "System: ✅ Healthy")
elif command == "help":
listener.send_message(chat_id, "Commands: /status, /help, /ping")
elif command == "ping":
listener.send_message(chat_id, f"Pong! Hi {user_name}")
@listener.image_handler()
async def process_image(image_path: Path) -> None:
size = image_path.stat().st_size
print(f"Received image: {size} bytes")
return None
@listener.mention_handler()
async def when_mentioned(text: str, message_id: str):
listener.bot.reply_message(
message_id,
"text",
{"text": "You mentioned me! How can I help?"}
)
@listener.reaction_handler()
async def on_reaction(action: str, emoji_type: str, message_id: str):
if action == "created" and emoji_type == "THUMBSUP":
print(f"Someone liked message {message_id}")
listener.run()
Example 4: File Processing Pipeline
from pywayne.lark_bot_listener import LarkBotListener
from pathlib import Path
import pandas as pd
listener = LarkBotListener(app_id="cli_xxx", app_secret="sec_xxx")
@listener.file_handler()
async def process_csv(file_path: Path, chat_id: str, user_name: str) -> Optional[Path]:
if not str(file_path).endswith('.csv'):
listener.send_message(chat_id, f"{user_name}, please send a CSV file")
return None
try:
df = pd.read_csv(file_path)
summary = df.describe()
summary_text = f"""
**CSV Analysis for {user_name}**
Rows: {len(df)}
Columns: {len(df.columns)}
Statistics:
{summary.to_markdown()}
"""
listener.send_message(chat_id, summary_text)
output_path = file_path.with_name(f"{file_path.stem}_summary.csv")
summary.to_csv(output_path)
return output_path
except Exception as e:
listener.send_message(chat_id, f"Error processing CSV: {e}")
return None
listener.run()
Example 5: Event-Driven Automation
from pywayne.lark_bot_listener import LarkBotListener
listener = LarkBotListener(app_id="cli_xxx", app_secret="sec_xxx")
@listener.member_changed_handler()
async def welcome_members(action: str, chat_id: str, users: list):
if action == "added":
listener.bot.send_markdown_to_chat(
chat_id,
md_text=f"""
# 👋 Welcome!
Welcome to our group! We have {len(users)} new member(s).
**Getting Started**:
- Read pinned messages
- Check group description
- Introduce yourself
Type `/help` for bot commands.
""",
title="Welcome"
)
@listener.reaction_handler()
async def track_reactions(action: str, emoji_type: str, message_id: str, user_id: str):
if action == "created":
print(f"Reaction: {user_id} added {emoji_type} to {message_id}")
@listener.bot_added_handler()
async def on_added(chat_id: str, name: str):
listener.bot.send_markdown_to_chat(
chat_id,
md_text=f"""
# 🤖 Bot Activated
Thanks for adding me to **{name}**!
**I can help with**:
- Auto-reply to messages
- Process images and files
- Answer questions
- Provide status updates
@mention me or type `/help` to get started.
""",
title="Hello!"
)
listener.run()
Example 6: Advanced Reaction Workflow
import asyncio
listener = LarkBotListener(app_id="cli_xxx", app_secret="sec_xxx")
@listener.listen(message_type="text")
async def process_with_reaction(ctx: MessageContext):
reaction = listener.bot.add_reaction(ctx.message_id, "WITTY")
reaction_id = reaction.get("reaction_id")
try:
await asyncio.sleep(2)
listener.bot.reply_message(
ctx.message_id,
"text",
{"text": f"Processed: {ctx.content}"}
)
finally:
if reaction_id:
listener.bot.delete_reaction(ctx.message_id, reaction_id)
listener.run()
Example 7: Interactive Card with Callback
from pywayne.lark_bot import CardContentV2
listener = LarkBotListener(app_id="cli_xxx", app_secret="sec_xxx")
@listener.bot_added_handler()
async def send_welcome_card(chat_id: str):
card = {
"header": {
"title": {"content": "Welcome Survey", "tag": "plain_text"},
"template": "blue"
},
"elements": [
{
"tag": "markdown",
"content": "Please complete our welcome survey"
},
{
"tag": "action",
"actions": [
{
"tag": "button",
"text": {"content": "Start Survey", "tag": "plain_text"},
"type": "primary",
"value": "start_survey"
}
]
}
]
}
listener.bot.send_interactive_to_chat(chat_id, card)
@listener.card_action_handler(verification_token="your_token")
def on_button_click(card_event):
action_value = card_event.action.value
message_id = card_event.event.context.open_message_id
if action_value == "start_survey":
updated_card = CardContentV2(title="Thank You!", template="green")
updated_card.add_markdown("✅ Survey link sent via private message")
listener.bot.update_interactive_card(message_id, updated_card.get_card())
return {"toast": {"type": "success", "content": "Survey started!"}}
listener.run()
Best Practices
1. Handler Parameter Selection
Declare only parameters you need:
@listener.text_handler()
async def simple(text: str):
print(text)
@listener.text_handler()
async def contextual(text: str, chat_id: str, user_name: str):
print(f"{user_name} said {text} in {chat_id}")
@listener.text_handler()
async def wasteful(text: str, chat_id: str, is_group: bool, group_name: str, user_name: str):
print(text)
2. Resource Handling
Let handlers auto-cleanup:
@listener.image_handler()
async def process(image_path: Path) -> Path:
result = process_image(image_path)
return result
@listener.file_handler()
async def log_only(file_path: Path) -> None:
log_file_info(file_path)
return None
3. Error Handling
Handlers have isolated exception handling:
@listener.text_handler()
async def safe_handler(text: str, chat_id: str):
try:
result = risky_operation(text)
listener.send_message(chat_id, result)
except Exception as e:
listener.send_message(chat_id, f"Error: {e}")
4. Async vs Sync
Both work, but async preferred:
@listener.text_handler()
async def async_handler(text: str):
await some_async_operation()
@listener.text_handler()
def sync_handler(text: str):
some_sync_operation()
5. Use Built-in Bot for Advanced Features
@listener.text_handler()
async def advanced(text: str, chat_id: str):
msg = listener.bot.send_text_to_chat(chat_id, "Processing...")
listener.bot.update_message(
msg["message_id"],
"text",
{"text": "Done!"}
)
listener.bot.add_reaction(msg["message_id"], "THUMBSUP")
Important Notes
-
Message Deduplication:
- Each handler maintains its own deduplication cache
- Same message can trigger multiple handlers
- Default expiry: 60 seconds (configurable in constructor)
- Prevents duplicate processing of the same message by same handler
-
Temp File Management:
- Files downloaded to
{system_temp}/lark_bot_temp
- Auto-cleanup after handler completes
- Returned files also cleaned up after upload
-
Async Compatibility:
- All handlers support both
async def and def
- Internal async/sync detection and wrapping
- Prefer
async def for consistency
-
Group/Private Filtering:
group_only=True: Only process group messagesuser_only=True: Only process private messages- Both False (default): Process all messages
-
Parameter Flexibility:
- Handlers inspect function signature
- Only declared parameters are passed
- Allows concise handlers with minimal boilerplate
-
Multiple Handler Registration:
- Can register multiple handlers for same message type
- All matching handlers execute independently
- Useful for different aspects (logging, processing, analytics)
-
Card Callbacks:
- Require separate HTTP endpoint (Flask/FastAPI/etc.)
- Use
get_card_action_handler() to get handler
- Mount handler at configured callback URL
Dependencies
lark-oapi>=1.2.0
pywayne>=0.1.0 (for LarkBot)
cv2 (optional, for image processing)
Integration with LarkBot
LarkBotListener uses LarkBot internally. Access it via listener.bot:
listener = LarkBotListener(app_id="xxx", app_secret="xxx")
listener.bot.send_text_to_chat("oc_xxx", "Message")
listener.bot.reply_message("om_xxx", "text", {"text": "Reply"})
listener.bot.add_reaction("om_xxx", "THUMBSUP")
listener.bot.create_chat("New Chat", ["ou_a", "ou_b"])
Complete Minimal Example
from pywayne.lark_bot_listener import LarkBotListener
listener = LarkBotListener(
app_id="cli_xxxxxxxxxxxx",
app_secret="your_app_secret"
)
@listener.text_handler()
async def echo(text: str, chat_id: str):
listener.send_message(chat_id, f"You said: {text}")
listener.run()
