Metadata-Version: 2.5
Name: agf-sdk
Version: 0.9.0
Summary: INTREXA AXIS SDK — MIT-licensed Python SDK for integrating agents and applications with INTREXA AXIS
Project-URL: Homepage, https://agentgovernancefoundation.com
Project-URL: Documentation, https://agentgovernancefoundation.com/docs
License: MIT
License-File: LICENSE
Requires-Python: >=3.10
Requires-Dist: httpx>=0.27
Requires-Dist: pyjwt[crypto]>=2.8
Provides-Extra: all
Requires-Dist: crewai>=0.28; extra == 'all'
Requires-Dist: langchain-core>=0.2; extra == 'all'
Requires-Dist: langgraph>=0.2; extra == 'all'
Requires-Dist: nest-asyncio>=1.5; extra == 'all'
Requires-Dist: openai-agents>=0.7; extra == 'all'
Requires-Dist: playwright>=1.40; extra == 'all'
Provides-Extra: browser
Requires-Dist: nest-asyncio>=1.5; extra == 'browser'
Requires-Dist: playwright>=1.40; extra == 'browser'
Provides-Extra: crewai
Requires-Dist: crewai>=0.28; extra == 'crewai'
Provides-Extra: dev
Requires-Dist: awslambdaric>=2.0; extra == 'dev'
Requires-Dist: mcp<2,>=1.9; extra == 'dev'
Requires-Dist: pytest-asyncio>=0.23; extra == 'dev'
Requires-Dist: pytest>=8.0; extra == 'dev'
Provides-Extra: langchain
Requires-Dist: langchain-core>=0.2; extra == 'langchain'
Provides-Extra: langgraph
Requires-Dist: langgraph>=0.2; extra == 'langgraph'
Provides-Extra: openai-agents
Requires-Dist: openai-agents>=0.7; extra == 'openai-agents'
Description-Content-Type: text/markdown

# agf-sdk

Python SDK for INTREXA AXIS, built on the open Agent Authorization Protocol (AAP) maintained by [AGF](https://agentgovernancefoundation.com). Enforce identity, trust, and policy controls on every action your AI agents take.

## Installation

```bash
pip install agf-sdk
```

With LangChain support:

```bash
pip install agf-sdk[langchain]
```

With CrewAI support:

```bash
pip install agf-sdk[crewai]
```

## Quick start

```python
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:

```python
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`:

```python
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:

```python
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:

```python
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:

```python
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:

```python
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:

```python
@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()`:

```python
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:

```python
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:

```python
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:

```python
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

```python
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()`:

```python
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`:

```python
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 `await`ed. 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:

```python
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:

```python
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)

```python
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:

```python
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

```python
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:

```python
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](https://agentgovernancefoundation.com/docs)
- [Quick start](https://agentgovernancefoundation.com/docs/quick-start)
- [API reference](https://agentgovernancefoundation.com/docs/api)
