Custom Connectors

Custom ETL connectors let you integrate any pipeline orchestration tool with Monte Carlo. You implement two Python methods — fetch_metadata and fetch_run_details — that talk to your vendor's API, and the agent framework handles scheduling, pagination, and delivery to Monte Carlo.

Your connector runs inside a custom agent image deployed to your infrastructure. The agent is egress-only — it calls out to the vendor API and to Monte Carlo.

Architecture

flowchart LR
    subgraph agent ["Custom Agent (Docker)"]
        F["Agent Framework<br>scheduling, pagination,<br>MC API delivery"] --> C["Your Connector<br>connector.py"]
    end
    C -->|"vendor API calls"| V["Vendor ETL Tool<br>(Airflow, Coalesce, etc.)"]
    F -->|"pushes data"| MC["Monte Carlo API"]
    MC -->|"webhook trigger"| F

The connector runs inside a custom agent image deployed to your infrastructure. The agent framework:

• Schedules metadata and run collection automatically (every 60 minutes for runs)
• Paginates through results by calling your methods with incrementing offset
• Delivers data to Monte Carlo's backend
• Triggers on-demand collection when a webhook fires

Prerequisites

• Docker — the only host dependency. Everything else runs inside containers.
• Vendor API credentials — access to the ETL tool's API for reading job metadata and run history.

Steps

1. Set up the repo — Clone the custom-connector-setup repository.
2. Implement your connector — Implement the four methods and pass the test suite.
3. Build the agent image — Export capabilities and build the deployable Docker image.
4. Deploy the agent — Push the image to a container registry and deploy using any supported generic agent platform.
5. Register the connector — Run agent validations so Monte Carlo discovers your custom ETL connector.
6. Add the connection — Connect through the Monte Carlo UI.

1. Set up the repo

Clone the custom-connector-setup repository:

git clone https://github.com/monte-carlo-data/custom-connector-setup.git
cd custom-connector-setup

The repository contains:

2. Implement your connector

Path 1: AI-assisted (recommended)

The repository includes Claude Code skills that automate the workflow:

The only manual step is filling in credentials.json when step 2 pauses.

Path 2: Manual

1. Scaffold your connector:

python scripts/create_connector.py <name>

2. Add your vendor SDK to connectors/<name>/requirements.txt:

apache-airflow-client==2.9.0

3. Fill in credentials in connectors/<name>/credentials.json:

{
  "connect_args": {
    "host": "https://airflow.example.com",
    "username": "monte_carlo",
    "password": "secret"
  }
}

The keys inside connect_args are whatever your setup_connection() method reads from self.credentials.

4. Implement your connector — edit connectors/<name>/connector.py:

import requests

class Connector:
    credentials: dict

    def setup_connection(self):
        self.base_url = self.credentials["host"]
        self.session = requests.Session()
        self.session.auth = (
            self.credentials["username"],
            self.credentials["password"],
        )

    def close_connection(self):
        self.session.close()

    def fetch_metadata(self, limit, offset):
        resp = self.session.get(
            f"{self.base_url}/api/v1/dags",
            params={"limit": limit, "offset": offset},
        )
        resp.raise_for_status()
        return [
            {
                "job_source_id": dag["dag_id"],
                "name": dag["dag_id"],
                "description": dag.get("description"),
                "is_paused": dag.get("is_paused", False),
                "schedule": {
                    "kind": "cron",
                    "cron_expression": dag["schedule_interval"],
                } if dag.get("schedule_interval") else None,
                "owner": {
                    "primary_name": dag.get("owners", [None])[0],
                },
            }
            for dag in resp.json().get("dags", [])
        ]

    def fetch_run_details(self, run_ids=None, lookback=None, limit=100, offset=0):
        if run_ids:
            # Webhook mode — fetch specific runs
            runs = []
            for run_id in run_ids:
                resp = self.session.get(
                    f"{self.base_url}/api/v1/dags/~/dagRuns/{run_id}"
                )
                resp.raise_for_status()
                runs.append(self._map_run(resp.json()))
            return runs

        # Polling mode — fetch runs within lookback window
        from datetime import datetime, timezone
        start = (datetime.now(timezone.utc) - lookback).isoformat()
        resp = self.session.get(
            f"{self.base_url}/api/v1/dags/~/dagRuns",
            params={
                "start_date_gte": start,
                "limit": limit,
                "offset": offset,
            },
        )
        resp.raise_for_status()
        return [self._map_run(r) for r in resp.json().get("dag_runs", [])]

    def _map_run(self, run):
        status_map = {"success": "success", "failed": "failed", "running": "in_progress"}
        mapped_status = status_map.get(run.get("state"), "unknown")
        event = {
            "job_source_id": run["dag_id"],
            "run_source_id": run["dag_run_id"],
            "status": mapped_status,
            "event_time": run.get("end_date") or run.get("start_date"),
            "start_time": run.get("start_date"),
            "end_time": run.get("end_date"),
            "trigger": "SCHEDULE" if run.get("external_trigger") is False else "MANUAL",
            "run_url": f"{self.base_url}/dags/{run['dag_id']}/grid?dag_run_id={run['dag_run_id']}",
        }
        if mapped_status in ("failed", "error"):
            event["error"] = {"message": run.get("note") or "Run failed"}
        return event

5. Build and test iteratively:

docker compose build

# Verify connection
CONNECTOR=<name> docker compose run --rm test -m etl_connection

# Metadata collection
CONNECTOR=<name> docker compose run --rm test -m etl_metadata

# Run details (polling + webhook modes)
CONNECTOR=<name> docker compose run --rm test -m etl_run_details

The test suite validates that your methods return correctly structured dicts matching the EtlAsset and EtlRunEvent schemas. Test lookback defaults to 7 days (configurable via ETL_TEST_LOOKBACK_HOURS env var).

3. Build the agent image

After all tests pass, build the deployable image:

python scripts/generate_agent_image.py

The script packages your connector into a Docker image layered on top of montecarlodata/agent:latest-generic:

What's included in the image:

• connector.py — your connector code
• manifest.json — connection type, name, terminology
• requirements.txt — vendor SDK dependencies

What's NOT included:

• credentials.json — never baked into the image. Provided at deploy time via self-hosted credentials.

Terminology mapping

The manifest includes a terminology field that maps Monte Carlo's generic concepts to the vendor's terms:

{
  "terminology": {
    "group": "Workspace",
    "job": "Pipeline",
    "task": "Step"
  }
}

This affects how assets are labeled in the Monte Carlo UI.

Custom icon

You can provide an icon URL during connector setup to display the vendor's logo alongside the connection in the Monte Carlo UI. The scaffolding step prompts for this — both the manual flow (python scripts/create_connector.py <name> --etl) and the Claude Code skill (/create-connector <name> --etl) ask for an icon URL and write it to the manifest automatically.

Provide a publicly accessible URL to an image (PNG or SVG recommended). If left blank, Monte Carlo displays a default connector icon.

4. Deploy the agent

1. Push to a container registry accessible from your deployment environment.
2. Configure self-hosted credentials in Monte Carlo using the same credentials.json format — swap in production values.
3. Deploy the agent using any of the supported generic agent platforms. The agent is egress-only.

5. Register the connector

After deploying the agent, Monte Carlo needs to discover your custom ETL connector. This happens automatically when you run agent validations:

Settings → Deployments → your agent → Validate

The Discover connectors step finds all custom connectors on the agent and triggers an async registration job (usually completes within a couple of minutes).

6. Add the connection

Once the connector is registered, add the connection through the Monte Carlo UI:

Settings → Integrations → Add → Orchestration → Custom ETL integration

The form lets you choose how Monte Carlo gets each type of data:

• Collect (pull) — Monte Carlo actively collects data using your custom ETL connector on a schedule. When you select this option, the UI provides the webhook URL and authentication token for configuring near-real-time webhook-triggered collection.
• Push — Data is pushed externally via the ETL Push Ingest API. When you select this option, the UI provides the ingestion URL and token information for pushing data.

Select your connector, provide credentials, and Monte Carlo handles validation and setup.

After setup, Monte Carlo begins collecting automatically:

• Metadata collection runs on the agent's schedule
• Run details are collected every 60 minutes via polling

Data collection modes

Polling (scheduled)

The default collection mode. The agent calls fetch_run_details(lookback=timedelta(minutes=60)) every 60 minutes, collecting all runs that completed or updated within that window.

• The polling interval is fixed at 60 minutes (not customer-configurable)
• Failures are detected within the polling window (worst case: up to 60 minutes latency)

Webhook collection

For near-real-time failure detection, configure your ETL tool to send a webhook event to Monte Carlo when a pipeline fails.

When you create the Custom ETL integration in Monte Carlo, you will be given:

• A webhook URL
• Ability to generate an integration key for authentication (key ID and key Secret)

Configure your ETL tool to POST to the webhook URL. The webhook goes to Monte Carlo's backend (not the agent directly) — Monte Carlo then triggers a run collection job through the agent.

The POST event must contain these headers:

By default the run collection triggered will collect all the runs from the last hour, to make sure we fetch the failed run event(s). You can optional include a query parameters filter on the request to scope the run collection to just the failed run:

The webhook ignores any request body — all parameters must be query strings.

FAQs

What ETL tools can I build a connector for?

Any tool that exposes job metadata and run history through an API or SDK. The framework is API-agnostic — you bring the SDK and implement the fetch methods.

How is this different from Custom Connectors?

Custom Connectors are for SQL databases — they implement SQL templates so Monte Carlo can execute queries. Custom ETL connectors are for pipeline tools — they implement API calls to fetch job metadata and run events.

Can I include multiple connectors in one agent?

Yes. Pass --connector multiple times when building the image:

python scripts/generate_agent_image.py --connector airflow --connector coalesce

How do I debug a failing connector?

The test suite output shows exactly which method failed and the error. Run individual test markers to isolate the problem:

CONNECTOR=<name> docker compose run --rm test -m etl_connection     # just connection
CONNECTOR=<name> docker compose run --rm test -m etl_metadata       # just metadata
CONNECTOR=<name> docker compose run --rm test -m etl_run_details    # just run details

How do I rotate credentials?

Update the secret in your secret store (AWS Secrets Manager, GCP Secret Manager, etc.) and the agent picks up the new value on its next connection. See self-hosted credentials for details.

What happens to my integration keys if I delete the integration?

Deleting an integration in Monte Carlo also deletes any integration keys and webhook credentials associated with it. Any webhook configurations pointing to the deleted integration will stop working. Before deleting an integration, make sure no active webhook configurations depend on it.