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. Policy is checked even if the agent's reasoning has been altered by prompt injection.

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

Exception handling

Use the lower-level AGFClient.decide() if you prefer exception-based control flow or need access to the raw response:

from agf.exceptions import AGFDeniedError, AGFReviewRequiredError

try:
    result = client.decide({
        "chain": ["did:agf:agt_01abc"],
        "action": { "type": "exec:shell", "resource": "prod-infra" },
        "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