EVE CoreGuard Integration Guide: Connecting Your LLM Pipeline in Under 30 Minutes

pip install eve-coreguard

eve-coreguard>=0.1.0,<1.0.0

# Required
COREGUARD_API_KEY=cg_live_sk_xxxxxxxxxxxxxxxxxxxxxxxx
COREGUARD_ENDPOINT=https://api.eveaicore.com/v1

# Optional — defaults shown
COREGUARD_POLICY_SET=lending_v1
COREGUARD_TIMEOUT_MS=5000
COREGUARD_MAX_RETRIES=3
COREGUARD_RETRY_BACKOFF_MS=200
COREGUARD_WEBHOOK_SECRET=whsec_xxxxxxxxxxxxxxxxxxxxxxxx

from eve_coreguard import CoreGuardClient, EvaluationRequest
from eve_coreguard.exceptions import CoreGuardBlockedError, CoreGuardError
import os

# Instantiate once at module level — the client manages connection pooling internally
client = CoreGuardClient(
    api_key=os.environ["COREGUARD_API_KEY"],
    endpoint=os.environ["COREGUARD_ENDPOINT"],
    policy_set=os.environ.get("COREGUARD_POLICY_SET", "lending_v1"),
)

def process_loan_request(user: dict, action: dict, context: dict) -> dict:
    # Step 1: Evaluate the proposed action before touching the model
    try:
        request = EvaluationRequest(
            policy_set="lending_v1",
            user=user,
            action=action,
            context=context,
        )
        result = client.evaluate(request)
    except CoreGuardBlockedError as e:
        # Hard block — return the policy violation to the caller
        return {"error": "blocked", "violations": [v.rule_id for v in e.violations]}
    except CoreGuardError as e:
        # Infrastructure error — log and fail closed (reject the action)
        log_error(e)
        return {"error": "governance_unavailable"}

    # Step 2: Apply any required modifications before invoking the model
    if result.decision.status == "MODIFIED":
        action = apply_modifications(action, result.decision.modifications)

    # Step 3: Invoke your LLM — governance is complete
    llm_response = your_llm_client.generate(action)

    # Step 4: Attach the audit certificate to your response record
    return {
        "response": llm_response,
        "audit_certificate_id": result.audit_record.certificate_id,
        "decision": result.decision.status,
    }

import os, asyncio
from fastapi import FastAPI, HTTPException
from eve_coreguard import CoreGuardClient
from eve_coreguard.exceptions import VetoError

app = FastAPI()

# Instantiate once at startup and reuse — the client pools HTTP connections.
cg = CoreGuardClient(api_key=os.environ["COREGUARD_API_KEY"])

@app.post("/api/v1/decisions")
async def make_decision(payload: dict):
    # Offload the blocking call to a thread so the event loop stays free.
    try:
        result = await asyncio.to_thread(
            cg.evaluate,
            policy_set=payload["policy_set"],
            user=payload["user"],
            proposed_action=payload["proposed_action"],
            model_output=payload.get("model_output"),
            context=payload.get("context", {}),
        )
    except VetoError as e:
        raise HTTPException(status_code=422, detail={"status": "BLOCKED", "detail": str(e)})

    return {
        "status": result.decision.status,
        "risk_level": result.risk.level,
    }

curl -X POST https://api.eveaicore.com/v1/webhooks \
  -H "Authorization: Bearer $COREGUARD_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{
    "url": "https://your-siem.example.com/coreguard-events",
    "events": ["decision.evaluated", "decision.blocked", "certificate.issued"],
    "secret": "whsec_xxxxxxxxxxxxxxxxxxxxxxxx"
  }'

import hashlib, hmac, os
from fastapi import Request, HTTPException

WEBHOOK_SECRET = os.environ["COREGUARD_WEBHOOK_SECRET"].encode()

async def verify_coreguard_webhook(request: Request) -> bytes:
    body = await request.body()
    expected = hmac.new(WEBHOOK_SECRET, body, hashlib.sha256).hexdigest()
    received = request.headers.get("X-CoreGuard-Signature", "")
    if not hmac.compare_digest(expected, received):
        raise HTTPException(status_code=401, detail="Invalid webhook signature")
    return body

@app.post("/coreguard-events")
async def coreguard_webhook(request: Request):
    body = await verify_coreguard_webhook(request)
    event = json.loads(body)
    # Forward to your SIEM or compliance store
    await siem_client.ingest(event)
    return {"received": True}

FROM python:3.11-slim

WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY . .

# COREGUARD_API_KEY and COREGUARD_ENDPOINT are injected at runtime — NOT here
CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8080"]

docker run \
  -e COREGUARD_API_KEY="$(vault kv get -field=api_key secret/coreguard)" \
  -e COREGUARD_ENDPOINT="https://api.eveaicore.com/v1" \
  -p 8080:8080 \
  your-org/your-llm-service:latest

apiVersion: apps/v1
kind: Deployment
spec:
  template:
    spec:
      containers:
      - name: llm-service
        image: your-org/llm-service:latest
        env:
        - name: COREGUARD_ENDPOINT
          value: "http://localhost:8001/v1"  # sidecar — loopback only
        - name: COREGUARD_API_KEY
          valueFrom:
            secretKeyRef:
              name: coreguard-credentials
              key: api-key

      - name: coreguard-sidecar
        image: evecore/coreguard-sidecar:1.3
        ports:
        - containerPort: 8001
          name: governance
        env:
        - name: COREGUARD_UPSTREAM
          value: "https://api.eveaicore.com/v1"
        - name: COREGUARD_API_KEY
          valueFrom:
            secretKeyRef:
              name: coreguard-credentials
              key: api-key
        resources:
          requests: { cpu: "50m", memory: "64Mi" }
          limits:  { cpu: "200m", memory: "128Mi" }

from eve_coreguard.testing import MockCoreGuardClient
import pytest

def test_blocked_loan_is_rejected():
    # Configure the mock to return BLOCKED for this scenario
    mock_client = MockCoreGuardClient(
        default_response="BLOCKED",
        violations=[{"rule_id": "ecoa.disparate_impact", "severity": "HIGH"}]
    )

    result = process_loan_request_with_client(
        client=mock_client,
        user={"id": "u_test"},
        action={"type": "loan_approval", "amount": 50000},
        context={"credit_score": 720},
    )

    assert result["error"] == "blocked"
    assert "ecoa.disparate_impact" in result["violations"]

def test_compliant_loan_is_allowed():
    mock_client = MockCoreGuardClient(default_response="ALLOWED")
    result = process_loan_request_with_client(
        client=mock_client,
        user={"id": "u_test"},
        action={"type": "loan_approval", "amount": 25000},
        context={"credit_score": 760},
    )
    assert "audit_certificate_id" in result
    assert result["decision"] == "ALLOWED"

from eve_coreguard import verify_decision_record, fetch_public_key_pem

# Load the stored decision record (database, object store, etc.)
record = load_decision_record(decision_id="dec_abc123")

# Ed25519 records verify with the PUBLIC key alone — no shared secret.
pubkey = fetch_public_key_pem(os.environ["COREGUARD_ENDPOINT"])  # /.well-known/eve-pubkey
result = verify_decision_record(record, public_key_pem=pubkey)
assert result.valid, "Record tampered — hash or signature mismatch"
assert result.independently_verifiable, "Expected an asymmetric (public-key) record"

# Legacy hmac- records are shared-secret only (NOT third-party verifiable):
# verify_decision_record(record, hmac_key=os.environ["COREGUARD_HMAC_KEY"])