Python SDK Reference¶

# Core SDK
pip install aegis-governance

# With optional dependencies
pip install aegis-governance[engine]       # scipy for z-score computation
pip install aegis-governance[telemetry]    # prometheus_client for metrics
pip install aegis-governance[config]       # pyyaml for YAML config loading
pip install aegis-governance[crypto]       # BIP-340 Schnorr signatures
pip install aegis-governance[persistence]  # SQLAlchemy async workflow storage
pip install aegis-governance[all]          # Everything

from aegis_governance import pcw_decide, PCWContext, PCWPhase, PCWStatus, AegisConfig

config = AegisConfig.default()
evaluator = config.create_gate_evaluator()

decision = pcw_decide(
    PCWContext(
        agent_id="my-agent",
        session_id="session-1",
        phase=PCWPhase.PLAN,
        proposal_summary="Add caching layer",
        estimated_impact="medium",
        risk_proposed=0.3,
        complexity_score=0.6,
    ),
    gate_evaluator=evaluator,
)

if decision.status == PCWStatus.PROCEED:
    print("Safe to proceed")

from aegis_governance import AegisConfig

config = AegisConfig.default()

config = AegisConfig.from_yaml("schema/interface-contract.yaml")

evaluator = config.create_gate_evaluator()
result = evaluator.evaluate_risk_gate(baseline=0.1, proposed=0.3)

monitor = config.create_drift_monitor()
monitor.set_baseline([0.1, 0.15, 0.12, 0.14, ...])  # 30+ observations

calculator = config.create_utility_calculator()

from aegis_governance import PCWContext, PCWPhase

def pcw_decide(
    context: PCWContext,
    gate_evaluator: GateEvaluator | None = None,
    prometheus_exporter: PrometheusExporter | None = None,
    rbac_enforcer: RBACEnforcer | None = None,
    *,
    shadow_mode: bool = False,
    drift_monitor: DriftMonitor | None = None,
    telemetry_emitter: TelemetryEmitter | None = None,
) -> PCWDecision:

from aegis_governance import (
    pcw_decide, PCWContext, PCWPhase, PCWStatus, AegisConfig
)

config = AegisConfig.default()
evaluator = config.create_gate_evaluator()
drift_monitor = config.create_drift_monitor()
drift_monitor.set_baseline([0.1, 0.12, 0.11, 0.13] * 10)

decision = pcw_decide(
    PCWContext(
        agent_id="deploy-bot",
        session_id="deploy-123",
        phase=PCWPhase.COMMIT,
        proposal_summary="Deploy v2.5 to production",
        estimated_impact="high",
        risk_baseline=0.1,
        risk_proposed=0.15,
        profit_baseline=1000.0,
        profit_proposed=1200.0,
        novelty_score=0.3,
        complexity_score=0.8,
        quality_score=0.9,
        requires_human_approval=True,
    ),
    gate_evaluator=evaluator,
    drift_monitor=drift_monitor,
)

print(f"Status: {decision.status.value}")
print(f"Confidence: {decision.confidence}")
print(f"Rationale: {decision.rationale}")
print(f"Next steps: {decision.next_steps}")

if decision.status == PCWStatus.PROCEED:
    # Safe to continue
    pass
elif decision.status == PCWStatus.HALT:
    # Must stop -- check rationale
    pass
elif decision.status in (PCWStatus.PAUSE, PCWStatus.ESCALATE):
    # Requires human review
    pass

def quick_risk_check(
    agent_id: str,
    action_description: str,
    risk_score: float,
    threshold: float = 0.5,
) -> bool:

from aegis_governance import quick_risk_check

if quick_risk_check("my-agent", "Update config file", risk_score=0.1):
    print("Safe to proceed")

def require_human_approval(
    context: PCWContext,
    gate_evaluator: GateEvaluator | None = None,
) -> bool:

from aegis_governance import require_human_approval, PCWContext, PCWPhase

ctx = PCWContext(
    agent_id="agent-1",
    session_id="sess-1",
    phase=PCWPhase.PLAN,
    proposal_summary="Delete production database",
    estimated_impact="critical",
    reversible=False,
)

if require_human_approval(ctx):
    print("Human approval required before proceeding")

from aegis_governance import GateEvaluator, AegisConfig

config = AegisConfig.default()
evaluator = config.create_gate_evaluator()

# Evaluate individual gates
risk_result = evaluator.evaluate_risk_gate(baseline=0.1, proposed=0.3)
profit_result = evaluator.evaluate_profit_gate(baseline=100, proposed=120)
novelty_result = evaluator.evaluate_novelty_gate(novelty_score=0.85)
complexity_result = evaluator.evaluate_complexity_gate(complexity_score=0.7)
quality_result = evaluator.evaluate_quality_gate(
    quality_score=0.8, quality_subscores=[0.9, 0.7, 0.8]
)

# Evaluate all gates at once
full_result = evaluator.evaluate_all(
    risk_baseline=0.1,
    risk_proposed=0.15,
    profit_baseline=100,
    profit_proposed=120,
    novelty_score=0.85,
    complexity_score=0.7,
    quality_score=0.8,
    quality_subscores=[0.9, 0.7, 0.8],
    utility_result=utility_result,
)

if full_result.all_passed:
    print("All gates passed")
else:
    for failure in full_result.get_failures():
        print(f"FAILED: {failure.gate_type.value} "
              f"(value={failure.value:.3f}, threshold={failure.threshold:.3f})")

from aegis_governance import HTTPEventSink

sink = HTTPEventSink(
    url="https://telemetry.example.com/events",
    headers={"Authorization": "Bearer ${TOKEN}"},
    timeout=10,
)

from aegis_governance import BatchHTTPSink

sink = BatchHTTPSink(
    url="https://telemetry.example.com/events/batch",
    batch_size=100,
    flush_interval_seconds=60,
    max_retries=1,
)

from aegis_governance import http_sink

# Single-event sink (no batching)
sink = http_sink(url="https://telemetry.example.com/events")

# Batching sink
sink = http_sink(
    url="https://telemetry.example.com/events",
    batch_size=50,
)

from aegis_governance import pcw_decide, PCWContext, PCWPhase, AegisConfig
from aegis_governance import HTTPEventSink
from telemetry.emitter import TelemetryEmitter

# Create emitter with HTTP sink
emitter = TelemetryEmitter(source="my-service")
emitter.add_sink(HTTPEventSink(
    url="https://telemetry.example.com/events",
    allow_insecure=False,
))

config = AegisConfig.default()
decision = pcw_decide(
    PCWContext(
        agent_id="agent-1",
        session_id="sess-1",
        phase=PCWPhase.PLAN,
        proposal_summary="Example",
        estimated_impact="medium",
    ),
    gate_evaluator=config.create_gate_evaluator(),
    telemetry_emitter=emitter,
)