DOCSEARCHbyAlgolia
Submit Docs

Feishu OpenPlatform Server SDK for Python

repository·v2_main·Indexed 19 days ago

https://github.com/larksuite/oapi-sdk-python

A Python wrapper for Feishu (Lark) server-side APIs that simplifies token management, request signing, encryption/decryption, and event dispatching. It provides typed request/response models and includes the FeishuChannel module for building conversational bots with support for WebSocket long connections, webhook lifecycles, streaming replies, and interactive cards. Requires Python 3.8 or later.

Tokens
24.7K
Snippets
62
Records
92
Agent score
67%

What's inside larksuite-oapi-sdk-python

  1. Understand the InboundMessage model

    v2_main

    When handling message events, you receive an InboundMessage object. Key fields include:

    • message_id / id: Feishu message id
    • create_time: Original event timestamp
    • conversation: Conversation(chat_id, chat_type, thread_id)
    • chat_id: Shortcut for conversation.chat_id
    • chat_type: p2p, group, topic, or unknown
    • sender: Identity for the sender
    • sender_id: Shortcut for sender.open_id
    • sender_name: Optional display name
    • mentions: List of Mention objects
    • mentioned_all: Whether the message mentioned all members
    • mentioned_bot: Whether the message mentioned the bot
    • reply_to_message_id: Parent message id when present
    • content: Typed MessageContent dataclass
    • content_text: Flattened markdown/XML-style text
    • resources: Resource descriptors for download
    • raw_content_type: Original Feishu message type
    • raw: Original event payload

  2. Use the FeishuChannel module for conversational bots

    v2_main

    The FeishuChannel class is a high-level entry point that encapsulates event listening, message normalization, security policies, outbound sending, media upload/download, card interaction, and streaming replies.

    Use FeishuChannel if you are developing conversational bots that need to handle normalized messages, replies, media, card callbacks, @ strategies, WebSocket long connections, or webhook lifecycles. If you only need raw event dispatching or direct OpenAPI calls, use WSClient, EventDispatcherHandler, or Client instead.

  3. Understand CardKit sequence semantics for streaming

    v2_main

    When using update_card_element_content(card_id, element_id, content, sequence), the sequence parameter must be a strictly increasing integer per card_id.

    • The first patch must have sequence >= 1.
    • Every subsequent patch must have a sequence greater than the previous one (e.g., 1, 3, 5 is valid, but 1, 1 is not).
    • finish_streaming_card(card_id, sequence) must also use a sequence number that exceeds the largest sequence used in any previous update call for that card.
    seq = 0

async def patch(text):
    nonlocal seq
    seq += 1
    await channel.update_card_element_content(card_id, "main", text, sequence=seq)

# ... stream tokens, calling patch() ...

seq += 1
await channel.finish_streaming_card(card_id, sequence=seq)

  4. Use ClientAssertion keyless mode

    v2_main

    If your self-built application uses an external service to provide a client_assertion, you can exchange it for a tenant token without configuring an app_secret. In this mode, the SDK does not generate, parse, or sign JWTs, nor does it store private keys. You must implement a provider that returns the assertion string.

    Note: This mode only supports self-built applications and does not support APIs that rely solely on AppAccessToken. If using a custom OpenAPI domain, you must also configure oauth_base_url(...) so the SDK can correctly generate the OAuth audience.

    import os

import lark_oapi as lark
from lark_oapi.core.client_assertion import ClientAssertionToken


class EnvClientAssertionProvider:
    def retrieve_token(self, aud: str) -> ClientAssertionToken:
        return ClientAssertionToken(os.environ["LARK_CLIENT_ASSERTION"])


client = lark.Client.builder() \
    .app_id(os.environ["LARK_APP_ID"]) \
    .client_assertion_provider(EnvClientAssertionProvider()) \
    .build()

  5. Difference between finish_streaming_card and update_card

    v2_main

    These methods serve different purposes and cannot be used interchangeably:

    1. finish_streaming_card(card_id, sequence): Used when a streaming session is complete. It sets config.streaming_mode = false on the preallocated card.
    2. update_card(message_id, card): Used for one-shot card replacement. It replaces the entire card payload of an existing message. It uses message_id (not card_id) and does not use a sequence.

    Important: If you need to update a card after you have called finish_streaming_card, you must use update_card with the message_id returned from the initial send_card_by_reference(...) call.

  6. Configure Markdown conversion modes in FeishuChannel

    v2_main

    You can control how markdown is converted into Feishu post messages by configuring the MarkdownConverter within the OutboundConfig of a FeishuChannel.

    There are two primary modes:

    • structured (default): Parses markdown into explicit post nodes (e.g., tag:text, tag:a, tag:code_block). Use this for predictable, deterministic rendering across different clients and for testable post ASTs. Note that some constructs like headings or blockquotes are flattened or approximated.
    • native: Wraps markdown into tag:md nodes, delegating rendering to the Feishu client. Use this when user-facing markdown structure (like headings and quotes) is more important than cross-client parity.

    To send plain text without any markdown conversion, do not rely on disabling the converter; instead, explicitly send a message using the {"text": "..."} format.

    from lark_oapi.channel import FeishuChannel, MarkdownConverter, OutboundConfig

channel = FeishuChannel(
    app_id="cli_xxx",
    app_secret="***",
    outbound=OutboundConfig(
        markdown_converter=MarkdownConverter(tag_md_mode="native"),
    ),
)

  7. How the Channel Webhook Server Adapter works

    v2_main

    The Channel SDK does not include a built-in HTTP server. Instead, it provides an asynchronous request entry point that you must integrate into your own web framework (like aiohttp or FastAPI).

    Core Logic

    Channel exposes the following async method to process incoming requests:

    status, body_bytes = await channel.handle_webhook_request(headers, body)

    When encrypt_key is configured, handle_webhook_request performs the following:

    • Decrypts the request body.
    • Validates the verification_token.
    • Verifies request signatures for non-challenge events.
    • Routes the event to handlers registered via channel.on(...).

    Note: If encrypt_key is not configured, the dispatcher treats the request as plaintext and does not verify signature headers.

    Initialization

    You must initialize the channel before processing any requests:

    • Async frameworks: Use await channel.connect_until_ready() during application startup.
    • Synchronous setup: Use channel.start().

    Warning: Calling handle_webhook_request(...) before startup will raise FeishuChannelError(code=not_connected).

  8. Understand the two-layer Channel dedup architecture

    v2_main

    The Channel SDK uses two distinct layers to prevent duplicate message processing:

    1. Pipeline layer: Uses a DedupStore within the InboundPipeline. This layer catches webhook retries and WebSocket reconnection backfills before full message normalization.
    2. Safety layer: Uses a SeenCache within the SafetyPipeline. This layer catches duplicate dispatches to user handlers and can optionally use a shared ICache for cross-worker coordination.

    Because these layers run at different stages of the pipeline, they use different protocols and interfaces.

  9. Use FeishuChannel for event-driven messaging

    v2_main

    The FeishuChannel class is the primary entry point for managing Feishu/Lark bot interactions. It handles WebSocket/webhook transport, message normalization, safety policies, deduplication, and outbound messaging.

    Minimal Implementation Example:

    import asyncio
import os
from lark_oapi.channel import FeishuChannel

channel = FeishuChannel(
    app_id=os.environ["LARK_APP_ID"],
    app_secret=os.environ["LARK_APP_SECRET"],
)

async def on_message(msg):
    # msg is an InboundMessage
    await channel.send(
        msg.chat_id,
        {"markdown": f"received: {msg.content_text}"},
        {"reply_to": msg.message_id},
    )

channel.on("message", on_message)

asyncio.run(channel.connect())
    import asyncio
import os

from lark_oapi.channel import FeishuChannel

channel = FeishuChannel(
    app_id=os.environ["LARK_APP_ID"],
    app_secret=os.environ["LARK_APP_SECRET"],
)

async def on_message(msg):
    await channel.send(
        msg.chat_id,
        {"markdown": f"received: {msg.content_text}"},
        {"reply_to": msg.message_id},
    )

channel.on("message", on_message)

asyncio.run(channel.connect())

  11. Migrate from lark_oapi.channel to lark-channel-sdk

    v2_main

    The Channel capabilities have migrated to a standalone package: lark-channel-sdk. The new import path is lark_channel.

    Important Notes:

    • The existing lark_oapi.channel module is kept for backward compatibility.
    • New Channel features will only be released to lark-channel-sdk.
    • Critical bug fixes for lark_oapi.channel will be evaluated for backporting, but the maintenance window ends on 2027-06-02.
    • Refer to the Migration Manual and SecurityConfig for details.

  12. Migrate from legacy Channel module to lark-channel-sdk

    v2_main

    The lark_oapi.channel module is a legacy entry point maintained for compatibility until 2027-06-02. New Channel features are developed in the lark-channel-sdk package.

    To migrate, install the new SDK:

    pip install lark-channel-sdk

    Then use the lark_channel import path. The new SDK's SecurityConfig defaults to compatibility mode to facilitate a smooth transition.

    from lark_channel import FeishuChannel