Skip to content
TapPass

Agent client

This page covers the governed chat client in the tappass Python SDK (v0.10.1, Python ≥ 3.11): the Agent and AsyncAgent classes, the typed ChatResponse they return, the GovernanceDecision exception tree they raise, and the small TapPass runtime admin client.

Install

Section titled “Install”
Terminal window
pip install tappass

Agent / AsyncAgent

Section titled “Agent / AsyncAgent”
from tappass import Agent, AsyncAgent


agent = Agent(
    "http://localhost:9620",          # server URL (positional)
    "tp_dev_...",                     # api_key (positional or keyword)
    model="gpt-4o-mini",              # default model (keyword-only)
    respect_breakglass=False,         # raise BreakGlassActive during emergencies (default)
    raise_on_redaction=False,         # raise RedactionApplied instead of returning redacted content
)

AsyncAgent has the same parameters, but api_key is keyword-only:

agent = AsyncAgent("http://localhost:9620", api_key="tp_dev_...")

The constructor probes the server and raises TapPassConfigError if it is older than 0.6.0.

Parameter Type Default Description
url str required TapPass server base URL
api_key str "" Agent API key (tp_...). Keyword-only on AsyncAgent
model str "gpt-4o-mini" Default model for chat/stream
respect_breakglass bool False When True, break-glass state is returned as metadata instead of raising BreakGlassActive
raise_on_redaction bool False When True, redacted responses raise RedactionApplied instead of returning mutated content

Methods

Section titled “Methods”

chat(prompt, *, model=None, metadata=None) -> ChatResponse

Section titled “chat(prompt, *, model=None, metadata=None) -> ChatResponse”

Send a governed chat completion. Blocks on the governance decision; returns a typed response.

response = agent.chat("What are the GDPR requirements?")
print(response.content)                   # LLM answer
print(response.pipeline.classification)   # "INTERNAL"
print(response.usage.total_tokens)        # 156
print(response.session_id)                # correlation ID for this session

stream(prompt, *, model=None, metadata=None) -> Iterator[ChatChunk]

Section titled “stream(prompt, *, model=None, metadata=None) -> Iterator[ChatChunk]”

Stream the completion as server-sent events. Each ChatChunk has delta (text), finished (bool), and correlation fields (session_id, task_id, agent_uuid, pipeline_uuid, audit_url) that may be empty on later chunks.

for chunk in agent.stream("Write a report"):
    print(chunk.delta, end="", flush=True)

poll_approval(request_id) -> dict

Section titled “poll_approval(request_id) -> dict”

One-shot check on a pending approval. Returns the server-side approval row. Use for dashboards that render approval state without blocking.

wait_approval(request_id, timeout_seconds=300.0) -> dict

Section titled “wait_approval(request_id, timeout_seconds=300.0) -> dict”

Block until the approval resolves (server long-poll translated into a single call). Use when you have a request_id from an escalate decision and want to wait for the human decision.

close() / context managers

Section titled “close() / context managers”
with Agent("http://localhost:9620", "tp_dev_...") as agent:
    response = agent.chat("Hello")


async with AsyncAgent("http://localhost:9620", api_key="tp_dev_...") as agent:
    response = await agent.chat("Hello")

AsyncAgent.close() is named aclose().

ChatResponse

Section titled “ChatResponse”
Field Type Description
content str Assistant response text
pipeline PipelineResult Govern-call execution summary (see below)
session_id str Session correlation ID
task_id str Task correlation ID for this turn
agent_uuid str Agent that processed the request
pipeline_uuid str Authoring artifact UUID — which policy bundle ran
audit_url str Direct URL to the audit trail for this session
model str Model that generated the response
tool_calls list[ToolCall] Tool calls requested by the model
usage Usage prompt_tokens, completion_tokens, total_tokens
token CapabilityToken | None Capability token minted when the kernel allowed the call
raw dict Complete raw server response

str(response) returns response.content.

PipelineResult

Section titled “PipelineResult”
Field Type Description
blocked bool Whether the govern call blocked the request
blocked_by str Step that blocked (empty if allowed)
classification str Data classification (PUBLIC, INTERNAL, …)
steps_run int Number of decision steps executed
total_duration_ms float Total decision latency
steps list[StepResult] Per-step results (step, action, detected, message, duration_ms)
shadow_mode bool Decision computed in shadow mode
degraded bool Response served in degraded mode (TapPass unreachable)
degraded_reason str Human-readable degraded-mode reason

Exceptions

Section titled “Exceptions”
from tappass import (
    TapPassError, TapPassConnectionError, TapPassConfigError,
    GovernanceDecision, PolicyBlockError, RedactionApplied,
    ApprovalRequired, AuthorizationRequired, TrustTierDenied,
    BreakGlassActive, ToolIntegrityViolation,
)


try:
    response = agent.chat("Show me all credit cards")
except PolicyBlockError as e:
    print(e.blocked_by)   # step that blocked
    print(e.reason)       # human-readable reason
except ApprovalRequired as e:
    print(e.approval_url) # deep-link for a human approver
except TapPassConnectionError:
    print("TapPass server unreachable")

Hierarchy:

TapPassError
├── TapPassConnectionError        # server unreachable
├── TapPassConfigError            # invalid/missing configuration
└── GovernanceDecision            # server-returned runtime decision
    ├── PolicyBlockError          # +blocked_by, classification, details
    ├── RedactionApplied          # +redactions, redacted_content
    ├── ApprovalRequired          # +approval_url, resume_token, expires_at
    ├── AuthorizationRequired     # +provider, authorize_url, scopes
    ├── TrustTierDenied           # +required_tier, actual_tier
    ├── BreakGlassActive          # +level, expires_at
    └── ToolIntegrityViolation    # +tool_name, violation

Every GovernanceDecision carries the correlation IDs the server returned: reason, session_id, task_id, agent_uuid, pipeline_uuid, audit_url. All subclasses are picklable — they survive Celery/RQ/multiprocessing worker boundaries with every field intact.

PolicyBlockError is a deliberate policy decision. Do not retry it blindly.

Runtime admin: TapPass / AsyncTapPass

Section titled “Runtime admin: TapPass / AsyncTapPass”

A narrow client for agent boot-time operations. Dashboard-level admin (policy CRUD, compliance reports, retention) is not in the SDK.

from tappass import TapPass


with TapPass("http://localhost:9620", admin_key="tp_...") as tp:
    health = tp.health()                       # HealthStatus: status, version, storage, agents
    info = tp.agent(agent_uuid)                # AgentInfo lookup
    cred = tp.register_agent("my-service", framework="langchain")  # self-registration
Method Returns Description
health() HealthStatus Server health and version
agent(uuid) AgentInfo Look up a registered agent
register_agent(agent_id, **kwargs) dict Register this agent at startup

Next steps

Section titled “Next steps”