Integration Guide

pip install agentgate-pdp

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=False,# True = enable prompt injection scanning
    requires_human_approval=False,   # True = always ESCALATE to human
)

from agentgate.exceptions import AgentGateDenied, AgentGateEscalated

# Basic authorization check
result = gate.authorize(
    action="read",
    resource="/reports/q3.pdf",
    justification="User requested Q3 quarterly summary",
)

print(result["decision"])          # "PERMIT" | "ESCALATE" | "DENY"
print(result["trust_breakdown"])   # {"identity": 0.92, "delegation": 1.0, "purpose": 0.87, "behavioral": 0.95}
print(result["explanation"])       # Human-readable reasoning string
print(result["attack_flags"])      # [] or ["HIGH_VELOCITY", "SCOPE_VIOLATION", ...]

# Exception-based pattern
try:
    gate.authorize("read", "/confidential/salary.txt")
except AgentGateDenied as e:
    print(f"Denied: {e}")
    # Agent never reads the file

# Decorator pattern — cleaner for wrapping existing functions
@gate.guard("read", resource_arg="path")
def read_document(path: str) -> str:
    return open(path).read()  # only runs if PERMIT

# Context manager pattern
with gate.operation("delete", "/temp/export.csv"):
    os.remove("/temp/export.csv")  # only runs if PERMIT

# Async version (for LangGraph, CrewAI, AutoGen)
from agentgate import AsyncAgentGate

async def main():
    agate = AsyncAgentGate("http://localhost:8000", api_key="your-key")
    await agate.register("async_bot", "AsyncBot", "Process reports", ["/reports/*"], ["read"])
    result = await agate.authorize("read", "/reports/q4.pdf")

# Content scanning (prompt injection detection)
scan = gate.scan(email_body)
if scan["level"] == "injection":
    raise ValueError(f"Injection detected: {scan['evidence']}")

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