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:

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.

Building a chain from your keypair

AgentGovernance self-signs a chain for you automatically, but if you're calling AGFClient/SyncAGFClient directly (or a guard — AGFGuardedTool, AGFCrewAITool, guard_tool(), guard_action()) and hold an EC P-256 keypair, build the chain= argument yourself with build_self_signed_chain:

from agf import build_self_signed_chain, generate_keypair, AGFClient

private_key_pem, public_key_pem = generate_keypair()  # persist and reuse

chain = build_self_signed_chain(
    private_key_pem,
    agent_id="did:agf:my-agent-1",
    action="file:write",
)

client = AGFClient(api_key=os.environ["AGF_API_KEY"])
result = await client.decide("file:write", "s3://corp-data/q2.csv", chain=chain)

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}")

Execution-time authorization validation

decide() evaluates a Decision once, at issuance — dispatching it later is not a new authorization check (Spec 30). If real dispatch can happen seconds or hours after decide() returns (a queued job, a long-running agent), call validate_execution() on the Decision's artifact_id immediately before dispatch to re-check platform-halt state, revocation, and expiry:

result = await client.decide("file:write", "s3://corp-data/q2.csv", chain=chain)

# ... time passes, e.g. a queued job picks this up later ...

check = await client.validate_execution(result.artifact_id)
if check.result == "invalid":
    raise PermissionError(f"No longer valid: {check.reasons}")
dispatch_the_action()

Unlike decide(), validate_execution() never raises for an invalid result — inspect .result/.reasons yourself. SyncAGFClient.validate_execution() is the sync equivalent.

Every guard in this SDK — AGFGuardedTool, AGFCrewAITool, guard_tool(), and AgentGovernance.authorize() — also accepts an opt-in validate_execution=True to run this check automatically right before dispatch, raising AGFDeniedError (or, for authorize(), returning a denied AuthResult) if it comes back invalid. Off by default.

Execution receipts

A Receipt (Spec 00 §3.5) is evidence, not authority — a signed record of what actually happened after a Decision, correlated back to it by decision_ref. Fetch receipts for a Decision:

receipts = await client.list_receipts(result.artifact_id)
for r in receipts:
    print(r.outcome, r.attempted, r.completed_at)

receipt = await client.get_receipt(receipts[0].receipt_id)  # raises AGFError (404) if not found

SyncAGFClient.list_receipts()/.get_receipt() are the sync equivalents.

Important: receipts are currently only emitted by AGF's Gateway proxies (MCP/A2A/HTTP) — calls through MCPGatewayClient/SyncMCPGatewayClient below, not a direct decide()/ validate_execution() call. guard_tool(), AgentGovernance.authorize(), and the LangChain/ CrewAI guards do not currently cause a receipt to be emitted; list_receipts() for one of their Decisions will correctly return an empty list, not an error.

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. The browser extra also pulls in nest_asyncio, since sync Playwright (playwright.sync_api) runs its own event loop under the hood, and GuardedPage's sync path needs to run an async policy check from inside it.

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