agf-sdk

Python SDK for the Agent Governance Foundation authorization service. Enforce identity, trust, and policy controls on every action your AI agents take.

Installation

pip install agf-sdk

With LangChain support:

pip install agf-sdk[langchain]

With CrewAI support:

pip install agf-sdk[crewai]

Quick start

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="file:write",
    resource="s3://corp-data/q2.csv",
)

if result.allowed:
    write_file()
else:
    raise PermissionError(f"Denied: {result.reason}")

Authorization results

authorize() never raises for deny/review — it always returns an AuthResult:

Field	Type	Description
allowed	bool	True when the PDP issued ALLOW
denied	bool	True when the PDP issued DENY
review_required	bool	True when HITL approval is needed
reason	str	Human-readable denial reason
artifact_id	str	Signed audit artifact ID
risk_score	float	0.0–1.0
trust_score	int	0–100
approval_request_id	str	HITL request ID (review_required only)

Auto-discovery & self-signed chains

Calling authorize() without a chain requires a private_key_pem — the SDK self-signs a minimal single-hop chain (iss == sub == agent_id) rather than silently failing. Generate a keypair once and reuse the same private key across restarts:

from agf import AgentGovernance, generate_keypair

private_key_pem, public_key_pem = generate_keypair()  # persist private_key_pem yourself

agf = AgentGovernance(
    api_key=os.environ["AGF_API_KEY"],
    auto_discover=True,
    private_key_pem=private_key_pem,
)

result = agf.authorize("did:agf:my-agent-1", "file:write", "s3://corp-data/q2.csv")

With auto_discover=True, the first authorize() call for a given agent_id also submits it to AGF's Agent Discovery (discovery_source="sdk") — it shows up in the dashboard's Discovery page as a shadow agent, blocked from acting until an operator enrolls it. Discovery submission is best-effort and never blocks or fails the authorization call itself.

Important: reuse the same private_key_pem across process restarts. A freshly generated key each run won't match the public key AGF already has on file for that agent's DID, and real chain validation (which happens after enrollment) will fail.

Async client

For async frameworks (FastAPI, async Django, etc.) use AGFClient directly:

from agf import AGFClient, AGFDeniedError

async def handle_request():
    async with AGFClient(api_key="agfk_...") as client:
        try:
            result = await client.decide(
                action_type="file:write",
                resource="s3://corp-data/q2.csv",
                chain=[root_jwt, agent_jwt],
            )
        except AGFDeniedError as exc:
            print(f"Denied — artifact: {exc.artifact_id}")

LangChain integration

Authorization gate tool (recommended for most agents)

Add an authorization tool to your agent's tool list. The agent calls it before performing sensitive operations:

from agf import AgentGovernance
from langchain.agents import initialize_agent, AgentType
from langchain_openai import ChatOpenAI

agf = AgentGovernance(api_key="agfk_...", org_id="org_acme")
agf_tool = agf.langchain_tool(agent_id="did:agf:agt_01abc")

agent = initialize_agent(
    tools=[agf_tool, *your_other_tools],
    llm=ChatOpenAI(),
    agent=AgentType.OPENAI_FUNCTIONS,
)

Per-tool guard (enforces policy on every tool call)

Wrap individual tools so no call can bypass the policy check:

from langchain_community.tools import ShellTool
from agf.langchain import AGFGuardedTool
from agf import AGFClient

client = AGFClient(api_key="agfk_...")

guarded_shell = AGFGuardedTool(
    tool=ShellTool(),
    client=client,
    agent_id="did:agf:my-assistant",
    action_type="exec:shell",
    resource="local-shell",
)

CrewAI integration

from crewai import Agent
from crewai.tools import BaseTool as CrewBaseTool
from agf.crewai import AGFCrewAITool
from agf import AGFClient

client = AGFClient(api_key="agfk_...")

class MyDBTool(CrewBaseTool):
    name: str = "database_query"
    description: str = "Query the production database"

    def _run(self, query: str) -> str:
        return db.execute(query)

guarded = AGFCrewAITool(
    tool=MyDBTool(),
    client=client,
    agent_id="did:agf:crew-researcher",
    action_type="query:database",
    resource="prod-db",
)

crew_agent = Agent(tools=[guarded], ...)

MCP integration

No new runtime dependency — both halves ship in core agf-sdk, no extra required.

Server-side guard (writing an MCP server)

Gate a tool function with an AGF policy check before it runs, in-process — the MCP analog of AGFGuardedTool/AGFCrewAITool. Apply guard_tool() before @mcp.tool() (closer to def) so FastMCP's schema introspection still sees the real signature:

from mcp.server.fastmcp import FastMCP
from agf import AGFClient
from agf.mcp import guard_tool

mcp = FastMCP("my-server")
client = AGFClient(api_key="agfk_...")

@mcp.tool()
@guard_tool(client, agent_id="did:agf:my-server", action_type="execute")
async def write_file(path: str, content: str) -> str:
    ...  # only runs on ALLOW

Client-side Gateway client (calling MCP tools through the runtime's MCP Gateway)

from agf.mcp import SyncMCPGatewayClient
from agf import AGFDeniedError

gw = SyncMCPGatewayClient(api_key="agfk_...", gateway_id="gw_01abc")

try:
    result = gw.call_tool("write_file", {"path": "a.txt"}, chain=[root_jwt, agent_jwt])
except AGFDeniedError as exc:
    print(f"Denied — artifact: {exc.artifact_id}")

Browser agent integration

No new runtime dependency required for the core primitive — GuardedPage needs the browser extra (pip install agf-sdk[browser]) only to talk to a real Playwright Page; unlike MCP/A2A/HTTP, a browser-automation agent has no downstream server for agf-runtime to front, so this is an SDK-side guard, not a gateway.

GuardedPage wraps a Playwright Page (sync or async) and gates a curated set of high-governance-relevance actions — goto, click, fill, set_input_files — with an AGF policy check before they run. Everything else passes through untouched:

from playwright.sync_api import sync_playwright
from agf import SyncAGFClient
from agf.browser import GuardedPage

client = SyncAGFClient(api_key="agfk_...")

with sync_playwright() as p:
    browser = p.chromium.launch()
    page = GuardedPage(browser.new_page(), client, agent_id="did:agf:my-browser-agent")
    page.goto("https://example.com")   # only navigates on ALLOW
    page.click("#submit")

Webhook verification

from agf import verify_signature, parse_event, AGFWebhookVerificationError

# FastAPI example
from fastapi import FastAPI, Request, HTTPException

app = FastAPI()

@app.post("/agf-webhook")
async def handle(request: Request):
    body = await request.body()
    try:
        verify_signature(body, request.headers["X-AGF-Signature"], WEBHOOK_SECRET)
    except AGFWebhookVerificationError:
        raise HTTPException(status_code=400, detail="Invalid signature")

    event = parse_event(body)
    if event.type == "decision.deny":
        print(f"Agent {event.agent_id} was denied — artifact {event.artifact_id}")

Sync client

For scripts, Django views, or any non-async context:

from agf import SyncAGFClient

with SyncAGFClient(api_key="agfk_...") as client:
    result = client.decide("file:write", "s3://bucket/file.csv")
    agents = client.list_agents(status="active")

Environment variable

Set AGF_API_KEY in your environment and pass it via os.environ["AGF_API_KEY"]. The SDK does not auto-read environment variables — this keeps the dependency graph minimal and the behaviour explicit.

Requirements

• Python 3.10+
• httpx >= 0.27
• langchain-core >= 0.2 (optional, agf-sdk[langchain])
• crewai >= 0.28 (optional, agf-sdk[crewai])

Links

• Documentation
• Quick start
• API reference