Skip to content

quickstart.md

Latest commit

 

History

History
194 lines (144 loc) · 4.83 KB

quickstart.md

File metadata and controls

194 lines (144 loc) · 4.83 KB
Raw

Channel Quickstart

This guide gets a minimal Channel echo bot running. For the full API surface, see the Channel reference.

Install

pip install lark-channel-sdk

Prepare the Bot

In the Feishu developer console:

  • Create a bot application.
  • Enable event subscriptions.
  • Enable the WebSocket event subscription channel for local development.
  • Subscribe to message receive events.
  • Grant bot message send/receive scopes such as im:message and im:message:send_as_bot.
  • Re-install the app into the tenant after changing scopes.

The SDK defaults to the Feishu OpenAPI domain. For Lark tenants, pass the Lark domain explicitly:

channel = FeishuChannel(
    app_id=os.environ["LARK_APP_ID"],
    app_secret=os.environ["LARK_APP_SECRET"],
    domain="https://open.larksuite.com",
)

Export credentials before running:

export LARK_APP_ID=cli_xxx
export LARK_APP_SECRET=your_app_secret

Run the Echo Bot

python samples/channel/echo_bot.py

The sample is intentionally small:

import asyncio
import os

from lark_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,
        {"text": f"echo: {msg.content_text}"},
    )

channel.on("message", on_message)

asyncio.run(channel.connect())

connect() starts the WebSocket transport and keeps the process running. Use await channel.disconnect() during graceful shutdown if your application owns the event loop.

In WebSocket mode, the SDK requests domain + "/callback/ws/endpoint" to get the server-provided WebSocket connection URL.

Register an error handler when you want centralized observability:

async def on_error(err):
    print("channel error:", err)

channel.on("error", on_error)

Send a Reply

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

channel.send(to, message, opts=None) accepts dict inputs, typed Outbound* dataclasses, or a bare markdown string.

Stream a Reply

async def produce(stream):
    for token in ["hello", " ", "world"]:
        await stream.append(token)

await channel.stream(
    msg.chat_id,
    {"markdown": produce},
    {"reply_to": msg.message_id},
)

For lower-level CardKit controls, see Streaming with CardKit.

Webhook Transport

For HTTP callbacks, construct the channel with transport="webhook" and pass each HTTP request to handle_webhook_request(headers, body).

pip install "lark-channel-sdk[aiohttp]"
from aiohttp import web
from lark_channel import FeishuChannel

channel = FeishuChannel(
    app_id="cli_xxx",
    app_secret="***",
    encrypt_key="...",
    verification_token="...",
    transport="webhook",
)

async def on_message(msg):
    await channel.send(msg.chat_id, {"text": f"echo: {msg.content_text}"})

channel.on("message", on_message)

async def webhook(request):
    status, body = await channel.handle_webhook_request(
        headers=dict(request.headers),
        body=await request.read(),
    )
    return web.Response(status=status, body=body, content_type="application/json")

async def init():
    app = web.Application()
    app.router.add_post("/feishu/webhook", webhook)
    await channel.connect_until_ready()
    return app

web.run_app(init())

The SDK does not ship a built-in HTTP server. Keep TLS termination, rate limiting, IP allowlisting, and anomaly tracking in your web framework or gateway layer. See Webhook server adapter.

Webhook routes are owned by your application. Register the full public callback URL for that route in the developer console; the SDK only processes the request after your web framework passes headers and body bytes to handle_webhook_request(...).

Security Mode

The default SecurityConfig uses compatibility mode so existing bots can migrate without behavior changes. For staging and production, start with audit mode and then move to strict mode after reviewing audit events:

from lark_channel import FeishuChannel, SecurityConfig

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

Strict mode enforces webhook signature checks for encrypted events, rejects remote insecure WebSocket endpoints by default, and hides detailed webhook/card error responses. See Security configuration.

Next Steps