Skip to content
TapPass

Governance from code

This page covers governing tool calls directly from Python code — without an LLM in the loop: tappass.govern() for wrapping any framework’s tools against POST /v1/govern, the tappass_session context managers for ambient correlation, and the exceptions you handle in enforce mode.

govern() — wrap tools for audit and enforcement

Section titled “govern() — wrap tools for audit and enforcement”
import tappass


tools = tappass.govern(
    [search, send_email, read_file],
    url="http://localhost:9620",       # or TAPPASS_URL
    api_key="tp_dev_...",              # or TAPPASS_API_KEY
    agent_id="spiffe://acme/bot-1",
    mode="enforce",
)

govern() works with CrewAI, LangChain, LlamaIndex, Pydantic AI, and plain callables — it introspects each tool object and wraps its hookable function (func, _fn, function, or _run). In mode="enforce", every invocation posts a TOOL_CALL Behavior to /v1/govern before running, and the tool never executes if the kernel blocks it.

Parameter Type Default Description
tools list required Tool objects (any framework) or callables
url str TAPPASS_URL or http://localhost:9620 TapPass server URL
api_key str TAPPASS_API_KEY or "" Bearer key
agent_id str "" Agent identifier. Required in enforce mode for per-agent policy scoping; ignored in audit mode
user_id str "" Optional user ID for audit attribution
mode "audit" | "enforce" "audit" audit = fire-and-forget execution reports; enforce = synchronous pre-call decision
session_id str auto (sdk-<12 hex> in enforce mode) Logical session for path-dependent policy; shared across all wrapped tools
org_id str "" Tenant ID, surfaced in policy input for per-tenant rules
on_pending "raise" | "return" "raise" Behaviour on outcome="needs_approval" (see below)

Audit mode

Section titled “Audit mode”

The default. Every tool execution is reported to the TapPass audit trail asynchronously (tool name, truncated arguments, duration, outcome, error). No decision is requested; the tool always runs.

Enforce mode

Section titled “Enforce mode”

Before the tool runs, the SDK synchronously posts the Behavior to /v1/govern and acts on the Decision:

Server outcome SDK behaviour
allow Tool executes
block Raises GovernanceBlocked; tool does not execute
needs_approval Raises ApprovalPending (with on_pending="raise") or returns the pending Decision dict (on_pending="return"). Resume by re-submitting the identical call, which returns allow once a human approves
escalate Legacy path: transparently long-polls the approval (wait=True) and converts approve → allow, deny/timeout/cancel → GovernanceBlocked
unreachable / 5xx Raises GovernanceUnavailable — fail-closed. Set TAPPASS_FAIL_OPEN=1 to allow instead (the synthetic allow is flagged with details.fail_open=True in audit)

Enforce-mode exceptions

Section titled “Enforce-mode exceptions”
from tappass import ApprovalPending, GovernanceBlocked, GovernanceUnavailable


try:
    governed_tools[0](query="...")
except GovernanceBlocked as e:
    print(e.reason)          # why the kernel blocked
    print(e.blocking_step)   # which step blocked (may be None)
    print(e.behavior_id)     # correlation back to the Behavior
except ApprovalPending as e:
    print(e.request_id)      # pending approval request
    print(e.fingerprint)     # approval fingerprint
    print(e.reason)
    # Suspend; re-submit the identical call later — it returns allow
    # once a human approves. Agent.wait_approval(e.request_id) blocks
    # until the human decides.
except GovernanceUnavailable as e:
    # /v1/govern unreachable or 5xx. Fail-closed by default.
    raise

All three derive from RuntimeError (not from TapPassError) because they are raised by the tool wrapper, not by the Agent chat path. A 4xx from /v1/govern is surfaced as GovernanceBlocked — it indicates a bad payload or auth failure, not a control-plane outage, so fail-open never applies to it.

ApprovalPending fields: request_id, fingerprint, reason, behavior_id. GovernanceBlocked fields: reason, blocking_step, behavior_id.

Ambient sessions: tappass_session

Section titled “Ambient sessions: tappass_session”
from tappass import tappass_session, tappass_session_async, current_session


with tappass_session(url="http://localhost:9620", api_key="tp_dev_...") as session:
    agent.chat("...")          # tagged with session.id
    my_tool("...")             # govern()-wrapped tools inherit it


async with tappass_session_async(url="...", api_key="tp_dev_...") as session:
    await agent.chat("...")

Sessions propagate ambiently to Agent, govern(), and the reporter. Async-safe: each task gets its own copy; threads do not share the ambient session by default.

Pass correlation_id= to override the generated session ID. On exit, the session is popped and the reporter is flushed, even on exception.

current_session() returns the active Session or None.

Testing: bypass_scope

Section titled “Testing: bypass_scope”

tappass.guard.bypass_scope() disables @guard enforcement for tests. It only works when TAPPASS_ENV=test and a test runner (pytest/unittest) is detected — defense in depth against accidental production bypass:

import os
os.environ["TAPPASS_ENV"] = "test"


from tappass import bypass_scope


def test_tool_runs():
    with bypass_scope():
        read_file("/data/report.pdf")   # no authorization check

Without TAPPASS_ENV=test, entering the scope raises immediately.

Next steps

Section titled “Next steps”