Crm Integration

1. Close CRM - Daily driver for SMB sales (simplest API, best value)
2. HubSpot - Marketing + Sales alignment with rich ecosystem
3. Salesforce - Enterprise requirements and complex workflows
4. Cross-CRM Sync - Bidirectional sync with conflict resolution

Key deliverables:

• API client setup with proper authentication
• CRUD operations for leads, contacts, deals, activities
• Webhook handlers for real-time sync
• Pipeline automation and reporting

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"]

Usage

close = CloseClient(os.environ["CLOSE_API_KEY"]) leads = close.search_leads("company:Coperniq")

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

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

# Create contact
contact = client.crm.contacts.basic_api.create(
    SimplePublicObjectInputForCreate(properties={
        "email": "user@example.com",
        "firstname": "Jane",
        "lastname": "Smith"
    })
)
print(f"Created: {contact.id}")

Salesforce (JWT Bearer):

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"]

Entity Mapping

Pipeline Stage Mapping

Note: Tim's primary CRM is HubSpot via Epiphan CRM MCP. Close CRM patterns are retained for reference but are not the active workflow.

See reference/close-crm-examples.md for query language examples, core operations (leads, opportunities, activities), and rate-limit headers.

See reference/close-deep-dive.md for query language, Smart Views, sequences, and reporting.

Python SDK Pattern

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"])

# Create deal with association
deal = client.crm.deals.basic_api.create(
    SimplePublicObjectInputForCreate(properties={
        "dealname": "Enterprise Deal",
        "amount": "50000",
        "dealstage": "appointmentscheduled",
        "pipeline": "default"
    })
)

# Search contacts by email domain
search = PublicObjectSearchRequest(
    filter_groups=[{
        "filters": [{
            "propertyName": "email",
            "operator": "CONTAINS",
            "value": "@example.com"
        }]
    }],
    properties=["email", "firstname", "lastname"],
    limit=50
)
results = client.crm.contacts.search_api.do_search(search)

Association Types

See reference/hubspot-patterns.md for batch operations, custom properties, and workflows.

SOQL Query Patterns

-- 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)

REST API v59.0

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()

# Composite API (batch up to 200 records)
def composite_create(self, records: list) -> dict:
    return self.client.post(
        f"{self.instance_url}/services/data/v59.0/composite/sobjects",
        json={"allOrNone": False, "records": records}
    )

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

See reference/webhook-handlers.md for the Close webhook FastAPI handler (with signature verification) and the full webhook event list.

Architecture

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

Sync Record Schema

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);

Conflict Resolution

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

See reference/sync-patterns.md for deduplication, migration scripts, and bulk sync.

HubSpot / Epiphan CRM

Clay MCP Enrichment (mcp__claude_ai_Clay__)

Pattern: HubSpot → Apollo → Clay → HubSpot sync

Cost Model:

• Apollo: Free (but rate-limited)
• Clay: Credits-based (~$150-300/month typical usage for BDR teams)
• Waterfall strategy: Try Apollo first (fast, free), fallback to Clay for missing data (phones, emails, work history)

Call Data & Intelligence

CRM-Specific:

• reference/close-deep-dive.md - Query language, Smart Views, sequences, reporting
• reference/close-crm-examples.md - Close query language, core operations, rate-limit headers
• reference/webhook-handlers.md - Close webhook FastAPI handler + event list
• reference/hubspot-patterns.md - SDK patterns, batch operations, workflows
• reference/salesforce-patterns.md - JWT auth, SOQL, Platform Events, bulk API

Operations:

• reference/sync-patterns.md - Cross-CRM sync, deduplication, migration
• reference/automation.md - Webhook setup, sequences, workflows

Templates:

• templates/close-client.py - Full Close API client
• templates/hubspot-client.py - HubSpot SDK wrapper
• templates/sync-service.py - Cross-CRM sync service

User wants CRM integration: → Default to HubSpot (Epiphan CRM MCP) for Tim's BDR workflow → Provide auth setup + basic CRUD

User wants enrichment / contact data: → Use Clay MCP waterfall pattern (preferred for waterfall, credits OK) → Workflow: find-and-enrich-contacts-at-company → add-contact-data-points → poll get-task for results → Fallback: Apollo MCP for quick free enrichment (no polling needed) → Reference: See "Clay MCP Enrichment" in Integration Points above → Cost: Apollo free, Clay $150-300/mo estimate

User wants Close CRM: → Provide API key setup, query language → Reference: reference/close-deep-dive.md

User wants HubSpot: → Use Epiphan CRM MCP tools for direct integration → Available tools: hubspot_search_companies, hubspot_search_contacts, hubspot_search_deals, hubspot_get_company, hubspot_get_contact, hubspot_get_deal → Company identification: crm_search_customers (fuzzy matching) → Activity data: ask_agent (CRM/analytics AI queries) → Enrichment: Clay MCP for waterfall enrichment before writeback to HubSpot → Reference: reference/hubspot-patterns.md

User wants Salesforce: → Provide JWT auth, SOQL patterns → Reference: reference/salesforce-patterns.md

User wants sync between CRMs: → Provide sync architecture, conflict resolution → Reference: reference/sync-patterns.md

User wants webhooks: → Provide handler pattern for specified CRM → Include signature verification

User wants phone verification / waterfall enrichment: → Use Clay MCP after Apollo: find-and-enrich-contacts-at-company → add-contact-data-points for Email/Phone → poll results → Clay aggregates 50+ providers for high match rates on phones and emails → See phone-verification-waterfall-skill for full implementation

Emit Outcome Sidecar

Write to ~/.claude/skill-analytics/last-outcome-crm-integration.json: {"ts":"[UTC ISO8601]","skill":"crm-integration","version":"1.0.0","variant":"default","status":"[success|partial|error]","runtime_ms":[ms],"metrics":{"integrations_configured":[n],"records_synced":[n]},"error":null,"session_id":"[YYYY-MM-DD]"}