CrewAI integration

Install

pip install agf-sdk[crewai]

Requires Python ≥ 3.10 and crewai ≥ 0.28.

Set your API key

export AGF_API_KEY=agfk_your_key_here

Quick authorization check

Use AgentGovernance.authorize() to check policy before running a crew. Always returns an AuthResult — never raises on DENY.

import os
from agf import AgentGovernance

agf = AgentGovernance(
    api_key=os.environ["AGF_API_KEY"],
    org_id="org_acme",
)

result = agf.authorize(
    agent_id="did:agf:agt_01abc",
    action="read:data",
    resource="customer-records",
)

if result.allowed:
    print("Authorized")
else:
    print(f"Blocked: {result.reason}")

Per-tool guard

AGFCrewAITool wraps any CrewAI BaseTool and intercepts every _run() call. If the PDP returns DENY, the tool raises AGFDeniedError before execution.

import os
from agf import AGFClient
from agf.crewai import AGFCrewAITool
from crewai_tools import FileReadTool

client = AGFClient(api_key=os.environ["AGF_API_KEY"])

# Wraps a CrewAI BaseTool — every _run() call goes through the PDP
guarded_file_tool = AGFCrewAITool(
    tool=FileReadTool(),
    client=client,
    agent_id="did:agf:agt_01abc",
    action_type="read:file",
    resource="filesystem",
)

Single-agent crew

from crewai import Agent, Task, Crew

researcher = Agent(
    role="Research Analyst",
    goal="Gather and analyse customer data",
    backstory="Expert data analyst with access to customer records.",
    tools=[guarded_file_tool],
    verbose=True,
)

task = Task(
    description="Read last month's churn report and summarise key findings.",
    agent=researcher,
    expected_output="A 3-bullet summary of churn drivers.",
)

crew = Crew(agents=[researcher], tasks=[task])
crew.kickoff()

Every time the researcher agent calls FileReadTool, the PDP is consulted first. If the policy denies the action, the tool raises before any file is read.

Multi-agent crew with delegation

For crews where agents pass context between tasks, register each agent separately and use a delegation chain in the authorization request. This lets the PDP evaluate the full authority chain, not just the immediate caller.

from crewai import Agent, Task, Crew

writer = Agent(
    role="Report Writer",
    goal="Write a report based on research findings",
    backstory="Senior technical writer.",
    tools=[guarded_file_tool, guarded_write_tool],
)

task1 = Task(description="Gather customer churn data.", agent=researcher, expected_output="Raw data.")
task2 = Task(description="Write a report from the data.", agent=writer, expected_output="Draft report.", context=[task1])

crew = Crew(agents=[researcher, writer], tasks=[task1, task2])
crew.kickoff()

Notes

• CrewAI tasks don’t have a direct intercept point — the guard wraps at the tool level, which is the right interception boundary.
• AGFCrewAITool wraps _run() synchronously. CrewAI’s async support varies by version; the guard uses run_coroutine_threadsafe internally for Python 3.10+ compatibility.
• Every PDP call produces a signed audit artifact stored in your org’s audit log. Export from the dashboard or GET /v1/audit.