Connector Layer

Audience: Engineers implementing or swapping vendor adapters.
What you will accomplish: Understand the ports-and-adapters architecture — how to implement a new adapter, test it with contract tests, and register it without touching core logic.
Prerequisites: Read Architecture and Core Services first.
Estimated time: ~10 minutes.

The connector layer implements ports-and-adapters architecture. Workflows import only Protocol definitions (the ports); each vendor system has an adapter that maps between the domain model and the vendor's API.

What is a connector?

A connector is the bridge between the server's domain model and an external vendor API (Clio, Salesforce, Microsoft Graph, etc.). The system uses ports and adapters: a port is a Python Protocol (interface) that defines what operations are available; an adapter is the concrete implementation that talks to a specific vendor.

Workflows import only the port. Swapping vendors means implementing one adapter file — zero workflow code changes.

Port definitions

class CaseConnector(Protocol):
    async def get_matter(self, id: str) -> Matter: ...
    async def create_matter(self, data: MatterDraft) -> Matter: ...
    async def list_deadlines(self, matter_id: str) -> list[Deadline]: ...
 
class CRMConnector(Protocol):
    async def upsert_contact(self, c: Contact) -> Contact: ...
    async def find_contact(self, q: str) -> list[Contact]: ...
 
class EmailConnector(Protocol):
    async def create_draft(self, c: Communication) -> str: ...
    async def send(self, draft_id: str, idem_key: str) -> str: ...
 
class DocStoreConnector(Protocol):
    async def put(self, doc: Document, content: bytes) -> Document: ...
    async def get(self, id: str) -> tuple[Document, bytes]: ...
    async def move(self, id: str, folder: str, acl: ACL) -> Document: ...

The ACL value type is defined with precise typing:

class ACL(BaseModel):
    principals: list[str]                          # user IDs or role names
    permission: Literal["read", "write", "none"]   # access level
    external: bool                                  # whether this is an external share

The external field is critical for the privilege gate: a document with privileged=True can never be moved with an ACL where external=True.

Workflows depend on the port, never the adapter. Swapping Clio → MyCase touches one file.

Adapter requirements

Every connector adapter must:

1. Map vendor payloads ↔ domain model — no vendor JSON reaches the core
2. Use httpx + tenacity — exponential backoff, jitter, honour rate limits
3. Be idempotent on writes — pass/forward idempotency keys
4. Surface typed ConnectorError — auth, rate-limit, not-found, transient, fatal
5. Ship contract tests — against recorded VCR fixtures (respx/vcr)
6. Ship CONNECTOR.md — auth model, scopes, endpoints, quirks, webhook events

Retry policy defaults

All outbound connector calls use httpx + tenacity with these defaults:

RateLimitError delays respect the vendor's Retry-After header when present, overriding the calculated backoff.

Connector registry

Connectors are registered at startup through a ConnectorRegistry:

from cam.connectors import ConnectorRegistry, ReferenceCaseConnector, ReferenceCRMConnector
 
registry = ConnectorRegistry()
registry.register("case", ReferenceCaseConnector())
registry.register("crm", ReferenceCRMConnector())

The ServiceContainer receives the registry, and workflows access connectors by category name (services.connectors.case, services.connectors.crm, etc.) — never by adapter class name.

ConnectorError taxonomy

The orchestrator uses the error type to decide retry vs gate vs fail:

Reference adapters

All four connector categories ship reference adapters — in-memory implementations that satisfy the Protocol for testing and development:

These enable full-fidelity workflow testing without live vendor credentials or network calls.

v1 candidate adapters

Webhook ingestion

The FastAPI sidecar exposes /webhooks/{connector}. Each inbound event:

1. Verified — HMAC signature or shared secret per provider
2. Normalised — mapped to internal Event model
3. Deduplicated — provider event id → Redis (TTL = 7 days)
4. Enqueued — passed to orchestrator as a workflow trigger

Verification

@router.post("/webhooks/{connector}")
async def receive_webhook(
    connector: str,
    request: Request,
    x_hub_signature: str = Header(None),
):
    body = await request.body()
    verifier = get_verifier(connector)
    if not verifier.verify(body, x_hub_signature):
        raise HTTPException(401, "Invalid signature")
    
    event = normalise(connector, body)
    if await is_duplicate(event.id):
        return {"status": "duplicate"}
    
    await enqueue_trigger(event)
    return {"status": "accepted"}

Dedup TTL

Default: 7 days. Configurable via CAM_WEBHOOK_DEDUP_TTL_SECONDS.

Circuit breaker

Each connector has a ConnectorHealth circuit breaker:

• CLOSED — normal operation
• OPEN — after failure_threshold errors in window_seconds; all requests fail fast
• HALF_OPEN — after cool_down_seconds; allows a probe request

For multi-worker deployments, circuit state is backed by Redis sorted sets (Phase 2).

Isolation guarantee

One connector failing does not cascade to others. Each connector has its own circuit breaker, connection pool, and health state. If the CRM adapter goes OPEN, the case management adapter continues operating normally.

Connection pooling

Per-connector httpx.AsyncClient with pool limits aligned to vendor rate limits:

_http_client = httpx.AsyncClient(
    limits=httpx.Limits(
        max_connections=50,
        max_keepalive_connections=20,
        keepalive_expiry=30.0,
    ),
    timeout=httpx.Timeout(connect=5.0, read=30.0, write=10.0, pool=2.0),
)

Testing connectors

Contract tests use respx to mock vendor HTTP APIs and verify the adapter maps correctly:

import respx
 
@respx.mock
def test_clio_get_matter():
    route = respx.get("https://app.clio.com/api/v4/matters/123.json").mock(
        return_value=httpx.Response(200, json={"data": {...}})
    )
    
    adapter = ClioAdapter(token="test")
    matter = asyncio.run(adapter.get_matter("123"))
    
    assert matter.reference == "REF-123"
    assert route.called

Common mistakes

Verification

After reading this page, verify your understanding:

1. Name the four connector port Protocols and their key methods
2. Explain what happens when a connector returns TransientError vs AuthError
3. Show how the circuit breaker state machine works (CLOSED → OPEN → HALF_OPEN)
4. Describe the webhook ingestion pipeline: verification → normalisation → dedup → enqueue

Next steps

• MCP Surface — Tools that invoke connectors
• Workflow Engine — How connectors are called from steps
• Core Services — What happens to connector outputs

Last updated: 2026-06-01