# Help Scout connector

Source: /docs/connectors/help-scout.md

Use this as an implementation starting point: adapt the worked example to the requested dataset and the source documentation. See [Custom API](/docs/connectors/custom-api) for the shared SDK pattern and document-upload workflow.

## What this connector is for [#what-this-connector-is-for]

Shared inbox conversations, customers, mailboxes, and support workflows.

A connector owns access to the source. Separate syncs shape selected resources into tables for analysis and downstream transformations using the [Python SDK](/docs/reference/python/connectors-and-syncs).

## Data and tables you can pull [#data-and-tables-you-can-pull]

These are representative dataset families to consider, not a claim that the worked example implements every resource. Access, fields, and availability depend on the source account, product edition, and permissions. Use the [vendor reference](https://developer.helpscout.com/mailbox-api/) to select concrete resources.

| Suggested table | One row represents |
| --- | --- |
| `mailboxes` | a mailboxe |
| `conversations` | a conversation |
| `threads` | a thread |
| `customers` | a customer |
| `users` | a user |
| `tags` | a tag |

### Relationships and extraction considerations [#relationships-and-extraction-considerations]

Choose stable object IDs and preserve references between related resources. Nested arrays may need child tables. The listed names are proposed output datasets; validate the actual endpoint, fields, permissions, and pagination before adding each sync.

## Connecting to Help Scout [#connecting-to-help-scout]

Connects through the source's HTTP API using OAuth client credentials; the connector obtains and refreshes access tokens before API requests. Set `client_id`, `client_secret` in the connection configuration.

Additional resources may require additional scopes. Connection settings belong to the connector; dataset filters and resource selection belong to each sync. See the [CLI workflow](/docs/reference/cli/overview) for registration and configuration.

### Connection settings used by the example [#connection-settings-used-by-the-example]

| Setting | Requirement | Purpose |
| --- | --- | --- |
| `client_id` | Required; secret | Credential or scope value used by the Help Scout API example. |
| `client_secret` | Required; secret | Credential or scope value used by the Help Scout API example. |

### requirements.in [#requirementsin]

Save these dependencies in `requirements.in` alongside `connector.py`; the example includes packages used by its sync. Install the published `vortexify` package for local development. Keep sync-specific dependencies in the sync's own `requirements.in` when splitting the implementation into separate resources.

```text
vortexify==0.1.1
httpx>=0.27,<1
pandas>=2,<3
```

No separately installed database driver is needed by this HTTP example.

## Worked example: sync `conversations` [#worked-example-sync-conversations]

**`conversations`** — Reads records from the `conversations` API resource. Includes `type`, `closedBy`, `createdAt`, `folderId`, `id` and the complete record as JSON in `document`.

The complete files below implement only this example resource. They illustrate the current SDK signatures, a separate preview, bounded DataFrame output, and source-specific extraction. They do not establish complete source coverage. Live credentials and platform ingestion have not been used to validate this guide.

With default paths, place the connector files in `connectors/help-scout/` and sync files in `connectors/help-scout/syncs/conversations/`. Use your project configuration if it specifies different paths.

### Connection schema: `config.json` [#connection-schema-configjson]

This file contains a JSON Schema, not actual credential values. Provide those through platform configuration.

```json
{
  "type": "object",
  "properties": {
    "client_id": {
      "type": "string",
      "title": "Client Id",
      "description": "Credential or scope value used by the Help Scout API example.",
      "x-vortexify-secret": true,
      "order": 0
    },
    "client_secret": {
      "type": "string",
      "title": "Client Secret",
      "description": "Credential or scope value used by the Help Scout API example.",
      "x-vortexify-secret": true,
      "order": 1
    }
  },
  "required": [
    "client_id",
    "client_secret"
  ]
}
```

### Connector: `connector.py` [#connector-connectorpy]

```python
from math import isfinite
from threading import Lock
from time import time
import httpx
from vortexify import Column, Connector, DiscoveredTable


class HelpScoutConnector(Connector):
    vendor_domain = "helpscout.com"

    def client(self):
        if self._client is None or self._client.is_closed:
            config = self.config
            refresh_lock = Lock()
            token_cache = {}

            def authenticate(request: httpx.Request) -> None:
                with refresh_lock:
                    if (
                        not token_cache.get("access_token")
                        or token_cache.get("expires_at", 0) <= time() + 60
                    ):
                        cfg = config
                        requested_at = time()
                        token_response = httpx.post(
                            "https://api.helpscout.net/v2/oauth2/token",
                            data={
                                "grant_type": "client_credentials",
                                "client_id": cfg["client_id"],
                                "client_secret": cfg["client_secret"],
                            },
                            timeout=30,
                        )
                        token_response.raise_for_status()
                        tokens = token_response.json()
                        access_token = tokens["access_token"]
                        expires_in = float(tokens["expires_in"])
                        if not isinstance(access_token, str) or not access_token:
                            raise ValueError("Invalid OAuth access token")
                        if not isfinite(expires_in) or expires_in <= 60:
                            raise ValueError("Invalid OAuth token lifetime")
                        token_cache.update(
                            access_token=access_token,
                            expires_at=requested_at + expires_in,
                        )
                        pass
                    request.headers["Authorization"] = (
                        "Bearer " + token_cache["access_token"]
                    )

            self._client = httpx.Client(
                base_url="https://api.helpscout.net",
                headers={"Accept": "application/json"},
                event_hooks={"request": [authenticate]},
                timeout=60,
                follow_redirects=False,
            )
        return self._client

    def check(self) -> bool:
        cfg = self.config
        page_size = 1
        path = "/v2/conversations"
        params = {
            "status": "all",
            "sortOrder": "asc",
            "embed": "threads",
            "sortField": "modifiedAt",
        }
        response = self.client().get(path, params=params)
        response.raise_for_status()
        payload = response.json()
        values = [payload]
        for part in ["_embedded", "conversations"]:
            children = []
            for value in values:
                if part == "*":
                    if isinstance(value, list):
                        children.extend(value)
                    elif isinstance(value, dict):
                        children.extend(value.values())
                elif isinstance(value, dict) and part in value:
                    children.append(value[part])
                elif isinstance(value, dict) and value:
                    raise ValueError("Missing API record container")
            values = children
        records = []
        for value in values:
            if isinstance(value, list):
                records.extend(value)
            elif isinstance(value, dict):
                records.append(value)
            elif value is not None:
                raise ValueError("Unexpected API record container")
        return True

    def discover(self) -> list[DiscoveredTable]:
        return [
            DiscoveredTable(
                name="conversations",
                schema=[
                    Column(name="type", type="string"),
                    Column(name="closedBy", type="number"),
                    Column(name="createdAt", type="string"),
                    Column(name="folderId", type="number"),
                    Column(name="id", type="number"),
                    Column(name="document", type="string"),
                ],
            )
        ]

    def close(self) -> None:
        if self._client is not None:
            try:
                self._client.close()
            finally:
                self._client = None
```

### Sync schema: `syncs/conversations/config.json` [#sync-schema-syncsconversationsconfigjson]

```json
{
  "type": "object",
  "properties": {},
  "required": []
}
```

### Sync: `syncs/conversations/sync.py` [#sync-syncsconversationssyncpy]

```python
from collections.abc import Iterator
from math import isfinite
import json
import pandas as pd
from vortexify import Sync


class HelpScoutConversationsSync(Sync):

    def preview(self) -> pd.DataFrame:
        cfg = self.connector.config
        page_size = 50
        path = "/v2/conversations"
        params = {
            "status": "all",
            "sortOrder": "asc",
            "embed": "threads",
            "sortField": "modifiedAt",
        }
        response = self.client.get(path, params=params)
        response.raise_for_status()
        payload = response.json()
        values = [payload]
        for part in ["_embedded", "conversations"]:
            children = []
            for value in values:
                if part == "*":
                    if isinstance(value, list):
                        children.extend(value)
                    elif isinstance(value, dict):
                        children.extend(value.values())
                elif isinstance(value, dict) and part in value:
                    children.append(value[part])
                elif isinstance(value, dict) and value:
                    raise ValueError("Missing API record container")
            values = children
        records = []
        for value in values:
            if isinstance(value, list):
                records.extend(value)
            elif isinstance(value, dict):
                records.append(value)
            elif value is not None:
                raise ValueError("Unexpected API record container")
        if not isinstance(records, list) or any(
            (not isinstance(record, dict) for record in records)
        ):
            raise ValueError(
                "Unexpected API record shape; refusing a partial successful sync"
            )
        records = records[:100]
        rows = []
        for record in records:
            row = {
                "type": record.get("type"),
                "closedBy": record.get("closedBy"),
                "createdAt": record.get("createdAt"),
                "folderId": record.get("folderId"),
                "id": record.get("id"),
                "document": json.dumps(record, ensure_ascii=False, default=str),
            }
            rows.append(
                {
                    key: (
                        json.dumps(value, ensure_ascii=False, default=str)
                        if isinstance(value, (dict, list))
                        else value
                    )
                    for key, value in row.items()
                }
            )
        return pd.DataFrame(
            rows,
            columns=["type", "closedBy", "createdAt", "folderId", "id", "document"],
            dtype=object,
        )

    def extract(
        self, *, avg_row_size_bytes: float | None = None
    ) -> Iterator[pd.DataFrame]:
        memory_mib = self.compute.memory if self.compute else None
        memory_mib = memory_mib if memory_mib is not None and memory_mib > 0 else 1024
        target_bytes = min(16 * 1024 * 1024, max(1, memory_mib * 1024 * 1024 // 20))
        batch_size = 100
        if avg_row_size_bytes is not None:
            if not isfinite(avg_row_size_bytes) or avg_row_size_bytes <= 0:
                raise ValueError("avg_row_size_bytes must be finite and positive")
            batch_size = max(1, int(target_bytes / max(1.0, avg_row_size_bytes) / 4))
        batch_size = min(batch_size, 50)
        cfg = self.connector.config
        page_size = batch_size
        path = "/v2/conversations"
        params = {
            "status": "all",
            "sortOrder": "asc",
            "embed": "threads",
            "sortField": "modifiedAt",
        }
        seen_pages = set()
        while True:
            response = self.client.get(path, params=params)
            response.raise_for_status()
            payload = response.json()
            values = [payload]
            for part in ["_embedded", "conversations"]:
                children = []
                for value in values:
                    if part == "*":
                        if isinstance(value, list):
                            children.extend(value)
                        elif isinstance(value, dict):
                            children.extend(value.values())
                    elif isinstance(value, dict) and part in value:
                        children.append(value[part])
                    elif isinstance(value, dict) and value:
                        raise ValueError("Missing API record container")
                values = children
            records = []
            for value in values:
                if isinstance(value, list):
                    records.extend(value)
                elif isinstance(value, dict):
                    records.append(value)
                elif value is not None:
                    raise ValueError("Unexpected API record container")
            if not isinstance(records, list) or any(
                (not isinstance(record, dict) for record in records)
            ):
                raise ValueError(
                    "Unexpected API record shape; refusing a partial successful sync"
                )
            rows = []
            for record in records:
                row = {
                    "type": record.get("type"),
                    "closedBy": record.get("closedBy"),
                    "createdAt": record.get("createdAt"),
                    "folderId": record.get("folderId"),
                    "id": record.get("id"),
                    "document": json.dumps(record, ensure_ascii=False, default=str),
                }
                rows.append(
                    {
                        key: (
                            json.dumps(value, ensure_ascii=False, default=str)
                            if isinstance(value, (dict, list))
                            else value
                        )
                        for key, value in row.items()
                    }
                )
            for start in range(0, len(rows), batch_size):
                yield pd.DataFrame(
                    rows[start : start + batch_size],
                    columns=[
                        "type",
                        "closedBy",
                        "createdAt",
                        "folderId",
                        "id",
                        "document",
                    ],
                    dtype=object,
                )
            next_token = payload.get("_links", {}).get("next", {}).get("href")
            if payload.get("_links", {}).get("next") is None:
                break
            if next_token is None or next_token == "" or next_token == {}:
                break
            marker = json.dumps(next_token, sort_keys=True)
            if marker in seen_pages:
                raise ValueError("API repeated a pagination cursor")
            seen_pages.add(marker)
            next_address = self.client.base_url.join(str(next_token))
            if (next_address.scheme, next_address.host, next_address.port) != (
                self.client.base_url.scheme,
                self.client.base_url.host,
                self.client.base_url.port,
            ):
                raise ValueError("Unexpected pagination origin")
            path = str(next_address)
            params = {}
```

### Running and extending the example [#running-and-extending-the-example]

Register the connector before the sync and commit each resource separately:

```sh
vortexify create connector help-scout
vortexify create sync help-scout/conversations
```

Edit the files, inspect `vortexify status`, and use the [CLI guide](/docs/reference/cli/overview) to commit and configure them. Registration alone does not run extraction or deploy the connector.

Preview must run with a fresh connector and disposable state. The extraction implementation determines full/incremental behavior; merely setting a mode does not implement change capture. API/driver failures propagate unless explicitly handled in the code. Check source limits and single-record/response sizes as well as output batch size.

## Sources and further reading [#sources-and-further-reading]

- [Vendor API and data reference](https://developer.helpscout.com/mailbox-api/) — Resource definitions, authentication, fields, and source-specific limits. Check the selected resource's reference when extending the example.
- [Vortexify connector and sync contracts](/docs/reference/python/connectors-and-syncs) — Client ownership, discovery, preview isolation, and DataFrame extraction.
- [Vortexify CLI workflow](/docs/reference/cli/overview) — Project paths, resource registration, source commits, and credential configuration.