Permissions

Permission policies for agent backends. See the Permissions guide.

agentflow.backends.base

Backend protocols and permission policies.

A backend drives one underlying engine — an agent process (Kiro, Codex, Claude Code) or an LLM endpoint (Ollama, OpenAI-compatible). The graph engine depends only on :class:Backend; user code that needs agent- or LLM-specific features constructs the concrete subtype and gets the narrowed method.

This module imports only :mod:agentflow.events and :mod:agentflow.errors — never the graph engine. Backends are a leaf layer.

PermissionPolicy

Bases: Protocol

Decides how to answer an agent's permission request.

Source code in agentflow/backends/base.py

@runtime_checkable
class PermissionPolicy(Protocol):
    """Decides how to answer an agent's permission request."""

    async def decide(self, req: PermissionRequest) -> PermissionDecision: ...

AllowAll

Approve every request. Auto-approve — only for a trusted local sandbox.

Source code in agentflow/backends/base.py

class AllowAll:
    """Approve every request. Auto-approve — only for a trusted local sandbox."""

    async def decide(self, req: PermissionRequest) -> PermissionDecision:
        return Allow()

DenyAll

Reject every request.

Source code in agentflow/backends/base.py

class DenyAll:
    """Reject every request."""

    async def decide(self, req: PermissionRequest) -> PermissionDecision:
        return Deny(reason="denied by policy")

ToolAllowlist

Approve only requests whose tool is on the allowlist; deny the rest.

A pragmatic least-privilege default: name the tools an agent may use and everything else is refused. Matching is exact on the request's tool.

policy = ToolAllowlist({"read_file", "list_directory"})

fallback decides unlisted tools (default :class:DenyAll); pass :class:Interactive to ask a human for anything not pre-approved.

Source code in agentflow/backends/base.py

class ToolAllowlist:
    """Approve only requests whose tool is on the allowlist; deny the rest.

    A pragmatic least-privilege default: name the tools an agent may use and
    everything else is refused. Matching is exact on the request's ``tool``.

        policy = ToolAllowlist({"read_file", "list_directory"})

    ``fallback`` decides unlisted tools (default :class:`DenyAll`); pass
    :class:`Interactive` to ask a human for anything not pre-approved.
    """

    def __init__(self, allowed, *, fallback: PermissionPolicy | None = None) -> None:
        self.allowed = set(allowed)
        self.fallback = fallback or DenyAll()

    async def decide(self, req: PermissionRequest) -> PermissionDecision:
        if req.tool in self.allowed:
            return Allow()
        return await self.fallback.decide(req)

Interactive

Escalate a permission request to a human via an engine interrupt.

When resolved inside a running graph, this calls ctx.interrupt(req) on the node's :class:~agentflow.runtime.Context, which suspends the run and persists an interrupted checkpoint. The human inspects the request and calls app.resume(thread, value=...); the resumed value becomes this policy's :class:PermissionDecision. A plain :class:Allow/:class:Deny or a bare truthy/None value is accepted and normalized.

The node's context is found via :data:agentflow.runtime.current_context, which the runtime sets around each node call. Because a node and any backend generator it iterates share one task, the policy — invoked while the node consumes the backend's event stream — sees the right context. Outside a graph run (no current context) it falls back to raising :class:~agentflow.errors.InterruptError so misuse fails loudly.

Source code in agentflow/backends/base.py

class Interactive:
    """Escalate a permission request to a human via an engine interrupt.

    When resolved inside a running graph, this calls ``ctx.interrupt(req)`` on
    the node's :class:`~agentflow.runtime.Context`, which suspends the run and
    persists an interrupted checkpoint. The human inspects the request and
    calls ``app.resume(thread, value=...)``; the resumed value becomes this
    policy's :class:`PermissionDecision`. A plain :class:`Allow`/:class:`Deny`
    or a bare truthy/``None`` value is accepted and normalized.

    The node's context is found via :data:`agentflow.runtime.current_context`,
    which the runtime sets around each node call. Because a node and any
    backend generator it iterates share one task, the policy — invoked while
    the node consumes the backend's event stream — sees the right context.
    Outside a graph run (no current context) it falls back to raising
    :class:`~agentflow.errors.InterruptError` so misuse fails loudly.
    """

    async def decide(self, req: PermissionRequest) -> PermissionDecision:
        # Imported lazily to keep the backend layer importable without the
        # engine, and to avoid any import-order sensitivity.
        from agentflow.runtime import current_context

        ctx = current_context.get()
        if ctx is None:
            raise InterruptError("<permission>", req)

        answer = await ctx.interrupt(req)
        return _normalize_decision(answer)

Callback

Ask a human (or any resolver) inline, without interrupting the graph.

Unlike :class:Interactive — which suspends the whole run via ctx.interrupt and requires a separate resume() — this awaits a caller-supplied async function within the same turn and answers the agent's permission request immediately. That matters for a persistent session backend (Kiro's ACP): the agent is holding one turn open waiting for the decision, so the answer must go back in that turn rather than abandoning it and starting a new one.

resolver is async (PermissionRequest) -> decision where the decision is an :class:Allow/:class:Deny, or a value normalized the same way as a resume value ("y"/"n", None, an option id, ...).

async def ask(req):
    answer = await asyncio.to_thread(input, f"allow {req.tool}? [y/N] ")
    return answer

backend = KiroBackend("vibe", permission=Callback(ask))

Source code in agentflow/backends/base.py

class Callback:
    """Ask a human (or any resolver) inline, without interrupting the graph.

    Unlike :class:`Interactive` — which suspends the whole run via
    ``ctx.interrupt`` and requires a separate ``resume()`` — this awaits a
    caller-supplied async function *within the same turn* and answers the
    agent's permission request immediately. That matters for a persistent
    session backend (Kiro's ACP): the agent is holding one turn open waiting
    for the decision, so the answer must go back in that turn rather than
    abandoning it and starting a new one.

    ``resolver`` is ``async (PermissionRequest) -> decision`` where the
    decision is an :class:`Allow`/:class:`Deny`, or a value normalized the same
    way as a resume value (``"y"``/``"n"``, ``None``, an option id, ...).

        async def ask(req):
            answer = await asyncio.to_thread(input, f"allow {req.tool}? [y/N] ")
            return answer

        backend = KiroBackend("vibe", permission=Callback(ask))
    """

    def __init__(self, resolver: Callable[[PermissionRequest], Awaitable[Any]]) -> None:
        self._resolver = resolver

    async def decide(self, req: PermissionRequest) -> PermissionDecision:
        answer = await self._resolver(req)
        return _normalize_decision(answer)