Python SDK Reference

# Core SDK (from repo root)
pip install -e "."

# With optional dependencies
pip install -e ".[engine]"       # scipy for z-score computation
pip install -e ".[telemetry]"    # prometheus_client for metrics
pip install -e ".[config]"       # pyyaml for YAML config loading
pip install -e ".[crypto]"       # BIP-340 Schnorr signatures
pip install -e ".[persistence]"  # SQLAlchemy async workflow storage
pip install -e ".[kms]"          # AWS KMS envelope encryption
pip install -e ".[hsm]"          # HSM PKCS#11 envelope encryption
pip install -e ".[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,
)

from aegis_governance import get_kek_provider

provider = get_kek_provider()          # auto-detect
provider = get_kek_provider("kms")     # force AWS KMS
provider = get_kek_provider("hsm")     # force PKCS#11 HSM

from aegis_governance import AWSKMSKEKProvider

# From explicit parameters
provider = AWSKMSKEKProvider(
    kms_key_id="arn:aws:kms:us-west-2:123456789012:key/...",
    wrapped_private_key=b"...",
    public_key_bytes=b"...",
)

# From environment variables (recommended)
provider = AWSKMSKEKProvider.from_environment()

from aegis_governance import HSMKEKProvider

# From environment variables (recommended)
provider = HSMKEKProvider.from_environment()

from aegis_governance import KMS_KEK_AVAILABLE, HSM_KEK_AVAILABLE, HYBRID_KEM_AVAILABLE

if KMS_KEK_AVAILABLE:
    provider = get_kek_provider("kms")

from aegis_governance import Approver
from crypto.bip322_provider import BIP322Provider

approver = Approver(
    actor_id="risk-lead-1",
    name="Risk Lead",
    is_risk_lead=True,
    signature_verifier=BIP322Provider(),
)

result = approver.authorize_override(
    proposal_id="prop-001",
    failed_gates=["risk"],
    justification="Emergency hotfix required",
    signature=risk_sig_b64,       # base64-encoded BIP-322 signature
    public_key=risk_public_key,   # 32-byte x-only public key
)

from aegis_governance import DualSignatureValidator, OverrideWorkflow

# Create validator and message hash
validator = DualSignatureValidator()
msg_hash = validator.create_message_hash(
    proposal_id="prop-001",
    justification="Emergency hotfix",
    failed_gates=["risk"],
)

# Use OverrideWorkflow for full two-key flow
workflow = OverrideWorkflow(
    proposal_id="prop-001",
    justification="Emergency hotfix",
    failed_gates=["risk"],
    requested_by="engineer-1",
)

# Each custodian adds their signature
workflow.add_signature("risk-lead-1", "risk_lead", risk_sig_b64, risk_pub_b64)
workflow.add_signature("sec-lead-1", "security_lead", sec_sig_b64, sec_pub_b64)

result = workflow.get_result()
if result.approved:
    print(f"Override approved until {result.expires_at}")