import httpx

class CloseClient:
    BASE_URL = "https://api.close.com/api/v1"

    def __init__(self, api_key: str):
        self.client = httpx.Client(
            base_url=self.BASE_URL,
            auth=(api_key, ""),  # Basic auth, password empty
            timeout=30.0,
        )

    def create_lead(self, data: dict) -> dict:
        response = self.client.post("/lead/", json=data)
        response.raise_for_status()
        return response.json()

    def search_leads(self, query: str) -> list:
        response = self.client.post("/data/search/", json={
            "query": {"type": "query_string", "value": query},
            "results_limit": 100
        })
        return response.json()["data"]

**HubSpot (Python SDK):**
```python
from hubspot import HubSpot
from hubspot.crm.contacts import SimplePublicObjectInputForCreate

client = HubSpot(access_token=os.environ["HUBSPOT_ACCESS_TOKEN"])

**Salesforce (JWT Bearer):**
```python
import jwt
from datetime import datetime, timedelta

class SalesforceClient:
    def __init__(self, client_id: str, username: str, private_key: str):
        self.auth_url = "https://login.salesforce.com"
        self._authenticate(client_id, username, private_key)

    def _authenticate(self, client_id, username, private_key):
        payload = {
            "iss": client_id,
            "sub": username,
            "aud": self.auth_url,
            "exp": int((datetime.utcnow() + timedelta(minutes=3)).timestamp())
        }
        assertion = jwt.encode(payload, private_key, algorithm="RS256")

        response = httpx.post(f"{self.auth_url}/services/oauth2/token", data={
            "grant_type": "urn:ietf:params:oauth:grant-type:jwt-bearer",
            "assertion": assertion
        })
        self.access_token = response.json()["access_token"]
        self.instance_url = response.json()["instance_url"]

undefined

undefined

undefined

undefined

X-RateLimit-Limit: 100
X-RateLimit-Remaining: 95
X-RateLimit-Reset: 1704067200

from hubspot import HubSpot
from hubspot.crm.deals import SimplePublicObjectInputForCreate
from hubspot.crm.contacts import PublicObjectSearchRequest

client = HubSpot(access_token=os.environ["HUBSPOT_ACCESS_TOKEN"])

undefined

-- Parent-child relationship (Contacts of Account)
SELECT Id, Name, (SELECT LastName, Email FROM Contacts)
FROM Account WHERE Industry = 'Technology'

-- Child-parent relationship
SELECT Id, FirstName, Account.Name, Account.Industry
FROM Contact WHERE Account.Industry = 'Technology'

-- Semi-join (Accounts with open Opportunities)
SELECT Id, Name FROM Account
WHERE Id IN (SELECT AccountId FROM Opportunity WHERE IsClosed = false)

def create_opportunity(self, data: dict) -> dict:
    """Required: Name, StageName, CloseDate."""
    response = self.client.post(
        f"{self.instance_url}/services/data/v59.0/sobjects/Opportunity/",
        headers={"Authorization": f"Bearer {self.access_token}"},
        json=data
    )
    return response.json()

> See `reference/salesforce-patterns.md` for JWT setup, Platform Events, and bulk API.
</salesforce_patterns>

<webhook_patterns>

from fastapi import FastAPI, Request, HTTPException
import hmac, hashlib

app = FastAPI()

@app.post("/webhooks/close")
async def close_webhook(request: Request):
    body = await request.body()
    signature = request.headers.get("Close-Sig")

    expected = hmac.new(
        CLOSE_WEBHOOK_SECRET.encode(), body, hashlib.sha256
    ).hexdigest()

    if not hmac.compare_digest(signature, expected):
        raise HTTPException(401, "Invalid signature")

    data = await request.json()
    event_type = data["event"]["event_type"]

    handlers = {
        "lead.created": handle_lead_created,
        "opportunity.status_changed": handle_opp_stage_change,
    }

    if handler := handlers.get(event_type):
        await handler(data["event"]["data"])

    return {"status": "ok"}

lead.created, lead.updated, lead.deleted, lead.status_changed
contact.created, contact.updated
opportunity.created, opportunity.status_changed
activity.note.created, activity.call.created, activity.email.created
unsubscribed_email.created

┌─────────────┐     ┌──────────────┐     ┌─────────────┐
│   Close     │────▶│  Sync Layer  │◀────│  HubSpot    │
│  (Primary)  │◀────│  (Postgres)  │────▶│  (Marketing)│
└─────────────┘     └──────────────┘     └─────────────┘

CREATE TABLE crm_sync_records (
    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    entity_type VARCHAR(50) NOT NULL,
    close_id VARCHAR(100) UNIQUE,
    hubspot_id VARCHAR(100) UNIQUE,
    salesforce_id VARCHAR(100) UNIQUE,
    email VARCHAR(255),
    company_name VARCHAR(255),
    last_synced_at TIMESTAMPTZ,
    sync_source VARCHAR(50),
    sync_hash VARCHAR(64)
);

CREATE INDEX idx_sync_email ON crm_sync_records(email);

from enum import Enum

class ConflictStrategy(Enum):
    CLOSE_WINS = "close"      # Close is source of truth
    LAST_WRITE_WINS = "lww"   # Most recent update wins

def resolve_conflict(close_record, hubspot_record, strategy):
    if strategy == ConflictStrategy.CLOSE_WINS:
        merged = close_record.copy()
        for key, value in hubspot_record.items():
            if key not in merged or not merged[key]:
                merged[key] = value
        return merged

undefined

```bash

</env_setup>

<example_session>

async def push_to_close(close_client, enriched_data: dict) -> str:
    lead_data = {
        "name": enriched_data["company_name"],
        "url": enriched_data.get("website"),
        "custom.cf_tier": enriched_data["tier"],
        "custom.cf_source": "sales-agent",
        "contacts": [{
            "name": c["name"],
            "title": c.get("title"),
            "emails": [{"email": c["email"]}] if c.get("email") else []
        } for c in enriched_data.get("contacts", [])]
    }

    result = close_client.create_lead(lead_data)
    return result["id"]