Python Integration¶

import os
import httpx

# Configuration
VARGATE_URL = os.environ.get("VARGATE_URL", "https://vargate.ai/api")
API_KEY = os.environ["VARGATE_API_KEY"]  # Never hardcode

client = httpx.Client(
    base_url=VARGATE_URL,
    headers={"X-API-Key": API_KEY},
    timeout=30,
)


def governed_tool_call(tool: str, method: str, params: dict) -> dict:
    """Submit a governed tool call to Vargate."""
    response = client.post("/mcp/tools/call", json={
        "agent_id": "my-agent-v1",
        "agent_type": "autonomous",
        "agent_version": "1.0.0",
        "tool": tool,
        "method": method,
        "params": params,
    })

    if response.status_code == 200:
        result = response.json()
        print(f"Allowed: {result['action_id']}")
        return result
    elif response.status_code == 403:
        detail = response.json()["detail"]
        print(f"Denied: {detail['violations']}")
        raise PermissionError(f"Action denied: {detail['violations']}")
    elif response.status_code == 202:
        result = response.json()
        print(f"Pending approval: {result['action_id']}")
        return result
    else:
        response.raise_for_status()


def check_audit_integrity() -> dict:
    """Verify the audit hash chain is intact."""
    response = client.get("/audit/verify")
    response.raise_for_status()
    return response.json()


def get_merkle_proof(record_hash: str) -> dict:
    """Get a Merkle inclusion proof for an audit record."""
    response = client.get(f"/audit/merkle/proof/{record_hash}")
    response.raise_for_status()
    return response.json()


def replay_decision(action_id: str) -> dict:
    """Replay a historical decision against current policy."""
    response = client.post("/audit/replay", json={"action_id": action_id})
    response.raise_for_status()
    return response.json()


# Usage
if __name__ == "__main__":
    # Send a governed action
    result = governed_tool_call("http", "GET", {
        "url": "https://api.example.com/data",
    })

    # Verify audit integrity
    integrity = check_audit_integrity()
    print(f"Chain valid: {integrity['valid']}, records: {integrity['record_count']}")

    # Replay the decision
    if result.get("action_id"):
        replay = replay_decision(result["action_id"])
        print(f"Replay consistent: {replay.get('consistent')}")

import httpx
import asyncio

async def governed_action_async():
    async with httpx.AsyncClient(
        base_url="https://vargate.ai/api",
        headers={"X-API-Key": API_KEY},
        timeout=30,
    ) as client:
        response = await client.post("/mcp/tools/call", json={
            "agent_id": "async-agent",
            "agent_type": "autonomous",
            "agent_version": "1.0.0",
            "tool": "http",
            "method": "GET",
            "params": {"url": "https://api.example.com/data"},
        })
        return response.json()

result = asyncio.run(governed_action_async())

import httpx

try:
    result = governed_tool_call("stripe", "create_transfer", {
        "amount": 50000,
        "destination": "acct_xyz",
    })
except PermissionError as e:
    # Handle policy denial
    print(f"Policy blocked this action: {e}")
except httpx.HTTPStatusError as e:
    if e.response.status_code == 429:
        print("Rate limited — back off and retry")
    elif e.response.status_code == 502:
        print("OPA unavailable — check failure mode config")
    else:
        raise