SDK Reference

Install

pip install tappass

Requires Python 3.11+.

Agent

from tappass import Agent

agent = Agent(
    url="https://tappass.example.com",
    api_key="tp_...",
    model="gpt-4o-mini",            # default model (optional)
    flags={"pii": "mask"},          # default governance flags (optional)
    timeout=30,                     # request timeout in seconds
    max_retries=2,                  # retry on transient errors
)

Methods

agent.chat(message, **kwargs)

response = agent.chat("What are the GDPR requirements?")
print(response.content)                   # LLM answer
print(response.pipeline.classification)   # "PUBLIC"
print(response.pipeline.blocked)          # False
print(response.usage.total_tokens)        # 156

Per-call overrides:

response = agent.chat("Complex analysis", model="gpt-4o", flags={"mode": "lockdown"})

agent.stream(message, **kwargs)

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

agent.achat(message, **kwargs) / agent.astream(message, **kwargs)

Async variants:

response = await agent.achat("Analyze the data")

async for chunk in agent.astream("Summarize"):
    print(chunk, end="", flush=True)

agent.reset()

Clear conversation history:

agent.chat("My name is Alice")
agent.chat("What's my name?")  # "Alice"
agent.reset()
agent.chat("What's my name?")  # doesn't know

agent.govern(tools)

Wrap callables for audited execution:

tools = agent.govern([search, send_email, read_file])

agent.secure(tier="standard")

Apply kernel-level sandbox. Irreversible for the current process.

agent.secure()                    # default sandbox
agent.secure(tier="worker")       # read/write workspace, safe commands only
agent.secure(tier="observer")     # read-only, no shell, no network

Response object

Field	Type	Description
response.content	str	LLM response text
response.model	str	Model used
response.usage.total_tokens	int	Total tokens
response.usage.prompt_tokens	int	Prompt tokens
response.usage.completion_tokens	int	Completion tokens
response.pipeline.classification	str	PUBLIC, INTERNAL, CONFIDENTIAL, RESTRICTED
response.pipeline.blocked	bool	Whether the request was blocked
response.pipeline.flags	dict	Active governance flags

Errors

from tappass import PolicyBlockError
from tappass.errors import TapPassConnectionError

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 TapPassConnectionError:
    print("TapPass server unreachable")

PolicyBlockError is never retried.

Context manager

with Agent("https://tappass.example.com", "tp_...") as agent:
    response = agent.chat("Hello")

Environment variables

Variable	Description
TAPPASS_URL	Server URL
TAPPASS_API_KEY	Agent API key
TAPPASS_FLAGS	Default flags (e.g. mode=observe, pii=mask)
TAPPASS_MODEL	Default model