CrewAI

Wrap a CrewAI crew in a TapPass session. Every tool call emits a structured audit event with correlation IDs. Every server-side governance decision surfaces as a typed exception inside the crew.

Install

pip install 'tappass[crewai]'

Govern a crew

from crewai import Agent, Task, Crew
from tappass.integrations.crewai import guard_crew
from tappass.constraints import Pattern, Subpath

crew = Crew(agents=[researcher, writer], tasks=[research_task, write_task])

with guard_crew(
    crew,
    tappass_url="https://tappass.example.com",
    api_key="tp_...",
    constraints={
        "search_web": {"query": Pattern("*financial*")},
        "read_file": {"path": Subpath("/data")},
    },
) as session:
    result = session.kickoff()
    print(session.id, session.audit_url)

Async variant

from tappass.integrations.crewai import guard_crew_async

async with guard_crew_async(crew, tappass_url=..., api_key="tp_...") as session:
    result = session.kickoff()

Handling governance decisions

from tappass import GovernanceDecision, PolicyBlockError

try:
    with guard_crew(crew, tappass_url=..., api_key="tp_...") as session:
        session.kickoff()
except PolicyBlockError as e:
    print(f"Blocked by {e.blocked_by}: {e.reason}")
    print(f"See {e.audit_url}")
except GovernanceDecision as e:
    print(f"Governance: {e}")

Migrating from 0.5

The 0.5 fire-and-forget guard_crew(crew, constraints=...) that mutated crew in place is gone. See the migration guide.