Building a Custom Integration

If your tool doesn’t have a managed connector, you can build a full two-way integration in under an hour using:

Ingestion API — push issues from your system into Causeloop
Inbound webhooks — let your system push events to Causeloop in real time
Outbound webhooks — let Causeloop notify your system when patterns are detected or syncs complete

This walkthrough uses a hypothetical internal deployment tracker called “Deploy Monitor” as the example source.

Before you start

You need:

A Causeloop API key with scopes: connectors:admin, ingest:write, webhooks:admin
An HTTPS endpoint on your system that can receive webhook deliveries
Ability to generate HMAC-SHA256 signatures in your source system

If you don’t have an API key, go to Settings → API Keys in the Causeloop app and create one with the scopes above.

Step 1: Create a connector record

Even for custom integrations, creating a connector gives you a stable identifier to associate issues with and filter webhook deliveries by.

curl -X POST https://api.causeloop.ai/v1/connectors \
  -H "Authorization: Bearer $TOKEN" \
  -H "Content-Type: application/json" \
  -d '{
    "type": "custom_api",
    "display_name": "Deploy Monitor",
    "credentials": {}
  }'

Save the returned id — you’ll use it throughout:

{
  "id": "con_custom_deploy_01",
  "type": "custom_api",
  "display_name": "Deploy Monitor",
  "status": "connected"
}

Step 2: Push issues from your system

Whenever something noteworthy happens in Deploy Monitor (a failed deploy, a rollback, a health check timeout), push it to Causeloop using the Ingestion API.

Option A: Push a single event

Best for real-time event dispatch from within your application.

import hashlib, hmac, json
import httpx

CAUSELOOP_TOKEN = "..."

def push_deployment_failure(deploy_id: str, service: str, error: str, occurred_at: str):
    resp = httpx.post(
        "https://api.causeloop.ai/v1/ingest/events",
        headers={"Authorization": f"Bearer {CAUSELOOP_TOKEN}"},
        json={
            "source": "custom_api",
            "external_id": f"deploy-{deploy_id}",
            "title": f"Deploy failed: {service}",
            "narrative": error,
            "occurred_at": occurred_at,
            "severity": "p1",
            "external_url": f"https://deploys.example.com/{deploy_id}",
            "metadata": {
                "service": service,
                "deploy_id": deploy_id,
            },
        },
    )
    resp.raise_for_status()
    return resp.json()["job_id"]

Option B: Push a batch at the end of a pipeline run

Best for nightly jobs or migration scripts that produce many issues at once.

def push_deploy_batch(failures: list[dict]):
    records = [
        {
            "external_id": f"deploy-{f['id']}",
            "source": "custom_api",
            "title": f"Deploy failed: {f['service']}",
            "narrative": f['error'],
            "occurred_at": f["timestamp"],
            "severity": f.get("severity", "p2"),
            "team": f.get("team"),
            "source_created_at": f["timestamp"],
        }
        for f in failures
    ]

    resp = httpx.post(
        "https://api.causeloop.ai/v1/ingest/batch",
        headers={"Authorization": f"Bearer {CAUSELOOP_TOKEN}"},
        json={"records": records},
    )
    resp.raise_for_status()
    result = resp.json()
    print(f"accepted={result['accepted']}, rejected={len(result['rejected'])}")
    return result["job_id"]

Step 3: Set up an inbound webhook (real-time push)

If Deploy Monitor can send webhook events, register an inbound webhook so Causeloop receives them the moment they fire — no polling needed.

3a. Register the inbound webhook

curl -X POST https://api.causeloop.ai/v1/webhooks \
  -H "Authorization: Bearer $TOKEN" \
  -H "Content-Type: application/json" \
  -d '{
    "url": "https://api.causeloop.ai/v1/webhooks/inbound/con_custom_deploy_01",
    "connector_id": "con_custom_deploy_01",
    "description": "Deploy Monitor real-time events"
  }'

Copy the signing_secret from the response and store it in Deploy Monitor’s webhook configuration.

3b. Configure Deploy Monitor to sign its payloads

In Deploy Monitor, add HMAC signing to every outgoing webhook:

import crypto from 'crypto';

const CAUSELOOP_SECRET = process.env.CAUSELOOP_INBOUND_SECRET;
const CAUSELOOP_URL = 'https://api.causeloop.ai/v1/webhooks/inbound/con_custom_deploy_01';

async function sendToCAuseloop(event) {
  const body = JSON.stringify(event);
  const signature = crypto
    .createHmac('sha256', CAUSELOOP_SECRET)
    .update(body)
    .digest('hex');

  const resp = await fetch(CAUSELOOP_URL, {
    method: 'POST',
    headers: {
      'Content-Type': 'application/json',
      'x-causeloop-signature': signature,
    },
    body,
  });

  if (!resp.ok) throw new Error(`Causeloop rejected: ${resp.status}`);
}

// Call this from your deployment pipeline:
sendToCAuseloop({
  event_type: 'deploy.failed',
  deploy_id: 'DEPLOY-9999',
  service: 'payment-service',
  severity: 'p1',
});

Compute the HMAC over the raw JSON bytes you send. Do not re-serialize the object — the byte sequence must match exactly what Causeloop receives.

Step 4: Receive outbound events from Causeloop

4a. Register the outbound webhook

curl -X POST https://api.causeloop.ai/v1/webhooks \
  -H "Authorization: Bearer $TOKEN" \
  -H "Content-Type: application/json" \
  -d '{
    "url": "https://deploy-monitor.example.com/hooks/causeloop",
    "event_types": [
      "connector.sync.failed",
      "issue.ingested",
      "issue.batch_ingested"
    ],
    "connector_id": "con_custom_deploy_01",
    "description": "Notify Deploy Monitor of Causeloop events"
  }'

Save the signing_secret.

4b. Verify incoming deliveries on your endpoint

import hashlib
import hmac
import json
from fastapi import FastAPI, Request, HTTPException

app = FastAPI()
CAUSELOOP_SECRET = "whsec_..."  # from step 4a

@app.post("/hooks/causeloop")
async def handle_causeloop_event(request: Request):
    raw_body = await request.body()
    provided = request.headers.get("x-causeloop-signature", "")

    expected = hmac.new(
        CAUSELOOP_SECRET.encode("utf-8"),
        raw_body,
        hashlib.sha256,
    ).hexdigest()

    if not hmac.compare_digest(expected, provided):
        raise HTTPException(status_code=401, detail="Invalid signature")

    event = json.loads(raw_body)
    event_type = event.get("event_type")

    if event_type == "connector.sync.failed":
        # Page the on-call team
        await alert_oncall(event)
    elif event_type == "issue.ingested":
        # Update Deploy Monitor's issue tracker
        await link_causeloop_issue(event)

    return {"received": True}

Step 5: Test the integration end to end

Test the connector connection

curl -X POST https://api.causeloop.ai/v1/connectors/con_custom_deploy_01/test \
  -H "Authorization: Bearer $TOKEN"

Expect "success": true.

Push a test issue

curl -X POST https://api.causeloop.ai/v1/ingest/events \
  -H "Authorization: Bearer $TOKEN" \
  -H "Content-Type: application/json" \
  -d '{
    "source": "custom_api",
    "external_id": "TEST-001",
    "title": "Test issue from Deploy Monitor",
    "narrative": "Manual test event to verify the integration end to end.",
    "occurred_at": "2026-06-14T09:00:00Z",
    "severity": "p3"
  }'

Expect a 202 response with a job_id. Poll GET /v1/ingest/status/{job_id} and verify the issue appears in the Causeloop dashboard.

Send a test inbound webhook

From your Deploy Monitor system, fire a test event to the inbound webhook URL. Check Settings → Integrations → [connector] → Sync History to see the delivery recorded.

Verify outbound delivery

Check your endpoint’s logs for a delivery from Causeloop. Confirm the signature verification passes.

Reference: all endpoints used in this walkthrough

Action	Endpoint	Scope
Create connector	`POST /v1/connectors`	`connectors:admin`
Test connector	`POST /v1/connectors/{id}/test`	`connectors:admin`
Push single event	`POST /v1/ingest/events`	`ingest:write`
Push batch	`POST /v1/ingest/batch`	`ingest:write`
Check job status	`GET /v1/ingest/status/{job_id}`	`ingest:write`
Register webhook	`POST /v1/webhooks`	`webhooks:admin`
Rotate secret	`POST /v1/webhooks/{id}/rotate`	`connectors:admin`
View deliveries	`GET /v1/webhooks/{id}/deliveries`	`webhooks:read`
Inbound receiver	`POST /v1/webhooks/inbound/{connector_id}`	HMAC only

Integrations

Data

Building a Custom Integration

Before you start

Step 1: Create a connector record

Step 2: Push issues from your system

Option A: Push a single event

Option B: Push a batch at the end of a pipeline run

Step 3: Set up an inbound webhook (real-time push)

3a. Register the inbound webhook

3b. Configure Deploy Monitor to sign its payloads

Step 4: Receive outbound events from Causeloop

4a. Register the outbound webhook

4b. Verify incoming deliveries on your endpoint

Step 5: Test the integration end to end

Reference: all endpoints used in this walkthrough

Further reading

​Before you start

​Step 1: Create a connector record

​Step 2: Push issues from your system

​Option A: Push a single event

​Option B: Push a batch at the end of a pipeline run

​Step 3: Set up an inbound webhook (real-time push)

​3a. Register the inbound webhook

​3b. Configure Deploy Monitor to sign its payloads

​Step 4: Receive outbound events from Causeloop

​4a. Register the outbound webhook

​4b. Verify incoming deliveries on your endpoint

​Step 5: Test the integration end to end

​Reference: all endpoints used in this walkthrough

​Further reading

Before you start

Step 1: Create a connector record

Step 2: Push issues from your system

Option A: Push a single event

Option B: Push a batch at the end of a pipeline run

Step 3: Set up an inbound webhook (real-time push)

3a. Register the inbound webhook

3b. Configure Deploy Monitor to sign its payloads

Step 4: Receive outbound events from Causeloop

4a. Register the outbound webhook

4b. Verify incoming deliveries on your endpoint

Step 5: Test the integration end to end

Reference: all endpoints used in this walkthrough

Further reading