complianceguard 0.1.1

Python SDK client for the ComplianceGuard HTTP API

complianceguard

Python client for the ComplianceGuard HTTP API.

This SDK supports the current MVP evidence contract:

• decision_event() / decision_event_async()
• metric_snapshot() / metric_snapshot_async()
• human_intervention() / human_intervention_async()
• dataset_registration()
• model_release()
• log_human_feedback()

Async event delivery is disk-backed. log_async, decision_event_async, metric_snapshot_async, and human_intervention_async durably spool events, replay pending files on startup, retry with backoff, and move exhausted deliveries to a dead-letter directory instead of silently dropping them.

Install

pip install complianceguard

Decision events

from complianceguard import ComplianceGuardClient

client = ComplianceGuardClient(
    base_url="http://localhost:5214",
    api_key="<environment-api-key>",
    spool_directory="/var/lib/complianceguard-sdk",
)

client.decision_event_async(
    model_id="credit-score-v1",
    version="2026.04.15",
    input={"applicant_id": "A-1001", "income": 65000},
    output={"score": 0.82, "decision": "approve"},
    metadata={"team": "risk", "feature_set": "baseline"},
    correlation_id="decision-1001",
    model_release_id="<model-release-id>",
)

client.flush()

log() and log_async() remain as backward-compatible aliases for typed decision events.

Metric snapshots

client.metric_snapshot(
    model_id="credit-score-v1",
    version="2026.04.15",
    metrics={"accuracy": 0.98, "drift": 0.03},
    metadata={"window": "5m", "service": "monitoring"},
    model_release_id="<model-release-id>",
)

Human intervention

client.human_intervention(
    model_id="credit-score-v1",
    version="2026.04.15",
    input={"applicant_id": "A-1001"},
    original_output={"decision": "deny"},
    final_output={"decision": "manual_review"},
    reviewer_display="Senior Analyst",
    intervention_type="overrode",
    notes="Escalated due to conflicting source records.",
    model_release_id="<model-release-id>",
)

Governance registration

dataset = client.dataset_registration(
    name="credit-review-corpus",
    version="2026.04",
    usage="validation",
    source="approved analyst review queue",
    provenance="sampled from manually reviewed decisions",
    record_count=3200,
    metadata={"region": "eu"},
)

release = client.model_release(
    environment_id="<environment-id>",
    name="credit-score-prod",
    model_id="credit-score-v1",
    version="2026.04.15",
    status="active",
    architecture_summary="transformer ensemble",
    intended_use="credit underwriting support",
    system_version="sys-2026.04.15",
    parameter_count=125000000,
    training_data_hash="abc123",
)

Backend responses use the API's snake_case JSON contract. The SDK normalizes responses before returning them so Python callers receive stable camelCase dictionaries.

Decorator tracing

from complianceguard import ComplianceGuardClient

client = ComplianceGuardClient(
    base_url="http://localhost:5214",
    api_key="<environment-api-key>",
    spool_directory="/var/lib/complianceguard-sdk",
)

@client.trace(
    model_id="credit-score-v1",
    version="2026.04.15",
    metadata={"service": "underwriting"},
)
def score_application(application: dict) -> dict:
    return {"score": 0.82, "decision": "approve"}

score_application({"applicant_id": "A-1001", "income": 65000})
client.flush()

Context manager tracing

from complianceguard import ComplianceGuardClient

client = ComplianceGuardClient(
    base_url="http://localhost:5214",
    api_key="<environment-api-key>",
    spool_directory="/var/lib/complianceguard-sdk",
)

with client.trace_context(
    model_id="credit-score-v1",
    version="2026.04.15",
    input={"applicant_id": "A-1001", "income": 65000},
    metadata={"service": "underwriting"},
    correlation_id="decision-1001",
) as trace:
    trace.output = {"score": 0.82, "decision": "approve"}

client.flush()

Delivery semantics

• Synchronous log() posts directly and generates an idempotency key if one is not provided.
• Asynchronous log_async() writes a spool file before background delivery begins.
• Pending spool files are replayed when the client starts again with the same spool_directory.
• Exhausted deliveries are moved to dead-letter/.

Human feedback

client.log_human_feedback(
    event_id="<event-id>",
    reviewer_display="Senior Analyst",
    label="approved",
    score=1.0,
    notes="Manual review confirmed the decision.",
    feedback={"review_ticket": "REV-1024"},
)

For correlation-level feedback:

client.log_human_feedback(
    correlation_id="decision-1001",
    environment_id="<environment-id>",
    reviewer_display="Senior Analyst",
    label="escalated",
    notes="Grouped review across the full decision chain.",
    feedback={"reason": "manual escalation"},
)

Test

python -m unittest discover -s tests -v