| name | pywayne-lark-bot |
| description | Feishu/Lark Bot API wrapper for full-featured Feishu bot interactions. Use when users need to send messages (text, image, audio, file, rich_text, card, share), especially Markdown delivery via send_markdown_message_to_chat with card_v2/post routing, table fallback, and auto chunking; build or update schema 2.0 cards; send in-place streaming reply cards with reply_streaming_card, update_streaming_card, recolor_streaming_card, stream_reply_card, or astream_reply_card; manage files (upload/download); query user/group info; reply to messages; forward/recall/update messages; edit previously sent text/rich_text/card messages via edit_text_message, edit_post_message, edit_card_message; add reactions; pin messages; manage chats (create, delete, update, members, admins); get message history; batch send; handle read receipts and urgent notifications. |
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 handlingCardContentV2: Schema 2.0 card builderLarkBotListener: Event listener for incoming messages (separate skill)
Installation
pip install pywayne lark-oapi
Quick Start
from pywayne.lark_bot import LarkBot
bot = LarkBot(
app_id="cli_xxxxxxxxxxxx",
app_secret="your_app_secret"
)
bot.send_text_to_user("ou_xxxxxxxx", "Hello from LarkBot!")
bot.send_text_to_chat("oc_xxxxxxxx", "Hello, everyone!")
LarkBot Class
Constructor
bot = LarkBot(
app_id: str,
app_secret: str
)
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
at_all = TextContent.make_at_all_pattern()
at_user = TextContent.make_at_someone_pattern("ou_xxxx", "John", "open_id")
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")
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 = post.make_text_content("Text", styles=["bold", "underline", "lineThrough", "italic"])
link = post.make_link_content("Display text", "https://example.com")
at = post.make_at_content("ou_xxxx", styles=["bold"])
img = post.make_image_content("img_key")
media = post.make_media_content(file_key="file_xxx", image_key="thumb_xxx")
emoji = post.make_emoji_content("THUMBSUP")
hr = post.make_hr_content()
code = post.make_code_block_content(language="python", text='print("hello")')
md = post.make_markdown_content("**Bold** and *italic*")
Adding Content:
post.add_content_in_line(content_dict)
post.add_contents_in_line([content1, content2])
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 |
"""
post.add_markdown(
md_text,
table_as="code_block",
max_chunk_bytes=8000,
mono_max_col_width=40
)
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")
post = PostContent(title="Release Report")
post.add_content_in_new_line(
post.make_text_content("Version 1.2.0 Released", styles=["bold"])
)
post.add_contents_in_new_line([
post.make_at_content("ou_xxx"),
post.make_text_content(" "),
post.make_emoji_content("OK")
])
post.add_content_in_new_line(
post.make_link_content("View release notes", "https://example.com/release/1.2.0")
)
post.add_content_in_new_line(
post.make_code_block_content("bash", "deploy.sh --env prod --version 1.2.0")
)
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",
template="blue"
)
Methods:
card.add_markdown(md_text: str, *, max_chunk_bytes: int = 18_000)
card.add_hr()
card.add_image(img_key: str, *, size: str = "large", preview: bool = True)
templates = CardContentV2.list_header_templates()
card_json = card.get_card()
Common Header Templates:
bluewathetturquoisegreenyelloworangeredcarminevioletpurpleindigogrey
Example: Daily Report Card:
from pywayne.lark_bot import LarkBot, CardContentV2
bot = LarkBot(app_id="cli_xxx", app_secret="sec_xxx")
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")
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",
table_fallback: str = "code_block",
max_message_bytes: Optional[int] = None
) -> List[Dict]
Parameters:
chat_id: Target chat IDmd_text: Markdown contenttitle: Message titleprefer: 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",
table_fallback="code_block"
)
Example 3: Long Markdown Auto-Chunking:
long_md = "\n".join([f"## Section {i}\n\n" + "- " * 50 for i in range(50)])
responses = bot.send_markdown_message_to_chat(
"oc_xxx",
md_text=long_md,
title="Long Report",
prefer="card_v2",
max_message_bytes=10000
)
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
bot.send_text_to_user(user_open_id: str, text: str = '') -> Dict
bot.send_text_to_chat(chat_id: str, text: str = '') -> Dict
Examples:
bot.send_text_to_user("ou_xxx", "Hello!")
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
image_key = bot.upload_image(image_path: str) -> str
bot.send_image_to_user(user_open_id: str, image_key: str) -> Dict
bot.send_image_to_chat(chat_id: str, image_key: str) -> Dict
bot.download_image(image_key: str, image_save_path: str) -> None
Example:
image_key = bot.upload_image("/tmp/report.png")
if image_key:
bot.send_image_to_chat("oc_xxx", image_key)
Audio Messages
audio_key = bot.upload_file(file_path: str, file_type: str = "opus") -> str
bot.send_audio_to_user(user_open_id: str, file_key: str) -> Dict
bot.send_audio_to_chat(chat_id: str, file_key: str) -> Dict
Media Messages (Video)
video_key = bot.upload_file(file_path: str, file_type: str = "mp4") -> str
bot.send_media_to_user(user_open_id: str, file_key: str) -> Dict
bot.send_media_to_chat(chat_id: str, file_key: str) -> Dict
File Messages
file_key = bot.upload_file(
file_path: str,
file_type: str = 'stream'
) -> str
bot.send_file_to_user(user_open_id: str, file_key: str) -> Dict
bot.send_file_to_chat(chat_id: str, file_key: str) -> Dict
bot.download_file(file_key: str, file_save_path: str) -> None
Example:
pdf_key = bot.upload_file("/tmp/report.pdf", file_type="pdf")
bot.send_file_to_chat("oc_xxx", pdf_key)
bot.download_file(pdf_key, "/save/path/report.pdf")
Post Messages (Rich Text)
bot.send_rich_text_to_user(user_open_id: str, rich_text_content: Dict) -> Dict
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
bot.send_card_to_user(user_open_id: str, card: Dict) -> Dict
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"]
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
bot.share_chat_to_user(user_open_id: str, shared_chat_id: str) -> Dict
bot.share_chat_to_chat(chat_id: str, shared_chat_id: str) -> Dict
bot.share_user_to_user(user_open_id: str, shared_user_id: str) -> Dict
bot.share_user_to_chat(chat_id: str, shared_user_id: str) -> Dict
System Messages
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,
content: Union[str, Dict[str, Any], List[Any]],
*,
reply_in_thread: bool = False,
uuid: str = ""
) -> Dict
Examples:
bot.reply_message("om_xxx", "text", {"text": "Received your message"})
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())
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",
uuid: str = ""
) -> Dict
Example:
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:
msg = bot.send_text_to_chat("oc_xxx", "Processing...")
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,
end_time: str,
*,
sort_type: str = "",
page_size: int = 50,
page_token: str = ""
) -> Dict
Example:
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 messagesedit_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
card = CardContentV2(title="Task Status", template="blue")
card.add_markdown("⏳ Processing your request...")
msg = bot.send_card_to_chat("oc_xxx", card.get_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
) -> Dict
Available emoji codes (partial list):
THUMBSUP, THUMBSDOWNOK, HEART, HAHAWITTY, SURPRISED, FLUSHEDSPEECHLESS, TEARING, ANGRY
Note: Use Feishu's emoji codes, not Unicode characters. Full list: PostContent.list_emoji_types() opens documentation.
Example:
reaction = bot.add_reaction("om_xxx", "THUMBSUP")
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 = "",
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 = bot.reply_message("om_xxx", "text", {"text": "Official answer: ..."})
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,
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:
msg = bot.send_text_to_chat("oc_xxx", "🔴 Production database down!")
bot.urgent_message(
msg["message_id"],
urgent_type="phone",
user_open_ids=["ou_oncall_xxx"]
)
Read Receipt Events
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:
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
) -> Dict
Example:
bot.set_chat_admin("oc_xxx", ["ou_leader"], is_admin=True)
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]],
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,
save_path: str,
file_key: str = None
) -> bool
Example:
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,
save_dir: str
) -> Dict[str, str]
Example:
msg = bot.get_message("om_xxx")
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:
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:
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 IDsdepartment_ids: List of department IDsuser_ids: List of user IDsunion_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:
bot.batch_send_message(
"text",
content="System maintenance tonight at 23:00",
user_open_ids=["ou_a", "ou_b", "ou_c"]
)
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"
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"]
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())
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())
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")
reply_msg = bot.reply_message(
"om_user_question",
"text",
{"text": "Processing your request..."}
)
bot.add_reaction("om_user_question", "THUMBSUP")
answer = "The solution is to restart the service"
bot.edit_text_message(reply_msg["message_id"], f"✅ {answer}")
bot.pin_message(reply_msg["message_id"])
Example 3: Alert Forwarding with Urgency
alert_msg = bot.send_text_to_chat(
"oc_alerts",
"🔴 CRITICAL: Database connection timeout"
)
bot.forward_message(
alert_msg["message_id"],
"ou_oncall",
receive_id_type="open_id"
)
bot.urgent_message(
alert_msg["message_id"],
"app",
["ou_oncall"]
)
Example 4: Dynamic Chat Creation and Management
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"]
bot.set_chat_admin(chat_id, ["ou_engineer_a"], is_admin=True)
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())
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")
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
"""
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
status_msg = bot.send_text_to_chat("oc_xxx", "⏳ Processing...")
import time
time.sleep(3)
bot.recall_message(status_msg["message_id"])
bot.send_text_to_chat("oc_xxx", "✅ Processing complete")
Pattern: Reaction-Based Workflow
reaction = bot.add_reaction("om_xxx", "WITTY")
try:
result = process_request()
bot.reply_message("om_xxx", "text", {"text": f"Result: {result}"})
finally:
bot.delete_reaction("om_xxx", reaction["reaction_id"])
Pattern: Reply and Pin for Visibility
reply = bot.reply_message(
"om_question",
"text",
{"text": "Official policy: ..."}
)
bot.pin_message(reply["message_id"])
Pattern: Message with Read Receipt Check
msg = bot.send_text_to_chat("oc_xxx", "Please review the updated policy")
receipts = bot.get_message_read_users(msg["message_id"])
readers = [reader["user_id"] for reader in receipts.get("items", [])]
all_members = bot.get_chat_members("oc_xxx")
non_readers = [m["member_id"] for m in all_members if m["member_id"] not in readers]
for user_id in non_readers:
bot.send_text_to_user(user_id, "Reminder: please review the updated policy")
Dependencies
lark-oapi>=1.2.0
pywayne>=0.1.0 (for tools integration)
Important Notes
-
Message Content Encoding:
reply_message, edit_text_message, edit_post_message, edit_card_message automatically JSON-encode content- Text messages typically use
{"text": "message"}
- Interactive cards pass the card JSON directly
-
Message IDs:
- Always store
message_id from send responses for later operations
- Message IDs are required for reply, forward, recall, update, reactions, pins
-
Message Editing Limits:
- A single message can be edited at most 20 times
- Only messages sent by the current bot/app can be edited
- Recalled, deleted, or expired messages cannot be edited
edit_text_message / edit_post_message are only for text / post, i.e. text / rich_text updatesedit_card_message is for cards and should not be mixed with text/rich_text updates
-
Reaction Emoji Codes:
- Use Feishu's emoji type strings (e.g.,
"THUMBSUP"), not Unicode emojis
- Get full list:
PostContent.list_emoji_types() opens documentation
-
Batch Send Limitations:
- Cannot reply to or update batch-sent messages
- Only supports user/department targets
- Different from normal group messages
-
Chat Announcement:
set_chat_announcement uses Feishu's patch API format- Requires understanding of Feishu announcement patch operations
- Refer to Feishu OpenAPI documentation for patch request structure
-
send_markdown_message_to_chat Advantages:
- Automatic byte-based chunking for large content
- Table fallback for reliable rendering
- Dual routing (card_v2/rich_text) flexibility
- Recommended for all Markdown sending
Integration with LarkBotListener
For receiving and processing incoming messages, use LarkBotListener (separate skill):
from pywayne.lark_bot_listener import LarkBotListener
listener = LarkBotListener(app_id="cli_xxx", app_secret="sec_xxx")
bot = listener.bot
@listener.text_handler()
async def handle_text(text: str, chat_id: str, message_id: str):
listener.bot.reply_message(message_id, "text", {"text": f"Echo: {text}"})
listener.run()
See pywayne-lark-bot-listener skill for complete listener documentation.
