Creating Custom Probes

metapod is extensible. Add your own probes for internal systems, custom APIs, or niche data sources.

The BaseProbe interface

Every probe implements three methods:

from core.base_probe import BaseProbe
from core.registry import register

@register
class MySourceTables(BaseProbe):
    @staticmethod
    def probe_name() -> str:
        """Fully qualified name: source.category.leaf"""
        return "mysource.metadata.tables"

    @staticmethod
    def description() -> str:
        """One-line description shown in probe catalog."""
        return "Extract table inventory from MySource API"

    def execute(self, connection, schema: str, **kwargs) -> list[dict]:
        """Run the probe. Return list of record dicts."""
        # connection = whatever your connection module returns
        # schema = the --schema CLI argument
        response = connection.get(f"/api/tables?schema={schema}")
        return [
            {
                "table_name": t["name"],
                "row_count": t["rows"],
                "created": t["created_at"],
            }
            for t in response.json()
        ]

The @register decorator adds the probe to the global registry. It will appear in metapod list-probes and can be matched by glob patterns.

Connection module

Each source needs a connection module at probes/<source>/connection.py:

# probes/mysource/connection.py
import httpx
from core.config import SourceConfig

def connect(config: SourceConfig) -> httpx.Client:
    """Create a connection to MySource."""
    return httpx.Client(
        base_url=config.host,
        headers={"Authorization": f"Bearer {config.token}"},
        timeout=30,
    )

def source_id(config: SourceConfig) -> str:
    """Return a human-readable source identifier."""
    return f"mysource://{config.host}"

Register the connection in core/connections.py:

CONNECTION_REGISTRY["mysource"] = ("probes.mysource.connection", "connect", "source_id")

File structure

probes/
  mysource/
    __init__.py          # empty
    connection.py        # connect() + source_id()
    metadata/
      __init__.py        # empty
      tables.py          # @register class MySourceTables(BaseProbe)
      columns.py         # @register class MySourceColumns(BaseProbe)

ProbeResult envelope

BaseProbe.run() wraps your execute() with timing, error handling, and the standard envelope:

result = probe.run(connection, schema, source_id="mysource://host")
# result.probe       = "mysource.metadata.tables"
# result.source_type = "mysource"
# result.source_id   = "mysource://host"
# result.timestamp   = "2026-03-29T12:00:00+00:00"
# result.records     = [{...}, {...}]
# result.record_count = 2
# result.duration_ms  = 45
# result.error        = None

If execute() raises, result.error contains the exception message and result.records is empty.

Plugin system (entry points)

External packages can register probes without modifying metapod source code.

In your package’s pyproject.toml:

[project.entry-points."metapod.probes"]
mysource = "my_metapod_plugin.probes"

metapod discovers entry points at startup and registers all BaseProbe subclasses found in the specified module.

Testing your probe

# tests/test_mysource.py
from unittest.mock import MagicMock
from probes.mysource.metadata.tables import MySourceTables

def test_returns_tables():
    mock_conn = MagicMock()
    mock_conn.get.return_value.json.return_value = [
        {"name": "users", "rows": 100, "created_at": "2026-01-01"},
    ]
    probe = MySourceTables()
    result = probe.run(mock_conn, "default")
    assert result.record_count == 1
    assert result.records[0]["table_name"] == "users"

Run with:

python3 -m pytest tests/test_mysource.py -v