Skip to main content

Splunk Shadow AI Ingestion

Poll AI-usage / CASB-style activity events out of a Splunk index and forward them to Credo AI's Shadow AI Discovery surfaces.

This cookbook uses raw HTTP calls rather than an SDK method, since neither SDK (pycredoai, @credo-ai/sdk) has a shadow_ai method yet.

Preview — talks to a backend API directly

Shadow AI Discovery is a Research Preview feature and has no endpoint on the public Integration Service yet. This cookbook gets its bearer token the normal way — POST /auth/token on the Integration Service, same X-API-Key/X-Tenant auth shape as every other cookbook on this site — but sends the bulk event POST straight to the backend instead: POST /api/v2/{tenant}/shadow_ai/ai_events/bulk.

That backend endpoint is not part of the stable public API contract — it can change shape or move without notice. Once Shadow AI ships on the public Integration Service, this cookbook should be migrated to use it instead.

How It Works​

Your Splunk (index, sourcetype) ──▶ This Poller ──▶ Credo AI (Shadow AI events)

There's no webhook to register — Splunk doesn't push. This poller pulls on an interval using the same /services/search/jobs/export REST API any external tool would use to read Splunk data back out, transforms results into Credo AI's event schema, and posts batches to the bulk endpoint.

A few reliability details worth knowing:

  • Checkpoint saved per batch, not per cycle. If the process is interrupted mid-cycle, at most the one batch in flight at that moment can be re-sent — everything already delivered earlier in the cycle isn't re-fetched.
  • Permanently rejected batches don't stall the poller. If Credo AI rejects a batch outright (entitlement off, bad event data), that batch is logged and skipped, and the poller moves on to the next one instead of stalling the whole cycle on it. A transient failure (5xx, timeout), by contrast, holds the checkpoint back and retries the same batch next cycle.
  • Client-side dedup. Since there's no backend idempotency key yet, the poller keeps an in-memory fingerprint of everything it has delivered in the current run, so a batch that's accidentally re-fetched (e.g. a checkpoint boundary tie) isn't sent twice in the same run.

The Poller​

Runs as a long-lived process, not a request handler — there's nothing to expose or tunnel.

import json

import httpx

CREDO_BASE_URL = "https://your-tenant.credo.ai/api/v1/integration"
CREDO_API_KEY = "your-api-key"
CREDO_TENANT = "your-tenant-name"
# Backend host for the Shadow AI bulk endpoint — see the preview note above.
# Not the same host as CREDO_BASE_URL. Ask your Credo AI representative.
CREDO_BACKEND_BASE_URL = "https://your-backend-host"

SPLUNK_BASE_URL = "https://your-splunk-host:8089"
SPLUNK_INDEX = "your-index-name"
SPLUNK_SOURCETYPE = "your-sourcetype-name"

# Credo AI field -> Splunk field. Adjust the right side to match your data.
FIELD_MAP = {
    "user_email": "user_email",
    "app_name": "app_name",
    "category": "category",
    "action": "action",
    "risk_score": "risk_score",
}


def fetch_events(earliest: str) -> list[dict]:
    fields = ", ".join(FIELD_MAP.values())
    spl = (
        f"search index={SPLUNK_INDEX} sourcetype={SPLUNK_SOURCETYPE} "
        f"earliest={earliest} | fields _time, {fields}"
    )
    resp = httpx.post(
        f"{SPLUNK_BASE_URL}/services/search/jobs/export",
        auth=("splunk-service-account", "splunk-password"),
        data={"search": spl, "output_mode": "json"},
        timeout=30,
    )
    resp.raise_for_status()
    return [
        json.loads(line)["result"]
        for line in resp.text.splitlines()
        if line.strip()
    ]


def send_batch(events: list[dict], token: str) -> str:
    """Returns "delivered", "dropped" (404/422 — permanent, safe to skip and
    advance past), or "retry" (5xx — transient, caller should hold its
    checkpoint and retry this batch next cycle).

    Token comes from CREDO_BASE_URL's /auth/token (Integration Service), but
    the bulk POST itself goes to the backend directly — see the preview note
    above."""
    resp = httpx.post(
        f"{CREDO_BACKEND_BASE_URL}/api/v2/{CREDO_TENANT}/shadow_ai/ai_events/bulk",
        headers={"Authorization": f"Bearer {token}", "Content-Type": "application/json"},
        json={"events": events},
        timeout=60,
    )
    if resp.status_code in (404, 422):
        return "dropped"
    if resp.status_code >= 500:
        return "retry"
    resp.raise_for_status()
    return "delivered"

The full, tested version of this poller — with checkpointing, retry/backoff, and graceful handling of a dropped batch — lives in credoai-recipes/integrations/splunk/shadow-ai-event-ingestion/server/python/main.py. The snippet above is illustrative; run the real file for anything beyond reading the shape of the integration.

Testing Locally​

  1. Install dependencies and run the poller:

    cd server/python
    pip install -r requirements.txt
    python main.py

  2. No ngrok needed — this isn't an inbound webhook receiver, it reaches out to both Splunk and Credo AI. You do need a Splunk instance with matching test data to poll against.

  3. Confirm the Splunk side independently first:

    curl -sk -u "$SPLUNK_USERNAME:$SPLUNK_PASSWORD" \
      "https://$SPLUNK_HOST:$SPLUNK_PORT/services/search/jobs/export" \
      -d "search=search index=$SPLUNK_INDEX sourcetype=$SPLUNK_SOURCETYPE earliest=-1h | head 5" \
      -d "output_mode=json"

    If this returns nothing, fix the index/sourcetype/credentials before going further.

See credoai-recipes/integrations/splunk/shadow-ai-event-ingestion/TESTING.md for the full test plan, including checkpoint-behavior and failure-mode checks.

Configure Splunk​

  1. Create a read-only Splunk service account, scoped via Indexes searched by default to just the index carrying your AI-activity events — never an admin account.
  2. Point .env at your Splunk instance's actual SPLUNK_HOST, SPLUNK_INDEX, and SPLUNK_SOURCETYPE.
  3. Map your Splunk field names to Credo AI's schema in FIELD_MAP if they differ from the defaults.

Production​

Run the poller as a long-lived process — a systemd service, a container, or a scheduled task — wherever you run other internal tooling. There's no port to expose.