Admin Client

The TapPass class provides admin operations for CISOs and platform teams.

from tappass import TapPass

with TapPass("http://localhost:9620", admin_key="tp_admin_...") as tp:
    health = tp.health()
    print(health.status)   # "healthy"

Server health

health = tp.health()
print(health.status)    # "healthy"
print(health.version)   # "0.2.0"
print(health.storage)   # "postgresql"

Pipelines

# Create
tp.create_pipeline("strict", steps={
    "detect_pii": {"on_detection": "block"},
    "detect_injection": {"on_detection": "block", "threshold": 0.5},
    "rate_limit": {"max_calls": 50, "window_seconds": 3600},
})

# Assign to agent
tp.assign("my-agent", "strict")

# List
for p in tp.pipelines():
    print(f"{p.pipeline_id}: {len(p.steps)} steps, {len(p.agents)} agents")

Agents

for a in tp.agents():
    print(f"{a.agent_id} ({a.framework}). active={a.active}")

Audit trail

events = tp.audit(agent_id="my-agent", limit=10)
for event in events:
    print(f"[{event.timestamp}] {event.event_type}: {event.agent_id}")

Context manager

# Sync
with TapPass("http://localhost:9620", admin_key="tp_admin_...") as tp:
    tp.create_pipeline("test", steps={})

# Async
async with TapPass("http://localhost:9620") as tp:
    health = await tp.ahealth()