跳转至

mail

feishu.agent.toolkit.mail

邮箱工具工厂:列邮件、读邮件、总结邮件、发邮件。详见 feishu.agent.toolkit

list_mail_messages

Python
list_mail_messages(*, description: str, name: str = 'list_mail_messages', auth_scopes: Sequence[str] = ()) -> Tool

读类工厂:列出当前用户邮箱中的邮件 ID,返回一个 feishu.agent.tools.Tool

工具始终以请求用户身份访问固定邮箱 "me",不会让模型指定任意 user_mailbox_id

参数:

名称 类型 描述 默认

description

str

工具描述(产品本地化文案)。

必需

name

str

工具名。默认为 "list_mail_messages"

'list_mail_messages'

auth_scopes

Sequence[str]

缺少授权时申请的飞书邮箱权限范围。

()

返回:

类型 描述
Tool

示例:

Python Console Session
1
2
3
>>> tool = list_mail_messages(description="列出邮件")
>>> tool.name, tool.requires_approval
('list_mail_messages', False)
源代码位于: feishu/agent/toolkit/mail.py
Python
def list_mail_messages(
    *,
    description: str,
    name: str = "list_mail_messages",
    auth_scopes: Sequence[str] = (),
) -> Tool:
    r"""
    读类工厂:列出当前用户邮箱中的邮件 ID,返回一个 [feishu.agent.tools.Tool][]。

    工具始终以请求用户身份访问固定邮箱 `"me"`,不会让模型指定任意 `user_mailbox_id`。

    Args:
        description: 工具描述(产品本地化文案)。
        name: 工具名。默认为 `"list_mail_messages"`。
        auth_scopes: 缺少授权时申请的飞书邮箱权限范围。

    Returns:
        可注册到 [feishu.agent.tools.ToolRegistry][] 的 [feishu.agent.tools.Tool][]。

    Examples:
        >>> tool = list_mail_messages(description="列出邮件")
        >>> tool.name, tool.requires_approval
        ('list_mail_messages', False)
    """

    async def handler(**arguments: Any) -> ToolResult:
        mail = await _user_mail_client(auth_scopes)
        if isinstance(mail, ToolResult):
            return mail
        messages = await mail.messages.list(
            _CURRENT_USER_MAILBOX_ID,
            page_size=arguments.get("page_size", 20),
            max_items=arguments.get("max_items"),
            folder_id=arguments.get("folder_id"),
            only_unread=arguments.get("only_unread"),
            label_id=arguments.get("label_id"),
        )
        return ToolResult(ToolOutcome.COMPLETED, content={"message_ids": messages})

    return Tool(
        name=name,
        description=description,
        input_schema={
            "type": "object",
            "properties": {
                "page_size": {"type": "integer", "description": "每页数量,飞书邮箱接口上限为 20。"},
                "max_items": {"type": "integer", "description": "最多返回多少封邮件;为空表示按接口翻完。"},
                "folder_id": {"type": "string", "description": "文件夹 ID,例如 INBOX。"},
                "only_unread": {"type": "boolean", "description": "是否只列未读邮件。"},
                "label_id": {"type": "string", "description": "标签 ID,例如 FLAGGED。"},
            },
            "additionalProperties": False,
        },
        handler=handler,
        auth_scopes=tuple(auth_scopes),
    )

search_mail_messages

Python
search_mail_messages(*, description: str, name: str = 'search_mail_messages', auth_scopes: Sequence[str] = ()) -> Tool

读类工厂:搜索当前用户邮箱中的邮件,返回一个 feishu.agent.tools.Tool

搜索范围固定为请求用户邮箱 "me";过滤条件直接透传给飞书邮箱 search 接口。

参数:

名称 类型 描述 默认

description

str

工具描述(产品本地化文案)。

必需

name

str

工具名。默认为 "search_mail_messages"

'search_mail_messages'

auth_scopes

Sequence[str]

缺少授权时申请的飞书邮箱权限范围。

()

返回:

类型 描述
Tool

示例:

Python Console Session
1
2
3
>>> tool = search_mail_messages(description="搜索邮件")
>>> tool.name
'search_mail_messages'
源代码位于: feishu/agent/toolkit/mail.py
Python
def search_mail_messages(
    *,
    description: str,
    name: str = "search_mail_messages",
    auth_scopes: Sequence[str] = (),
) -> Tool:
    r"""
    读类工厂:搜索当前用户邮箱中的邮件,返回一个 [feishu.agent.tools.Tool][]。

    搜索范围固定为请求用户邮箱 `"me"`;过滤条件直接透传给飞书邮箱 search 接口。

    Args:
        description: 工具描述(产品本地化文案)。
        name: 工具名。默认为 `"search_mail_messages"`。
        auth_scopes: 缺少授权时申请的飞书邮箱权限范围。

    Returns:
        可注册到 [feishu.agent.tools.ToolRegistry][] 的 [feishu.agent.tools.Tool][]。

    Examples:
        >>> tool = search_mail_messages(description="搜索邮件")
        >>> tool.name
        'search_mail_messages'
    """

    async def handler(**arguments: Any) -> ToolResult:
        mail = await _user_mail_client(auth_scopes)
        if isinstance(mail, ToolResult):
            return mail
        messages = await mail.messages.search(
            _CURRENT_USER_MAILBOX_ID,
            query=arguments.get("query"),
            filter=arguments.get("filter"),
            page_size=arguments.get("page_size", 15),
            max_items=arguments.get("max_items"),
        )
        return ToolResult(ToolOutcome.COMPLETED, content={"messages": messages})

    return Tool(
        name=name,
        description=description,
        input_schema={
            "type": "object",
            "properties": {
                "query": {"type": "string", "description": "搜索关键词,例如主题、正文或联系人关键字。"},
                "filter": {
                    "type": "object",
                    "description": "飞书邮箱搜索过滤条件,例如 from、to、folder、is_unread、create_time。",
                },
                "page_size": {"type": "integer", "description": "每页数量,飞书邮箱 search 接口上限为 15。"},
                "max_items": {"type": "integer", "description": "最多返回多少条搜索结果;为空表示按接口翻完。"},
            },
            "additionalProperties": False,
        },
        handler=handler,
        auth_scopes=tuple(auth_scopes),
    )

get_mail_message

Python
get_mail_message(*, description: str, name: str = 'get_mail_message', auth_scopes: Sequence[str] = ()) -> Tool

读类工厂:读取当前用户邮箱中的一封邮件详情,返回一个 feishu.agent.tools.Tool

参数:

名称 类型 描述 默认

description

str

工具描述(产品本地化文案)。

必需

name

str

工具名。默认为 "get_mail_message"

'get_mail_message'

auth_scopes

Sequence[str]

缺少授权时申请的飞书邮箱权限范围。

()

返回:

类型 描述
Tool

示例:

Python Console Session
1
2
3
>>> tool = get_mail_message(description="读取邮件")
>>> tool.name
'get_mail_message'
源代码位于: feishu/agent/toolkit/mail.py
Python
def get_mail_message(
    *,
    description: str,
    name: str = "get_mail_message",
    auth_scopes: Sequence[str] = (),
) -> Tool:
    r"""
    读类工厂:读取当前用户邮箱中的一封邮件详情,返回一个 [feishu.agent.tools.Tool][]。

    Args:
        description: 工具描述(产品本地化文案)。
        name: 工具名。默认为 `"get_mail_message"`。
        auth_scopes: 缺少授权时申请的飞书邮箱权限范围。

    Returns:
        可注册到 [feishu.agent.tools.ToolRegistry][] 的 [feishu.agent.tools.Tool][]。

    Examples:
        >>> tool = get_mail_message(description="读取邮件")
        >>> tool.name
        'get_mail_message'
    """

    async def handler(**arguments: Any) -> ToolResult:
        mail = await _user_mail_client(auth_scopes)
        if isinstance(mail, ToolResult):
            return mail
        message = await mail.messages.get(
            _CURRENT_USER_MAILBOX_ID,
            arguments["message_id"],
            format=arguments.get("format"),
        )
        return ToolResult(ToolOutcome.COMPLETED, content=message)

    return Tool(
        name=name,
        description=description,
        input_schema={
            "type": "object",
            "properties": {
                "message_id": {"type": "string", "description": "邮件 ID。"},
                "format": {
                    "type": "string",
                    "enum": ["full", "plain_text_full", "metadata"],
                    "description": "返回内容格式;不确定时可省略。",
                },
            },
            "required": ["message_id"],
            "additionalProperties": False,
        },
        handler=handler,
        auth_scopes=tuple(auth_scopes),
    )

summarize_mail_message

Python
summarize_mail_message(*, description: str, text_summarizer: Callable[..., Any] | None, name: str = 'summarize_mail_message', auth_scopes: Sequence[str] = (), max_body_chars: int = 4000, max_summary_chars: int = 2000) -> Tool

读类工厂:读取并总结当前用户邮箱中的一封邮件,返回一个 feishu.agent.tools.Tool

邮件全文只交给注入的快速文本摘要器,主模型只收到摘要和元数据。

参数:

名称 类型 描述 默认

description

str

工具描述(产品本地化文案)。

必需

text_summarizer

Callable[..., Any] | None

纯文本摘要器,通常由 fast/flash 模型提供。

必需

name

str

工具名。默认为 "summarize_mail_message"

'summarize_mail_message'

auth_scopes

Sequence[str]

缺少授权时申请的飞书邮箱权限范围。

()

max_body_chars

int

传给摘要器的单封邮件正文最大字符数。

4000

max_summary_chars

int

摘要最大字符数。

2000

返回:

类型 描述
Tool

示例:

Python Console Session
1
2
3
>>> tool = summarize_mail_message(description="总结邮件", text_summarizer=None)
>>> tool.name
'summarize_mail_message'
源代码位于: feishu/agent/toolkit/mail.py
Python
def summarize_mail_message(
    *,
    description: str,
    text_summarizer: Callable[..., Any] | None,
    name: str = "summarize_mail_message",
    auth_scopes: Sequence[str] = (),
    max_body_chars: int = 4000,
    max_summary_chars: int = 2000,
) -> Tool:
    r"""
    读类工厂:读取并总结当前用户邮箱中的一封邮件,返回一个 [feishu.agent.tools.Tool][]。

    邮件全文只交给注入的快速文本摘要器,主模型只收到摘要和元数据。

    Args:
        description: 工具描述(产品本地化文案)。
        text_summarizer: 纯文本摘要器,通常由 fast/flash 模型提供。
        name: 工具名。默认为 `"summarize_mail_message"`。
        auth_scopes: 缺少授权时申请的飞书邮箱权限范围。
        max_body_chars: 传给摘要器的单封邮件正文最大字符数。
        max_summary_chars: 摘要最大字符数。

    Returns:
        可注册到 [feishu.agent.tools.ToolRegistry][] 的 [feishu.agent.tools.Tool][]。

    Examples:
        >>> tool = summarize_mail_message(description="总结邮件", text_summarizer=None)
        >>> tool.name
        'summarize_mail_message'
    """

    async def handler(**arguments: Any) -> ToolResult:
        if text_summarizer is None:
            return _mail_summarizer_missing_result()
        message_id = arguments["message_id"]
        mail = await _user_mail_client(auth_scopes)
        if isinstance(mail, ToolResult):
            return mail
        messages, errors = await _read_mail_summary_items(
            mail,
            _CURRENT_USER_MAILBOX_ID,
            [message_id],
            max_body_chars=max_body_chars,
        )
        if not messages:
            return ToolResult(
                ToolOutcome.FAILED,
                content={
                    "summary": "邮件读取失败,暂时无法总结。",
                    "message": {"message_id": message_id},
                    "errors": errors,
                },
                is_error=True,
            )
        request = TextSummaryRequest(
            kind="mail_message",
            instruction="总结这封邮件。提炼主题、关键事实、待办、紧急程度和建议回复。不要输出原始全文。",
            text=_mail_digest_text(messages),
            max_chars=max_summary_chars,
        )
        summary = await _call_text_summarizer(text_summarizer, request)
        if not summary:
            return ToolResult(
                ToolOutcome.FAILED,
                content={
                    "summary": "邮件已读取,但 fast 文本摘要器暂时不可用。",
                    "message": _mail_public_item(messages[0]),
                    "errors": errors,
                },
                is_error=True,
            )
        return ToolResult(
            ToolOutcome.COMPLETED,
            content={"summary": summary, "message": _mail_public_item(messages[0]), "errors": errors},
        )

    return Tool(
        name=name,
        description=description,
        input_schema={
            "type": "object",
            "properties": {
                "message_id": {"type": "string", "description": "邮件 ID。"},
            },
            "required": ["message_id"],
            "additionalProperties": False,
        },
        handler=handler,
        auth_scopes=tuple(auth_scopes),
    )

summarize_mail_messages

Python
summarize_mail_messages(*, description: str, text_summarizer: Callable[..., Any] | None, name: str = 'summarize_mail_messages', auth_scopes: Sequence[str] = (), default_max_items: int = 10, max_items_limit: int = 20, max_body_chars: int = 4000, max_summary_chars: int = 2000) -> Tool

读类工厂:读取并总结当前用户邮箱中的多封邮件,返回一个 feishu.agent.tools.Tool

未显式给出 message_ids 时,工具会按筛选条件读取最近邮件再汇总。

参数:

名称 类型 描述 默认

description

str

工具描述(产品本地化文案)。

必需

text_summarizer

Callable[..., Any] | None

纯文本摘要器,通常由 fast/flash 模型提供。

必需

name

str

工具名。默认为 "summarize_mail_messages"

'summarize_mail_messages'

auth_scopes

Sequence[str]

缺少授权时申请的飞书邮箱权限范围。

()

default_max_items

int

未指定数量时默认总结的邮件数。

10

max_items_limit

int

单次可总结邮件数上限。

20

max_body_chars

int

传给摘要器的单封邮件正文最大字符数。

4000

max_summary_chars

int

摘要最大字符数。

2000

返回:

类型 描述
Tool

示例:

Python Console Session
1
2
3
>>> tool = summarize_mail_messages(description="总结最近邮件", text_summarizer=None)
>>> tool.name
'summarize_mail_messages'
源代码位于: feishu/agent/toolkit/mail.py
Python
def summarize_mail_messages(
    *,
    description: str,
    text_summarizer: Callable[..., Any] | None,
    name: str = "summarize_mail_messages",
    auth_scopes: Sequence[str] = (),
    default_max_items: int = 10,
    max_items_limit: int = 20,
    max_body_chars: int = 4000,
    max_summary_chars: int = 2000,
) -> Tool:
    r"""
    读类工厂:读取并总结当前用户邮箱中的多封邮件,返回一个 [feishu.agent.tools.Tool][]。

    未显式给出 `message_ids` 时,工具会按筛选条件读取最近邮件再汇总。

    Args:
        description: 工具描述(产品本地化文案)。
        text_summarizer: 纯文本摘要器,通常由 fast/flash 模型提供。
        name: 工具名。默认为 `"summarize_mail_messages"`。
        auth_scopes: 缺少授权时申请的飞书邮箱权限范围。
        default_max_items: 未指定数量时默认总结的邮件数。
        max_items_limit: 单次可总结邮件数上限。
        max_body_chars: 传给摘要器的单封邮件正文最大字符数。
        max_summary_chars: 摘要最大字符数。

    Returns:
        可注册到 [feishu.agent.tools.ToolRegistry][] 的 [feishu.agent.tools.Tool][]。

    Examples:
        >>> tool = summarize_mail_messages(description="总结最近邮件", text_summarizer=None)
        >>> tool.name
        'summarize_mail_messages'
    """

    async def handler(**arguments: Any) -> ToolResult:
        if text_summarizer is None:
            return _mail_summarizer_missing_result()
        limit = _bounded_count(arguments.get("max_items"), default=default_max_items, upper=max_items_limit)
        mail = await _user_mail_client(auth_scopes)
        if isinstance(mail, ToolResult):
            return mail
        raw_ids: Sequence[Any]
        message_ids = arguments.get("message_ids")
        if message_ids:
            raw_ids = message_ids
        else:
            raw_ids = await mail.messages.list(
                _CURRENT_USER_MAILBOX_ID,
                page_size=min(limit, 20),
                max_items=limit,
                folder_id=arguments.get("folder_id"),
                only_unread=arguments.get("only_unread"),
            )
        ids = []
        for item in raw_ids:
            message_id = _mail_message_id(item)
            if message_id:
                ids.append(message_id)
        ids = ids[:limit]
        if not ids:
            return ToolResult(
                ToolOutcome.COMPLETED,
                content={"summary": "没有找到符合条件的邮件。", "message_count": 0, "messages": []},
            )
        messages, errors = await _read_mail_summary_items(
            mail,
            _CURRENT_USER_MAILBOX_ID,
            ids,
            max_body_chars=max_body_chars,
        )
        if not messages:
            return ToolResult(
                ToolOutcome.FAILED,
                content={
                    "summary": "邮件读取失败,暂时无法总结。",
                    "message_count": 0,
                    "messages": [],
                    "errors": errors,
                },
                is_error=True,
            )
        request = TextSummaryRequest(
            kind="mail_digest",
            instruction=(
                "总结这些邮件。按重要性提炼:总体概览、每封邮件的一句话摘要、需要用户处理的待办、"
                "紧急程度和建议回复。不要输出原始全文。"
            ),
            text=_mail_digest_text(messages),
            max_chars=max_summary_chars,
        )
        summary = await _call_text_summarizer(text_summarizer, request)
        if not summary:
            return ToolResult(
                ToolOutcome.FAILED,
                content={
                    "summary": "邮件已读取,但 fast 文本摘要器暂时不可用。",
                    "message_count": len(messages),
                    "messages": [_mail_public_item(item) for item in messages],
                    "errors": errors,
                },
                is_error=True,
            )
        return ToolResult(
            ToolOutcome.COMPLETED,
            content={
                "summary": summary,
                "message_count": len(messages),
                "messages": [_mail_public_item(item) for item in messages],
                "errors": errors,
            },
        )

    return Tool(
        name=name,
        description=description,
        input_schema={
            "type": "object",
            "properties": {
                "message_ids": {
                    "type": "array",
                    "items": {"type": "string"},
                    "description": "要总结的邮件 ID 列表;为空时读取最近邮件。",
                },
                "max_items": {"type": "integer", "description": "要总结的邮件数量,默认 10,上限 20。"},
                "folder_id": {"type": "string", "description": "文件夹 ID,例如 INBOX。"},
                "only_unread": {"type": "boolean", "description": "是否只总结未读邮件。"},
            },
            "additionalProperties": False,
        },
        handler=handler,
        auth_scopes=tuple(auth_scopes),
    )

list_mail_folders

Python
list_mail_folders(*, description: str, name: str = 'list_mail_folders', auth_scopes: Sequence[str] = ()) -> Tool

读类工厂:列出当前用户邮箱的文件夹,返回一个 feishu.agent.tools.Tool

参数:

名称 类型 描述 默认

description

str

工具描述(产品本地化文案)。

必需

name

str

工具名。默认为 "list_mail_folders"

'list_mail_folders'

auth_scopes

Sequence[str]

缺少授权时申请的飞书邮箱权限范围。

()

返回:

类型 描述
Tool

示例:

Python Console Session
1
2
3
>>> tool = list_mail_folders(description="列出邮箱文件夹")
>>> tool.name
'list_mail_folders'
源代码位于: feishu/agent/toolkit/mail.py
Python
def list_mail_folders(
    *,
    description: str,
    name: str = "list_mail_folders",
    auth_scopes: Sequence[str] = (),
) -> Tool:
    r"""
    读类工厂:列出当前用户邮箱的文件夹,返回一个 [feishu.agent.tools.Tool][]。

    Args:
        description: 工具描述(产品本地化文案)。
        name: 工具名。默认为 `"list_mail_folders"`。
        auth_scopes: 缺少授权时申请的飞书邮箱权限范围。

    Returns:
        可注册到 [feishu.agent.tools.ToolRegistry][] 的 [feishu.agent.tools.Tool][]。

    Examples:
        >>> tool = list_mail_folders(description="列出邮箱文件夹")
        >>> tool.name
        'list_mail_folders'
    """

    async def handler(**arguments: Any) -> ToolResult:
        mail = await _user_mail_client(auth_scopes)
        if isinstance(mail, ToolResult):
            return mail
        folders = await mail.folders.list(_CURRENT_USER_MAILBOX_ID, folder_type=arguments.get("folder_type"))
        return ToolResult(ToolOutcome.COMPLETED, content={"folders": folders})

    return Tool(
        name=name,
        description=description,
        input_schema={
            "type": "object",
            "properties": {
                "folder_type": {"type": "integer", "description": "文件夹类型;为空时不筛选。"},
            },
            "additionalProperties": False,
        },
        handler=handler,
        auth_scopes=tuple(auth_scopes),
    )

send_mail_message

Python
send_mail_message(*, description: str, name: str = 'send_mail_message', requires_approval: bool = True, auth_scopes: Sequence[str] = ()) -> Tool

写类工厂:以当前用户身份发送一封邮件,返回一个需审批的 feishu.agent.tools.Tool

requires_approval=True 时,feishu.agent.loop.AgentEngine 会先发确认卡;用户批准后才执行发送。

参数:

名称 类型 描述 默认

description

str

工具描述(产品本地化文案)。

必需

name

str

工具名。默认为 "send_mail_message"

'send_mail_message'

requires_approval

bool

是否需用户审批后执行。默认为 True

True

auth_scopes

Sequence[str]

缺少授权时申请的飞书邮箱发送权限范围。

()

返回:

类型 描述
Tool

示例:

Python Console Session
1
2
3
>>> tool = send_mail_message(description="发送邮件")
>>> tool.name, tool.requires_approval
('send_mail_message', True)
源代码位于: feishu/agent/toolkit/mail.py
Python
def send_mail_message(
    *,
    description: str,
    name: str = "send_mail_message",
    requires_approval: bool = True,
    auth_scopes: Sequence[str] = (),
) -> Tool:
    r"""
    写类工厂:以当前用户身份发送一封邮件,返回一个需审批的 [feishu.agent.tools.Tool][]。

    `requires_approval=True` 时,[feishu.agent.loop.AgentEngine][] 会先发确认卡;用户批准后才执行发送。

    Args:
        description: 工具描述(产品本地化文案)。
        name: 工具名。默认为 `"send_mail_message"`。
        requires_approval: 是否需用户审批后执行。默认为 `True`。
        auth_scopes: 缺少授权时申请的飞书邮箱发送权限范围。

    Returns:
        可注册到 [feishu.agent.tools.ToolRegistry][] 的需审批 [feishu.agent.tools.Tool][]。

    Examples:
        >>> tool = send_mail_message(description="发送邮件")
        >>> tool.name, tool.requires_approval
        ('send_mail_message', True)
    """

    async def handler(**arguments: Any) -> ToolResult:
        to = arguments.get("to")
        subject = arguments.get("subject")
        raw = arguments.get("raw")
        if raw is None and (not to or subject is None):
            raise ToolValidationError("tool 'send_mail_message' requires 'raw' or both 'to' and 'subject'")
        mail = await _user_mail_client(auth_scopes)
        if isinstance(mail, ToolResult):
            return mail
        result = await mail.messages.send(
            _CURRENT_USER_MAILBOX_ID,
            subject=subject,
            to=to,
            raw=raw,
            cc=arguments.get("cc"),
            bcc=arguments.get("bcc"),
            body_plain_text=arguments.get("body_plain_text"),
            body_html=arguments.get("body_html"),
            attachments=arguments.get("attachments"),
            dedupe_key=arguments.get("dedupe_key"),
            head_from=arguments.get("head_from"),
        )
        return ToolResult(ToolOutcome.COMPLETED, content=result)

    return Tool(
        name=name,
        description=description,
        input_schema={
            "type": "object",
            "properties": {
                "to": {"type": "array", "items": _MAIL_ADDRESS_SCHEMA, "description": "收件人列表。"},
                "cc": {"type": "array", "items": _MAIL_ADDRESS_SCHEMA, "description": "抄送人列表。"},
                "bcc": {"type": "array", "items": _MAIL_ADDRESS_SCHEMA, "description": "密送人列表。"},
                "subject": {"type": "string", "description": "邮件主题。"},
                "raw": {"type": "string", "description": "base64url 编码后的 MIME/EML 原文。"},
                "body_plain_text": {"type": "string", "description": "纯文本正文。"},
                "body_html": {"type": "string", "description": "HTML 正文。"},
                "attachments": {"type": "array", "items": _MAIL_ATTACHMENT_SCHEMA, "description": "附件列表。"},
                "dedupe_key": {"type": "string", "description": "可选去重键。"},
                "head_from": {"type": "object", "description": "可选发件人显示信息。"},
            },
            "additionalProperties": False,
        },
        handler=handler,
        requires_approval=requires_approval,
        auth_scopes=tuple(auth_scopes),
    )