LangChain integration

Install

pip install agf-sdk[langchain]

Requires Python ≥ 3.10 and langchain-core ≥ 0.2.

Set your API key

export AGF_API_KEY=agfk_your_key_here

Generate a key from Settings → API Keys.

Pattern 1 — Gate tool

A BaseToolthe agent calls before sensitive operations. The agent asks "am I allowed?" and receives a decision. Good when you want the agent to reason about its own authorization.

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

agf = AgentGovernance(
    api_key=os.environ["AGF_API_KEY"],
    org_id="org_acme",
)

# agf_tool is a BaseTool — the agent calls it before sensitive operations
agf_tool = agf.langchain_tool(agent_id="did:agf:agt_01abc")

agent = initialize_agent(
    tools=[agf_tool, *your_other_tools],
    llm=ChatOpenAI(model="gpt-4o"),
    agent=AgentType.OPENAI_FUNCTIONS,
)

agent.run("Summarise last quarter's sales and write to /tmp/report.txt")

Or check authorization yourself before running the agent:

result = agf.authorize(
    agent_id="did:agf:agt_01abc",
    action="write:file",
    resource="/tmp/report.txt",
)

if result.allowed:
    # proceed
    pass
else:
    print(f"Blocked: {result.reason}")

Pattern 2 — Per-tool guard

AGFGuardedTool wraps any BaseTooland enforces policy on every invocation — including async calls. The policy check runs whenever the tool is invoked, whether or not the agent chose to ask for it. The decision is made from the action type and resource set when the tool is wrapped, plus the delegation chain; by default it does not evaluate the individual call's arguments. To include selected arguments in the decision, see Pattern 2b (agf-sdk 0.11.0 or later).

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

client = AGFClient(api_key=os.environ["AGF_API_KEY"])

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

agent = initialize_agent(
    tools=[guarded_shell],
    llm=ChatOpenAI(model="gpt-4o"),
    agent=AgentType.OPENAI_FUNCTIONS,
)

Pattern 2b — Per-call context (agf-sdk ≥ 0.11.0)

By default the guard sends no call-specific data. To include selected arguments in the decision, pass context_provider: a synchronous function that receives the same arguments the tool receives and returns the fields to send as context. context_allowlist is required with it and lists the only keys that may be sent.

from pydantic import BaseModel

class PayInput(BaseModel):
    amount: str      # a decimal string, for example "500.00"
    currency: str    # an ISO 4217 code, for example "INR"

guarded_pay = AGFGuardedTool(
    tool=PayInvoiceTool(),        # your own BaseTool
    client=client,
    agent_id="did:agf:agt_01abc",
    action_type="pay:invoice",
    resource="invoices",
    args_schema=PayInput,         # not copied from the wrapped tool: pass it explicitly
    context_provider=lambda amount, currency: {"amount": amount, "currency": currency},
    context_allowlist=["amount", "currency"],
)

Pattern 3 — Execution-time re-validation

A Decision is evaluated once, at decide() time — dispatching it later is not a new authorization check. If a tool call can happen seconds or hours after the guard's decision (a queued job, a long-running agent), pass validate_execution=True to re-check platform-halt state, revocation, and expiry immediately before the tool actually runs:

guarded_shell = AGFGuardedTool(
    tool=ShellTool(),
    client=client,
    agent_id="did:agf:agt_01abc",
    action_type="exec:shell",
    resource="local-shell",
    validate_execution=True,   # re-check right before dispatch
)

# If the underlying Decision has gone stale by dispatch time —
# revoked, expired, or the platform is under an emergency halt —
# AGFDeniedError is raised before the tool ever runs.

Exception handling

Use the lower-level AGFClient.decide() if you prefer exception-based control flow or need the full decision result:

from agf.exceptions import AGFDeniedError, AGFReviewRequiredError

# client is the AGFClient from Pattern 2; decide() is a coroutine, so await it
# (inside an async function).
try:
    result = await client.decide(
        "exec:shell",
        "prod-infra",
        chain=["did:agf:agt_01abc"],
        audience="api.internal",
        context={"environment": "production"},
    )
except AGFDeniedError as e:
    print(f"Denied: {e}")
except AGFReviewRequiredError as e:
    print(f"Approval required: {e.approval_request_id}")

Pattern comparison