Designing a Vendor-Agnostic Python Connector Interface

Good to see the repository taking shape. In the previous lesson, you created the FastAPI and Next.js application roots and proved the frontend can call a versioned API endpoint. The next risk is architectural: if FastAPI routes and future growth workflows call GoHighLevel, Airtable, or another vendor SDK directly, vendor-specific assumptions will quickly spread through the product.

This lesson creates a narrow Python connector contract for CRM contact synchronization. It will give the application a stable, tenant-bound interface while leaving GoHighLevel’s endpoints, authentication, payload shapes, and error responses inside a future adapter. The same pattern will support opportunities, Airtable imports, campaign tools, and additional CRMs without turning the command center into a collection of vendor conditionals.

---

Ports, adapters, and the boundary that matters

A connector interface is an application port: a statement of what the product needs an external system to do. An adapter is the vendor-specific implementation that fulfills that port.

For this command center, the application should be able to say:

“Synchronize this canonical contact with the configured CRM.”

It should not need to know:

• which GoHighLevel URL or location identifier to use;
• how GoHighLevel represents custom fields;
• whether the CRM creates, updates, or looks up a contact internally;
• how a vendor encodes rate-limit, authentication, or validation failures;
• whether a future client uses GoHighLevel, HubSpot, or another supported CRM.

Read this short AWS explanation before implementing the pattern.

The diagram has two useful implications for your SaaS:

1. The port reflects a product capability, not a vendor API. A ContactConnector is meaningful because the product synchronizes contacts. A GoHighLevelClient is not a port because it exposes a particular vendor.
2. The test adapter is a first-class adapter. A fake connector can satisfy the same contract as a live GoHighLevel adapter, allowing workflow tests to run without credentials, network calls, or a sandbox CRM.

This separation is especially valuable when Claude Code helps implement features. A small, explicit contract gives it a firm boundary: it can change GoHighLevel mapping code without “helpfully” leaking GoHighLevel fields into qualification, outreach, or dashboard code.

For a concise Python-oriented walkthrough of the dependency direction, watch these three segments.

---

Design the contract from the workflow, not the vendor

A common first attempt looks like this:

async def sync_to_gohighlevel(contact: dict[str, object]) -> dict[str, object]:
    ...

This seems fast, but it imports vendor language into the product’s workflow. Eventually, routes and services begin to depend on GoHighLevel-specific concepts such as locationId, endpoint paths, raw JSON field names, and status codes. Replacing or adding a CRM then becomes a broad rewrite.

Instead, define a contract from the operation the command center needs now: upsert one email-addressable canonical contact into a tenant’s configured CRM.

“Upsert” is intentionally a product-level operation. An adapter may realize it with a vendor’s native upsert endpoint, or with a provider-specific lookup followed by a create or update. That decision belongs to the adapter.

The contract needs four categories of information:

Category	Product-facing form	Must remain vendor-specific
Input	Canonical contact ID, email, name, company, phone, approved attributes	JSON keys, custom-field IDs, location IDs
Context	A connector already bound to one organization and its integration	Access token, API base URL, account or subaccount identifiers
Output	Provider name, external contact ID, normalized action, sync time	Raw response body, provider status codes
Failure	Normalized retryable or non-retryable connector errors	SDK exception types, raw error payloads

Two choices deserve attention.

Keep connectors tenant-bound

Do not write a method such as:

async def upsert_contact(
    organization_id: UUID,
    access_token: str,
    command: ContactUpsertCommand,
) -> ContactSyncResult:
    ...

That signature lets callers supply arbitrary organization identifiers and credentials. It also makes it too easy for a future background job to accidentally pair one tenant’s contact with another tenant’s integration.

Instead, a server-side factory will eventually resolve the active organization, load its encrypted integration credentials, and construct a connector already bound to that organization. The contact operation then accepts only the data to be synchronized.

You will implement credential storage and a tenant-aware connector factory in Module 5. For now, this contract makes the desired security boundary explicit.

Prefer small capability-specific ports

A giant CRMConnector with contacts, opportunities, campaigns, analytics, and arbitrary vendor methods is a “god interface.” It forces every integration to pretend it supports every feature.

Start with ContactConnector. When the product needs opportunity synchronization, define an OpportunityConnector with its own commands and results. A GoHighLevel adapter may implement both, while Airtable may instead implement a distinct lead-import capability. The application depends only on the capability required by the workflow.

---

Add the connector contract

From apps/api, create the first connector package:

mkdir -p app/connectors
touch app/connectors/__init__.py

Create apps/api/app/connectors/contracts.py:

from dataclasses import dataclass, field
from datetime import datetime
from typing import Literal, Mapping, Protocol
from uuid import UUID


@dataclass(frozen=True, slots=True)
class ContactUpsertCommand:
    """Canonical contact data approved for synchronization to a CRM."""

    canonical_contact_id: UUID
    email: str
    idempotency_key: str
    first_name: str | None = None
    last_name: str | None = None
    company_name: str | None = None
    phone: str | None = None
    attributes: Mapping[str, str] = field(default_factory=dict)

    def __post_init__(self) -> None:
        if not self.email.strip():
            raise ValueError("email must not be blank")
        if not self.idempotency_key.strip():
            raise ValueError("idempotency_key must not be blank")


@dataclass(frozen=True, slots=True)
class ContactSyncResult:
    """The normalized outcome of a successful contact synchronization."""

    provider: str
    external_contact_id: str
    action: Literal["created", "updated", "unchanged"]
    synced_at: datetime


class ContactConnector(Protocol):
    """A tenant-bound port for synchronizing contacts to a CRM."""

    @property
    def provider(self) -> str:
        """Return a stable provider identifier for observability."""
        ...

    async def upsert_contact(
        self,
        command: ContactUpsertCommand,
    ) -> ContactSyncResult:
        """Synchronize one canonical contact using the configured CRM."""
        ...

This is deliberately plain Python:

• dataclass provides explicit, framework-independent input and output models.
• Mapping[str, str] captures approved canonical attributes without exposing vendor custom-field IDs.
• Literal constrains the successful outcome to behavior the application understands.
• Protocol describes required behavior without requiring an implementation to inherit from a shared base class.
• async reflects the fact that a real connector will call a network service.

The idempotency_key identifies one logical product operation. It does not claim that every CRM has a native idempotency header. A future adapter must translate this guarantee into the safest vendor-specific mechanism available, such as an external reference, a deterministic lookup, or an integration-side idempotency record.

Why use a protocol?

Python protocols use structural typing. A class satisfies ContactConnector if it provides compatible provider and upsert_contact members; it does not need to inherit from ContactConnector.

Avoid decorating this protocol with @runtime_checkable simply to call isinstance(connector, ContactConnector). Runtime protocol checks only establish that members exist; they do not verify compatible method signatures. Static type checking is the correct protection for this architectural contract.

---

Define safe, normalized connector failures

A real CRM adapter will encounter network timeouts, expired credentials, rate limits, and vendor validation errors. The rest of the application should not need to know a provider SDK’s exception hierarchy.

Create apps/api/app/connectors/errors.py:

class ConnectorError(Exception):
    """A normalized failure from an external connector."""

    def __init__(
        self,
        *,
        provider: str,
        operation: str,
        retryable: bool,
        safe_message: str,
    ) -> None:
        super().__init__(safe_message)
        self.provider = provider
        self.operation = operation
        self.retryable = retryable
        self.safe_message = safe_message


class ConnectorRejectedError(ConnectorError):
    """The provider rejected validly delivered input; retrying will not help."""

    def __init__(
        self,
        *,
        provider: str,
        operation: str,
        safe_message: str,
    ) -> None:
        super().__init__(
            provider=provider,
            operation=operation,
            retryable=False,
            safe_message=safe_message,
        )


class ConnectorRateLimitedError(ConnectorError):
    """The provider requests delayed retry of an operation."""

    def __init__(
        self,
        *,
        provider: str,
        operation: str,
        safe_message: str,
        retry_after_seconds: int | None = None,
    ) -> None:
        super().__init__(
            provider=provider,
            operation=operation,
            retryable=True,
            safe_message=safe_message,
        )
        self.retry_after_seconds = retry_after_seconds

An adapter will later translate provider-specific failures into these application errors. For example:

• an invalid or revoked GoHighLevel credential becomes a non-retryable connector error;
• a 429 response becomes ConnectorRateLimitedError;
• a temporary network failure becomes a retryable ConnectorError;
• a CRM rejection caused by an invalid field value becomes ConnectorRejectedError.

safe_message is intentional. Never place access tokens, unfiltered vendor responses, or personally identifiable contact data into errors that may later be shown in the client dashboard, stored in an automation run, or emitted to logs.

The API route or background worker will decide how to present or retry a normalized failure. The adapter only translates between the vendor’s behavior and this contract; it should not raise FastAPI HTTPException objects.

---

Prove the contract with a test adapter

A protocol has its greatest value when code can operate against it without calling a real provider. Add a test package and a fake connector:

mkdir -p tests/connectors

Create apps/api/tests/connectors/test_contact_connector.py:

import asyncio
from dataclasses import dataclass, field
from datetime import UTC, datetime
from uuid import UUID, uuid4

from app.connectors.contracts import (
    ContactConnector,
    ContactSyncResult,
    ContactUpsertCommand,
)


@dataclass(slots=True)
class FakeContactConnector:
    provider: str = "fake-crm"
    received: list[ContactUpsertCommand] = field(default_factory=list)

    async def upsert_contact(
        self,
        command: ContactUpsertCommand,
    ) -> ContactSyncResult:
        self.received.append(command)

        return ContactSyncResult(
            provider=self.provider,
            external_contact_id=f"fake-{command.canonical_contact_id}",
            action="created",
            synced_at=datetime.now(UTC),
        )


def requires_contact_connector(connector: ContactConnector) -> None:
    """This function represents any future contact-sync workflow."""
    del connector


def test_fake_connector_satisfies_contact_contract() -> None:
    connector = FakeContactConnector()
    requires_contact_connector(connector)

    command = ContactUpsertCommand(
        canonical_contact_id=uuid4(),
        email="ada@example.com",
        first_name="Ada",
        last_name="Lovelace",
        company_name="Analytical Engines Ltd",
        idempotency_key="contact-sync-001",
        attributes={"job_title": "Founder"},
    )

    result = asyncio.run(connector.upsert_contact(command))

    assert connector.received == [command]
    assert result.provider == "fake-crm"
    assert result.external_contact_id == f"fake-{command.canonical_contact_id}"
    assert result.action == "created"

requires_contact_connector() has no runtime behavior. Its role is to make the expected dependency visible to the type checker. If a future fake or GoHighLevel adapter omits upsert_contact, has the wrong parameter type, or returns the wrong result type, static checking should reject it.

Add mypy to the dev dependency group in apps/api/pyproject.toml:

[dependency-groups]
dev = [
  "httpx>=0.27.0,<1.0.0",
  "mypy>=1.13.0,<2.0.0",
  "pytest>=8.0.0,<9.0.0",
  "ruff>=0.8.0,<1.0.0",
]

Then add this configuration:

[tool.mypy]
python_version = "3.12"
strict = true

Synchronize dependencies and validate the new boundary:

cd apps/api

uv sync
uv run ruff check .
uv run mypy app tests
uv run pytest

These checks prove different things:

Check	What it protects
ruff check	Basic style and likely code mistakes
mypy app tests	Protocol compatibility and typed contract usage
pytest	Observable behavior of the fake adapter and command/result models

The test does not prove that a real CRM behaves correctly. It proves that application code can rely on the contract without network access. In Module 5, provider contract tests will add pagination, rate-limit, and transient-failure cases around the real adapter.

---

Keep vendor details on one side of the boundary

Your next implementation will add a GoHighLevel contact adapter. Its internal flow may need to:

1. Obtain the organization’s server-side integration configuration.
2. Translate ContactUpsertCommand into GoHighLevel’s request payload and custom-field identifiers.
3. Perform the provider-specific lookup, create, or update operation.
4. Translate the provider response into ContactSyncResult.
5. Convert vendor failures into ConnectorError subclasses.

Only the adapter should know those details.

A healthy future file layout will therefore look like this:

apps/api/app/
├── connectors/
│   ├── contracts.py
│   ├── errors.py
│   └── gohighlevel/
│       └── contacts.py
├── services/
│   └── contact_sync.py
└── api/
    └── v1/
        └── routes/

The eventual contact_sync.py service will depend on ContactConnector, not on GoHighLevelContactAdapter. The FastAPI dependency layer or worker composition code will choose which concrete adapter to provide for the authenticated organization.

Before committing, use this boundary review:

• Does contracts.py avoid imports from FastAPI, httpx, GoHighLevel SDKs, and database libraries?
• Does the port describe a product capability rather than a vendor endpoint?
• Is the connector tenant-bound rather than passed arbitrary credentials or organization IDs per call?
• Do commands and results use canonical names rather than raw vendor JSON?
• Do errors expose only safe, normalized operational details?
• Can a fake adapter exercise a workflow without network access?
• Does no route or service import a vendor SDK directly?

Commit the contract as a focused architectural change:

git add apps/api
git commit -m "feat: define contact connector contract"

---

Key takeaways

You now have the first integration boundary for the Growth Command Center:

• A port defines what the product needs from an external CRM.
• A vendor adapter translates between that stable contract and provider-specific API behavior.
• ContactConnector is narrow and capability-focused, avoiding a universal CRM interface.
• ContactUpsertCommand and ContactSyncResult use product language, not GoHighLevel payload shapes.
• A connector should be constructed with server-side, tenant-specific integration context rather than receive credentials on each call.
• Protocol plus strict static typing verifies structural compatibility without inheritance.
• A fake adapter gives you fast, credential-free tests.

Next, you will implement the first concrete adapter behavior: creating and updating GoHighLevel contacts through this connector layer.