feishu-streaming-card 0.1.0

Feishu/Lark CardKit streaming-card helper — typewriter-style card updates for AI bots in a few lines.

feishu-streaming-card

Minimal helper for Feishu / Lark CardKit streaming cards — give your AI bot a native typewriter effect with a coloured final card swap-in, in ~10 lines.

import lark_oapi as lark
from feishu_streaming_card import StreamingCard

client = lark.Client.builder().app_id(APP_ID).app_secret(APP_SECRET).build()

async with await StreamingCard.open(client=client, chat_id=USER_OPEN_ID) as card:
    buffer = ""
    async for delta in your_llm.stream(...):
        buffer += delta
        await card.push(buffer)
# context-manager exit auto-swaps the streaming card for a coloured rich card

That's it. No need to deal with card_id, sequence, throttling, or the dual-call quirks of cardkit.v1.card.update vs cardkit.v1.card.settings.

---

Why this exists

Feishu's CardKit streaming gives you a beautiful typewriter effect in chat — but the SDK requires you to:

1. Create a card entity (cardkit.v1.card.create)
2. Send the entity as an interactive message (im.v1.message.create)
3. Push cumulative text to the card element on every delta, with a strictly monotonic sequence (cardkit.v1.card_element.content)
4. Throttle pushes under 10 QPS/card or get rate-limited
5. Call cardkit.v1.card.update (or .settings) at the end to leave streaming mode and swap in a richer final layout — otherwise the chat preview stays stuck on [生成中...] and the card can't be forwarded

This library does all five for you. You just push cumulative text and exit the context manager.

---

Install

pip install feishu-streaming-card

Requires Python 3.9+ and lark_oapi >= 1.4.

---

Setup checklist

In the Feishu open-platform admin console:

• Enable the bot capability
• Grant scopes:
  • im:message:send_as_bot — send cards as the bot
  • im:message — receive / reply to messages (only needed if you reply to incoming)
  • cardkit:card:write — required for streaming cards

Without cardkit:card:write, StreamingCard.open() raises StreamingCardError.

The Feishu client must be version 7.20+ — older clients only see the card title and a "please upgrade" placeholder.

---

Quick start

1) Plain text stream

import asyncio, os, lark_oapi as lark
from feishu_streaming_card import StreamingCard, StreamingCardConfig

async def main():
    client = (
        lark.Client.builder()
        .app_id(os.environ["FEISHU_APP_ID"])
        .app_secret(os.environ["FEISHU_APP_SECRET"])
        .build()
    )

    cfg = StreamingCardConfig(title="Bot", final_header_template="green")

    async with await StreamingCard.open(
        client=client,
        chat_id=os.environ["FEISHU_RECEIVE_ID"],   # open_id or chat_id
        receive_id_type="open_id",                  # or "chat_id"
        config=cfg,
    ) as card:
        # push cumulative — NOT the delta. The library handles throttling.
        buffer = ""
        for ch in "你好，我是一个流式 AI 卡片 demo":
            buffer += ch
            await card.push(buffer)
            await asyncio.sleep(0.03)

See examples/quickstart.py.

2) Drive directly from an async for LLM stream

from feishu_streaming_card import stream_into_card

card = await stream_into_card(
    client=feishu_client,
    chat_id=open_id,
    deltas=your_async_llm_stream(),
)
print(card.message_id)   # use this to thread follow-ups

See examples/with_openai.py for a real OpenAI integration.

---

Public API

Symbol	Type	Purpose
StreamingCard.open(...)	classmethod (async)	Create + deliver the card; returns ready-to-use instance
StreamingCard.push(text, *, force=False)	async method	Push cumulative text; throttled under min_push_interval_s
StreamingCard.close(*, final_text=None, final_card=None)	async method	Flush + swap in the rich final card; idempotent
StreamingCard.message_id / .card_id / .chat_id / .closed	properties	Inspect state
stream_into_card(...)	async function	One-shot helper: open → drive → close
StreamingCardConfig	dataclass	Tuning knobs (see below)
build_streaming_card_json(...)	function	Customise the streaming-phase card JSON
build_final_rich_card(...)	function	Customise the final swap-in card
StreamingCardError	exception	Raised when card creation / message send fails

Tuning (StreamingCardConfig)

Field	Default	Effect
element_id	"answer"	id of the markdown block updated mid-stream
title	"AI"	header title shown above streaming text
final_header_template	"blue"	colour template for the rich final card
streaming_summary	"[生成中...]"	chat-preview text while streaming
print_step	1	client animates this many chars per frame
print_frequency_ms	50	ms between client-side animation frames
print_strategy	"delay"	"delay" queues every char (no skips); "fast" flushes the queue on each push
min_push_interval_s	0.12	server-side throttle (CardKit caps at 10 pushes/s/card)
final_max_body_chars	28000	cap for the markdown body of the swap-in card

A typical LLM (~80 chars/s) feels best with the defaults. If your LLM bursts faster, raise print_step to 2–3 and lower print_frequency_ms to 30 so the client keeps up.

---

Custom card layouts

Pass initial_card= (streaming phase) or final_card= (swap-in) to use your own v2 card JSON:

async with await StreamingCard.open(
    client=client, chat_id=open_id,
    initial_card={
        "schema": "2.0",
        "header": {"title": {"tag": "plain_text", "content": "Custom"}},
        "config": {
            "streaming_mode": True,
            "summary": {"content": "thinking…"},
        },
        "body": {"elements": [
            {"tag": "markdown", "element_id": "answer", "content": ""},
        ]},
    },
) as card:
    ...
    await card.close(final_card={
        "schema": "2.0",
        "header": {"title": {"tag": "plain_text", "content": "Done"},
                   "template": "purple"},
        "config": {"streaming_mode": False, "enable_forward": True},
        "body": {"elements": [
            {"tag": "markdown", "content": final_text},
            {"tag": "action", "actions": [
                {"tag": "button", "text": {"tag": "plain_text", "content": "👍"},
                 "value": {"vote": "up"}},
                {"tag": "button", "text": {"tag": "plain_text", "content": "👎"},
                 "value": {"vote": "down"}},
            ]},
        ]},
    })

The streaming phase only allows a single markdown element — that's a Feishu limit, not ours. Buttons / panels / multiple sections must wait for the swap-in.

---

Limits & gotchas

• Single markdown element during streaming. CardKit rejects collapsible panels, action blocks, and additional elements while streaming_mode: true. Use final_card=... to add them after.
• message_id ↔ card_id are different things. message_id is the IM message; card_id is the CardKit entity. Use message_id to thread replies, card_id if you need to call CardKit APIs directly.
• A card entity can only be sent once. To deliver into another chat, open a new StreamingCard.
• Forwarding is disabled while streaming. close() calls cardkit.card.update which clears streaming_mode; only then does the user see a Forward button.
• push(text) takes cumulative text, not deltas. Same-prefix pushes drive the typewriter; different-prefix pushes re-render the whole markdown.
• CardKit auto-disables streaming after 10 minutes. Always call close() (or use the context manager) anyway — leftover [生成中...] previews degrade UX.

---

Troubleshooting

Symptom	Likely cause	Fix
StreamingCardError: ... code=99991663	App missing cardkit:card:write scope	Add scope, re-publish app version
code=230003 schema error	You passed v1 card JSON	All cards must have "schema": "2.0"
Card stays stuck on [生成中...]	close() never ran	Use the async with form, or wrap your loop in try/finally
Typewriter "bursts then pauses"	LLM faster than client render	Raise print_step to 2–3 and/or lower print_frequency_ms
code=230020 sequence error	Concurrent producers without serialisation	Library serialises automatically; only happens if you call internals directly

---

License

MIT — see LICENSE.

Author

Kai Sleung <[e-mail]>