跳转至

im

feishu.im

forward_message

Python
forward_message(message_id: str, receive_id: str, receive_id_type: str | None = None, uuid: str = '', **kwargs)
Python
forward_message(message_id_list: list[str], receive_id: str, receive_id_type: str | None = None, uuid: str = '', **kwargs)
Python
forward_message(message_id: str | list[str], receive_id: str, receive_id_type: str | None = None, uuid: str = '', **kwargs) -> NestedDict

转发消息

当 message_id 为列表时，合并转发消息。 否则，转发消息。

Parameters:

Name Type Description Default

message_id

str | list[str]

消息 ID

 required

receive_id

str

接收者 ID

 required

receive_id_type

str | None

接收者 ID 类型。会根据receive_id来自动推断。

 None
飞书文档

转发消息

合并转发消息

功能 实现函数
转发消息 feishu.im.messages.forward_message
合并转发多条消息 feishu.im.messages.forward_message_list
Source code in feishu/im/messages.py
Python
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
def forward_message(
    message_id: str | list[str], receive_id: str, receive_id_type: str | None = None, uuid: str = "", **kwargs
) -> NestedDict:
    r"""
    转发消息

    当 `message_id` 为列表时，合并转发消息。
    否则，转发消息。

    Args:
        message_id: 消息 ID
        receive_id: 接收者 ID
        receive_id_type: 接收者 ID 类型。会根据`receive_id`来自动推断。

    飞书文档:
        [转发消息](https://open.feishu.cn/document/server-docs/im-v1/message/forward)

        [合并转发消息](https://open.feishu.cn/document/server-docs/im-v1/message/merge_forward)

    | 功能       | 实现函数                                        |
    |----------|---------------------------------------------|
    | 转发消息     | [feishu.im.messages.forward_message][]      |
    | 合并转发多条消息 | [feishu.im.messages.forward_message_list][] |
    """
    if isinstance(message_id, list):
        return forward_message_list(message_id, receive_id, receive_id_type, uuid, **kwargs)
    if receive_id_type is None:
        receive_id_type = infer_receive_id_type(receive_id)
    message = post(
        f"im/v1/messages/{message_id}/forward",
        {"receive_id": receive_id},
        {"receive_id_type": receive_id_type, "uuid": uuid},
        **kwargs,
    )
    message.data.body.content = convert_json_to_dict(message.data.body.content)
    return message

get_message

Python
get_message(message_id: str, file_key: str | None = None, file_type: str | None = None, **kwargs)

获取消息

当 file_key 不为空时，获取消息中的资源文件。 否则，获取消息内容。

Parameters:

Name Type Description Default

message_id

str

消息 ID

 required

file_key

str | None

文件 Key

 None

file_type

str | None

文件类型

 None
飞书文档

获取指定消息的内容

获取消息中的资源文件

功能 实现函数
获取消息内容 feishu.im.messages.get_message_content
获取消息中的资源文件 feishu.im.messages.get_message_resource
Source code in feishu/im/messages.py
Python
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
def get_message(message_id: str, file_key: str | None = None, file_type: str | None = None, **kwargs) -> NestedDict:
    r"""
    获取消息

    当 `file_key` 不为空时，获取消息中的资源文件。
    否则，获取消息内容。

    Args:
        message_id: 消息 ID
        file_key: 文件 Key
        file_type: 文件类型

    飞书文档:
        [获取指定消息的内容](https://open.feishu.cn/document/server-docs/im-v1/message/get)

        [获取消息中的资源文件](https://open.feishu.cn/document/server-docs/im-v1/message/get-2)

    | 功能         | 实现函数                                        |
    |------------|---------------------------------------------|
    | 获取消息内容     | [feishu.im.messages.get_message_content][]  |
    | 获取消息中的资源文件 | [feishu.im.messages.get_message_resource][] |
    """
    if file_key is not None:
        return get_message_resource(message_id, file_key, file_type, **kwargs)  # type: ignore[arg-type]
    return get_message_content(message_id, **kwargs)

get_messages

Python
get_messages(id: str, container_id_type: str = 'chat', max_num_messages: int | float = float('inf'), max_message_length: int | float = float('inf'), **kwargs)
Python
get_messages(id: str, container_id_type: str = 'chat', start_time: int | None = None, end_time: int | None = None, sort_type: str = 'ByCreateTimeAsc', page_size: int = 50, page_token: str | None = None, **kwargs)
Python
get_messages(id: str, container_id_type: str = 'chat', sort_type: str = 'ByCreateTimeDesc', max_num_messages: int | float = float('inf'), max_message_length: int | float = float('inf'), start_time: int | None = None, end_time: int | None = None, page_size: int = 50, page_token: str | None = None, **kwargs) -> NestedDict

获取消息

当 id 以 om_ 开头时，获取消息及其回复链中所有消息的内容。 当 id 以其他方式开头时，获取会话历史消息。

Parameters:

Name Type Description Default

id

str

消息 ID / 会话 ID

 required

container_id_type

str

容器 ID 类型。默认为 chat。 只用于 id 以其他方式开头时。

 'chat'

sort_type

str

排序方式。默认为 ByCreateTimeAsc。

 'ByCreateTimeDesc'

max_num_messages

int | float

最大历史消息数量。默认为正无穷。 只用于 id 以 om_ 开头时。

 float('inf')

max_message_length

int | float

最大历史消息长度。默认为正无穷。 只用于 id 以 om_ 开头时。

 float('inf')

start_time

int | None

开始时间。默认为 None。 只用于 id 以其他方式开头时。

 None

end_time

int | None

结束时间。默认为 None。 只用于 id 以其他方式开头时。

 None

page_size

int

每页数量。默认为 50。 只用于 id 以其他方式开头时。

 50

page_token

str | None

分页标识。默认为 None。 只用于 id 以其他方式开头时。

 None
飞书文档

获取指定消息的内容

获取会话历史消息

功能 实现函数
获取消息及其回复链中所有消息的内容 feishu.im.messages.get_messages_chain
获取会话历史消息 feishu.im.messages.get_messages_history
Source code in feishu/im/messages.py
Python
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
def get_messages(
    id: str,
    container_id_type: str = "chat",
    sort_type: str = "ByCreateTimeDesc",
    max_num_messages: int | float = float("inf"),
    max_message_length: int | float = float("inf"),
    start_time: int | None = None,
    end_time: int | None = None,
    page_size: int = 50,
    page_token: str | None = None,
    **kwargs,
) -> NestedDict:
    r"""
    获取消息

    当 `id` 以 `om_` 开头时，获取消息及其回复链中所有消息的内容。
    当 `id` 以其他方式开头时，获取会话历史消息。

    Args:
        id: 消息 ID / 会话 ID
        container_id_type: 容器 ID 类型。默认为 `chat`。
            只用于 `id` 以其他方式开头时。
        sort_type: 排序方式。默认为 `ByCreateTimeAsc`。
        max_num_messages: 最大历史消息数量。默认为正无穷。
            只用于 `id` 以 `om_` 开头时。
        max_message_length: 最大历史消息长度。默认为正无穷。
            只用于 `id` 以 `om_` 开头时。
        start_time: 开始时间。默认为 `None`。
            只用于 `id` 以其他方式开头时。
        end_time: 结束时间。默认为 `None`。
            只用于 `id` 以其他方式开头时。
        page_size: 每页数量。默认为 50。
            只用于 `id` 以其他方式开头时。
        page_token: 分页标识。默认为 `None`。
            只用于 `id` 以其他方式开头时。

    飞书文档:
        [获取指定消息的内容](https://open.feishu.cn/document/server-docs/im-v1/message/get)

        [获取会话历史消息](https://open.feishu.cn/document/server-docs/im-v1/message/list)

    | 功能                | 实现函数                                        |
    |-------------------|---------------------------------------------|
    | 获取消息及其回复链中所有消息的内容 | [feishu.im.messages.get_messages_chain][]   |
    | 获取会话历史消息          | [feishu.im.messages.get_messages_history][] |
    """
    if id.startswith("om_"):
        return get_messages_chain(
            message_id=id,
            sort_type=sort_type,
            max_num_messages=max_num_messages,
            max_message_length=max_message_length,
            **kwargs,
        )
    return get_messages_history(
        container_id=id,
        container_id_type=container_id_type,
        sort_type=sort_type,
        start_time=start_time,
        end_time=end_time,
        page_size=page_size,
        page_token=page_token,
        **kwargs,
    )

push_follow_up

Python
push_follow_up(message_id: str, follow_ups: str | dict, **kwargs)

添加跟随气泡

Parameters:

Name Type Description Default

message_id

str

消息 ID

 required

follow_ups

str | dict

跟随气泡内容

 required
飞书文档

添加跟随气泡

Source code in feishu/im/messages.py
Python
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
def push_follow_up(message_id: str, follow_ups: str | dict, **kwargs) -> NestedDict:
    r"""
    添加跟随气泡

    Args:
        message_id: 消息 ID
        follow_ups: 跟随气泡内容

    飞书文档:
        [添加跟随气泡](https://open.feishu.cn/document/server-docs/im-v1/message/push_follow_up)
    """
    if isinstance(follow_ups, str):
        follow_ups = {
            "follow_ups": [
                {
                    "content": follow_ups,
                }
            ]
        }
    return post(f"im/v1/messages/{message_id}/push_follow_up", follow_ups, **kwargs)

read_users

Python
read_users(message_id: str, user_id_type: str = 'open_id', **kwargs)

查询消息已读信息

Parameters:

Name Type Description Default

message_id

str

消息 ID

 required

user_id_type

str

用户 ID 类型。默认为 open_id。 可以是 open_id 或 “union_id” 或 user_id。

 'open_id'
飞书文档

查询消息已读信息

Source code in feishu/im/messages.py
Python
596
597
598
599
600
601
602
603
604
605
606
607
608
609
@pagination
def read_users(message_id: str, user_id_type: str = "open_id", **kwargs) -> NestedDict:
    r"""
    查询消息已读信息

    Args:
        message_id: 消息 ID
        user_id_type: 用户 ID 类型。默认为 `open_id`。
            可以是 `open_id` 或 "union_id" 或 `user_id`。

    飞书文档:
        [查询消息已读信息](https://open.feishu.cn/document/server-docs/im-v1/message/read_users)
    """
    return get(f"im/v1/messages/{message_id}/read_users", {"user_id_type": user_id_type}, **kwargs)

recall_message

Python
recall_message(message_id: str, **kwargs)

撤回消息

Parameters:

Name Type Description Default

message_id

str

消息 ID

 required
飞书文档

撤回消息

Source code in feishu/im/messages.py
Python
308
309
310
311
312
313
314
315
316
317
318
def recall_message(message_id: str, **kwargs) -> NestedDict:
    r"""
    撤回消息

    Args:
        message_id: 消息 ID

    飞书文档:
        [撤回消息](https://open.feishu.cn/document/server-docs/im-v1/message/recall)
    """
    return delete(f"im/v1/messages/{message_id}", **kwargs)

send_message

Python
send_message(message: str, receive_id: str | None = None, receive_id_type: str | None = None, uuid: str = '', **kwargs)
Python
send_message(message: dict, receive_id: str | None = None, receive_id_type: str | None = None, uuid: str = '', **kwargs)
Python
send_message(message: Stream, receive_id: str, receive_id_type: str | None = None, uuid: str = '', **kwargs)
Python
send_message(message: str, message_id: str, **kwargs)
Python
send_message(message: dict, message_id: str, **kwargs)
Python
send_message(message: Stream, message_id: str, **kwargs)
Python
send_message(message: str | dict | Stream, id: str | None = None, receive_id_type: str | None = None, message_type: str | None = None, uuid: str = '', **kwargs) -> NestedDict

发送消息

当 id 以 om_ 开头时，回复消息。 否则，发送消息。

Parameters:

Name Type Description Default

message

str | dict | Stream

消息内容

 required

id

str | None

接收者 ID 或消息 ID

 None

receive_id_type

str | None

接收者 ID 类型。会根据receive_id来自动推断。

 None

message_type

str | None

消息类型。默认为 text。

 None

uuid

str

消息唯一标识，用于消息去重。

 ''
飞书文档

发送消息

回复消息

功能 实现函数
发送内容消息 feishu.im.messages.send_message_content
流式消息 feishu.im.messages.stream_message
回复消息 feishu.im.messages.reply_message
Source code in feishu/im/messages.py
Python
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
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
def send_message(
    message: str | dict | Stream,
    id: str | None = None,
    receive_id_type: str | None = None,
    message_type: str | None = None,
    uuid: str = "",
    **kwargs,
) -> NestedDict:
    r"""
    发送消息

    当 `id` 以 `om_` 开头时，回复消息。
    否则，发送消息。

    Args:
        message: 消息内容
        id: 接收者 ID 或消息 ID
        receive_id_type: 接收者 ID 类型。会根据`receive_id`来自动推断。
        message_type: 消息类型。默认为 `text`。
        uuid: 消息唯一标识，用于消息去重。

    飞书文档:
        [发送消息](https://open.feishu.cn/document/server-docs/im-v1/message/create)

        [回复消息](https://open.feishu.cn/document/server-docs/im-v1/message/reply)

    | 功能     | 实现函数                                        |
    |--------|---------------------------------------------|
    | 发送内容消息 | [feishu.im.messages.send_message_content][] |
    | 流式消息   | [feishu.im.messages.stream_message][]       |
    | 回复消息   | [feishu.im.messages.reply_message][]        |
    """
    if isinstance(message, str):
        message = {
            "content": json.dumps({"text": message}),
        }
    if id is None:
        id = message.pop("receive_id", kwargs.pop("receive_id", kwargs.pop("message_id")))
    if id is None:
        raise ValueError("Unable to identify receiver")
    if openai_available and isinstance(message, Stream):
        return stream_message(stream=message, receive_id=id, receive_id_type=receive_id_type, uuid=uuid, **kwargs)
    if id.startswith("om_"):
        return reply_message(message, id, uuid=uuid, **kwargs)
    return send_message_content(
        message=message,
        receive_id=id,
        receive_id_type=receive_id_type,
        message_type=message_type,
        uuid=uuid,
        **kwargs,
    )

update_message

Python
update_message(message: str | dict, message_id: str, **kwargs)

编辑消息

Parameters:

Name Type Description Default

message

str | dict

消息内容

 required

message_id

str

消息 ID

 required
飞书文档

编辑消息

Source code in feishu/im/messages.py
Python
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
def update_message(message: str | dict, message_id: str, **kwargs) -> NestedDict:
    r"""
    编辑消息

    Args:
        message: 消息内容
        message_id: 消息 ID

    飞书文档:
        [编辑消息](https://open.feishu.cn/document/server-docs/im-v1/message/update)
    """
    if isinstance(message, str):
        message = {
            "msg_type": "text",
            "content": json.dumps({"text": message}),
        }
    message = put(f"im/v1/messages/{message_id}", message, **kwargs)
    message.data.body.content = convert_json_to_dict(message.data.body.content)  # type: ignore[union-attr]
    return message

get_message_text

Python
get_message_text(message: NestedDict) -> str

读取消息中的文本

Source code in feishu/im/utils.py
Python
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
def get_message_text(message: NestedDict) -> str:
    r"""
    读取消息中的文本
    """
    message = message.get("event", message).get("message", message)
    message = message.get("data", message)
    content = message.get("body", message).content
    if not content:
        return ""
    if isinstance(content, str):
        content = NestedDict.from_jsons(content)
    text = ""
    if "text" in content:
        text = content["text"]
    elif "elements" in content:
        text = "\n\n".join([e["text"] for l in content["elements"] for e in l if "text" in e])  # noqa: E741
        if content.get("title"):
            text = f"## {content['title']}\n\n" + text
    if mentions := message.get("mentions"):
        for mention in mentions:
            text = text.replace(mention["key"], f"@{mention['name']}")
    return text

is_mentioned

Python
is_mentioned(message: NestedDict) -> bool

判断消息是否被@了

Source code in feishu/im/utils.py
Python
57
58
59
60
61
62
63
64
65
66
67
def is_mentioned(message: NestedDict) -> bool:
    r"""
    判断消息是否被@了
    """
    message = message.get("event", message).get("message", message)
    message = message.get("data", message)
    if not message or not message.get("message_id"):
        raise ValueError("Not a valid message")
    if mentions := message.get("mentions"):
        return any(_is_mentioned(mention) for mention in mentions)
    return False