EVE Core › Docs › CoreGuard Integration Guide
Integration Guide

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

This guide walks through the complete process of integrating CoreGuard's AI governance API into an existing LLM pipeline. Whether you are running a FastAPI service, a batch inference worker, or a real-time chat application, the same integration pattern applies: evaluate the proposed action before it reaches the model, and enforce the decision on the way out. By the end of this guide your pipeline will produce ALLOWED, BLOCKED, or MODIFIED decisions with HMAC-SHA256 signed audit certificates on every request. For a complete API reference, see the CoreGuard API Reference.

Contents

  1. Prerequisites
  2. Installation
  3. Environment Variables
  4. Synchronous Integration Pattern
  5. Asynchronous Integration Pattern
  6. Webhook Setup for Audit Events
  7. Docker Deployment
  8. Kubernetes Sidecar Pattern
  9. Error Handling
  10. Testing Your Integration
  11. Frequently Asked Questions

1. Prerequisites

Before you begin, ensure you have the following:

Sandbox vs. Production: The sandbox endpoint shares the same API contract as production but uses a fixed, non-rotating verification key. Certificates issued by the sandbox are clearly marked with "environment": "sandbox" and are not accepted by EVE Proof's production verification endpoint. Do not use sandbox credentials in production.

2. Installation

Install the CoreGuard Python SDK from PyPI:

Terminal
pip install eve-coreguard

To pin a specific version in a requirements.txt or pyproject.toml:

requirements.txt
eve-coreguard>=0.1.0,<1.0.0

The SDK has minimal runtime dependencies: httpx for HTTP transport (supports both sync and async), cryptography for offline certificate verification, and pydantic for request and response model validation. No additional system libraries are required.

Offline verification: The cryptography dependency enables offline HMAC certificate verification — you can verify audit certificates without making any API call. This is important for regulated environments where audit logs must be verifiable independently of EVE Core infrastructure.

3. Environment Variables

CoreGuard reads configuration from environment variables. Set these before starting your application:

.env
# 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

Secret management: Never commit COREGUARD_API_KEY or COREGUARD_WEBHOOK_SECRET to source control. Use a secrets manager (AWS Secrets Manager, GCP Secret Manager, HashiCorp Vault, or Kubernetes Secrets) and inject at runtime. The SDK will raise CoreGuardConfigError at import time if the required variables are absent.

4. Synchronous Integration Pattern

The synchronous client is the simplest integration path. Insert one evaluate() call before each LLM invocation in your existing pipeline:

Python — Synchronous Pattern
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,
    }

The result object contains three top-level fields: result.decision (status, risk level, violations, modifications), result.audit_record (certificate ID, timestamp, HMAC signature), and result.metadata (latency, policy version, request ID). Store certificate_id alongside every decision record — it is your regulatory evidence handle.

5. Asynchronous Integration Pattern

For FastAPI, aiohttp, or any other asyncio-based application, use AsyncCoreGuardClient. The interface is identical to the synchronous client; all methods are coroutines:

Python — Async Pattern (FastAPI)
from fastapi import FastAPI, HTTPException
from eve_coreguard import AsyncCoreGuardClient, EvaluationRequest
from eve_coreguard.exceptions import CoreGuardBlockedError
import os

app = FastAPI()

# Shared client — AsyncCoreGuardClient uses an aiohttp session pool internally
cg = AsyncCoreGuardClient(
    api_key=os.environ["COREGUARD_API_KEY"],
    endpoint=os.environ["COREGUARD_ENDPOINT"],
)

@app.post("/api/v1/decisions")
async def make_decision(payload: dict):
    try:
        result = await cg.evaluate(EvaluationRequest(**payload))
    except CoreGuardBlockedError as e:
        raise HTTPException(status_code=422, detail={
            "status": "BLOCKED",
            "violations": [v.model_dump() for v in e.violations],
        })

    return {
        "status": result.decision.status,
        "certificate_id": result.audit_record.certificate_id,
        "risk_level": result.decision.risk_level,
    }

@app.on_event("shutdown")
async def shutdown():
    # Close the aiohttp session cleanly on shutdown
    await cg.close()

Connection lifecycle: Instantiate AsyncCoreGuardClient once at application startup, not per-request. The client holds an aiohttp ClientSession with a connection pool. Creating a new session per request is a common mistake that causes connection exhaustion under load.

6. Webhook Setup for Audit Events

CoreGuard can push signed audit events to your SIEM, data warehouse, or compliance platform in real time. Configure a webhook endpoint to receive decision.evaluated, decision.blocked, and certificate.issued event types.

Registering a Webhook Endpoint

cURL — Register webhook
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"
  }'

Verifying Webhook Signatures

Every webhook delivery includes an X-CoreGuard-Signature header containing an HMAC-SHA256 signature over the raw request body, computed using your COREGUARD_WEBHOOK_SECRET. Verify this signature before processing the event:

Python — Webhook Signature Verification
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}

7. Docker Deployment

Pass CoreGuard credentials as environment variables at container runtime. Never bake them into the image:

Dockerfile
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 — inject secrets at runtime
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

8. Kubernetes Sidecar Pattern

The recommended production architecture runs CoreGuard as a sidecar container in the same pod as your LLM service. The sidecar listens on localhost:8001 and intercepts governance requests before they leave the pod network. This eliminates the external network hop and keeps median evaluation latency under 1 ms.

Kubernetes — Pod spec with CoreGuard sidecar
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" }

9. Error Handling

The SDK raises typed exceptions that map directly to governance outcomes and infrastructure conditions. Your error handler should distinguish between policy blocks (expected, handle gracefully) and infrastructure errors (unexpected, fail closed):

Exception Meaning Recommended Action
CoreGuardBlockedError Policy hard block — action is not permitted Reject the action; return policy violations to caller
CoreGuardModifiedError Policy requires modifications before proceeding Apply modifications from e.modifications, then continue
CoreGuardTimeoutError Evaluation exceeded COREGUARD_TIMEOUT_MS Fail closed — reject the action; log and alert
CoreGuardConnectionError Network or TLS failure reaching the API Fail closed — reject the action; activate circuit breaker
CoreGuardConfigError Missing or invalid environment variables Fail fast at startup — do not serve requests
CoreGuardRateLimitError API rate limit exceeded (HTTP 429) Back off using e.retry_after_seconds; queue the request

Fail-closed discipline: CoreGuard is a mandatory gateway, not an optional filter. If the governance layer is unreachable, the action must be rejected — not allowed through on a best-effort basis. Fail-open behavior in a governed pipeline creates an audit gap that regulators treat as a compliance failure.

10. Testing Your Integration

The SDK ships with a built-in mock client for unit and integration testing. The mock accepts the same EvaluationRequest objects and returns deterministic fixture responses without making any network calls:

Python — Unit Test with MockCoreGuardClient
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"

For end-to-end integration tests, use the sandbox endpoint with your sandbox API key. The sandbox returns realistic decision responses and issues real HMAC-signed certificates that can be verified offline using verify_certificate() from the SDK:

Python — Offline Certificate Verification
from eve_coreguard import verify_certificate

# Load the certificate from your storage (database, object store, etc.)
cert = load_certificate(certificate_id="cert_abc123")

result = verify_certificate(cert, verification_key=os.environ["COREGUARD_VERIFY_KEY"])
assert result.valid, f"Certificate tampered: {result.reason}"
assert result.chain_intact, "Certificate chain broken"

Frequently Asked Questions

How long does it take to integrate CoreGuard into an existing LLM pipeline?

Most teams complete a working integration in under 30 minutes using the Python SDK. The critical path is: install the package (pip install eve-coreguard), set two environment variables (COREGUARD_API_KEY and COREGUARD_ENDPOINT), and insert one evaluate() call before your LLM invocation. The SDK handles authentication, serialization, HMAC certificate generation, and retries automatically. More complex integrations — async pipelines, webhook audit delivery, or Kubernetes sidecar deployment — typically take 2 to 4 hours.

What environment variables does CoreGuard require?

CoreGuard requires two environment variables: COREGUARD_API_KEY (your organization's Bearer token) and COREGUARD_ENDPOINT (the base URL for your CoreGuard instance). Optional variables include COREGUARD_TIMEOUT_MS (default: 5000), COREGUARD_POLICY_SET (default: lending_v1), and COREGUARD_WEBHOOK_SECRET for webhook signature verification.

Can CoreGuard be integrated into an async Python application?

Yes. The SDK provides both synchronous (CoreGuardClient) and asynchronous (AsyncCoreGuardClient) clients. The async client uses aiohttp internally and is compatible with FastAPI, aiohttp servers, and any asyncio-based pipeline. The evaluate() method on the async client is a coroutine and must be awaited. Both clients support connection pooling and automatic retry with exponential backoff.

How do I deploy CoreGuard in a Kubernetes environment?

The recommended pattern is a sidecar container running alongside each LLM-serving pod. The sidecar listens on loopback port 8001 and receives governance requests from the main application container — no external network hop required. This keeps evaluation latency under 1 ms and ensures the governance layer cannot be bypassed. EVE Core provides a Helm chart and reference init container that configures mandatory routing through the sidecar for all outbound LLM traffic.

Ready to Integrate?

Get your sandbox API key and start testing CoreGuard in your pipeline today. No credit card required — provisioned in under 24 hours.

Request API Access API Reference