CrewAI Integration

The governance API is accessed via HTTP. Use any HTTP client — the examples below use httpx.

Install

pip install crewai httpx

Usage

import httpx
from crewai import Agent, Task, Crew

GOVERNANCE_API = "https://agentgovernance.dev/api/v1"
API_KEY = "ag_..."  # from /api/v1/api-keys/register

# Register your agent
with httpx.Client() as client:
    agent = client.post(
        f"{GOVERNANCE_API}/agents",
        headers={"Authorization": f"Bearer {API_KEY}"},
        json={"name": "researcher", "framework": "crewai"},
    ).raise_for_status().json()

    # Start a governance run
    run = client.post(
        f"{GOVERNANCE_API}/govern/runs",
        headers={"Authorization": f"Bearer {API_KEY}"},
        json={"agentId": agent["id"]},
    ).raise_for_status().json()

# Run your crew
researcher = Agent(role="Researcher", goal="Find data", backstory="...")
task = Task(description="Research Q1 trends", agent=researcher)
crew = Crew(agents=[researcher], tasks=[task])
result = crew.kickoff()

# Assess the output — fail open if API is unavailable
try:
    with httpx.Client(timeout=1.0) as client:
        assessment = client.post(
            f"{GOVERNANCE_API}/observe/assess",
            headers={"Authorization": f"Bearer {API_KEY}"},
            json={"text": str(result)[:2000], "agentId": agent["id"], "runId": run["id"]},
        ).raise_for_status().json()
    print(f"Patterns detected: {assessment['summary']['patternsDetected']}")
except Exception:
    pass  # governance is a sidecar — do not block on failure

Note

The governance call uses a 1-second timeout and fails open. See the Integration Guide for the recommended production pattern.