CoreGuard API Tutorial: Implementing AI Policy Enforcement in Your LLM Pipeline

Every regulated AI deployment shares a common architectural problem: the gap between what an LLM generates and what compliance policy allows is not enforced at the model layer — it is enforced, if at all, by the application wrapping the model. Most teams solve this with manual review queues, post-hoc audits, or brittle prompt engineering. None of these approaches meet the bar required by ECOA, SR 11-7, the EU AI Act, or HIPAA enforcement guidance.

CoreGuard closes that gap deterministically. It sits between your application logic and your LLM output pathway, evaluating every proposed action against a machine-readable policy pack before the response reaches your user or triggers a downstream process. This tutorial walks through a complete integration from first API call to production-grade error handling, including the signed audit certificate that proves each decision was governed.

1. Understanding the CoreGuard Decision Model

Before writing a single line of code, it helps to understand what CoreGuard is actually doing. CoreGuard is not an output filter that scans for keywords. It is a decision enforcement engine that evaluates structured proposals — an action type, a set of parameters, contextual metadata, and the identity of the requester — against a versioned policy pack that encodes the compliance rules for a specific regulated domain.

The evaluation pipeline is deterministic and synchronous. Given the same inputs, CoreGuard always produces the same outputs. There is no probabilistic model in the enforcement path. This matters because your auditors need to be able to reproduce any decision from the audit record, and stochastic enforcement cannot provide that guarantee.

Each evaluation returns one of three dispositions:

The MODIFY disposition is particularly important for regulated lending and insurance workflows. Rather than hard-blocking a loan approval that slightly exceeds a safe-harbor ceiling, CoreGuard can return the maximum permissible amount with a note explaining the adjustment. Your application can then present the modified offer to the user, maintaining compliance while preserving the user experience.

2. Authentication and Base URLs

CoreGuard uses API key authentication via the Authorization header. Keys are scoped to an organization and a set of permitted policy packs. Your key is prefixed with cg_live_ for production and cg_test_ for sandbox.

3. Your First Request: POST /v1/decisions/evaluate

The primary endpoint is POST /v1/decisions/evaluate. Every CoreGuard evaluation, regardless of domain, uses this single endpoint. The policy pack you specify in the request body determines which rules apply.

Request Schema

cURL Example

# Evaluate a loan approval action against the lending_v1 policy pack
curl -X POST https://sandbox.api.eveaicore.com/v1/decisions/evaluate \
  -H "Authorization: Bearer cg_test_YOUR_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{
    "policy_set": "lending_v1",
    "request_id": "req_2026_abc123",
    "user": {
      "id": "officer_007",
      "role": "loan_officer",
      "tenant_id": "bank_midwest_001"
    },
    "action": {
      "type": "loan_approval",
      "amount": 75000,
      "product": "auto_loan",
      "term_months": 60,
      "apr": 7.25
    },
    "context": {
      "credit_score": 720,
      "dti_ratio": 0.38,
      "income_verified": true,
      "application_id": "app_20260505_001"
    }
  }'

Python Example

import requests
import json

COREGUARD_API_KEY = "cg_test_YOUR_API_KEY"
COREGUARD_BASE_URL = "https://sandbox.api.eveaicore.com"

def evaluate_loan_approval(application: dict) -> dict:
    """Evaluate a loan approval action against CoreGuard lending_v1 policy."""
    payload = {
        "policy_set": "lending_v1",
        "request_id": f"req_{application['application_id']}",
        "user": {
            "id": application["officer_id"],
            "role": "loan_officer",
            "tenant_id": application["branch_id"]
        },
        "action": {
            "type": "loan_approval",
            "amount": application["requested_amount"],
            "product": application["product_type"],
            "term_months": application["term_months"],
            "apr": application["offered_apr"]
        },
        "context": {
            "credit_score": application["credit_score"],
            "dti_ratio": application["dti_ratio"],
            "income_verified": application["income_verified"],
            "application_id": application["application_id"]
        },
        "strict_mode": True
    }
    response = requests.post(
        f"{COREGUARD_BASE_URL}/v1/decisions/evaluate",
        headers={
            "Authorization": f"Bearer {COREGUARD_API_KEY}",
            "Content-Type": "application/json"
        },
        json=payload,
        timeout=5  # Always set a timeout — see fail-closed section
    )
    response.raise_for_status()
    return response.json()

4. Parsing the Response

A successful response always returns HTTP 200, regardless of whether the disposition is ALLOW, BLOCK, or MODIFY. HTTP error codes are reserved for API-level failures (authentication, malformed request, service unavailable). This design is intentional: it keeps your application logic clean — you always parse a valid decision object, never catch an exception to determine compliance status.

{
  "decision_id": "dec_2xK9p_20260505T143022Z",
  "request_id": "req_app_20260505_001",
  "policy_set": "lending_v1",
  "policy_version": "1.4.2",
  "timestamp": "2026-05-05T14:30:22.441Z",
  "decision": {
    "status": "ALLOW",
    "risk_level": "LOW",
    "rules_evaluated": 12,
    "rules_triggered": 0,
    "violations": [],
    "modified_params": null,
    "reasoning": "All 12 lending_v1 rules evaluated. No violations detected."
  },
  "certificate": {
    "algorithm": "HMAC-SHA256",
    "signature": "3a7f2c9e1b4d8a6f0e3c5b2d9a7e4c1f...",
    "payload_hash": "sha256:e3b0c44298fc1c149afbf...",
    "signed_at": "2026-05-05T14:30:22.448Z"
  },
  "latency_ms": 7
}

{
  "decision_id": "dec_7mQ3r_20260505T143155Z",
  "request_id": "req_app_20260505_002",
  "policy_set": "lending_v1",
  "policy_version": "1.4.2",
  "timestamp": "2026-05-05T14:31:55.021Z",
  "decision": {
    "status": "BLOCK",
    "risk_level": "HIGH",
    "rules_evaluated": 12,
    "rules_triggered": 2,
    "violations": [
      {
        "rule_id": "lending_v1.dti_ceiling",
        "severity": "HIGH",
        "message": "DTI ratio 0.52 exceeds safe-harbor ceiling of 0.43 per lending_v1 §4.2",
        "regulatory_reference": "CFPB Ability-to-Repay Rule, 12 CFR 1026.43"
      },
      {
        "rule_id": "lending_v1.adverse_action_required",
        "severity": "MEDIUM",
        "message": "Adverse action notice required under ECOA. See adverse_action_template.",
        "regulatory_reference": "15 U.S.C. § 1691(d), Regulation B"
      }
    ],
    "modified_params": null,
    "reasoning": "DTI ratio exceeds maximum permitted under ability-to-repay rules. Adverse action notice required."
  },
  "certificate": {
    "algorithm": "HMAC-SHA256",
    "signature": "b9e4d2a7c5f0e3b8a1d4c7f2e5b8a3d6...",
    "payload_hash": "sha256:a1b2c3d4e5f6a7b8c9d0...",
    "signed_at": "2026-05-05T14:31:55.029Z"
  },
  "latency_ms": 9
}

Notice that BLOCK responses also include a signed certificate. This is intentional: the certificate proves that a governance decision was made, not merely that the system was unavailable. Your audit trail needs to record both the decision and the proof that the enforcement infrastructure was functioning correctly at the time the decision was made.

5. Handling the MODIFY Disposition

The MODIFY disposition is the most operationally complex of the three. When CoreGuard returns MODIFY, it means the original action parameters violated at least one policy rule, but a compliant version of the action is possible. The modified_params field contains the adjusted values.

{
  "decision": {
    "status": "MODIFY",
    "risk_level": "MEDIUM",
    "rules_evaluated": 12,
    "rules_triggered": 1,
    "violations": [
      {
        "rule_id": "lending_v1.amount_ceiling",
        "severity": "MEDIUM",
        "message": "Requested amount $85,000 exceeds computed ceiling of $72,400 for this credit profile",
        "regulatory_reference": "Internal credit policy §7.4.1"
      }
    ],
    "modified_params": {
      "amount": 72400,
      "modification_reason": "Amount reduced to maximum permissible for credit profile",
      "original_amount": 85000
    }
  }
}

def process_loan_decision(application: dict) -> dict:
    result = evaluate_loan_approval(application)
    decision = result["decision"]
    status = decision["status"]

    if status == "ALLOW":
        # Proceed with original parameters
        return {
            "approved": True,
            "amount": application["requested_amount"],
            "decision_id": result["decision_id"],
            "certificate": result["certificate"]
        }

    elif status == "MODIFY":
        # Use modified parameters — do NOT use original values
        modified = decision["modified_params"]
        return {
            "approved": True,
            "amount": modified["amount"],
            "modification_note": modified["modification_reason"],
            "decision_id": result["decision_id"],
            "certificate": result["certificate"]
        }

    elif status == "BLOCK":
        # Collect violations for adverse action notice
        violation_reasons = [
            v["message"] for v in decision["violations"]
        ]
        return {
            "approved": False,
            "denial_reasons": violation_reasons,
            "adverse_action_required": any(
                "adverse_action_required" in v["rule_id"]
                for v in decision["violations"]
            ),
            "decision_id": result["decision_id"],
            "certificate": result["certificate"]
        }

    else:
        raise ValueError(f"Unknown disposition: {status}")

6. Verifying HMAC-SHA256 Audit Certificates

Every CoreGuard decision includes a cryptographic certificate. This certificate is an HMAC-SHA256 signature computed over the canonical JSON representation of the decision record. It proves two things: (1) that a specific decision was made at a specific time, and (2) that the decision record has not been tampered with since it was signed.

Certificate verification is offline — it does not require a round-trip to the CoreGuard API. You need the signing key associated with your environment. In production, this is derived from your organization's HSM-backed key. In sandbox, it is your API key's associated signing secret, available in the dashboard under "Certificate Keys."

import hashlib
import hmac
import json
import base64

def verify_decision_certificate(decision_response: dict, signing_key: bytes) -> bool:
    """
    Verify a CoreGuard HMAC-SHA256 decision certificate offline.

    Args:
        decision_response: The full JSON response from CoreGuard
        signing_key: Your organization's certificate signing key (bytes)

    Returns:
        True if the certificate is valid, False if tampered or invalid
    """
    certificate = decision_response.get("certificate", {})
    received_signature = certificate.get("signature", "")

    # Build the canonical payload (same construction as CoreGuard server)
    canonical = {
        "decision_id": decision_response["decision_id"],
        "request_id": decision_response["request_id"],
        "policy_set": decision_response["policy_set"],
        "policy_version": decision_response["policy_version"],
        "timestamp": decision_response["timestamp"],
        "decision": decision_response["decision"]
    }
    # JCS-style: sorted keys, no whitespace
    canonical_bytes = json.dumps(canonical, sort_keys=True, separators=(',', ':')).encode("utf-8")

    # Compute expected signature
    expected = hmac.new(signing_key, canonical_bytes, hashlib.sha256).hexdigest()

    # Constant-time comparison prevents timing attacks
    return hmac.compare_digest(received_signature, expected)


# Usage
SIGNING_KEY = bytes.fromhex("YOUR_CERTIFICATE_SIGNING_KEY_HEX")

response = evaluate_loan_approval(application)
is_valid = verify_decision_certificate(response, SIGNING_KEY)

if not is_valid:
    raise SecurityError("Certificate verification failed — decision record may be tampered")

7. Production Error Handling and Fail-Closed Behavior

Your integration must handle three distinct failure scenarios correctly. Getting this wrong is not a theoretical risk — in regulated environments, an LLM pipeline that allows actions through during a governance system outage creates real compliance exposure.

HTTP 4xx Errors — Client Errors

These indicate problems with your request. Common causes and remediation:

HTTP 5xx Errors and Timeouts — Fail-Closed Pattern

When CoreGuard returns a 5xx or your request times out, your application must decide: allow the action or block it? For regulated workflows, the answer is always block — a governance system that fails open during outages provides no compliance guarantee.

import requests
from requests.exceptions import Timeout, ConnectionError, HTTPError
import logging

logger = logging.getLogger("coreguard.integration")

def evaluate_with_failsafe(payload: dict) -> dict:
    """
    Evaluate an action with fail-closed error handling.
    Returns a synthetic BLOCK decision if CoreGuard is unreachable.
    """
    try:
        response = requests.post(
            f"{COREGUARD_BASE_URL}/v1/decisions/evaluate",
            headers={"Authorization": f"Bearer {COREGUARD_API_KEY}"},
            json=payload,
            timeout=5
        )
        response.raise_for_status()
        return response.json()

    except Timeout:
        logger.error("CoreGuard timeout after 5s — applying fail-closed BLOCK",
                      extra={"request_id": payload.get("request_id")})
        return _synthetic_block("GOVERNANCE_TIMEOUT", payload)

    except ConnectionError:
        logger.error("CoreGuard unreachable — applying fail-closed BLOCK")
        return _synthetic_block("GOVERNANCE_UNREACHABLE", payload)

    except HTTPError as e:
        if e.response.status_code >= 500:
            logger.error(f"CoreGuard server error {e.response.status_code}")
            return _synthetic_block("GOVERNANCE_SERVER_ERROR", payload)
        raise  # Re-raise 4xx — these are caller errors


def _synthetic_block(reason: str, payload: dict) -> dict:
    """Synthetic BLOCK decision for fail-closed scenarios — no certificate."""
    return {
        "decision_id": "synthetic",
        "request_id": payload.get("request_id"),
        "decision": {
            "status": "BLOCK",
            "risk_level": "HIGH",
            "violations": [{
                "rule_id": f"system.{reason.lower()}",
                "severity": "HIGH",
                "message": f"Governance system unavailable ({reason}) — action blocked"
            }]
        },
        "certificate": None,  # No certificate for synthetic blocks
        "synthetic": True
    }

8. Idempotency and Retry Strategy

CoreGuard supports idempotency through the request_id field. If you supply a request_id and submit the same request twice within the idempotency window (24 hours), CoreGuard returns the cached response rather than evaluating the request a second time. This is critical for retry logic: you should always include a stable request_id derived from the underlying business transaction, not generated freshly on each retry.

Recommended retry strategy for transient failures (5xx, timeouts): exponential backoff starting at 100ms, maximum 3 retries, maximum total wait 4.5 seconds. Do not retry 4xx errors — they indicate a request problem that will not resolve on retry.

9. Listing Available Policy Packs

To discover which policy packs are available and what action types they govern, use the GET /v1/policy-packs endpoint:

curl https://api.eveaicore.com/v1/policy-packs \
  -H "Authorization: Bearer cg_live_YOUR_KEY"

The response includes the pack identifier, current version, a list of supported action types, and the regulatory frameworks the pack encodes. Standard packs included with CoreGuard as of version 1.4.2:

• lending_v1 — ECOA, FCRA, CFPB Ability-to-Repay, Regulation B, CRA. Action types: loan_approval, credit_limit_change, adverse_action_notice, rate_adjustment
• hipaa_v1 — HIPAA Privacy Rule, Security Rule, Breach Notification Rule. Action types: phi_disclosure, clinical_recommendation, record_access, diagnosis_assist
• sr11_7_v1 — Federal Reserve SR 11-7 Model Risk Management. Action types: model_output_production, model_validation, override_review
• eu_ai_act_v1 — EU AI Act Articles 9–17 (high-risk AI systems). Action types: automated_decision, biometric_processing, critical_infrastructure_control
• custom — Customer-defined policy packs. Available on Enterprise tier.

10. Integrating CoreGuard into a FastAPI Application

The following example shows a complete FastAPI route that enforces CoreGuard governance on a loan origination endpoint. This pattern is production-ready and includes the certificate storage required for audit trail completeness.

from fastapi import FastAPI, HTTPException, Depends
from pydantic import BaseModel
import uuid

app = FastAPI()

class LoanApplicationRequest(BaseModel):
    applicant_id: str
    requested_amount: float
    product_type: str
    term_months: int
    offered_apr: float
    credit_score: int
    dti_ratio: float
    income_verified: bool

@app.post("/loans/evaluate")
async def evaluate_loan(req: LoanApplicationRequest, officer_id: str = Depends(get_officer)):
    application_id = str(uuid.uuid4())

    # Evaluate against CoreGuard before touching any downstream systems
    cg_response = evaluate_with_failsafe({
        "policy_set": "lending_v1",
        "request_id": application_id,
        "user": {"id": officer_id, "role": "loan_officer"},
        "action": {
            "type": "loan_approval",
            "amount": req.requested_amount,
            "product": req.product_type,
            "term_months": req.term_months,
            "apr": req.offered_apr
        },
        "context": {
            "credit_score": req.credit_score,
            "dti_ratio": req.dti_ratio,
            "income_verified": req.income_verified,
            "application_id": application_id
        }
    })

    decision = cg_response["decision"]

    # Always persist the decision record with certificate
    await store_governance_record(application_id, cg_response)

    if decision["status"] == "BLOCK":
        return {
            "approved": False,
            "application_id": application_id,
            "denial_reasons": [v["message"] for v in decision["violations"]],
            "decision_id": cg_response["decision_id"]
        }

    # ALLOW or MODIFY — proceed with effective amount
    effective_amount = (
        decision["modified_params"]["amount"]
        if decision["status"] == "MODIFY"
        else req.requested_amount
    )

    return {
        "approved": True,
        "application_id": application_id,
        "approved_amount": effective_amount,
        "decision_id": cg_response["decision_id"]
    }

Frequently Asked Questions