Skip to content

feishu

feishu

FeishuChannel — Feishu (Lark) adapter.

Classes

FeishuChannel 

FeishuChannel(
    app_id: str = "",
    *,
    app_secret: str = "",
    bus: Optional[EventBus] = None,
)

Bases: BaseChannel

Feishu (Lark) channel adapter.

PARAMETER DESCRIPTION
app_id

Feishu App ID. Falls back to FEISHU_APP_ID env var.

TYPE: str DEFAULT: ''

app_secret

Feishu App Secret. Falls back to FEISHU_APP_SECRET env var.

TYPE: str DEFAULT: ''

bus

Optional event bus for publishing channel events.

TYPE: Optional[EventBus] DEFAULT: None

Source code in src/diapason/channels/feishu.py 
37
38
39
40
41
42
43
44
45
46
47
48
def __init__(
    self,
    app_id: str = "",
    *,
    app_secret: str = "",
    bus: Optional[EventBus] = None,
) -> None:
    self._app_id = app_id or os.environ.get("FEISHU_APP_ID", "")
    self._app_secret = app_secret or os.environ.get("FEISHU_APP_SECRET", "")
    self._bus = bus
    self._handlers: List[ChannelHandler] = []
    self._status = ChannelStatus.DISCONNECTED
Methods:
connect 
connect() -> None

Mark as connected (send-only — no persistent connection).

Source code in src/diapason/channels/feishu.py 
52
53
54
55
56
57
58
def connect(self) -> None:
    """Mark as connected (send-only — no persistent connection)."""
    if not self._app_id or not self._app_secret:
        logger.warning("No Feishu app_id or app_secret configured")
        self._status = ChannelStatus.ERROR
        return
    self._status = ChannelStatus.CONNECTED
disconnect 
disconnect() -> None

Mark as disconnected.

Source code in src/diapason/channels/feishu.py 
60
61
62
def disconnect(self) -> None:
    """Mark as disconnected."""
    self._status = ChannelStatus.DISCONNECTED
send 
send(
    channel: str,
    content: str,
    *,
    conversation_id: str = "",
    metadata: Dict[str, Any] | None = None,
) -> bool

Send a message to a Feishu chat via the Open API.

Source code in src/diapason/channels/feishu.py 
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
def send(
    self,
    channel: str,
    content: str,
    *,
    conversation_id: str = "",
    metadata: Dict[str, Any] | None = None,
) -> bool:
    """Send a message to a Feishu chat via the Open API."""
    if not self._app_id or not self._app_secret:
        logger.warning("Cannot send: no Feishu credentials configured")
        return False

    try:
        import httpx

        # Obtain tenant_access_token
        token_url = (
            "https://open.feishu.cn/open-apis/auth/v3/tenant_access_token/internal"
        )
        token_resp = httpx.post(
            token_url,
            json={
                "app_id": self._app_id,
                "app_secret": self._app_secret,
            },
            timeout=10.0,
        )
        if token_resp.status_code >= 300:
            logger.warning(
                "Feishu token request returned status %d",
                token_resp.status_code,
            )
            return False
        tenant_token = token_resp.json().get("tenant_access_token", "")
        if not tenant_token:
            logger.warning("Feishu token response missing tenant_access_token")
            return False

        # Send message
        msg_url = (
            "https://open.feishu.cn/open-apis/im/v1/messages"
            "?receive_id_type=chat_id"
        )
        headers = {
            "Authorization": f"Bearer {tenant_token}",
            "Content-Type": "application/json",
        }
        payload: Dict[str, Any] = {
            "receive_id": channel,
            "msg_type": "text",
            "content": json.dumps({"text": content}),
        }

        resp = httpx.post(
            msg_url,
            json=payload,
            headers=headers,
            timeout=10.0,
        )
        if resp.status_code < 300:
            self._publish_sent(channel, content, conversation_id)
            return True
        logger.warning(
            "Feishu API returned status %d",
            resp.status_code,
        )
        return False
    except Exception:
        logger.debug("Feishu send failed", exc_info=True)
        return False
status 
status() -> ChannelStatus

Return the current connection status.

Source code in src/diapason/channels/feishu.py 
138
139
140
def status(self) -> ChannelStatus:
    """Return the current connection status."""
    return self._status
list_channels 
list_channels() -> List[str]

Return available channel identifiers.

Source code in src/diapason/channels/feishu.py 
142
143
144
def list_channels(self) -> List[str]:
    """Return available channel identifiers."""
    return ["feishu"]
on_message 
on_message(handler: ChannelHandler) -> None

Register a callback for incoming messages.

Source code in src/diapason/channels/feishu.py 
146
147
148
def on_message(self, handler: ChannelHandler) -> None:
    """Register a callback for incoming messages."""
    self._handlers.append(handler)