breach-intel-client 0.3.1

Python SDK for the Breach Intel Policy Agent — attach any AI agent to breach logging in one import

breach-intel-client

Python SDK for the Breach Intel Policy Agent. Drop into any Python AI agent in 2 lines.

pip install breach-intel-client

Quick Start

from breach_intel_client import PolicyAgentClient

# Initialize (agent_id from register() below)
client = PolicyAgentClient(
    base_url="http://your-policy-agent:8080",
    agent_id="your-agent-id",
    vertical="fintech",
    silent=True,   # never block your agent if policy agent is down
)

# Wrap your LLM calls
response = my_llm.generate(prompt)
result = client.emit_llm_response(response.text, tenant_id="customer-001")

if result.is_breach:
    print(f"[BREACH] {result.breach_types} — severity: {result.severity}")
    print(f"Breach ID for audit: {result.breach_id}")

Registration

Note: If you installed with ./install.sh and enabled auto-monitoring, registration happens automatically. The sitecustomize.py hook calls auto_attach() → _auto_register(), which registers the process and reports detected AI frameworks (OpenAI, Anthropic, LangChain) as declared capabilities. Manual registration is only needed if you want to declare custom capabilities or skip auto-monitoring.

Call once at agent startup:

reg = client.register(
    agent_name="Portfolio Advisor v2",
    owner_id="acme-corp",
    declared_capabilities=["read_portfolio", "get_market_data", "send_report"],
    approved_domains=["api.bloomberg.com", "smtp.acme.com"],
)
client.agent_id = reg.agent_id  # save this

Emitting Events

Method	Use when
emit_llm_response(text)	Any LLM output
emit_tool_call(name, args)	Before/after tool invocation
emit_api_request(method, url, body)	Any outbound HTTP call
emit(event_type, payload)	Any other event

Async Support

# pip install breach-intel-client[async]
from breach_intel_client import AsyncPolicyAgentClient

async with AsyncPolicyAgentClient("http://localhost:8080", agent_id="...") as client:
    result = await client.emit_llm_response_async(text, tenant_id="cust-001")

Querying Breaches

# All breaches for this agent
breaches = client.get_breaches(severity="CRITICAL")

# Single record
breach = client.get_breach("br-xyz")

# Tamper check
integrity = client.verify_breach("br-xyz")
print(integrity["tampered"])  # False = clean

# Summary counts
print(client.summary())