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

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