lark-bridge 0.6.0

Unofficial Feishu/Lark Web SDK - WebSocket listener, message search, send, and drive operations

lark-bridge

Unofficial Feishu/Lark Web SDK using cookie-based authentication. Provides real-time message listening via WebSocket, message search/fetch, sending, and Drive operations.

⚠️ This library is reverse-engineered from Feishu's web client. It is not an official API and may break without notice.

Installation

pip install lark-bridge

Quick Start

import asyncio
from lark_bridge import LarkBridge

bridge = LarkBridge("your_cookie_string_here")

# For enterprise tenants with custom domain:
bridge = LarkBridge("your_cookie_string_here", domain="yourcompany.feishu.cn")

# If WebSocket connection fails, you may need to provide app_key explicitly:
bridge = LarkBridge("your_cookie_string_here", domain="yourcompany.feishu.cn", app_key="your_app_key")

How to Get app_key

The app_key is used for WebSocket authentication. A default value is built-in, but if Feishu updates it you'll need to extract a new one:

1. Open your Feishu domain in Chrome (e.g. https://yourcompany.feishu.cn/messenger/)
2. Open DevTools → Sources → find lark.xxxxxxxx.js (under feishucdn.com)
3. Search for sass:" in that JS file — the 32-char hex string is your app_key

Alternatively, from the browser console Network tab, look for WebSocket connections to msg-frontier.feishu.cn and find the access_key parameter derivation.

Listen to Messages

async def main():
    async for msg in bridge.listen(watch_chats=["chat_id"]):
        print(f"[{msg['chat_id']}] {msg['from_id']}: {msg['text']}")

asyncio.run(main())

Search History

# Simple: auto-paginate up to limit
result = await bridge.search_messages(
    chat_id="7052636707732193282",
    start_time=1716192000,
    end_time=1716278400,
    limit=50,
)
print(result["msg_ids"])

# Manual pagination:
page = await bridge.search_messages_page(chat_id="7052636707732193282")
while page["has_more"]:
    page = await bridge.search_messages_page(
        chat_id="7052636707732193282",
        page_token=page["page_token"],
    )
    print(page["msg_ids"])

Fetch Messages

messages = await bridge.fetch_messages(["msg_id_1", "msg_id_2"])
for msg in messages:
    print(msg["text"])
# Each message dict contains:
# msg_id, type, from_id, chat_id, chat_type, parent_id, root_id, create_time, text
# chat_type: 0=unknown, 1=P2P(private), 2=GROUP, 3=TOPIC_GROUP

Send Message

await bridge.send_message(
    chat_id="7052636707732193282",
    text="Hello!",
    reply_id="optional_msg_id",
    at_user_ids=["user_id"],
)

Drive Operations

# Create folder
folder = await bridge.create_folder("My Folder", parent_token="root_token")
# folder = {"token": "...", "url": "https://domain/drive/folder/..."}

# Upload file
result = await bridge.upload_file(folder["token"], "report.txt", b"file content")
# result = {"file_token": "...", "node_token": "...", "url": "https://domain/file/..."}

# List my space root folders
items = await bridge.list_my_space()

# List shared folders
items = await bridge.list_shared_folders()

# List children of a folder
items = await bridge.list_children("fldcnfByPeMzPxrgtS5Xl06YRid")
# Each item: {"token", "obj_token", "name", "type", "type_code", "url", "edit_time"}

# Download a file (use obj_token, not node token)
content = await bridge.download_file("boxcnAbcDef123")
if content:
    Path("report.pdf").write_bytes(content)

# Delete files/folders (use node token)
await bridge.delete_nodes(["nodcnXxx123"])

Export Document

# Export a docx/wiki page as markdown
content = await bridge.export_document(
    token="GSeJdHQbQo3b2jxg8CYc1Zqqnah",
    type="docx",              # "docx" | "sheet" | "doc"
    file_extension="markdown",  # "markdown" | "docx" | "pdf" | "xlsx" | "csv"
)
if content:
    Path("output.md").write_bytes(content)

Supported combinations:

• type="docx" (wiki/docx pages) → markdown, docx, pdf
• type="sheet" → xlsx only

Import Document

# Import a markdown file as an online docx
result = await bridge.import_document(
    folder_token="fldcnfByPeMzPxrgtS5Xl06YRid",
    file_name="report.md",
    file_content=Path("report.md").read_bytes(),
    file_extension="md",      # source format
    target_type="docx",       # target online doc type
)
# result = {"token": "...", "url": "https://domain/docx/..."}

MCP Server

Expose lark-bridge as an MCP server so AI clients (Claude Desktop, Cursor, Kiro, etc.) can call Feishu tools directly.

Install

pip install lark-bridge[mcp]

Run

FEISHU_COOKIE="your_cookie_string" lark-bridge-mcp

For enterprise tenants with a custom domain:

FEISHU_COOKIE="your_cookie_string" FEISHU_DOMAIN="yourcompany.feishu.cn" lark-bridge-mcp

If your cookie changes frequently, use a file instead of an environment variable. The server reads the file on every tool call, so updates take effect immediately without restarting:

FEISHU_COOKIE_FILE="/path/to/cookie.txt" lark-bridge-mcp

MCP Client Config

{
  "mcpServers": {
    "lark-bridge": {
      "command": "lark-bridge-mcp",
      "env": {
        "FEISHU_COOKIE": "your_cookie_string"
      }
    }
  }
}

Available tools: send_message, search_messages, fetch_messages, create_folder, upload_file.

Cookie Setup

1. Open Feishu Web in your browser and log in
2. Open DevTools (F12) → Application → Cookies
3. Copy the full cookie string (all key=value pairs joined by ; )
4. Pass it to LarkBridge("your_cookie_string")

Example cookie format:

cookie = "passport_web_did=YOUR_DID; session=YOUR_SESSION; _csrf_token=YOUR_CSRF_TOKEN"

bridge = LarkBridge(cookie)

Cookie typically stays valid as long as the WebSocket connection is maintained.

License

MIT