Skip to content

CrewAI Integration

Gate CrewAI task assignment through pyhall governance. Before assigning a task to an agent, ask Hall Server whether the capability is permitted. Blocked capabilities never reach the crew.

Install

Terminal window
pip install pyhall-wcp crewai

Then start Hall Server:

Terminal window
pyhall hall start

Integration Pattern

import os, requests
from crewai import Agent, Task, Crew


HALL_URL = "http://localhost:8765"
HALL_TOKEN = os.environ.get("HALL_SESSION_TOKEN", "")


def governed_task(capability_id: str, tenant_id: str = "your-org") -> dict:
    """Check Hall Server before assigning a task to a CrewAI agent."""
    resp = requests.post(f"{HALL_URL}/api/route",
        headers={"Authorization": f"Bearer {HALL_TOKEN}"},
        json={
            "capability_id": capability_id,
            "env": "dev",
            "data_label": "PUBLIC",
            "tenant_id": tenant_id,
        })
    resp.raise_for_status()
    return resp.json()


# Use in a CrewAI workflow:
decision = governed_task("cap.doc.summarize")
if not decision["allowed"]:
    raise PermissionError(f"Capability blocked: {decision.get('deny_reason')}")
# assign task to agent

Notes

  • Call governed_task() once per capability before constructing the Task object
  • On denial, do not assign the task — raise or return early
  • deny_reason values are machine-readable strings: "risk_tier_exceeded", "capability_not_enrolled", "tenant_policy_block", etc.
  • For multi-agent crews, call governed_task() for each distinct capability in the plan

Reference