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.

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.

Receipts are emitted by AGF's Gateway proxies (MCP/A2A/HTTP) automatically. For a direct decide()/validate_execution() call — no Gateway proxy involved — nothing produces a receipt unless you ask for one:

result = await client.decide("tool:write_file", "write_file", chain=chain)
try:
    do_the_write()
except Exception:
    await client.report_outcome(result.artifact_id, "not_executed")
    raise
else:
    await client.report_outcome(result.artifact_id, "executed")

guard_tool() does this automatically when you pass report_outcome=True — no manual try/except needed:

@mcp.tool()
@guard_tool(client, agent_id="did:agf:my-server", action_type="tool:write_file",
            chain_provider=my_chain_provider, report_outcome=True)
def write_file(path: str, content: str) -> str:
    ...

The resulting receipt is marked gateway="self_reported" — deliberately distinct from a Gateway-observed one (mcp/a2a/http), since this is the caller claiming an outcome AGF never directly witnessed, not AGF observing it. report_outcome() is best-effort: a failure to record it is logged and swallowed, never raised into your code and never able to mask the guarded call's own result or exception. SyncAGFClient.report_outcome() is the sync equivalent. AgentGovernance.authorize() and the LangChain/CrewAI guards have no automatic equivalent — call client.report_outcome() yourself if you want one for those.

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], ...)

LangGraph integration

Two governance surfaces exist in a LangGraph agent:

Tool-calling nodes — reuse the LangChain guard, no new code

langgraph.prebuilt.ToolNode/create_react_agent accept plain langchain_core BaseTool instances. AGFGuardedTool (above) already is one, so wrap your tool with it as usual and hand the wrapped instance to ToolNode directly — this works today with zero LangGraph-specific code.

Graph nodes — guard_node

StateGraph.add_node(name, fn) accepts an arbitrary callable, not a BaseTool — gate one with guard_node, the same decorator shape as agf.mcp.guard_tool():

from agf import AGFClient
from agf.langgraph import guard_node

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

@guard_node(client, agent_id="did:agf:my-agent", action_type="node:issue_refund")
def issue_refund(state: State) -> dict:
    ...  # only runs on ALLOW

graph.add_node("issue_refund", issue_refund)

OpenAI Agents SDK integration

agents.tool.FunctionTool (what @function_tool builds) is guarded by wrapping its on_invoke_tool call boundary, the same "wrap the tool object" pattern as AGFGuardedTool/ AGFCrewAITool:

from agents import function_tool
from agf import AGFClient
from agf.openai_agents import guard_function_tool

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

@function_tool
def issue_refund(order_id: str) -> str:
    ...  # only runs on ALLOW

guarded = guard_function_tool(issue_refund, client, agent_id="did:agf:my-agent")
agent = Agent(name="support-agent", tools=[guarded])

AWS Lambda integration

No new module, no new extra — guard_tool() (below) already works unmodified on a raw Lambda handler. Verified directly against the real AWS Lambda Python Runtime Interface Client (awslambdaric): the runtime invokes a handler as a plain, synchronous, positional call — response = handler(event, context), never awaited. Python Lambda handlers are always sync at the runtime boundary (an async def handler is never awaited by AWS's own runtime and will fail to marshal). guard_tool()'s wrapping is already fully generic over (*args, **kwargs), so it applies unchanged:

from agf import AGFClient
from agf.mcp import guard_tool

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

@guard_tool(client, agent_id="did:agf:my-lambda-fn", action_type="lambda:issue_refund")
def handler(event, context):
    ...  # only runs on ALLOW

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])
• langgraph >= 0.2 (optional, agf-sdk[langgraph])
• openai-agents >= 0.7 (optional, agf-sdk[openai-agents])

Links

• Documentation
• Quick start
• API reference