peregrine/docs/developer-guide/adding-integrations.md

# Adding an Integration

Peregrine's integration system is auto-discovered — add a class and a config example, and it appears in the wizard and Settings automatically. No registration step is needed.

---

## Step 1 — Create the integration module

Create `scripts/integrations/myservice.py`:

```python
# scripts/integrations/myservice.py

from scripts.integrations.base import IntegrationBase


class MyServiceIntegration(IntegrationBase):
    name  = "myservice"      # must be unique; matches config filename
    label = "My Service"     # display name shown in the UI
    tier  = "free"           # "free" | "paid" | "premium"

    def fields(self) -> list[dict]:
        """Return form field definitions for the connection card in the wizard/Settings UI."""
        return [
            {
                "key":         "api_key",
                "label":       "API Key",
                "type":        "password",   # "text" | "password" | "url" | "checkbox"
                "placeholder": "sk-...",
                "required":    True,
                "help":        "Get your key at myservice.com/settings/api",
            },
            {
                "key":         "workspace_id",
                "label":       "Workspace ID",
                "type":        "text",
                "placeholder": "ws_abc123",
                "required":    True,
                "help":        "Found in your workspace URL",
            },
        ]

    def connect(self, config: dict) -> bool:
        """
        Store credentials in memory. Return True if all required fields are present.
        Does NOT verify credentials — call test() for that.
        """
        self._api_key      = config.get("api_key", "").strip()
        self._workspace_id = config.get("workspace_id", "").strip()
        return bool(self._api_key and self._workspace_id)

    def test(self) -> bool:
        """
        Verify the stored credentials actually work.
        Returns True on success, False on any failure.
        """
        try:
            import requests
            r = requests.get(
                "https://api.myservice.com/v1/ping",
                headers={"Authorization": f"Bearer {self._api_key}"},
                params={"workspace": self._workspace_id},
                timeout=5,
            )
            return r.ok
        except Exception:
            return False

    def sync(self, jobs: list[dict]) -> int:
        """
        Optional: push jobs to the external service.
        Return the count of successfully synced jobs.
        The default implementation in IntegrationBase returns 0 (no-op).
        Only override this if your integration supports job syncing
        (e.g. Notion, Airtable, Google Sheets).
        """
        synced = 0
        for job in jobs:
            try:
                self._push_job(job)
                synced += 1
            except Exception as e:
                print(f"[myservice] sync error for job {job.get('id')}: {e}")
        return synced

    def _push_job(self, job: dict) -> None:
        import requests
        requests.post(
            "https://api.myservice.com/v1/records",
            headers={"Authorization": f"Bearer {self._api_key}"},
            json={
                "workspace":  self._workspace_id,
                "title":      job.get("title", ""),
                "company":    job.get("company", ""),
                "status":     job.get("status", "pending"),
                "url":        job.get("url", ""),
            },
            timeout=10,
        ).raise_for_status()
```

---

## Step 2 — Create the config example file

Create `config/integrations/myservice.yaml.example`:

```yaml
# config/integrations/myservice.yaml.example
# Copy to config/integrations/myservice.yaml and fill in your credentials.
# This file is gitignored — never commit the live credentials.
api_key: ""
workspace_id: ""
```

The live credentials file (`config/integrations/myservice.yaml`) is gitignored automatically via the `config/integrations/` entry in `.gitignore`.

---

## Step 3 — Auto-discovery

No registration step is needed. The integration registry (`scripts/integrations/__init__.py`) imports all `.py` files in the `integrations/` directory and discovers subclasses of `IntegrationBase` automatically.

On next startup, `myservice` will appear in:
- The first-run wizard Step 7 (Integrations)
- **Settings → Integrations** with a connection card rendered from `fields()`

---

## Step 4 — Tier-gate new features (optional)

If you want to gate a specific action (not just the integration itself) behind a tier, add an entry to `app/wizard/tiers.py`:

```python
FEATURES: dict[str, str] = {
    # ...existing entries...
    "myservice_sync": "paid",   # or "free" | "premium"
}
```

Then guard the action in the relevant UI page:

```python
from app.wizard.tiers import can_use
from scripts.user_profile import UserProfile

user = UserProfile()
if can_use(user.tier, "myservice_sync"):
    # show the sync button
else:
    st.info("MyService sync requires a Paid plan.")
```

---

## Step 5 — Write a test

Create or add to `tests/test_integrations.py`:

```python
# tests/test_integrations.py (add to existing file)

import pytest
from unittest.mock import patch, MagicMock
from pathlib import Path
from scripts.integrations.myservice import MyServiceIntegration


def test_fields_returns_required_keys():
    integration = MyServiceIntegration()
    fields = integration.fields()
    assert len(fields) >= 1
    for field in fields:
        assert "key" in field
        assert "label" in field
        assert "type" in field
        assert "required" in field


def test_connect_returns_true_with_valid_config():
    integration = MyServiceIntegration()
    result = integration.connect({"api_key": "sk-abc", "workspace_id": "ws-123"})
    assert result is True


def test_connect_returns_false_with_missing_required_field():
    integration = MyServiceIntegration()
    result = integration.connect({"api_key": "", "workspace_id": "ws-123"})
    assert result is False


def test_test_returns_true_on_200(tmp_path):
    integration = MyServiceIntegration()
    integration.connect({"api_key": "sk-abc", "workspace_id": "ws-123"})

    mock_resp = MagicMock()
    mock_resp.ok = True

    with patch("scripts.integrations.myservice.requests.get", return_value=mock_resp):
        assert integration.test() is True


def test_test_returns_false_on_error(tmp_path):
    integration = MyServiceIntegration()
    integration.connect({"api_key": "sk-abc", "workspace_id": "ws-123"})

    with patch("scripts.integrations.myservice.requests.get", side_effect=Exception("timeout")):
        assert integration.test() is False


def test_is_configured_reflects_file_presence(tmp_path):
    config_dir = tmp_path / "config"
    config_dir.mkdir()
    (config_dir / "integrations").mkdir()

    assert MyServiceIntegration.is_configured(config_dir) is False

    (config_dir / "integrations" / "myservice.yaml").write_text("api_key: sk-abc\n")
    assert MyServiceIntegration.is_configured(config_dir) is True
```

---

## IntegrationBase Reference

All integrations inherit from `scripts/integrations/base.py`. Here is the full interface:

| Method / attribute | Required | Description |
|-------------------|----------|-------------|
| `name: str` | Yes | Machine key — must be unique. Matches the YAML config filename. |
| `label: str` | Yes | Human-readable display name for the UI. |
| `tier: str` | Yes | Minimum tier: `"free"`, `"paid"`, or `"premium"`. |
| `fields() -> list[dict]` | Yes | Returns form field definitions. Each dict: `key`, `label`, `type`, `placeholder`, `required`, `help`. |
| `connect(config: dict) -> bool` | Yes | Stores credentials in memory. Returns `True` if required fields are present. Does NOT verify credentials. |
| `test() -> bool` | Yes | Makes a real network call to verify stored credentials. Returns `True` on success. |
| `sync(jobs: list[dict]) -> int` | No | Pushes jobs to the external service. Returns count synced. Default is a no-op returning 0. |
| `config_path(config_dir: Path) -> Path` | Inherited | Returns `config_dir / "integrations" / f"{name}.yaml"`. |
| `is_configured(config_dir: Path) -> bool` | Inherited | Returns `True` if the config YAML file exists. |
| `save_config(config: dict, config_dir: Path)` | Inherited | Writes config dict to the YAML file. Call after `test()` returns `True`. |
| `load_config(config_dir: Path) -> dict` | Inherited | Loads and returns the YAML config, or `{}` if not configured. |

### Field type values

| `type` value | UI widget rendered |
|-------------|-------------------|
| `"text"` | Plain text input |
| `"password"` | Password input (masked) |
| `"url"` | URL input |
| `"checkbox"` | Boolean checkbox |