scheduled_policy

feishu.agent.scheduled_policy

Fail-closed tool access for scheduled executions.

ScheduledToolBlocked

Bases: PermissionError

A tool cannot run under the scheduled task’s policy.

源代码位于： feishu/agent/scheduled_policy.py

class ScheduledToolBlocked(PermissionError):
    """A tool cannot run under the scheduled task's policy."""

ScheduledContinuationNotReady

Bases: RuntimeError

The original scheduled turn has not finished persisting its wait.

源代码位于： feishu/agent/scheduled_policy.py

class ScheduledContinuationNotReady(RuntimeError):
    """The original scheduled turn has not finished persisting its wait."""

ScheduledDeliveryUnknown

Bases: RuntimeError

A scheduled card may have been delivered, but its outcome is unknown.

源代码位于： feishu/agent/scheduled_policy.py

class ScheduledDeliveryUnknown(RuntimeError):
    """A scheduled card may have been delivered, but its outcome is unknown."""

ScheduledPolicyRegistry

Expose only permitted tools and prevent direct execution of writes.

The wrapped registry is trusted configuration and must not be exposed to the model. confirm_writes presents non-read tools as requiring approval, and permits their dispatch only inside a validated, claimed approval or previously approved OAuth continuation.

源代码位于： feishu/agent/scheduled_policy.py

class ScheduledPolicyRegistry:
    """Expose only permitted tools and prevent direct execution of writes.

    The wrapped registry is trusted configuration and must not be exposed to
    the model. ``confirm_writes`` presents non-read tools as requiring approval,
    and permits their dispatch only inside a validated, claimed approval or
    previously approved OAuth continuation.
    """

    def __init__(
        self,
        registry: ToolRegistry,
        policy: Literal["read_only", "confirm_writes"],
        *,
        lease_validator: Callable[[], None] | None = None,
    ) -> None:
        if policy not in ("read_only", "confirm_writes"):
            raise ValueError("unsupported scheduled policy")
        self._registry = registry
        self.policy = policy
        self._lease_validator = lease_validator

    def specs(self) -> list[ToolSpec]:
        return [spec for spec in self._registry.specs() if self._allowed(spec.name)]

    def get(self, name: str) -> Tool:
        tool = self._registry.get(name)
        if self.policy == "read_only" and tool.scheduled_effect != "read":
            raise KeyError(name)
        if tool.scheduled_effect != "read" and not tool.requires_approval:
            return replace(tool, requires_approval=True)
        return tool

    async def dispatch(self, name: str, arguments: dict[str, Any]) -> ToolResult:
        if self._lease_validator is not None:
            self._lease_validator()
        tool = self._registry.get(name)
        if tool.scheduled_effect != "read" or tool.requires_approval:
            grant = _GRANT.get()
            users = set(user_identity_keys(current_tool_context().requesting_user()))
            if (
                (self.policy == "read_only" and tool.scheduled_effect != "read")
                or grant is None
                or not grant.ready
                or grant.used
                or grant.name != name
                or grant.digest != payload_sha256(arguments)
                or not (grant.owners & users)
            ):
                raise ScheduledToolBlocked(f"scheduled execution of {name!r} requires a trusted approval")
            grant.used = True
        return await self._registry.dispatch(name, arguments)

    def _allowed(self, name: str) -> bool:
        return self.policy == "confirm_writes" or self._registry.get(name).scheduled_effect == "read"

current_scheduled_registry

Python

current_scheduled_registry(registry: Any) -> Any

Return the registry scoped to this turn, without affecting other tasks.

源代码位于： feishu/agent/scheduled_policy.py

def current_scheduled_registry(registry: Any) -> Any:
    """Return the registry scoped to this turn, without affecting other tasks."""
    scoped = _SCOPED_REGISTRY.get()
    return scoped[1] if scoped is not None and scoped[0] is registry else registry

scheduled_registry_scope

Python

scheduled_registry_scope(registry: Any, policy: Literal['read_only', 'confirm_writes'], *, lease_validator: Callable[[], None] | None = None) -> Iterator[None]

Apply a scheduled policy only to calls made by the current async task.

源代码位于： feishu/agent/scheduled_policy.py

@contextmanager
def scheduled_registry_scope(
    registry: Any, policy: Literal["read_only", "confirm_writes"], *, lease_validator: Callable[[], None] | None = None
) -> Iterator[None]:
    """Apply a scheduled policy only to calls made by the current async task."""
    if isinstance(registry, ScheduledPolicyRegistry):
        if registry.policy != policy:
            raise ValueError("cannot replace an active scheduled policy")
        registry = registry._registry
    token = _SCOPED_REGISTRY.set((registry, ScheduledPolicyRegistry(registry, policy, lease_validator=lease_validator)))
    try:
        yield
    finally:
        _SCOPED_REGISTRY.reset(token)

scheduled_resume_lease async

Python

scheduled_resume_lease(agent: Any, pending: PendingApproval | PendingAuthorization) -> AsyncIterator[None]

Claim and renew the run while handling a persisted continuation.

源代码位于： feishu/agent/scheduled_policy.py

@asynccontextmanager
async def scheduled_resume_lease(agent: Any, pending: PendingApproval | PendingAuthorization) -> AsyncIterator[None]:
    """Claim and renew the run while handling a persisted continuation."""
    policy = _stored_policy(pending)
    if policy is None:
        yield
        return
    run_id = pending.extra["scheduled_run_id"]
    wait_id = pending.approval_id if isinstance(pending, PendingApproval) else pending.authorization_id
    expected = "awaiting_confirmation" if isinstance(pending, PendingApproval) else "awaiting_authorization"
    run = agent.scheduled_store.claim_resume(run_id, wait_id, expected_status=expected, now=agent._now())
    if not run.lease_token:
        raise RuntimeError("scheduled continuation did not receive a lease")
    owner_task = asyncio.current_task()

    async def heartbeat() -> None:
        try:
            while True:
                await asyncio.sleep(20)
                if not agent.scheduled_store.renew_lease(run_id, run.lease_token, now=agent._now()):
                    current = agent.scheduled_store.get_run(run_id)
                    if (
                        current is not None
                        and current.lease_token == run.lease_token
                        and current.status in ("succeeded", "failed", "awaiting_authorization", "awaiting_confirmation")
                    ):
                        return
                    if owner_task is not None:
                        owner_task.cancel()
                    return
        except Exception:
            if owner_task is not None:
                owner_task.cancel()

    renewal = asyncio.create_task(heartbeat())
    try:
        with scheduled_lease_scope(run_id, run.lease_token):
            yield
    finally:
        renewal.cancel()
        await asyncio.gather(renewal, return_exceptions=True)