harnest.agent.approval

All Python modules · Import and compatibility guidance

Approval request inputs

Input	Rule	Example
`action`	Stable and non-sensitive	`typescript.execute`
`arguments`	Serializable values identifying the operation	Source hash and capabilities
`message`	Public text shown to the approver	`Execute TypeScript with network access?`

Exports

Name	Kind
`ApprovalDenied`	Class
`ApprovalEnforcementError`	Class
`ApprovalError`	Class
`ApprovalExecution`	Class
`ApprovalExpired`	Class
`ApprovalPolicy`	Class
`ApprovalRequired`	Class
`ApprovalRun`	Class
`InMemoryApprovalStore`	Class
`PendingApproval`	Class
`approval_execution`	Function
`approval_policy`	Function
`authorize_mcp`	Function
`bind_tool_arguments`	Function
`record_approved_execution`	Function
`record_approved_failure`	Function
`request_human_approval`	Function
`require_human_approval`	Function
`wrap_approved_tool`	Function

ApprovalDenied

Raised inside a suspended execution when its principal denies the call.

Example

from harnest.agent.approval import ApprovalDenied

try:
    ...
except ApprovalDenied as error:
    ...

ApprovalEnforcementError

Raised when protected work lacks a managed approval boundary.

Example

from harnest.agent.approval import ApprovalEnforcementError

try:
    ...
except ApprovalEnforcementError as error:
    ...

ApprovalError

Base error for approval policy and state transitions.

Example

from harnest.agent.approval import ApprovalError

try:
    ...
except ApprovalError as error:
    ...

ApprovalExecution

Invocation identity and optional grant visible to protected callables.

ApprovalExecution(
    user_id: str,
    session_id: str,
    call_id: str,
    grant: PendingApproval | None = None,
    store: 'InMemoryApprovalStore | None' = None,
    run: 'ApprovalRun | None' = None,
) -> None

Fields (— means no declared default):

Field	Type	Default
`user_id`	`str`	`—`
`session_id`	`str`	`—`
`call_id`	`str`	`—`
`grant`	`PendingApproval \| None`	`None`
`store`	`'InMemoryApprovalStore \| None'`	`None`
`run`	`'ApprovalRun \| None'`	`None`

Parameters and example

Parameter	Type	Default
`user_id`	`str`	Required
`session_id`	`str`	Required
`call_id`	`str`	Required
`grant`	`PendingApproval \| None`	`None`
`store`	`'InMemoryApprovalStore \| None'`	`None`
`run`	`'ApprovalRun \| None'`	`None`

from harnest.agent.approval import ApprovalExecution

execution = ApprovalExecution(user_id=..., session_id=..., call_id=...)

ApprovalExecution.authorize

async ApprovalExecution.authorize(challenge: ApprovalChallenge) -> PendingApproval | None

Authorize a challenge through a live store or supplied one-time grant.

Parameters and example

Parameter	Type	Default
`challenge`	`ApprovalChallenge`	Required

from harnest.agent.approval import ApprovalExecution

execution: ApprovalExecution
approval = await execution.authorize(challenge=...)

ApprovalExpired

Raised inside a suspended execution when its decision deadline passes.

Example

from harnest.agent.approval import ApprovalExpired

try:
    ...
except ApprovalExpired as error:
    ...

ApprovalPolicy

Immutable policy attached to a local tool or MCP client factory.

ApprovalPolicy(
    message: str,
    timeout_seconds: int = 900,
    on_timeout: Literal['deny'] = 'deny',
    tools: tuple[str, ...] | None = None,
) -> None

Fields (— means no declared default):

Field	Type	Default
`message`	`str`	`—`
`timeout_seconds`	`int`	`900`
`on_timeout`	`Literal['deny']`	`'deny'`
`tools`	`tuple[str, ...] \| None`	`None`

Parameters and example

Parameter	Type	Default
`message`	`str`	Required
`timeout_seconds`	`int`	`900`
`on_timeout`	`Literal['deny']`	`'deny'`
`tools`	`tuple[str, ...] \| None`	`None`

from harnest.agent.approval import ApprovalPolicy

policy = ApprovalPolicy(message=...)

ApprovalPolicy.applies_to

ApprovalPolicy.applies_to(name: str) -> bool

Report whether this policy protects the named remote tool.

Parameters and example

Parameter	Type	Default
`name`	`str`	Required

from harnest.agent.approval import ApprovalPolicy

policy: ApprovalPolicy
result = policy.applies_to(name=...)

ApprovalRequired

Stop managed execution before a protected action is called.

ApprovalRequired(challenge: ApprovalChallenge) -> None

Example

from harnest.agent.approval import ApprovalRequired

try:
    ...
except ApprovalRequired as error:
    ...

ApprovalRun

One live invocation whose Python task may suspend at protected calls. Obtain this object from the owning Harnest API; its constructor includes runtime bindings. Fields (— means no declared default):

Field	Type	Default
`id`	`str`	`—`
`user_id`	`str`	`—`
`session_id`	`str`	`—`
`call_id`	`str`	`—`
`notifications`	`asyncio.Queue[tuple[str, Any]]`	`Queue()`
`activation`	`asyncio.Event`	`Event()`
`task`	`asyncio.Task[Any] \| None`	`None`

Example

from harnest.agent.approval import ApprovalRun

run: ApprovalRun  # returned by Harnest, not constructed directly

InMemoryApprovalStore

Development approval store with atomic one-time transitions.

InMemoryApprovalStore(
    *,
    clock: Callable[[], float] = time,
    max_pending: int = 1024,
    tombstone_seconds: float = 60,
) -> None

Parameters and example

Parameter	Type	Default
`clock`	`Callable[[], float]`	`time`
`max_pending`	`int`	`1024`
`tombstone_seconds`	`float`	`60`

from harnest.agent.approval import InMemoryApprovalStore

store = InMemoryApprovalStore()

InMemoryApprovalStore.create_run

InMemoryApprovalStore.create_run(*, user_id: str, session_id: str, call_id: str) -> ApprovalRun

Create an invocation-scoped approval suspension channel.

Parameters and example

Parameter	Type	Default
`user_id`	`str`	Required
`session_id`	`str`	Required
`call_id`	`str`	Required

from harnest.agent.approval import InMemoryApprovalStore

store: InMemoryApprovalStore
run = store.create_run(user_id=..., session_id=..., call_id=...)

InMemoryApprovalStore.request

InMemoryApprovalStore.request(
    challenge: ApprovalChallenge,
    *,
    user_id: str,
    session_id: str,
    call_id: str,
    invocation: Any = None,
) -> PendingApproval

Parameters and example

Parameter	Type	Default
`challenge`	`ApprovalChallenge`	Required
`user_id`	`str`	Required
`session_id`	`str`	Required
`call_id`	`str`	Required
`invocation`	`Any`	`None`

from harnest.agent.approval import InMemoryApprovalStore

store: InMemoryApprovalStore
approval = store.request(challenge=..., user_id=..., session_id=..., call_id=...)

InMemoryApprovalStore.suspend

async InMemoryApprovalStore.suspend(run: ApprovalRun, challenge: ApprovalChallenge) -> PendingApproval

Suspend an invocation until its challenge is approved or rejected.

Parameters and example

Parameter	Type	Default
`run`	`ApprovalRun`	Required
`challenge`	`ApprovalChallenge`	Required

from harnest.agent.approval import InMemoryApprovalStore

store: InMemoryApprovalStore
approval = await store.suspend(run=..., challenge=...)

InMemoryApprovalStore.decide

InMemoryApprovalStore.decide(
    approval_id: str,
    *,
    user_id: str,
    decision: Decision,
    deliver: bool = True,
) -> PendingApproval

Commit an approval decision for the matching user-owned challenge.

Parameters and example

Parameter	Type	Default
`approval_id`	`str`	Required
`user_id`	`str`	Required
`decision`	`Decision`	Required
`deliver`	`bool`	`True`

from harnest.agent.approval import InMemoryApprovalStore

store: InMemoryApprovalStore
approval = store.decide(approval_id=..., user_id=..., decision=...)

InMemoryApprovalStore.deliver_decision

InMemoryApprovalStore.deliver_decision(pending: PendingApproval, decision: Decision) -> None

Deliver a committed decision to its suspended invocation.

Parameters and example

Parameter	Type	Default
`pending`	`PendingApproval`	Required
`decision`	`Decision`	Required

from harnest.agent.approval import InMemoryApprovalStore

store: InMemoryApprovalStore
store.deliver_decision(pending=..., decision=...)

InMemoryApprovalStore.run_for

InMemoryApprovalStore.run_for(pending: PendingApproval) -> ApprovalRun | None

Return the live invocation associated with a pending challenge.

Parameters and example

Parameter	Type	Default
`pending`	`PendingApproval`	Required

from harnest.agent.approval import InMemoryApprovalStore

store: InMemoryApprovalStore
run = store.run_for(pending=...)

InMemoryApprovalStore.pending_for

InMemoryApprovalStore.pending_for(*, user_id: str, session_id: str) -> list[PendingApproval]

Expose only live challenges owned by this principal and session.

Parameters and example

Parameter	Type	Default
`user_id`	`str`	Required
`session_id`	`str`	Required

from harnest.agent.approval import InMemoryApprovalStore

store: InMemoryApprovalStore
result = store.pending_for(user_id=..., session_id=...)

InMemoryApprovalStore.assert_consumed

InMemoryApprovalStore.assert_consumed(pending: PendingApproval) -> None

Require resumed execution to have reached the approved action.

Parameters and example

Parameter	Type	Default
`pending`	`PendingApproval`	Required

from harnest.agent.approval import InMemoryApprovalStore

store: InMemoryApprovalStore
store.assert_consumed(pending=...)

InMemoryApprovalStore.executed

InMemoryApprovalStore.executed(pending: PendingApproval) -> None

Mark an approved action as successfully executed.

Parameters and example

Parameter	Type	Default
`pending`	`PendingApproval`	Required

from harnest.agent.approval import InMemoryApprovalStore

store: InMemoryApprovalStore
store.executed(pending=...)

InMemoryApprovalStore.failed

InMemoryApprovalStore.failed(pending: PendingApproval) -> None

Mark an approved action as failed during execution.

Parameters and example

Parameter	Type	Default
`pending`	`PendingApproval`	Required

from harnest.agent.approval import InMemoryApprovalStore

store: InMemoryApprovalStore
store.failed(pending=...)

InMemoryApprovalStore.cancel_run

InMemoryApprovalStore.cancel_run(run: ApprovalRun) -> None

Cancel all unresolved challenges and work for one invocation.

Parameters and example

Parameter	Type	Default
`run`	`ApprovalRun`	Required

from harnest.agent.approval import InMemoryApprovalStore

store: InMemoryApprovalStore
store.cancel_run(run=...)

InMemoryApprovalStore.retained_count

Return the number of live challenges and retained tombstones.

Example

from harnest.agent.approval import InMemoryApprovalStore

store: InMemoryApprovalStore
value = store.retained_count

PendingApproval

One principal-scoped invocation awaiting a one-time decision. Obtain this object from the owning Harnest API; its constructor includes runtime bindings. Fields (— means no declared default):

Field	Type	Default
`id`	`str`	`—`
`user_id`	`str`	`—`
`session_id`	`str`	`—`
`call_id`	`str`	`—`
`action`	`str`	`—`
`argument_hash`	`str`	`—`
`message`	`str`	`—`
`expires_at`	`float`	`—`
`status`	`str`	`'pending'`
`consumed`	`bool`	`False`
`finished_at`	`float \| None`	`None`

Example

from harnest.agent.approval import PendingApproval

approval: PendingApproval  # returned by Harnest, not constructed directly

PendingApproval.public

PendingApproval.public() -> dict[str, Any]

Return the client-safe fields needed to present an approval prompt.

Example

from harnest.agent.approval import PendingApproval

approval: PendingApproval
result = approval.public()

approval_execution

Activate approval authority for protected work in this context.

approval_execution(execution: ApprovalExecution) -> Iterator[None]

Parameters and example

Parameter	Type	Default
`execution`	`ApprovalExecution`	Required

from harnest.agent.approval import approval_execution

with approval_execution(execution=...):
    ...

approval_policy

Read the Harnest approval policy attached to a callable, if present.

approval_policy(value: Any) -> ApprovalPolicy | None

Parameters and example

Parameter	Type	Default
`value`	`Any`	Required

from harnest.agent.approval import approval_policy

policy = approval_policy(value=...)

authorize_mcp

Authorize one MCP tool call when the client policy selects it.

async authorize_mcp(
    client_name: str,
    tool_name: str,
    arguments: Mapping[str, Any],
    policy: ApprovalPolicy,
) -> PendingApproval | None

Parameters and example

Parameter	Type	Default
`client_name`	`str`	Required
`tool_name`	`str`	Required
`arguments`	`Mapping[str, Any]`	Required
`policy`	`ApprovalPolicy`	Required

from harnest.agent.approval import authorize_mcp

approval = await authorize_mcp(
    client_name=...,
    tool_name=...,
    arguments=...,
    policy=...,
)

bind_tool_arguments

Bind a tool call to canonical approval-safe argument values.

bind_tool_arguments(function: Callable[..., Any], args: tuple[Any, ...], kwargs: Mapping[str, Any]) -> dict[str, Any]

Parameters and example

Parameter	Type	Default
`function`	`Callable[..., Any]`	Required
`args`	`tuple[Any, ...]`	Required
`kwargs`	`Mapping[str, Any]`	Required

from harnest.agent.approval import bind_tool_arguments

result = bind_tool_arguments(function=..., args=..., kwargs=...)

record_approved_execution

Record successful completion for an optional one-time approval grant.

record_approved_execution(grant: PendingApproval | None) -> None

Parameters and example

Parameter	Type	Default
`grant`	`PendingApproval \| None`	Required

from harnest.agent.approval import record_approved_execution

record_approved_execution(grant=...)

record_approved_failure

Record failed execution for an optional one-time approval grant.

record_approved_failure(grant: PendingApproval | None) -> None

Parameters and example

Parameter	Type	Default
`grant`	`PendingApproval \| None`	Required

from harnest.agent.approval import record_approved_failure

record_approved_failure(grant=...)

request_human_approval

Protect only the dynamic operation inside this asynchronous block. Callers may evaluate policy or risk before entering the block. The approval is bound to a stable action name and canonical arguments, while the block boundary lets Harnest audit whether the approved operation completed.

request_human_approval(
    *,
    action: str,
    message: str,
    arguments: Mapping[str, Any] | None = None,
    timeout_seconds: int = 900,
    on_timeout: Literal['deny'] = 'deny',
) -> AsyncIterator[None]

Parameters and example

Parameter	Type	Default
`action`	`str`	Required
`message`	`str`	Required
`arguments`	`Mapping[str, Any] \| None`	`None`
`timeout_seconds`	`int`	`900`
`on_timeout`	`Literal['deny']`	`'deny'`

from harnest.agent.approval import request_human_approval

async with request_human_approval(action=..., message=...):
    ...

require_human_approval

Declare that a tool or MCP client() factory requires approval.

require_human_approval(
    *,
    message: str,
    timeout_seconds: int = 900,
    on_timeout: Literal['deny'] = 'deny',
    tools: Sequence[str] | None = None,
) -> Callable[[F], F]

Parameters and example

Parameter	Type	Default
`message`	`str`	Required
`timeout_seconds`	`int`	`900`
`on_timeout`	`Literal['deny']`	`'deny'`
`tools`	`Sequence[str] \| None`	`None`

from harnest.agent.approval import require_human_approval

@require_human_approval(message=...)
def my_function():
    ...

wrap_approved_tool

Install an execution gate after both decorator orders are known.

wrap_approved_tool(function: F) -> F

Parameters and example

Parameter	Type	Default
`function`	`F`	Required

from harnest.agent.approval import wrap_approved_tool

result = wrap_approved_tool(function=...)

​Approval request inputs

​Exports

​ApprovalDenied

​ApprovalEnforcementError

​ApprovalError

​ApprovalExecution

​ApprovalExecution.authorize

​ApprovalExpired

​ApprovalPolicy

​ApprovalPolicy.applies_to

​ApprovalRequired

​ApprovalRun

​InMemoryApprovalStore

​InMemoryApprovalStore.create_run

​InMemoryApprovalStore.request

​InMemoryApprovalStore.suspend

​InMemoryApprovalStore.decide

​InMemoryApprovalStore.deliver_decision

​InMemoryApprovalStore.run_for

​InMemoryApprovalStore.pending_for

​InMemoryApprovalStore.assert_consumed

​InMemoryApprovalStore.executed

​InMemoryApprovalStore.failed

​InMemoryApprovalStore.cancel_run

​InMemoryApprovalStore.retained_count

​PendingApproval

​PendingApproval.public

​approval_execution

​approval_policy

​authorize_mcp

​bind_tool_arguments

​record_approved_execution

​record_approved_failure

​request_human_approval

​require_human_approval

​wrap_approved_tool

Approval request inputs

Exports

ApprovalDenied

ApprovalEnforcementError

ApprovalError

ApprovalExecution

ApprovalExecution.authorize

ApprovalExpired

ApprovalPolicy

ApprovalPolicy.applies_to

ApprovalRequired

ApprovalRun

InMemoryApprovalStore

InMemoryApprovalStore.create_run

InMemoryApprovalStore.request

InMemoryApprovalStore.suspend

InMemoryApprovalStore.decide

InMemoryApprovalStore.deliver_decision

InMemoryApprovalStore.run_for

InMemoryApprovalStore.pending_for

InMemoryApprovalStore.assert_consumed

InMemoryApprovalStore.executed

InMemoryApprovalStore.failed

InMemoryApprovalStore.cancel_run

InMemoryApprovalStore.retained_count

PendingApproval

PendingApproval.public

approval_execution

approval_policy

authorize_mcp

bind_tool_arguments

record_approved_execution

record_approved_failure

request_human_approval

require_human_approval

wrap_approved_tool