跳转至

robot

feishu.robot

handle_chat

Python
handle_chat(request: dict, max_num_messages: int | None = ..., max_message_length: int | None = ..., stop_words: set[str] | None = ..., stream: bool = ...) -> dict
Python
handle_chat(request: dict, max_num_messages: int | None = ..., max_message_length: int | None = ..., stream: bool = ...) -> dict
Python
handle_chat(request: NestedDict, max_num_messages: int | None = None, max_message_length: int | None = None, stop_words: set[str] | None = None, stream: bool = True) -> dict

回复飞书消息。

这个函数会根据消息的类型选择不同的回复方式。

对于私聊,会访问所有聊天记录,直到遇到停用词,或者达到最大消息数量。 对于群聊,会访问消息及其回复链中所有的消息,直到达到最大消息数量。

Parameters:

Name Type Description Default

request

NestedDict

飞书请求。

required

max_num_messages

int | None

最大消息数量。默认为 feishu.variables 中的 MAX_NUM_MESSAGES

None

max_message_length

int | None

最大消息长度。默认为 feishu.variables 中的 MAX_MESSAGE_LENGTH

None

stop_words

set[str] | None

停用词集合。默认为 feishu.variables 中的 STOP_WORDS

None

stream

bool

是否使用流式 API。默认为 True

True
功能 实现函数
回复私聊消息 feishu.robot.chat.handle_chat_history
回复群聊消息 feishu.robot.chat.handle_chat_chain
Source code in feishu/robot/chat.py
Python
def handle_chat(
    request: NestedDict,
    max_num_messages: int | None = None,
    max_message_length: int | None = None,
    stop_words: set[str] | None = None,
    stream: bool = True,
) -> dict:
    r"""
    回复飞书消息。

    这个函数会根据消息的类型选择不同的回复方式。

    对于私聊,会访问所有聊天记录,直到遇到停用词,或者达到最大消息数量。
    对于群聊,会访问消息及其回复链中所有的消息,直到达到最大消息数量。

    Args:
        request: 飞书请求。
        max_num_messages: 最大消息数量。默认为 [feishu.variables][] 中的 `MAX_NUM_MESSAGES`。
        max_message_length: 最大消息长度。默认为 [feishu.variables][] 中的 `MAX_MESSAGE_LENGTH`。
        stop_words: 停用词集合。默认为 [feishu.variables][] 中的 `STOP_WORDS`。
        stream: 是否使用流式 API。默认为 `True`。

    | 功能     | 实现函数                                      |
    |--------|-------------------------------------------|
    | 回复私聊消息 | [feishu.robot.chat.handle_chat_history][] |
    | 回复群聊消息 | [feishu.robot.chat.handle_chat_chain][]   |
    """
    if not isinstance(request, NestedDict):
        request = NestedDict(request)
    chat_type = request.event.message.chat_type
    if chat_type in {
        "p2p",
    }:
        return handle_chat_history(request, max_num_messages, max_message_length, stop_words, stream)
    if chat_type in {"group", "meeting", "private", "public"}:
        return handle_chat_chain(request, max_num_messages, max_message_length, stream)
    return {"status": "error", "message": "Unsupported chat type"}

handle_verification

Python
handle_verification(request: dict) -> dict

处理飞书URL验证请求

飞书文档

配置订阅方式

Source code in feishu/robot/verification.py
Python
def handle_verification(request: dict) -> dict:
    r"""
    处理飞书URL验证请求

    飞书文档:
        [配置订阅方式](https://open.feishu.cn/document/server-docs/event-subscription-guide/event-subscription-configure-/request-url-configuration-case)
    """
    if request.get("token") and variables.VERIFICATION_TOKEN and request["token"] != variables.VERIFICATION_TOKEN:
        raise ValueError("Invalid verification token")
    return {"challenge": request["challenge"]}