跳转至

context

feishu.agent.context

工具的按轮上下文：让工具处理函数在不改变其参数签名的前提下访问飞书客户端与用户态客户端。

feishu.agent.loop.AgentEngine 在处理每条消息 / 卡片回调前，把一个 feishu.agent.context.ToolContext 设入一个 contextvars.ContextVar；工具处理函数（如 feishu.agent.toolkit 中的工厂所产出者）经 feishu.agent.context.current_tool_context 读取它。contextvars 的值会随 await 在同一任务内保持，并被 asyncio.to_thread（同步处理函数的执行方式）复制，因而读 / 写工具都能拿到正确的按轮上下文，而 feishu.agent.tools.ToolRegistry.dispatch 的签名保持不变。

ToolContext dataclass

一次工具调用的按轮上下文：租户客户端、触发事件、用户态 token 提供方与已解析的用户标识。

as_user 借助 user_tokens（feishu.auth 的用户态 token 提供方）解析「当前请求用户」的用户态飞书 客户端，用于需以用户身份执行的读写；缺少提供方或有效授权时返回 None，由工具自行决定降级或要求授权。

示例：

Python Console Session
1
2
3
>>> ctx = ToolContext(client="<tenant-client>")
>>> ctx.client
'<tenant-client>'
源代码位于： feishu/agent/context.py
Python
 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
 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
137
138
139
140
141
@dataclass
class ToolContext:
    r"""
    一次工具调用的按轮上下文：租户客户端、触发事件、用户态 token 提供方与已解析的用户标识。

    `as_user` 借助 `user_tokens`（[feishu.auth][] 的用户态 token 提供方）解析「当前请求用户」的用户态飞书
    客户端，用于需以用户身份执行的读写；缺少提供方或有效授权时返回 `None`，由工具自行决定降级或要求授权。

    Examples:
        >>> ctx = ToolContext(client="<tenant-client>")
        >>> ctx.client
        '<tenant-client>'
    """

    client: Any | None = None
    event: Any | None = None
    user_tokens: Any | None = None
    user: dict[str, Any] = field(default_factory=dict)
    authorize_url_builder: Any | None = None
    shared_files: Any | None = None  # a SharedFileResolver: the only path from a file_id to bytes
    payment_accounts: Any | None = None  # a PaymentAccountResolver: account_id handle -> account value
    timezone: str | Callable[..., Any] | None = None

    async def as_user(self) -> Any | None:
        r"""解析当前请求用户的用户态飞书客户端；无提供方 / 无授权时返回 `None`。"""
        if self.user_tokens is None:
            return None
        user = self.user or _user_from_event(self.event)
        if not user:
            return None
        return await self.user_tokens.as_user(user)

    async def has_user_auth(self, scopes: Sequence[str]) -> bool:
        r"""判断当前请求用户是否已有有效授权；提供方支持 scope 检查时要求覆盖 `scopes`。"""
        if not scopes:
            return True
        if self.user_tokens is None:
            return False
        user = self.user or _user_from_event(self.event)
        if not user:
            return False
        checker = getattr(self.user_tokens, "has_scopes", None)
        if checker is not None:
            result = checker(user, tuple(scopes))
            if inspect.isawaitable(result):
                result = await result
            return bool(result)
        return await self.as_user() is not None

    def requesting_user(self) -> dict[str, Any]:
        r"""
        返回发起本轮请求的用户标识（open_id / union_id / user_id 的子集）。

        优先使用显式传入的 `user`，否则从触发事件解析。用户主体类工具应据此把操作限定在「请求用户本人」，
        防止被越权指向他人（zero-trust / least-privilege）。
        """
        return self.user or _user_from_event(self.event)

    def authorize_url(self, scopes: Sequence[str] = ()) -> str | None:
        r"""
        为当前用户构造授权跳转 URL，供工具在缺少用户授权时回传给模型转述。

        URL 的构造（重定向地址、`state` 签名等）由产品侧注入的 `authorize_url_builder` 负责，SDK 不内置任何
        产品配置；未注入或无法解析用户时返回 `None`。

        Args:
            scopes: 本次操作所需的飞书权限范围。

        Returns:
            授权 URL，或在未配置 / 无用户身份时返回 `None`。
        """
        if self.authorize_url_builder is None:
            return None
        user = self.user or _user_from_event(self.event)
        if not user:
            return None
        return self.authorize_url_builder(user, tuple(scopes))

    async def current_timezone(self, default: str | None = None) -> str | None:
        r"""返回当前轮次的时区，优先使用飞书事件上下文，其次使用产品默认值。"""
        event_timezone = _timezone_from_event(self.event)
        if event_timezone:
            return event_timezone
        timezone = self.timezone
        if callable(timezone):
            try:
                timezone = timezone(self.event) if _accepts_positional_arguments(timezone, 1) else timezone()
                if inspect.isawaitable(timezone):
                    timezone = await timezone
            except Exception:
                logger.debug("tool context timezone resolver failed", exc_info=True)
                return default
        if timezone:
            return str(timezone)
        return default

as_user async

Python
as_user() -> Any | None

解析当前请求用户的用户态飞书客户端；无提供方 / 无授权时返回 None。

源代码位于： feishu/agent/context.py
Python
70
71
72
73
74
75
76
77
async def as_user(self) -> Any | None:
    r"""解析当前请求用户的用户态飞书客户端；无提供方 / 无授权时返回 `None`。"""
    if self.user_tokens is None:
        return None
    user = self.user or _user_from_event(self.event)
    if not user:
        return None
    return await self.user_tokens.as_user(user)

has_user_auth async

Python
has_user_auth(scopes: Sequence[str]) -> bool

判断当前请求用户是否已有有效授权；提供方支持 scope 检查时要求覆盖 scopes。

源代码位于： feishu/agent/context.py
Python
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
async def has_user_auth(self, scopes: Sequence[str]) -> bool:
    r"""判断当前请求用户是否已有有效授权；提供方支持 scope 检查时要求覆盖 `scopes`。"""
    if not scopes:
        return True
    if self.user_tokens is None:
        return False
    user = self.user or _user_from_event(self.event)
    if not user:
        return False
    checker = getattr(self.user_tokens, "has_scopes", None)
    if checker is not None:
        result = checker(user, tuple(scopes))
        if inspect.isawaitable(result):
            result = await result
        return bool(result)
    return await self.as_user() is not None

requesting_user

Python
requesting_user() -> dict[str, Any]

返回发起本轮请求的用户标识（open_id / union_id / user_id 的子集）。

优先使用显式传入的 user，否则从触发事件解析。用户主体类工具应据此把操作限定在「请求用户本人」， 防止被越权指向他人（zero-trust / least-privilege）。

源代码位于： feishu/agent/context.py
Python
 96
 97
 98
 99
100
101
102
103
def requesting_user(self) -> dict[str, Any]:
    r"""
    返回发起本轮请求的用户标识（open_id / union_id / user_id 的子集）。

    优先使用显式传入的 `user`，否则从触发事件解析。用户主体类工具应据此把操作限定在「请求用户本人」，
    防止被越权指向他人（zero-trust / least-privilege）。
    """
    return self.user or _user_from_event(self.event)

authorize_url

Python
authorize_url(scopes: Sequence[str] = ()) -> str | None

为当前用户构造授权跳转 URL，供工具在缺少用户授权时回传给模型转述。

URL 的构造（重定向地址、state 签名等）由产品侧注入的 authorize_url_builder 负责，SDK 不内置任何 产品配置；未注入或无法解析用户时返回 None。

参数：

名称 类型 描述 默认
scopes
Sequence[str]

本次操作所需的飞书权限范围。

 ()

返回：

类型 描述
str | None

授权 URL，或在未配置 / 无用户身份时返回 None。
源代码位于： feishu/agent/context.py
Python
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
def authorize_url(self, scopes: Sequence[str] = ()) -> str | None:
    r"""
    为当前用户构造授权跳转 URL，供工具在缺少用户授权时回传给模型转述。

    URL 的构造（重定向地址、`state` 签名等）由产品侧注入的 `authorize_url_builder` 负责，SDK 不内置任何
    产品配置；未注入或无法解析用户时返回 `None`。

    Args:
        scopes: 本次操作所需的飞书权限范围。

    Returns:
        授权 URL，或在未配置 / 无用户身份时返回 `None`。
    """
    if self.authorize_url_builder is None:
        return None
    user = self.user or _user_from_event(self.event)
    if not user:
        return None
    return self.authorize_url_builder(user, tuple(scopes))

current_timezone async

Python
current_timezone(default: str | None = None) -> str | None

返回当前轮次的时区，优先使用飞书事件上下文，其次使用产品默认值。

源代码位于： feishu/agent/context.py
Python
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
async def current_timezone(self, default: str | None = None) -> str | None:
    r"""返回当前轮次的时区，优先使用飞书事件上下文，其次使用产品默认值。"""
    event_timezone = _timezone_from_event(self.event)
    if event_timezone:
        return event_timezone
    timezone = self.timezone
    if callable(timezone):
        try:
            timezone = timezone(self.event) if _accepts_positional_arguments(timezone, 1) else timezone()
            if inspect.isawaitable(timezone):
                timezone = await timezone
        except Exception:
            logger.debug("tool context timezone resolver failed", exc_info=True)
            return default
    if timezone:
        return str(timezone)
    return default

current_tool_context

Python
current_tool_context() -> ToolContext

读取当前生效的 feishu.agent.context.ToolContext；未设置时返回一个空上下文。

返回：

类型 描述
ToolContext

当前任务内由 feishu.agent.context.use_tool_context 设入的上下文；未设置时返回 ToolContext()
ToolContext

（client 为 None），由工具自行处理缺省。

示例：

Python Console Session
1
2
>>> current_tool_context().client is None
True
源代码位于： feishu/agent/context.py
Python
147
148
149
150
151
152
153
154
155
156
157
158
159
def current_tool_context() -> ToolContext:
    r"""
    读取当前生效的 [feishu.agent.context.ToolContext][]；未设置时返回一个空上下文。

    Returns:
        当前任务内由 [feishu.agent.context.use_tool_context][] 设入的上下文；未设置时返回 `ToolContext()`
        （`client` 为 `None`），由工具自行处理缺省。

    Examples:
        >>> current_tool_context().client is None
        True
    """
    return _CURRENT.get() or ToolContext()

use_tool_context

Python
use_tool_context(context: ToolContext) -> Iterator[None]

在 with 作用域内将 context 设为当前的 feishu.agent.context.ToolContext，退出时恢复。

参数：

名称 类型 描述 默认

context

ToolContext

本轮生效的工具上下文。

 必需

示例：

Python Console Session
1
2
3
4
5
6
>>> ctx = ToolContext(client="c1")
>>> with use_tool_context(ctx):
...     current_tool_context().client
'c1'
>>> current_tool_context().client is None
True
源代码位于： feishu/agent/context.py
Python
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
@contextmanager
def use_tool_context(context: ToolContext) -> Iterator[None]:
    r"""
    在 `with` 作用域内将 `context` 设为当前的 [feishu.agent.context.ToolContext][]，退出时恢复。

    Args:
        context: 本轮生效的工具上下文。

    Examples:
        >>> ctx = ToolContext(client="c1")
        >>> with use_tool_context(ctx):
        ...     current_tool_context().client
        'c1'
        >>> current_tool_context().client is None
        True
    """
    token = _CURRENT.set(context)
    try:
        yield
    finally:
        _CURRENT.reset(token)