Pywayne Lark Bot

Pywayne Lark Bot - Full-Featured Feishu API Wrapper

Overview

LarkBot is a comprehensive Feishu (Lark) application bot wrapper that provides complete bidirectional interaction capabilities. It's designed for scenarios requiring full message lifecycle management, chat administration, and complex card-based interactions.

Key Capabilities:

• Send all message types (text, image, audio, video, file, rich_text, card)
• Reply, forward, recall, update messages
• Edit sent text/rich_text/card messages with semantic helper methods
• Build and update in-place streaming cards for long-running or LLM-style responses
• Reactions, pins, read receipts, urgent notifications
• Chat management (create, delete, update, members, admins, announcements)
• File upload/download with message resource handling
• User and group information queries
• Batch messaging to users/departments
• Recommended: send_markdown_message_to_chat with auto-chunking and table fallback

Companion Classes:

• TextContent: Quick text formatting (@mentions, bold, italic, links)
• PostContent: Rich text builder with Markdown table handling
• CardContentV2: Schema 2.0 card builder
• LarkBotListener: Event listener for incoming messages (separate skill)

Installation

pip install pywayne lark-oapi

Quick Start

from pywayne.lark_bot import LarkBot

# Initialize bot
bot = LarkBot(
    app_id="cli_xxxxxxxxxxxx",
    app_secret="your_app_secret"
)

# Send text to user
bot.send_text_to_user("ou_xxxxxxxx", "Hello from LarkBot!")

# Send text to chat group
bot.send_text_to_chat("oc_xxxxxxxx", "Hello, everyone!")

LarkBot Class

Constructor

bot = LarkBot(
    app_id: str,        # Feishu application ID
    app_secret: str     # Feishu application secret
)

Instance Attributes:

• client: Underlying lark.Client for advanced usage
• All methods return Dict with API response data

Helper Classes

TextContent - Quick Text Formatting

Static helper for creating formatted text patterns used in text messages.

Available Methods:

from pywayne.lark_bot import TextContent

# @mentions
at_all = TextContent.make_at_all_pattern()
at_user = TextContent.make_at_someone_pattern("ou_xxxx", "John", "open_id")

# Text styles
bold = TextContent.make_bold_pattern("Bold text")
italic = TextContent.make_italian_pattern("Italic text")
underline = TextContent.make_underline_pattern("Underlined text")
strikethrough = TextContent.make_delete_line_pattern("Strike text")

# Links
link = TextContent.make_url_pattern("https://example.com", "Click here")

Example: Formatted Notification:

from pywayne.lark_bot import LarkBot, TextContent

bot = LarkBot(app_id="cli_xxx", app_secret="sec_xxx")

message = (
    TextContent.make_at_someone_pattern("ou_xxxx", "Wayne", "open_id")
    + " "
    + TextContent.make_bold_pattern("Deployment completed")
    + " - "
    + TextContent.make_url_pattern("https://jenkins.example.com", "View build")
)

bot.send_text_to_chat("oc_xxxx", message)

PostContent - Rich Text Post Builder

Builder for complex structured rich text messages supporting text, links, @mentions, images, code blocks, and Markdown content.

Constructor:

from pywayne.lark_bot import PostContent

post = PostContent(title="Post Title")

Content Creation Methods:

# Text with optional styles
text = post.make_text_content("Text", styles=["bold", "underline", "lineThrough", "italic"])

# Hyperlink
link = post.make_link_content("Display text", "https://example.com")

# @mention
at = post.make_at_content("ou_xxxx", styles=["bold"])

# Image
img = post.make_image_content("img_key")

# Media (video/audio with thumbnail)
media = post.make_media_content(file_key="file_xxx", image_key="thumb_xxx")

# Emoji (Feishu emoji codes like "OK", "THUMBSUP", "HEART")
emoji = post.make_emoji_content("THUMBSUP")

# Horizontal rule
hr = post.make_hr_content()

# Code block
code = post.make_code_block_content(language="python", text='print("hello")')

# Markdown
md = post.make_markdown_content("**Bold** and *italic*")

Adding Content:

# Add to current line
post.add_content_in_line(content_dict)
post.add_contents_in_line([content1, content2])  # Multiple elements in same line

# Add to new line
post.add_content_in_new_line(content_dict)
post.add_contents_in_new_line([content1, content2])

Recommended: Add Markdown Directly:

md_text = """
## Section Title

- Item 1
- Item 2

| Column A | Column B |
| -------- | -------- |
| Data 1   | Data 2   |
"""

# Auto-chunk and handle tables
post.add_markdown(
    md_text,
    table_as="code_block",      # "code_block" or "md"
    max_chunk_bytes=8000,       # Max bytes per chunk
    mono_max_col_width=40       # Max column width for code_block mode
)

# Send
bot.send_rich_text_to_chat("oc_xxx", post.get_content())

Complete Example:

from pywayne.lark_bot import LarkBot, PostContent

bot = LarkBot(app_id="cli_xxx", app_secret="sec_xxx")

# Build post
post = PostContent(title="Release Report")

# Line 1: Title
post.add_content_in_new_line(
    post.make_text_content("Version 1.2.0 Released", styles=["bold"])
)

# Line 2: @mention with emoji
post.add_contents_in_new_line([
    post.make_at_content("ou_xxx"),
    post.make_text_content(" "),
    post.make_emoji_content("OK")
])

# Line 3: Link
post.add_content_in_new_line(
    post.make_link_content("View release notes", "https://example.com/release/1.2.0")
)

# Line 4: Code block
post.add_content_in_new_line(
    post.make_code_block_content("bash", "deploy.sh --env prod --version 1.2.0")
)

# Send
bot.send_rich_text_to_chat("oc_xxx", post.get_content())

CardContentV2 - Schema 2.0 Interactive Card Builder

Lightweight builder for Feishu schema 2.0 cards, ideal for announcements, reports, and status updates with Markdown content.

Constructor:

from pywayne.lark_bot import CardContentV2

card = CardContentV2(
    title="Card Title",      # Optional header title
    template="blue"          # Header color: "blue", "wathet", "turquoise", "green", "yellow", "orange", "red", "carmine", "violet", "purple", "indigo", "grey"
)

Methods:

# Add Markdown content (auto-chunks by bytes)
card.add_markdown(md_text: str, *, max_chunk_bytes: int = 18_000)

# Add horizontal divider
card.add_hr()

# Add image
card.add_image(img_key: str, *, size: str = "large", preview: bool = True)

# List commonly used header templates
templates = CardContentV2.list_header_templates()  # ["blue", "wathet", ...]

# Get complete card JSON
card_json = card.get_card()

Common Header Templates:

• blue
• wathet
• turquoise
• green
• yellow
• orange
• red
• carmine
• violet
• purple
• indigo
• grey

Example: Daily Report Card:

from pywayne.lark_bot import LarkBot, CardContentV2

bot = LarkBot(app_id="cli_xxx", app_secret="sec_xxx")

# Build card
card = CardContentV2(title="Daily Report", template="blue")

card.add_markdown("""
# Today's Progress

- ✅ API integration completed
- ✅ Fixed 3 critical bugs
- 🔄 Code review in progress
- 📝 Documentation updated
""")

card.add_hr()

card.add_markdown("**Next Steps**: Deploy to staging environment")

# Send
bot.send_card_to_chat("oc_xxx", card.get_card())

Core Messaging Methods

Recommended Entry Point: send_markdown_message_to_chat

The preferred high-level method for sending Markdown content with automatic chunking, table handling, and dual routing (card_v2/rich_text).

responses = bot.send_markdown_message_to_chat(
    chat_id: str,
    md_text: str,
    *,
    title: str = "",
    prefer: str = "card_v2",              # "card_v2" or "post"
    table_fallback: str = "code_block",   # "code_block" or "md" (for post route)
    max_message_bytes: Optional[int] = None
) -> List[Dict]

Parameters:

• chat_id: Target chat ID
• md_text: Markdown content
• title: Message title
• prefer: Route preference:
  • "card_v2" (default): Send as schema 2.0 card (supports most Markdown)
  • "post": Send as rich_text message (supports table fallback)
• table_fallback: How to render Markdown tables in rich_text route:
  • "code_block": Convert tables to fixed-width text blocks (stable, recommended)
  • "md": Keep tables as Markdown (may have layout issues)
• max_message_bytes: Per-message byte limit (defaults: 18k for card_v2, 8k for rich_text route)

Returns: List of API response dicts for all sent chunks

Example 1: Simple Markdown (Default card_v2):

md = """
# Deployment Complete

- API: v1.2.3
- Frontend: v2.4.5
- Database: migrated

✅ All services healthy
"""

bot.send_markdown_message_to_chat(
    "oc_xxx",
    md_text=md,
    title="Deployment Status"
)

Example 2: Markdown with Tables (Post route with fallback):

md = """
## Test Results

| Module   | Status | Coverage |
| -------- | ------ | -------- |
| Auth     | ✅     | 95%      |
| Payment  | ✅     | 87%      |
| API      | ⚠️     | 72%      |
"""

bot.send_markdown_message_to_chat(
    "oc_xxx",
    md_text=md,
    title="Test Report",
    prefer="post",                    # Use rich_text route for table support
    table_fallback="code_block"       # Convert table to fixed-width text
)

Example 3: Long Markdown Auto-Chunking:

# Very long markdown content
long_md = "\n".join([f"## Section {i}\n\n" + "- " * 50 for i in range(50)])

# Automatically split into multiple messages
responses = bot.send_markdown_message_to_chat(
    "oc_xxx",
    md_text=long_md,
    title="Long Report",
    prefer="card_v2",
    max_message_bytes=10000  # Custom chunk size
)

print(f"Sent {len(responses)} message chunks")

Why Use send_markdown_message_to_chat?

• Handles large content automatically
• Tables render reliably with fallback
• Single API for both card and rich_text routes
• No manual JSON construction
• Consistent chunking and encoding

Text Messages

# Send to user
bot.send_text_to_user(user_open_id: str, text: str = '') -> Dict

# Send to chat
bot.send_text_to_chat(chat_id: str, text: str = '') -> Dict

Examples:

# Simple text
bot.send_text_to_user("ou_xxx", "Hello!")

# With formatting (use TextContent helpers)
from pywayne.lark_bot import TextContent

msg = (
    TextContent.make_at_all_pattern() + " "
    + TextContent.make_bold_pattern("Important")
    + ": System maintenance tonight at 23:00"
)
bot.send_text_to_chat("oc_xxx", msg)

Image Messages

# Upload image
image_key = bot.upload_image(image_path: str) -> str

# Send to user
bot.send_image_to_user(user_open_id: str, image_key: str) -> Dict

# Send to chat
bot.send_image_to_chat(chat_id: str, image_key: str) -> Dict

# Download image
bot.download_image(image_key: str, image_save_path: str) -> None

Example:

# Upload and send
image_key = bot.upload_image("/tmp/report.png")
if image_key:
    bot.send_image_to_chat("oc_xxx", image_key)

Audio Messages

# Upload audio (typically .opus format)
audio_key = bot.upload_file(file_path: str, file_type: str = "opus") -> str

# Send to user
bot.send_audio_to_user(user_open_id: str, file_key: str) -> Dict

# Send to chat
bot.send_audio_to_chat(chat_id: str, file_key: str) -> Dict

Media Messages (Video)

# Upload video (typically .mp4 format)
video_key = bot.upload_file(file_path: str, file_type: str = "mp4") -> str

# Send to user
bot.send_media_to_user(user_open_id: str, file_key: str) -> Dict

# Send to chat
bot.send_media_to_chat(chat_id: str, file_key: str) -> Dict

File Messages

# Upload file
file_key = bot.upload_file(
    file_path: str,
    file_type: str = 'stream'  # 'stream', 'opus', 'mp4', 'pdf', 'doc', 'xls', 'ppt'
) -> str

# Send to user
bot.send_file_to_user(user_open_id: str, file_key: str) -> Dict

# Send to chat
bot.send_file_to_chat(chat_id: str, file_key: str) -> Dict

# Download file
bot.download_file(file_key: str, file_save_path: str) -> None

Example:

# Upload PDF and send
pdf_key = bot.upload_file("/tmp/report.pdf", file_type="pdf")
bot.send_file_to_chat("oc_xxx", pdf_key)

# Download file
bot.download_file(pdf_key, "/save/path/report.pdf")

Post Messages (Rich Text)

# Send to user
bot.send_rich_text_to_user(user_open_id: str, rich_text_content: Dict) -> Dict

# Send to chat
bot.send_rich_text_to_chat(chat_id: str, rich_text_content: Dict) -> Dict

Example (see PostContent section for builder usage):

from pywayne.lark_bot import PostContent

post = PostContent(title="Announcement")
post.add_markdown("**Important update**: System will be upgraded tonight")

bot.send_rich_text_to_chat("oc_xxx", post.get_content())

Interactive Card Messages

# Send to user
bot.send_card_to_user(user_open_id: str, card: Dict) -> Dict

# Send to chat
bot.send_card_to_chat(chat_id: str, card: Dict) -> Dict

Return Value:

• Both methods return a response Dict.
• When the send succeeds, the response includes the created message metadata, including message_id.
• Save that message_id if you plan to call edit_card_message(), pin_message(), or other message lifecycle methods later.

Example with Raw Card JSON:

card = {
    "header": {
        "title": {"content": "Approval Request", "tag": "plain_text"},
        "template": "red"
    },
    "elements": [
        {"tag": "markdown", "content": "**Ticket #1234** needs approval"},
        {
            "tag": "action",
            "actions": [
                {
                    "tag": "button",
                    "text": {"content": "Approve", "tag": "plain_text"},
                    "type": "primary",
                    "url": "https://example.com/approve/1234"
                }
            ]
        }
    ]
}

bot.send_card_to_chat("oc_xxx", card)

Example with CardContentV2 Builder:

from pywayne.lark_bot import CardContentV2

card = CardContentV2(title="Status Update", template="green")
card.add_markdown("All systems operational ✅")
card.add_hr()
card.add_image("img_xxx", size="large")

bot.send_card_to_chat("oc_xxx", card.get_card())

Example: Capture message_id for Later Update:

from pywayne.lark_bot import CardContentV2

card = CardContentV2(title="Deployment Status", template="blue")
card.add_markdown("⏳ Deployment started")

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

# ... perform the long-running task ...

done_card = CardContentV2(title="Deployment Status", template="green")
done_card.add_markdown("✅ Deployment completed successfully")

bot.edit_card_message(message_id, done_card.get_card())

Share Messages

# Share chat to user
bot.share_chat_to_user(user_open_id: str, shared_chat_id: str) -> Dict

# Share chat to chat
bot.share_chat_to_chat(chat_id: str, shared_chat_id: str) -> Dict

# Share user to user
bot.share_user_to_user(user_open_id: str, shared_user_id: str) -> Dict

# Share user to chat
bot.share_user_to_chat(chat_id: str, shared_user_id: str) -> Dict

System Messages

# Send system message to user (special divider-style message)
bot.send_system_message_to_user(user_open_id: str, system_msg_text: str) -> Dict

Message Lifecycle Management

Reply to Message

Reply to an existing message with quote/reference.

response = bot.reply_message(
    message_id: str,
    msg_type: str,                                    # "text", "image", "post", "interactive", etc.
    content: Union[str, Dict[str, Any], List[Any]],
    *,
    reply_in_thread: bool = False,                    # Reply in thread instead of main chat
    uuid: str = ""
) -> Dict

Examples:

# Reply with text
bot.reply_message("om_xxx", "text", {"text": "Received your message"})

# Reply with card
from pywayne.lark_bot import CardContentV2

card = CardContentV2(title="Processing")
card.add_markdown("Your request is being processed...")

bot.reply_message("om_xxx", "interactive", card.get_card())

# Reply in thread
bot.reply_message(
    "om_xxx",
    "text",
    {"text": "Thread reply"},
    reply_in_thread=True
)

Forward Message

Forward an existing message to another user or chat.

response = bot.forward_message(
    message_id: str,
    receive_id: str,
    *,
    receive_id_type: str = "chat_id",  # "chat_id" or "open_id"
    uuid: str = ""
) -> Dict

Example:

# Forward alert to on-call engineer
bot.forward_message(
    message_id="om_alert_xxx",
    receive_id="ou_engineer_xxx",
    receive_id_type="open_id"
)

Recall Message

Recall/delete a message sent by the bot.

response = bot.recall_message(message_id: str) -> Dict

Example:

# Send temporary status message
msg = bot.send_text_to_chat("oc_xxx", "Processing...")

# Later recall it
bot.recall_message(msg["message_id"])

Get Message

Retrieve details of a specific message.

response = bot.get_message(
    message_id: str,
    *,
    user_id_type: str = "open_id"
) -> Dict

Get Message List

Retrieve historical messages from a chat.

response = bot.get_message_list(
    chat_id: str,
    start_time: str,           # Unix timestamp in milliseconds
    end_time: str,             # Unix timestamp in milliseconds
    *,
    sort_type: str = "",       # "ByCreateTimeAsc" or "ByCreateTimeDesc"
    page_size: int = 50,
    page_token: str = ""
) -> Dict

Example:

# Get messages from last 24 hours
import time

end_time = str(int(time.time() * 1000))
start_time = str(int((time.time() - 86400) * 1000))

history = bot.get_message_list(
    chat_id="oc_xxx",
    start_time=start_time,
    end_time=end_time,
    sort_type="ByCreateTimeAsc"
)

for msg in history.get("items", []):
    print(msg["message_id"], msg["msg_type"])

Edit Text/Post Message

Edit a previously sent text or rich_text (post) message.

response = bot.edit_text_message(
    message_id: str,
    text: str
) -> Dict

response = bot.edit_post_message(
    message_id: str,
    post_content: Dict[str, Any]
) -> Dict

Edit Card Message

Update a card message in place.

response = bot.edit_card_message(
    message_id: str,
    card: Dict[str, Any]
) -> Dict

Important Limits:

• edit_text_message() / edit_post_message() are for text and post only, i.e. text and rich_text messages
• edit_card_message() is for cards
• A single message can be edited at most 20 times
• You can only edit messages sent by the current bot/app
• Recalled, deleted, or expired messages cannot be edited
• Text/rich_text and card update APIs are separate and must not be mixed

Example: Status Card Workflow:

from pywayne.lark_bot import CardContentV2

# Send initial "processing" card
card = CardContentV2(title="Task Status", template="blue")
card.add_markdown("⏳ Processing your request...")

msg = bot.send_card_to_chat("oc_xxx", card.get_card())

# ... perform task ...

# Update to "completed" card
completed_card = CardContentV2(title="Task Status", template="green")
completed_card.add_markdown("✅ Task completed successfully!")

bot.edit_card_message(msg["message_id"], completed_card.get_card())

In-Place Streaming Cards

Use these helpers when a reply should stay in one message while the content keeps growing, such as LLM output, multi-step jobs, or approval workflows.

card = bot.build_streaming_card(
    md_text: str,
    *,
    title: str = "",
    template: str = "blue",
    streaming: bool = True,
    status_text: str = "",
    max_chunk_bytes: int = 18_000
) -> Dict[str, Any]

reply = bot.reply_streaming_card(
    message_id: str,
    *,
    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 = bot.update_streaming_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 = bot.recolor_streaming_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 = bot.stream_reply_card(
    source_message_id: str,
    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 bot.astream_reply_card(
    source_message_id: str,
    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]

Important Behavior:

• update_streaming_card() expects the full current Markdown text, not only the newest delta chunk.
• stream_reply_card() and astream_reply_card() coerce each chunk to text, so str, bytes, and other printable values can all be streamed.
• Card updates are rate-limited by Feishu. Keep update_interval above zero and lower it only when the UX benefit is worth the extra traffic.
• final_template="green" is the easiest way to turn a running blue card into a completed green card automatically.
• CardContentV2.list_header_templates() gives you the built-in common template names when you want to switch status colors safely.

Example 1: Preview a Streaming Card Before Sending:

card = bot.build_streaming_card(
    md_text="Step 1 complete\nStep 2 running",
    title="Migration Progress",
    template="orange",
    streaming=True,
    status_text="Waiting for final checks..."
)

bot.send_card_to_chat("oc_xxx", card)

Example 2: Manual Start, Multiple Updates, Final Recolor:

reply = bot.reply_streaming_card(
    "om_xxx",
    title="Incident Analysis",
    template="blue",
    initial_md="Collecting logs...",
    status_text="Working..."
)

card_message_id = reply["message_id"]

bot.update_streaming_card(
    card_message_id,
    "Collecting logs...\n\n- API logs loaded\n- Worker logs loaded",
    title="Incident Analysis",
    template="blue"
)

bot.update_streaming_card(
    card_message_id,
    "Collecting logs...\n\n- API logs loaded\n- Worker logs loaded\n- Root cause isolated",
    title="Incident Analysis",
    template="blue",
    status_text="Preparing summary..."
)

bot.recolor_streaming_card(
    card_message_id,
    "## Incident Summary\n\n- Root cause: expired credential\n- Fix: rotated secret\n- Follow-up: add alerting",
    title="Incident Analysis",
    template="green",
    status_text="Completed"
)

Example 3: Synchronous Generator for Token-Style Output:

import time

def fake_stream():
    for chunk in ["Hello", ", ", "this ", "reply ", "streams ", "in place."]:
        time.sleep(0.2)
        yield chunk

result = bot.stream_reply_card(
    "om_xxx",
    fake_stream(),
    title="Assistant Reply",
    template="wathet",
    status_text="Generating...",
    final_status_text="Answer complete",
    final_template="green",
    update_interval=0.4
)

print(result["message_id"])
print(result["text"])

Example 4: Async Generator for LLM Streaming:

async def llm_stream():
    for chunk in ["## Findings\n", "- Issue reproduced\n", "- Fix validated\n"]:
        yield chunk

result = await bot.astream_reply_card(
    "om_xxx",
    llm_stream(),
    title="LLM Analysis",
    template="blue",
    status_text="Thinking...",
    final_status_text="Done",
    final_template="green"
)

Example 5: Fail Fast and Turn the Card Red:

reply = bot.reply_streaming_card(
    "om_xxx",
    title="Deployment Job",
    template="blue",
    initial_md="Starting deploy pipeline..."
)

card_message_id = reply["message_id"]
current_text = "Starting deploy pipeline...\n- Build passed\n- Smoke tests passed"

try:
    bot.update_streaming_card(
        card_message_id,
        current_text,
        title="Deployment Job",
        template="blue",
        status_text="Rolling out..."
    )
    raise RuntimeError("Canary health check failed")
except Exception as exc:
    bot.recolor_streaming_card(
        card_message_id,
        current_text + f"\n\n**Error**: {exc}",
        title="Deployment Job",
        template="red",
        status_text="Failed",
        done=True
    )

Reactions, Pins, and Urgency

Add Reaction

Add emoji reaction to a message.

response = bot.add_reaction(
    message_id: str,
    emoji_type: str  # Feishu emoji code: "THUMBSUP", "OK", "HEART", "HAHA", etc.
) -> Dict

Available emoji codes (partial list):

• THUMBSUP, THUMBSDOWN
• OK, HEART, HAHA
• WITTY, SURPRISED, FLUSHED
• SPEECHLESS, TEARING, ANGRY

Note: Use Feishu's emoji codes, not Unicode characters. Full list: PostContent.list_emoji_types() opens documentation.

Example:

# Add thumbs up
reaction = bot.add_reaction("om_xxx", "THUMBSUP")

# Store reaction_id for later removal
reaction_id = reaction["reaction_id"]

Delete Reaction

Remove a previously added reaction.

response = bot.delete_reaction(
    message_id: str,
    reaction_id: str
) -> Dict

List Reactions

Get all reactions on a message.

response = bot.list_reactions(
    message_id: str,
    *,
    reaction_type: str = "",       # Filter by emoji type
    user_id_type: str = "open_id",
    page_size: int = 50,
    page_token: str = ""
) -> Dict

Pin Message

Pin a message in chat.

response = bot.pin_message(message_id: str) -> Dict

Unpin Message

Unpin a message in chat.

response = bot.unpin_message(message_id: str) -> Dict

List Pinned Messages

Get all pinned messages in a chat.

response = bot.list_pinned_messages(
    chat_id: str,
    *,
    start_time: str = "",
    end_time: str = "",
    page_size: int = 50,
    page_token: str = ""
) -> Dict

Example: Pin Important Reply:

# Reply to user question
reply = bot.reply_message("om_xxx", "text", {"text": "Official answer: ..."})

# Pin the reply for visibility
bot.pin_message(reply["message_id"])

Get Message Read Users

Get list of users who read a message.

response = bot.get_message_read_users(
    message_id: str,
    *,
    user_id_type: str = "open_id",
    page_size: int = 50,
    page_token: str = ""
) -> Dict

Urgent Message

Send urgent notification for an existing message.

response = bot.urgent_message(
    message_id: str,
    urgent_type: str,              # "app", "phone", or "sms"
    user_open_ids: List[str],
    *,
    user_id_type: str = "open_id"
) -> Dict

Urgent Types:

• "app": In-app notification
• "phone": Phone call
• "sms": SMS text message

Example: Alert On-Call Engineer:

# Send alert message
msg = bot.send_text_to_chat("oc_xxx", "🔴 Production database down!")

# Urgent notify on-call engineer
bot.urgent_message(
    msg["message_id"],
    urgent_type="phone",
    user_open_ids=["ou_oncall_xxx"]
)

Read Receipt Events

# Get read receipt info
receipts = bot.get_message_read_users(message_id="om_xxx")
for reader in receipts.get("items", []):
    print(f"{reader['user_id']} read at {reader['read_time']}")

Chat Management

Create Chat

Create a new group chat.

response = bot.create_chat(
    name: str,
    user_open_ids: List[str],
    description: str = "",
    *,
    avatar: str = "",
    owner_open_id: str = "",
    bot_ids: Optional[List[str]] = None,
    set_bot_manager: bool = False,
    uuid: str = ""
) -> Dict

Example:

# Create project chat
chat = bot.create_chat(
    name="Project Alpha",
    user_open_ids=["ou_a", "ou_b", "ou_c"],
    description="Alpha project collaboration",
    owner_open_id="ou_a"
)

chat_id = chat["chat_id"]

Delete Chat

Delete a chat group.

response = bot.delete_chat(chat_id: str) -> Dict

Update Chat

Update chat information.

response = bot.update_chat(
    chat_id: str,
    *,
    name: str = "",
    description: str = "",
    avatar: str = "",
    owner_open_id: str = ""
) -> Dict

Example:

bot.update_chat(
    "oc_xxx",
    name="Project Alpha - Staging",
    description="Staging environment coordination"
)

Add Members to Chat

Add users to a chat group.

response = bot.add_members_to_chat(
    chat_id: str,
    user_open_ids: List[str],
    *,
    succeed_type: int = 0
) -> Dict

Remove Members from Chat

Remove users from a chat group.

response = bot.remove_members_from_chat(
    chat_id: str,
    user_open_ids: List[str]
) -> Dict

Set Chat Admin

Add or remove chat administrators.

response = bot.set_chat_admin(
    chat_id: str,
    user_open_ids: List[str],
    *,
    is_admin: bool = True  # True to add, False to remove
) -> Dict

Example:

# Add admin
bot.set_chat_admin("oc_xxx", ["ou_leader"], is_admin=True)

# Remove admin
bot.set_chat_admin("oc_xxx", ["ou_leader"], is_admin=False)

Transfer Chat Owner

Transfer chat ownership to another member.

response = bot.transfer_chat_owner(
    chat_id: str,
    new_owner_open_id: str
) -> Dict

Get Chat Announcement

Retrieve current chat announcement.

response = bot.get_chat_announcement(chat_id: str) -> Dict

Set Chat Announcement

Update chat announcement using Feishu's patch API.

response = bot.set_chat_announcement(
    chat_id: str,
    *,
    requests: Union[str, List[str]],  # Patch operations
    revision: str = ""
) -> Dict

Note: This uses Feishu's patch operation format. See Feishu documentation for announcement patch operations.

Message Resource Handling

Download Message Resource

Download a specific resource (image, file, audio, video) from a message.

success = bot.download_message_resource(
    message_id: str,
    resource_type: str,        # "image", "file", "audio", "video", "media"
    save_path: str,
    file_key: str = None       # Optional: specific resource key
) -> bool

Example:

# Download image from message
bot.download_message_resource(
    message_id="om_xxx",
    resource_type="image",
    save_path="/tmp/message_image.png",
    file_key="img_xxx"
)

Download All Message Resources

Download all resources embedded in a message.

resources = bot.download_message_resources(
    message_id: str,
    message_content: str,  # JSON string from message content
    save_dir: str
) -> Dict[str, str]  # Returns {resource_type: save_path}

Example:

# Get message first
msg = bot.get_message("om_xxx")

# Download all resources
resources = bot.download_message_resources(
    message_id="om_xxx",
    message_content=msg["body"]["content"],
    save_dir="/tmp/resources"
)

for resource_type, path in resources.items():
    print(f"Downloaded {resource_type}: {path}")

User and Group Queries

Get User Info

Query user information by email or mobile.

users = bot.get_user_info(
    emails: List[str],
    mobiles: List[str]
) -> Optional[Dict]

Example:

users = bot.get_user_info(
    emails=["alice@example.com"],
    mobiles=["13800138000"]
)

if users:
    for user in users:
        print(user["user_id"], user["name"])

Get Group List

Get list of all groups the bot is in.

groups = bot.get_group_list() -> List[Dict]

Get Group Chat ID by Name

Find chat IDs matching a group name.

chat_ids = bot.find_chat_ids_by_name(group_name: str) -> List[str]

Example:

# Find "Project Alpha" groups
chat_ids = bot.find_chat_ids_by_name("Project Alpha")

if chat_ids:
    bot.send_text_to_chat(chat_ids[0], "Hello, team!")

Get Members in Group

Get list of members in a chat group.

members = bot.get_chat_members(
    group_chat_id: str
) -> List[Dict]

Get Member Open ID by Name

Find member open IDs matching a name in a chat.

open_ids = bot.find_member_open_ids_by_name(
    group_chat_id: str,
    member_name: str
) -> List[str]

Example:

# Find member in group
chat_ids = bot.find_chat_ids_by_name("Project Alpha")
if chat_ids:
    member_ids = bot.find_member_open_ids_by_name(chat_ids[0], "Alice")
    if member_ids:
        bot.send_text_to_user(member_ids[0], "Hi Alice!")

Get Chat and User Name

Helper to get both chat name and user name in one call.

chat_name, user_name = bot.get_chat_and_user_name(
    chat_id: str,
    user_id: str
) -> Tuple[str, str]

Example:

chat_name, user_name = bot.get_chat_and_user_name("oc_xxx", "ou_xxx")
print(f"User {user_name} in chat {chat_name}")

Batch Messaging

Send messages to multiple users or departments at once.

response = bot.batch_send_message(
    msg_type: str,
    *,
    content: Optional[Union[str, Dict[str, Any], List[Any]]] = None,
    card: Optional[Dict[str, Any]] = None,
    user_open_ids: Optional[List[str]] = None,
    department_ids: Optional[List[str]] = None,
    user_ids: Optional[List[str]] = None,
    union_ids: Optional[List[str]] = None
) -> Dict

Parameters:

• msg_type: "text", "interactive", etc.
• content: Message content (for non-card types)
• card: Card content (for card type)
• Target lists (at least one required):
  • user_open_ids: List of user open IDs
  • department_ids: List of department IDs
  • user_ids: List of user IDs
  • union_ids: List of union IDs

Important Notes:

• Uses Feishu's /message/v4/batch_send/ endpoint
• Batch messages cannot be replied to or updated like normal messages
• Only supports user/department targets, not chat groups

Example: Send Notification to Multiple Users:

# Text to multiple users
bot.batch_send_message(
    "text",
    content="System maintenance tonight at 23:00",
    user_open_ids=["ou_a", "ou_b", "ou_c"]
)

# Card to department
from pywayne.lark_bot import CardContentV2

card = CardContentV2(title="Announcement", template="red")
card.add_markdown("**Important**: Please update your passwords")

bot.batch_send_message(
    "interactive",
    card=card.get_card(),
    department_ids=["od_engineering"]
)

Complete Usage Examples

Example 1: Comprehensive Release Workflow

from pywayne.lark_bot import LarkBot, CardContentV2

bot = LarkBot(app_id="cli_xxx", app_secret="sec_xxx")

chat_id = "oc_xxx"

# Step 1: Send initial status card
card = CardContentV2(title="Deployment Status", template="blue")
card.add_markdown("⏳ Deployment started...")

msg = bot.send_card_to_chat(chat_id, card.get_card())
message_id = msg["message_id"]

# Step 2: Update progress
import time
time.sleep(5)

progress_card = CardContentV2(title="Deployment Status", template="blue")
progress_card.add_markdown("📦 Building Docker images... 50%")

bot.edit_card_message(message_id, progress_card.get_card())

# Step 3: Update to completion
time.sleep(5)

done_card = CardContentV2(title="Deployment Status", template="green")
done_card.add_markdown("""
✅ Deployment completed successfully!

**Services Updated**:
- API: v1.2.3
- Frontend: v2.4.5
- Worker: v1.1.1

**Health Check**: All systems operational
""")

bot.edit_card_message(message_id, done_card.get_card())

# Step 4: Pin the final status
bot.pin_message(message_id)

Example 2: Interactive Support Ticket

from pywayne.lark_bot import LarkBot, TextContent

bot = LarkBot(app_id="cli_xxx", app_secret="sec_xxx")

# User asks question in chat
# (Triggered by LarkBotListener - see lark-bot-listener skill)

# Reply with acknowledgment
reply_msg = bot.reply_message(
    "om_user_question",
    "text",
    {"text": "Processing your request..."}
)

# Add reaction to original question
bot.add_reaction("om_user_question", "THUMBSUP")

# Process request...
answer = "The solution is to restart the service"

# Update reply with answer
bot.edit_text_message(reply_msg["message_id"], f"✅ {answer}")

# Pin the solution
bot.pin_message(reply_msg["message_id"])

# Remove processing reaction
# (reaction_id would be stored from add_reaction response)

Example 3: Alert Forwarding with Urgency

# Alert detected
alert_msg = bot.send_text_to_chat(
    "oc_alerts",
    "🔴 CRITICAL: Database connection timeout"
)

# Forward to on-call engineer
bot.forward_message(
    alert_msg["message_id"],
    "ou_oncall",
    receive_id_type="open_id"
)

# Send urgent notification
bot.urgent_message(
    alert_msg["message_id"],
    "app",
    ["ou_oncall"]
)

Example 4: Dynamic Chat Creation and Management

# Create incident response chat
incident_chat = bot.create_chat(
    name="Incident #1234 - DB Outage",
    user_open_ids=["ou_engineer_a", "ou_engineer_b", "ou_manager"],
    description="Emergency response for database outage",
    owner_open_id="ou_manager"
)

chat_id = incident_chat["chat_id"]

# Set admins
bot.set_chat_admin(chat_id, ["ou_engineer_a"], is_admin=True)

# Send initial briefing
from pywayne.lark_bot import CardContentV2

briefing = CardContentV2(title="Incident Briefing", template="red")
briefing.add_markdown("""
**Incident**: Database connection timeout
**Start Time**: 2026-03-12 14:35:00
**Impact**: Production API down
**Status**: Investigating

**Action Items**:
- Check database logs
- Review recent deployments
- Monitor connection pool
""")

bot.send_card_to_chat(chat_id, briefing.get_card())

# After incident resolved...
bot.update_chat(
    chat_id,
    name="[RESOLVED] Incident #1234 - DB Outage",
    description="Incident resolved - database connection restored"
)

Example 5: Using send_markdown_message_to_chat with Tables

from pywayne.lark_bot import LarkBot

bot = LarkBot(app_id="cli_xxx", app_secret="sec_xxx")

# Markdown report with table
report = """
# Weekly Test Report

## Summary
All critical tests passed this week.

## Results by Module

| Module      | Tests | Passed | Failed | Coverage |
| ----------- | ----- | ------ | ------ | -------- |
| Auth        | 145   | 145    | 0      | 95%      |
| Payment     | 89    | 87     | 2      | 87%      |
| API         | 234   | 230    | 4      | 92%      |
| Frontend    | 567   | 565    | 2      | 88%      |

## Next Steps
- Fix 8 failing tests
- Improve Payment module coverage
"""

# Send with post route and table fallback
bot.send_markdown_message_to_chat(
    "oc_xxx",
    md_text=report,
    title="Weekly Test Report",
    prefer="post",
    table_fallback="code_block"
)

Common Patterns

Pattern: Temporary Status Message

# Send status
status_msg = bot.send_text_to_chat("oc_xxx", "⏳ Processing...")

# Do work...
import time
time.sleep(3)

# Recall temporary message
bot.recall_message(status_msg["message_id"])

# Send final result
bot.send_text_to_chat("oc_xxx", "✅ Processing complete")

Pattern: Reaction-Based Workf

…(truncated)