agf-sdk

agf-sdk

Python SDK for INTREXA AXIS, built on the open Agent Authorization Protocol (AAP) maintained by AGF. 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.

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.

Approved execution (after a human approves a REVIEW_REQUIRED decision)

A REVIEW_REQUIRED decision can be executed after a human approves it, but only for the exact call that was reviewed, and only once (Spec 30 §3.5). No new decision is made: the reviewed one is executed. Approving it is not, on its own, permission to act.

Direct callers bind the exact call at decision time, then use execute_approved():

from agf.binding import direct_binding_sha256

payload = b'{"to":"acct_42","amount":250000.50}'          # the exact bytes you will send
digest = direct_binding_sha256(
    org_id=ORG_ID,                                           # your organisation id (required)
    audience="agf", chain=chain, action_type="payment:create", action_resource="payments/acct_42",
    destination="https://payments.example.org/v2/transfers", payload=payload,
)
try:
    await client.decide("payment:create", "payments/acct_42", chain=chain, binding_sha256=digest)
except AGFReviewRequiredError as review:
    ...  # later, once a different user has approved review.approval_request_id:
    run = await client.execute_approved(review.artifact_id, review.approval_request_id, digest,
                                        lambda: send(payload))
    value = run.result()          # the action's return value, or its original exception re-raised
    print(run.report.state)       # recorded | accepted_without_receipt | rejected | uncertain

• What the helper guarantees: at most one invocation of the action and at most one outcome-report attempt per call. Nothing is retried.
• Before running the action it requires result == "valid" for this decision, plus an execution claim for this approval with a signed execution_not_after that hasn't passed. Otherwise it raises AGFApprovedExecutionRefused. That exception's .report carries the status of the one not_executed report attempt made when the window has passed.
• Errors. Refusals from the runtime raise AGFApprovalExecutionError with a .code. APPROVAL_CONSUMED and APPROVAL_CLAIM_UNCERTAIN mean the approval may already be used up: never retry them. A validation that returned no usable answer raises AGFApprovalExecutionUncertain.
• Report status. run.report.state == "uncertain" means the report may or may not have been recorded. You may make one manual report_outcome(...); the SDK never retries.
• caller defaults to org_id, which is correct for API-key clients. Pass caller=<user id> when the decision is made under a user session instead. The same principal must request the decision and validate it, otherwise validation fails with APPROVAL_CALLER_MISMATCH.
• Limits. AXIS does not perform a direct caller's call, so it cannot observe the call or stop it being made elsewhere. The digest is your attestation of what you send. The report is your claim, not an observation.
• Caching. Passing binding_sha256 bypasses the local decision cache, so there is no offline fallback for these decisions.

MCP gateway calls are re-presented byte for byte:

try:
    await gw.call_tool("transfer", {"amount": 12.5}, chain=chain, session_id="sess-1")
except AGFReviewRequiredError as review:
    ...  # after approval:
    result = await gw.execute_approved(review.replay, review.approval_request_id, session_id="sess-1")

• Rebuilding the call. The replay holds the reviewed request exactly as sent: URL, body bytes (including the JSON-RPC id) and every header in order. execute_approved() rebuilds the request from it, never from current client defaults, and sends it once.
• Refusals. If the client's base URL, gateway, API key or session differs from the reviewed call, it raises AGFReplayContextChanged and sends nothing. A new session needs a new review.
• Keep the replay in memory. It is sensitive: its chain tokens and body may contain credentials or private data. It refuses to be pickled or copied, its repr() shows no values, and pickling the exception drops it.

SyncAGFClient.execute_approved() and SyncMCPGatewayClient.execute_approved() are the sync equivalents. For HTTP and A2A gateway calls, resend the identical request with the header X-AGF-Approval: <approval_request_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], ...)

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