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 — read by YOUR code and passed to CoreGuardClient(api_key=...)
COREGUARD_API_KEY=eve_sk_xxxxxxxxxxxxxxxxxxxxxxxx

# Optional — passed as base_url=...; omit to use the SDK default
COREGUARD_BASE_URL=https://api.evecore.com

import os
from eve_coreguard import CoreGuardClient, CoreGuardError

# Create once at module level and reuse.
client = CoreGuardClient(
    api_key=os.environ["COREGUARD_API_KEY"],
    base_url=os.environ.get("COREGUARD_BASE_URL", "https://api.evecore.com"),
)

def process_loan_request(user: dict, action: dict, context: dict) -> dict:
    # Step 1: evaluate the proposed action before touching the model.
    # action["type"] (e.g. "loan_approval") selects the bound pack server-side.
    try:
        result = client.evaluate(
            tenant_id="bank_001",
            user=user,
            proposed_action=action,
            context=context,
        )
    except CoreGuardError as e:
        # No decision was obtained (network, auth, quota) — fail closed.
        log_error(e)
        return {"error": "governance_unavailable"}

    if result.blocked:
        # Policy refusal — return the rule ids that fired.
        return {"error": "blocked",
                "violations": [v.policy_id for v in result.policy_violations]}

    # Step 2: apply any field edits CoreGuard made before invoking the model.
    if result.verdict == "MODIFIED":
        action = apply_modifications(action, result.modifications)

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

    # Step 4: keep the decision id with your record — it is your evidence handle.
    return {
        "response": llm_response,
        "decision_id": result.decision_id,
        "decision": result.verdict,
    }

import os, asyncio
from fastapi import FastAPI, HTTPException
from eve_coreguard import CoreGuardClient, CoreGuardError

app = FastAPI()
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,
            tenant_id=payload["tenant_id"],
            user=payload["user"],
            proposed_action=payload["proposed_action"],
            model_output=payload.get("model_output"),
            context=payload.get("context", {}),
        )
    except CoreGuardError:
        raise HTTPException(status_code=503, detail="governance unavailable")

    if result.blocked:
        raise HTTPException(status_code=422, detail={"status": "BLOCKED",
                            "decision_id": result.decision_id})
    return {"status": result.verdict, "risk_level": result.risk.level}

curl -X POST https://api.evecore.com/api/sdk/webhooks \
  -H "Authorization: Bearer $COREGUARD_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{
    "url": "https://your-siem.example.com/coreguard-events",
    "events": ["governance.charter.veto", "governance.action"]
  }'

import hashlib, hmac, json, 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-EVE-Webhook-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 optionally COREGUARD_BASE_URL) 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)" \
  -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_API_KEY
          valueFrom:
            secretKeyRef:
              name: coreguard-credentials
              key: api-key

from eve_coreguard import EvaluationResult

class StubClient:
    def __init__(self, response: dict):
        self._response = response
    def evaluate(self, **kwargs):
        return EvaluationResult.from_dict(self._response)

def test_blocked_loan_is_rejected(monkeypatch):
    stub = StubClient({
        "decision": {"status": "BLOCKED"},
        "policy_violations": [{"policy_id": "DTI_LIMIT", "description": "DTI above limit"}],
    })
    monkeypatch.setattr("myservice.client", stub)

    result = process_loan_request(
        user={"id": "u_test"},
        action={"type": "loan_approval", "amount": 50000},
        context={"credit_score": 720, "debt_to_income": 0.52},
    )
    assert result["error"] == "blocked"
    assert "DTI_LIMIT" in result["violations"]

import os
from eve_coreguard import CoreGuardClient, verify_decision_record, fetch_public_key_pem

client = CoreGuardClient(api_key=os.environ["COREGUARD_API_KEY"])
decision = client.evaluate(
    tenant_id="bank_001",
    proposed_action={"type": "loan_approval", "amount": 50000},
    context={"credit_score": 720, "debt_to_income": 0.30, "employment_verified": True},
    include_evidence=True,
)

# One call: fetches /.well-known/eve-pubkey if no key is supplied.
check = client.verify_evidence_offline(decision.signed_governance)
assert check.valid, check.reason

# Or verify a STORED record later, with an explicitly pinned public key:
pubkey = fetch_public_key_pem("https://api.evecore.com")
check = verify_decision_record(stored_record, public_key_pem=pubkey)
assert check.valid and check.independently_verifiable

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