Metadata-Version: 2.4
Name: agf-sdk
Version: 0.3.0
Summary: Agent Governance Foundation — Python SDK
Project-URL: Homepage, https://agentgovernancefoundation.com
Project-URL: Documentation, https://agentgovernancefoundation.com/docs
License: MIT
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: playwright>=1.40; extra == 'all'
Provides-Extra: browser
Requires-Dist: playwright>=1.40; extra == 'browser'
Provides-Extra: crewai
Requires-Dist: crewai>=0.28; extra == 'crewai'
Provides-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'
Description-Content-Type: text/markdown

# agf-sdk

Python SDK for the [Agent Governance Foundation](https://agentgovernancefoundation.com) authorization service. 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.

## 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}")
```

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

## 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.

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

## Links

- [Documentation](https://agentgovernancefoundation.com/docs)
- [Quick start](https://agentgovernancefoundation.com/docs/quick-start)
- [API reference](https://agentgovernancefoundation.com/docs/api)
