Integration Guide

# Python
pip install agentgate-pdp

# TypeScript / Node 18+
npm install agentgate-pdp

# The offline verifier, for whoever is checking your evidence.
# Needs no server and no account.
pip install "agentgate-pdp[verify]"

git clone https://github.com/ElamOlame31/agentgate-public
cd agentgate-public
pip install -r requirements.txt
python run.py

from agentgate import AgentGate

gate = AgentGate("http://localhost:8000", api_key="your-key")

# Register the agent with its declared scope
gate.register(
    agent_id="report_bot_001",
    name="ReportBot",
    declared_purpose="Read and summarize quarterly business reports for the executive team",
    authorized_resources=["/reports/*", "/documents/public/*"],
    authorized_actions=["read", "search"],
    delegation_depth=0,               # 0 = not a sub-agent
    processes_external_content=True,  # reads content it did not author
    requires_human_approval=False,    # True = always ESCALATE to human

    # Where this agent may send. Declared before any content arrives, which is
    # what makes it meaningful: once a session carries untrusted material, that
    # material must not get to choose the destination. Without this, every send
    # from a contaminated session is refused.
    allowed_destinations=["/outbox/*", "*@ourcompany.com"],
)

from agentgate.exceptions import AgentGateDenied

# Pass the material arguments, not just the resource. Anything left out is not
# bound by the receipt and may change between the decision and the call.
result = gate.authorize(
    "transfer",
    "/payments/outbound",
    arguments={"amount_minor": 25000, "currency": "EUR", "recipient": "acct_9931"},
)

# result["decision"]     PERMIT | ESCALATE | DENY | PENDING
# result["action_ref"]   digest of the operation, recomputable here
# result["flow"]         what the session was exposed to, and whether trusted
# result["receipt_sig"]  Ed25519, checkable by a third party

# Immediately before executing, not immediately after authorizing: the gap
# between the two is where the operation can change.
gate.check_binding(result, "transfer", "/payments/outbound", arguments)
gate.redeem(result)          # spends it, once. 409 on replay.

transfer(...)

# The decorator does all of it: authorize with every argument, recompute the
# digest before dispatch, spend the receipt, then run the body. If anything
# changed in between, the body does not run.

@gate.guard("transfer", resource_arg="account")
def transfer(account: str, amount_minor: int, recipient: str):
    ...

# An argument that cannot be serialized raises rather than being dropped —
# silently skipping is how one ends up unbound without anyone deciding to.
@gate.guard("write", resource_arg="path", exclude=("logger",))
def write_report(path: str, body: str, logger):
    ...

# Anyone can check a receipt. No account, no network, no shared secret,
# and no way to produce one from the published key.

pip install agentgate-pdp[verify]

curl -s https://your-pdp/receipts/public-key | jq -r .public_key_pem > agentgate.pem

python -m agentgate.verify receipt.json        --public-key agentgate.pem        --operation  what_actually_ran.json

#   PASS
#   signature     valid - signed by the holder of this key
#   binding       matches - this receipt authorizes exactly this operation
#
#   information flow
#     session had been exposed to public material
#     decision inputs were trusted

# Exit code is 1 on failure, so it drops into a pipeline unparsed.

from integrations.langchain_agentgate import AgentGateToolkit
from langchain_core.tools import tool
from langchain_anthropic import ChatAnthropic
from langgraph.prebuilt import create_react_agent

# Define your tools as normal
@tool
def read_document(path: str) -> str:
    """Read a document from the company file system."""
    return open(path).read()

@tool
def list_documents(directory: str) -> str:
    """List all documents in a directory."""
    import os
    return str(os.listdir(directory))

# AgentGateToolkit registers the agent AND wraps tools in one call
toolkit = AgentGateToolkit(
    agentgate_url="http://localhost:8000",
    agent_id="langchain_report_bot",
    name="LangChainReportBot",
    declared_purpose="Read and summarize quarterly business reports for the executive team",
    authorized_resources=["/documents/*"],
    authorized_actions=["read", "search"],
    api_key="your-key",
)

# wrap() returns drop-in replacements — every call goes through AgentGate
safe_tools = toolkit.wrap([read_document, list_documents])

llm = ChatAnthropic(model="claude-haiku-4-5-20251001", max_tokens=512)
agent = create_react_agent(llm, safe_tools)

# Run the agent — tool calls are intercepted transparently
result = agent.invoke({
    "messages": [{"role": "user", "content": "Summarize Q3 and Q4 reports"}]
})

# AutoGen integration — similar pattern
from integrations.autogen_agentgate import AgentGateToolkit as AutoGenToolkit

toolkit = AutoGenToolkit(
    agentgate_url="http://localhost:8000",
    agent_id="autogen_bot_001",
    name="AutoGenResearchBot",
    declared_purpose="Research and summarize public financial reports",
    authorized_resources=["/public/*"],
    authorized_actions=["read", "search"],
    api_key="your-key",
)
# Wrap your AutoGen tools the same way

# Scan any string — user input, tool response, email body, etc.
scan = gate.scan(content)

print(scan["level"])    # "clean" | "injection" | "suspicious"
print(scan["evidence"]) # list of matched patterns
print(scan["redacted"]) # sanitized version of content (PII/credentials replaced)

if scan["level"] == "injection":
    raise ValueError(f"Injection attempt blocked: {scan['evidence']}")

# For MCP (Model Context Protocol) tool-calling agents, route tool traffic
# through the AgentGate MCP proxy — it intercepts tool responses before
# the agent processes them. Configure your MCP client to use:
#   http://localhost:8000/mcp  (instead of the upstream MCP server URL)
# AgentGate forwards permitted requests and blocks INSTRUCTION_TAG / IMPERATIVE_INJECT patterns.

import requests

BASE = "http://localhost:8000"
HEADERS = {"X-API-Key": "your-key"}

# Check agent status
status = requests.get(f"{BASE}/agents/report_bot_001/status", headers=HEADERS).json()
print(status["quarantined"])       # True | False
print(status["attack_flags"])      # ["KILL_CHAIN_DETECTED", ...]
print(status["trust_score"])       # current composite score

# Manually quarantine an agent (admin action)
requests.post(f"{BASE}/agents/report_bot_001/quarantine", headers=HEADERS)

# Release from quarantine
requests.delete(f"{BASE}/agents/report_bot_001/quarantine", headers=HEADERS)

# View active contagion records (penalties propagated from neighbors)
contagion = requests.get(f"{BASE}/contagion", headers=HEADERS).json()
# [{"agent_id": "report_bot_001", "direction": "from_parent", "penalty": 30.0, "expires_in": 2847}, ...]

# Clear contagion penalties from a specific source agent
requests.post(f"{BASE}/agents/compromised_parent/contagion/clear", headers=HEADERS)

# Merkle audit trail — verify a specific decision record
proof = requests.get(f"{BASE}/audit/inclusion-proof/{{record_id}}", headers=HEADERS).json()
print(proof["merkle_root"])        # SHA-256 root hash
print(proof["proof_path"])         # O(log n) hashes for verification