Entity Mapping

Create mappings between Kruncher entities (person, company, opportunity) and external provider IDs. This enables bidirectional synchronization with CRM systems like Affinity, Attio, Pipedrive, and other third-party platforms.

Endpoint

POST https://api.kruncher.ai/api/integration/map

Headers

Header	Required	Description
`Authorization`	Yes	Your API key (format: `YOUR_API_KEY`)
`Content-Type`	Yes	Must be `application/json`

Request Body

Field	Type	Required	Description
`entityType`	string	Yes	Type of entity: `person`, `company`, or `opportunity`
`provider`	string	Yes	External provider name (e.g., `affinity`, `attio`, `pipedrive`, `zapier`)
`externalId`	string	Yes	ID used by the external provider for this entity
`kruncherId`	string	Yes	ID used internally by Kruncher for this entity

Entity Types

Type	Description	Example Use Case
`person`	Individual contacts, founders, investors	Sync contact information with CRM
`company`	Organizations, startups, portfolio companies	Link company records across systems
`opportunity`	Investment opportunities, deals, projects	Track deals in multiple platforms

Supported Providers

Provider	Platform	Integration Type
`affinity`	Affinity CRM	Relationship intelligence platform
`attio`	Attio	Modern CRM for startups
`pipedrive`	Pipedrive	Sales CRM and pipeline management
`zapier`	Zapier	Automation platform
`custom`	Custom Integration	Your own system

Note: Contact support to enable additional provider integrations.

Use Cases

When to Create Mappings

CRM Sync: Link Kruncher companies to CRM records
Bidirectional Updates: Keep data in sync across platforms
External References: Maintain links to external systems
Data Migration: Map existing records during migration
Automation: Enable Zapier or custom integrations
Multi-System Workflow: Connect multiple tools together

Code Examples

JavaScript/TypeScript

Basic

CODE

const API_KEY = "YOUR_API_KEY_HERE";
 
const response = await fetch("https://api.kruncher.ai/api/integration/map", {
  method: "POST",
  headers: {
    "Authorization": `${API_KEY}`,
    "Content-Type": "application/json"
  },
  body: JSON.stringify({
    entityType: "company",
    provider: "affinity",
    externalId: "affinity_12345",
    kruncherId: "kruncher_67890"
  })
});
 
const result = await response.json();
console.log("Mapping created:", result);

Result: Creates a mapping between Kruncher company and Affinity organization.

Bulk Mapping

CODE

const API_KEY = "YOUR_API_KEY_HERE";
const BASE_URL = "https://api.kruncher.ai/api";
 
async function createMapping(entityType, provider, externalId, kruncherId) {
  const response = await fetch(`${BASE_URL}/integration/map`, {
    method: "POST",
    headers: {
      "Authorization": `${API_KEY}`,
      "Content-Type": "application/json"
    },
    body: JSON.stringify({
      entityType,
      provider,
      externalId,
      kruncherId
    })
  });
 
  if (!response.ok) {
    throw new Error(`Mapping failed: ${response.status} ${response.statusText}`);
  }
 
  return await response.json();
}
 
async function bulkCreateMappings(mappings) {
  const results = {
    successful: [],
    failed: [],
    total: mappings.length
  };
 
  for (const mapping of mappings) {
    try {
      const result = await createMapping(
        mapping.entityType,
        mapping.provider,
        mapping.externalId,
        mapping.kruncherId
      );
      results.successful.push({ ...mapping, result });
      console.log(`✓ Mapped ${mapping.entityType}: ${mapping.kruncherId}`);
    } catch (error) {
      results.failed.push({ ...mapping, error: error.message });
      console.error(`✗ Failed to map ${mapping.entityType}: ${error.message}`);
    }
  }
 
  return results;
}
 
// Usage: Map multiple companies to Affinity
const companyMappings = [
  {
    entityType: "company",
    provider: "affinity",
    externalId: "affinity_12345",
    kruncherId: "proj_abc123"
  },
  {
    entityType: "company",
    provider: "affinity",
    externalId: "affinity_67890",
    kruncherId: "proj_def456"
  },
  {
    entityType: "person",
    provider: "affinity",
    externalId: "affinity_person_123",
    kruncherId: "person_xyz789"
  }
];
 
const results = await bulkCreateMappings(companyMappings);
console.log(`Success: ${results.successful.length}/${results.total}`);
console.log(`Failed: ${results.failed.length}/${results.total}`);

Result: Creates multiple mappings with error handling and reporting.

Advanced

CODE

const API_KEY = "YOUR_API_KEY_HERE";
const BASE_URL = "https://api.kruncher.ai/api";
 
class KruncherMappingManager {
  constructor(apiKey) {
    this.apiKey = apiKey;
    this.baseURL = BASE_URL;
    this.mappingCache = new Map();
  }
 
  async createMapping(entityType, provider, externalId, kruncherId) {
    const response = await fetch(`${this.baseURL}/integration/map`, {
      method: "POST",
      headers: {
        "Authorization": `${this.apiKey}`,
        "Content-Type": "application/json"
      },
      body: JSON.stringify({
        entityType,
        provider,
        externalId,
        kruncherId
      })
    });
 
    if (!response.ok) {
      const error = await response.json().catch(() => ({ 
        message: response.statusText 
      }));
      throw new Error(`Mapping failed: ${error.message || error.description}`);
    }
 
    const result = await response.json();
    
    // Cache the mapping
    const cacheKey = `${entityType}:${provider}:${kruncherId}`;
    this.mappingCache.set(cacheKey, externalId);
    
    return result;
  }
 
  async mapCompany(kruncherId, provider, externalId) {
    return await this.createMapping("company", provider, externalId, kruncherId);
  }
 
  async mapPerson(kruncherId, provider, externalId) {
    return await this.createMapping("person", provider, externalId, kruncherId);
  }
 
  async mapOpportunity(kruncherId, provider, externalId) {
    return await this.createMapping("opportunity", provider, externalId, kruncherId);
  }
 
  async bulkMap(mappings, options = {}) {
    const { 
      batchSize = 10, 
      delayMs = 100,
      stopOnError = false 
    } = options;
 
    const results = {
      successful: [],
      failed: [],
      total: mappings.length
    };
 
    for (let i = 0; i < mappings.length; i += batchSize) {
      const batch = mappings.slice(i, i + batchSize);
      
      const batchPromises = batch.map(async (mapping) => {
        try {
          const result = await this.createMapping(
            mapping.entityType,
            mapping.provider,
            mapping.externalId,
            mapping.kruncherId
          );
          return { success: true, mapping, result };
        } catch (error) {
          return { success: false, mapping, error: error.message };
        }
      });
 
      const batchResults = await Promise.all(batchPromises);
 
      for (const result of batchResults) {
        if (result.success) {
          results.successful.push(result);
        } else {
          results.failed.push(result);
          if (stopOnError) {
            return results;
          }
        }
      }
 
      // Delay between batches
      if (i + batchSize < mappings.length) {
        await new Promise(resolve => setTimeout(resolve, delayMs));
      }
    }
 
    return results;
  }
 
  async syncWithAffinity(kruncherCompanies, affinityOrganizations) {
    const mappings = kruncherCompanies.map((company, index) => ({
      entityType: "company",
      provider: "affinity",
      externalId: affinityOrganizations[index].id,
      kruncherId: company.id
    }));
 
    return await this.bulkMap(mappings);
  }
 
  async syncWithAttio(kruncherData, attioData) {
    const mappings = [];
 
    // Map companies
    kruncherData.companies?.forEach((company, index) => {
      mappings.push({
        entityType: "company",
        provider: "attio",
        externalId: attioData.organizations[index].id,
        kruncherId: company.id
      });
    });
 
    // Map people
    kruncherData.people?.forEach((person, index) => {
      mappings.push({
        entityType: "person",
        provider: "attio",
        externalId: attioData.people[index].id,
        kruncherId: person.id
      });
    });
 
    return await this.bulkMap(mappings, { batchSize: 20 });
  }
 
  getCachedExternalId(entityType, provider, kruncherId) {
    const cacheKey = `${entityType}:${provider}:${kruncherId}`;
    return this.mappingCache.get(cacheKey);
  }
 
  clearCache() {
    this.mappingCache.clear();
  }
 
  getMappingStats(results) {
    return {
      total: results.total,
      successful: results.successful.length,
      failed: results.failed.length,
      successRate: (results.successful.length / results.total * 100).toFixed(2) + '%',
      failedMappings: results.failed.map(f => ({
        entityType: f.mapping.entityType,
        kruncherId: f.mapping.kruncherId,
        error: f.error
      }))
    };
  }
}
 
// Usage examples
const manager = new KruncherMappingManager("YOUR_API_KEY_HERE");
 
// Map single company
await manager.mapCompany("proj_123", "affinity", "affinity_org_456");
 
// Map single person
await manager.mapPerson("person_789", "attio", "attio_person_012");
 
// Bulk mapping
const mappings = [
  {
    entityType: "company",
    provider: "affinity",
    externalId: "affinity_12345",
    kruncherId: "proj_abc123"
  },
  {
    entityType: "company",
    provider: "affinity",
    externalId: "affinity_67890",
    kruncherId: "proj_def456"
  }
];
 
const results = await manager.bulkMap(mappings, {
  batchSize: 5,
  delayMs: 200
});
 
const stats = manager.getMappingStats(results);
console.log("Mapping stats:", stats);
 
// Sync with Affinity
const kruncherCompanies = [
  { id: "proj_123", name: "Company A" },
  { id: "proj_456", name: "Company B" }
];
 
const affinityOrgs = [
  { id: "affinity_org_1", name: "Company A" },
  { id: "affinity_org_2", name: "Company B" }
];
 
await manager.syncWithAffinity(kruncherCompanies, affinityOrgs);

Result: Full-featured mapping manager with caching, batching, and CRM sync.

Python

Basic

CODE

import requests
 
API_KEY = "YOUR_API_KEY_HERE"
url = "https://api.kruncher.ai/api/integration/map"
 
headers = {
    "Authorization": f"{API_KEY}",
    "Content-Type": "application/json"
}
 
data = {
    "entityType": "company",
    "provider": "affinity",
    "externalId": "affinity_12345",
    "kruncherId": "kruncher_67890"
}
 
response = requests.post(url, headers=headers, json=data)
 
if response.status_code == 200:
    print("Mapping created:", response.json())
else:
    print(f"Error: {response.status_code} {response.text}")

Result: Creates a single entity mapping.

Bulk Mapping

CODE

import requests
 
API_KEY = "YOUR_API_KEY_HERE"
BASE_URL = "https://api.kruncher.ai/api"
 
def create_mapping(entity_type, provider, external_id, kruncher_id):
    """Create a single mapping."""
    url = f"{BASE_URL}/integration/map"
    
    headers = {
        "Authorization": f"{API_KEY}",
        "Content-Type": "application/json"
    }
    
    payload = {
        "entityType": entity_type,
        "provider": provider,
        "externalId": external_id,
        "kruncherId": kruncher_id
    }
    
    response = requests.post(url, headers=headers, json=payload)
    response.raise_for_status()
    return response.json()
 
def bulk_create_mappings(mappings):
    """Create multiple mappings with error handling."""
    results = {
        'successful': [],
        'failed': [],
        'total': len(mappings)
    }
    
    for mapping in mappings:
        try:
            result = create_mapping(
                mapping['entityType'],
                mapping['provider'],
                mapping['externalId'],
                mapping['kruncherId']
            )
            results['successful'].append({**mapping, 'result': result})
            print(f"✓ Mapped {mapping['entityType']}: {mapping['kruncherId']}")
        except Exception as e:
            results['failed'].append({**mapping, 'error': str(e)})
            print(f"✗ Failed to map {mapping['entityType']}: {e}")
    
    return results
 
# Usage
mappings = [
    {
        "entityType": "company",
        "provider": "affinity",
        "externalId": "affinity_12345",
        "kruncherId": "proj_abc123"
    },
    {
        "entityType": "company",
        "provider": "affinity",
        "externalId": "affinity_67890",
        "kruncherId": "proj_def456"
    },
    {
        "entityType": "person",
        "provider": "affinity",
        "externalId": "affinity_person_123",
        "kruncherId": "person_xyz789"
    }
]
 
results = bulk_create_mappings(mappings)
print(f"\nSuccess: {len(results['successful'])}/{results['total']}")
print(f"Failed: {len(results['failed'])}/{results['total']}")

Result: Bulk mapping with detailed error reporting.

Advanced

CODE

import requests
import time
from typing import Dict, List, Optional, Tuple
from dataclasses import dataclass
from concurrent.futures import ThreadPoolExecutor, as_completed
 
API_KEY = "YOUR_API_KEY_HERE"
BASE_URL = "https://api.kruncher.ai/api"
 
@dataclass
class MappingResult:
    entity_type: str
    provider: str
    external_id: str
    kruncher_id: str
    success: bool
    result: Optional[Dict] = None
    error: Optional[str] = None
 
class KruncherMappingManager:
    """Comprehensive mapping manager for Kruncher integrations."""
    
    def __init__(self, api_key: str = API_KEY):
        self.api_key = api_key
        self.base_url = BASE_URL
        self.session = requests.Session()
        self.session.headers.update({
            "Authorization": f"{api_key}",
            "Content-Type": "application/json"
        })
        self.mapping_cache = {}
    
    def create_mapping(
        self,
        entity_type: str,
        provider: str,
        external_id: str,
        kruncher_id: str
    ) -> Dict:
        """
        Create a mapping between Kruncher and external provider.
        
        Args:
            entity_type: Type of entity (person, company, opportunity)
            provider: External provider name
            external_id: ID in external system
            kruncher_id: ID in Kruncher
        
        Returns:
            API response dict
        
        Raises:
            requests.HTTPError: If mapping fails
        """
        payload = {
            "entityType": entity_type,
            "provider": provider,
            "externalId": external_id,
            "kruncherId": kruncher_id
        }
        
        response = self.session.post(
            f"{self.base_url}/integration/map",
            json=payload
        )
        response.raise_for_status()
        
        # Cache the mapping
        cache_key = f"{entity_type}:{provider}:{kruncher_id}"
        self.mapping_cache[cache_key] = external_id
        
        return response.json()
    
    def map_company(
        self,
        kruncher_id: str,
        provider: str,
        external_id: str
    ) -> Dict:
        """Create company mapping."""
        return self.create_mapping("company", provider, external_id, kruncher_id)
    
    def map_person(
        self,
        kruncher_id: str,
        provider: str,
        external_id: str
    ) -> Dict:
        """Create person mapping."""
        return self.create_mapping("person", provider, external_id, kruncher_id)
    
    def map_opportunity(
        self,
        kruncher_id: str,
        provider: str,
        external_id: str
    ) -> Dict:
        """Create opportunity mapping."""
        return self.create_mapping("opportunity", provider, external_id, kruncher_id)
    
    def bulk_map(
        self,
        mappings: List[Dict],
        batch_size: int = 10,
        delay_seconds: float = 0.1,
        parallel: bool = False,
        max_workers: int = 5
    ) -> List[MappingResult]:
        """
        Create multiple mappings with batching and optional parallelization.
        
        Args:
            mappings: List of mapping dicts
            batch_size: Number of mappings per batch
            delay_seconds: Delay between batches
            parallel: Whether to process in parallel
            max_workers: Max parallel workers
        
        Returns:
            List of MappingResult objects
        """
        results = []
        
        if parallel:
            results = self._bulk_map_parallel(mappings, max_workers)
        else:
            results = self._bulk_map_sequential(
                mappings,
                batch_size,
                delay_seconds
            )
        
        return results
    
    def _bulk_map_sequential(
        self,
        mappings: List[Dict],
        batch_size: int,
        delay_seconds: float
    ) -> List[MappingResult]:
        """Process mappings sequentially with batching."""
        results = []
        
        for i in range(0, len(mappings), batch_size):
            batch = mappings[i:i + batch_size]
            
            for mapping in batch:
                try:
                    result = self.create_mapping(
                        mapping['entityType'],
                        mapping['provider'],
                        mapping['externalId'],
                        mapping['kruncherId']
                    )
                    results.append(MappingResult(
                        entity_type=mapping['entityType'],
                        provider=mapping['provider'],
                        external_id=mapping['externalId'],
                        kruncher_id=mapping['kruncherId'],
                        success=True,
                        result=result
                    ))
                except Exception as e:
                    results.append(MappingResult(
                        entity_type=mapping['entityType'],
                        provider=mapping['provider'],
                        external_id=mapping['externalId'],
                        kruncher_id=mapping['kruncherId'],
                        success=False,
                        error=str(e)
                    ))
            
            # Delay between batches
            if i + batch_size < len(mappings):
                time.sleep(delay_seconds)
        
        return results
    
    def _bulk_map_parallel(
        self,
        mappings: List[Dict],
        max_workers: int
    ) -> List[MappingResult]:
        """Process mappings in parallel."""
        results = []
        
        def process_mapping(mapping):
            try:
                result = self.create_mapping(
                    mapping['entityType'],
                    mapping['provider'],
                    mapping['externalId'],
                    mapping['kruncherId']
                )
                return MappingResult(
                    entity_type=mapping['entityType'],
                    provider=mapping['provider'],
                    external_id=mapping['externalId'],
                    kruncher_id=mapping['kruncherId'],
                    success=True,
                    result=result
                )
            except Exception as e:
                return MappingResult(
                    entity_type=mapping['entityType'],
                    provider=mapping['provider'],
                    external_id=mapping['externalId'],
                    kruncher_id=mapping['kruncherId'],
                    success=False,
                    error=str(e)
                )
        
        with ThreadPoolExecutor(max_workers=max_workers) as executor:
            futures = {
                executor.submit(process_mapping, mapping): mapping
                for mapping in mappings
            }
            
            for future in as_completed(futures):
                results.append(future.result())
        
        return results
    
    def sync_with_affinity(
        self,
        kruncher_companies: List[Dict],
        affinity_organizations: List[Dict]
    ) -> List[MappingResult]:
        """Sync Kruncher companies with Affinity organizations."""
        mappings = [
            {
                "entityType": "company",
                "provider": "affinity",
                "externalId": affinity_org['id'],
                "kruncherId": kruncher_company['id']
            }
            for kruncher_company, affinity_org 
            in zip(kruncher_companies, affinity_organizations)
        ]
        
        return self.bulk_map(mappings)
    
    def sync_with_attio(
        self,
        kruncher_data: Dict,
        attio_data: Dict
    ) -> List[MappingResult]:
        """Sync Kruncher data with Attio."""
        mappings = []
        
        # Map companies
        if 'companies' in kruncher_data and 'organizations' in attio_data:
            for kruncher_co, attio_org in zip(
                kruncher_data['companies'],
                attio_data['organizations']
            ):
                mappings.append({
                    "entityType": "company",
                    "provider": "attio",
                    "externalId": attio_org['id'],
                    "kruncherId": kruncher_co['id']
                })
        
        # Map people
        if 'people' in kruncher_data and 'people' in attio_data:
            for kruncher_person, attio_person in zip(
                kruncher_data['people'],
                attio_data['people']
            ):
                mappings.append({
                    "entityType": "person",
                    "provider": "attio",
                    "externalId": attio_person['id'],
                    "kruncherId": kruncher_person['id']
                })
        
        return self.bulk_map(mappings, batch_size=20)
    
    def get_cached_external_id(
        self,
        entity_type: str,
        provider: str,
        kruncher_id: str
    ) -> Optional[str]:
        """Get cached external ID."""
        cache_key = f"{entity_type}:{provider}:{kruncher_id}"
        return self.mapping_cache.get(cache_key)
    
    def get_mapping_stats(self, results: List[MappingResult]) -> Dict:
        """Get statistics from mapping results."""
        successful = [r for r in results if r.success]
        failed = [r for r in results if not r.success]
        
        return {
            'total': len(results),
            'successful': len(successful),
            'failed': len(failed),
            'success_rate': f"{len(successful) / len(results) * 100:.2f}%" if results else "0%",
            'by_entity_type': self._count_by_field(results, 'entity_type'),
            'by_provider': self._count_by_field(results, 'provider'),
            'failed_mappings': [
                {
                    'entity_type': r.entity_type,
                    'kruncher_id': r.kruncher_id,
                    'error': r.error
                }
                for r in failed
            ]
        }
    
    def _count_by_field(
        self,
        results: List[MappingResult],
        field: str
    ) -> Dict[str, int]:
        """Count results by field."""
        counts = {}
        for result in results:
            value = getattr(result, field)
            counts[value] = counts.get(value, 0) + 1
        return counts
    
    def clear_cache(self):
        """Clear mapping cache."""
        self.mapping_cache.clear()
 
# Usage examples
manager = KruncherMappingManager()
 
# Map single company
manager.map_company("proj_123", "affinity", "affinity_org_456")
 
# Map single person
manager.map_person("person_789", "attio", "attio_person_012")
 
# Bulk mapping
mappings = [
    {
        "entityType": "company",
        "provider": "affinity",
        "externalId": "affinity_12345",
        "kruncherId": "proj_abc123"
    },
    {
        "entityType": "company",
        "provider": "affinity",
        "externalId": "affinity_67890",
        "kruncherId": "proj_def456"
    }
]
 
results = manager.bulk_map(mappings, batch_size=5, delay_seconds=0.2)
stats = manager.get_mapping_stats(results)
print(f"Mapping stats: {stats}")
 
# Parallel bulk mapping
results_parallel = manager.bulk_map(
    mappings,
    parallel=True,
    max_workers=5
)
 
# Sync with Affinity
kruncher_companies = [
    {"id": "proj_123", "name": "Company A"},
    {"id": "proj_456", "name": "Company B"}
]
 
affinity_orgs = [
    {"id": "affinity_org_1", "name": "Company A"},
    {"id": "affinity_org_2", "name": "Company B"}
]
 
sync_results = manager.sync_with_affinity(kruncher_companies, affinity_orgs)
print(f"Synced {len([r for r in sync_results if r.success])} companies")

Result: Production-ready manager with parallel processing and CRM sync.

cURL

Company

CODE

curl -X POST "https://api.kruncher.ai/api/integration/map" \
  -H "Authorization: YOUR_API_KEY_HERE" \
  -H "Content-Type: application/json" \
  -d '{
    "entityType": "company",
    "provider": "affinity",
    "externalId": "affinity_12345",
    "kruncherId": "proj_abc123"
  }'

Person

CODE

curl -X POST "https://api.kruncher.ai/api/integration/map" \
  -H "Authorization: YOUR_API_KEY_HERE" \
  -H "Content-Type: application/json" \
  -d '{
    "entityType": "person",
    "provider": "attio",
    "externalId": "attio_person_123",
    "kruncherId": "person_xyz789"
  }'

Opportunity

CODE

curl -X POST "https://api.kruncher.ai/api/integration/map" \
  -H "Authorization: YOUR_API_KEY_HERE" \
  -H "Content-Type: application/json" \
  -d '{
    "entityType": "opportunity",
    "provider": "pipedrive",
    "externalId": "pipedrive_deal_456",
    "kruncherId": "opportunity_def456"
  }'

Response

Success Response (200 OK)

CODE

{
  "code": "1000",
  "title": "Success",
  "description": "Mapping created successfully",
  "data": {
    "entityType": "company",
    "provider": "affinity",
    "externalId": "affinity_12345",
    "kruncherId": "proj_abc123",
    "createdAt": "2024-01-15T10:30:00Z"
  }
}

Error Responses

400 Bad Request

CODE

{
  "code": "4000",
  "title": "Bad Request",
  "description": "Invalid entity type or missing required fields"
}

409 Conflict

CODE

{
  "code": "4090",
  "title": "Conflict",
  "description": "Mapping already exists for this entity"
}

Common Workflows

Workflow 1: Sync Kruncher with Affinity

CODE

// Fetch companies from both systems
const kruncherCompanies = await getKruncherCompanies();
const affinityOrgs = await getAffinityOrganizations();
 
// Match by name or other criteria
const matched = matchCompanies(kruncherCompanies, affinityOrgs);
 
// Create mappings
const manager = new KruncherMappingManager(API_KEY);
const results = await manager.syncWithAffinity(
  matched.kruncher,
  matched.affinity
);
 
console.log(`Mapped ${results.successful.length} companies`);

Workflow 2: Bidirectional Sync

CODE

# Create mapping
manager = KruncherMappingManager()
manager.map_company("proj_123", "affinity", "affinity_org_456")
 
# Now updates can flow both ways:
# - Update in Kruncher → sync to Affinity using external ID
# - Update in Affinity → sync to Kruncher using kruncher ID
 
# Get external ID for syncing
external_id = manager.get_cached_external_id("company", "affinity", "proj_123")
update_affinity(external_id, data)

Workflow 3: Migration Mapping

CODE

# Map all existing records during migration
import csv
 
manager = KruncherMappingManager()
mappings = []
 
with open('migration_mapping.csv') as f:
    reader = csv.DictReader(f)
    for row in reader:
        mappings.append({
            "entityType": row['entity_type'],
            "provider": row['provider'],
            "externalId": row['external_id'],
            "kruncherId": row['kruncher_id']
        })
 
# Process in batches
results = manager.bulk_map(mappings, batch_size=50, delay_seconds=0.5)
stats = manager.get_mapping_stats(results)
 
print(f"Migration complete: {stats['success_rate']}")
if stats['failed_mappings']:
    print(f"Failed mappings: {len(stats['failed_mappings'])}")
    for failed in stats['failed_mappings']:
        print(f"  - {failed['kruncher_id']}: {failed['error']}")

Integration Examples

Affinity CRM Sync

CODE

async function syncAffinityCompanies() {
  const manager = new KruncherMappingManager(API_KEY);
  
  // Get companies from Kruncher
  const kruncherResponse = await fetch(
    `${BASE_URL}/integration/projects`,
    { headers: { "Authorization": `${API_KEY}` } }
  );
  const kruncherData = await kruncherResponse.json();
  
  // Get organizations from Affinity (using their API)
  const affinityOrgs = await fetchAffinityOrganizations();
  
  // Match by company name
  const mappings = [];
  for (const company of kruncherData.data) {
    const affinityOrg = affinityOrgs.find(
      org => org.name.toLowerCase() === company.companyName.toLowerCase()
    );
    
    if (affinityOrg) {
      mappings.push({
        entityType: "company",
        provider: "affinity",
        externalId: affinityOrg.id.toString(),
        kruncherId: company.id
      });
    }
  }
  
  const results = await manager.bulkMap(mappings);
  return results;
}

Attio Integration

CODE

def sync_attio_contacts():
    """Sync Kruncher people with Attio contacts."""
    manager = KruncherMappingManager()
    
    # Get people from both systems
    kruncher_people = get_kruncher_people()
    attio_contacts = get_attio_contacts()
    
    # Match by email
    mappings = []
    for person in kruncher_people:
        attio_contact = next(
            (c for c in attio_contacts if c['email'] == person['email']),
            None
        )
        
        if attio_contact:
            mappings.append({
                "entityType": "person",
                "provider": "attio",
                "externalId": attio_contact['id'],
                "kruncherId": person['id']
            })
    
    results = manager.bulk_map(mappings)
    return results

Best Practices

Matching Strategy

Use unique identifiers (email, website, name + domain)
Validate matches before creating mappings
Log unmatched entities for manual review
Consider fuzzy matching for name variations

Error Handling

Check for existing mappings before creating
Handle 409 Conflict errors gracefully
Retry failed mappings with exponential backoff
Log all mapping operations for audit

Performance

Batch mappings to reduce API calls
Use parallel processing for large datasets
Add delays between batches to avoid rate limits
Cache mappings locally

Data Integrity

Verify entity exists in both systems
Validate external IDs format
Keep mapping records for rollback
Monitor sync success rates

Create Company - Create Kruncher entities to map
Find Companies - Get Kruncher IDs for mapping
Configuration - Get provider list
Integrations - Provider-specific documentation

Troubleshooting

”Mapping already exists”

Check if mapping was created previously
Delete old mapping if updating
Use different external ID if needed

Invalid Entity Type

Use exactly: person, company, or opportunity
Check for typos or capitalization

Provider Not Supported

Verify provider name spelling
Contact support to enable provider
Use custom for custom integrations

IDs Not Matching

Verify entity exists in both systems
Check ID format (string vs number)
Ensure IDs are from correct environment (staging vs production)

Need Help?

New provider? Contact support to enable integrations
Bulk migration? Ask about assisted migration services
Custom mapping? Discuss custom provider integration
Sync issues? Check provider-specific integration docs

Entity Mapping

Endpoint

Headers

Request Body

Entity Types

Supported Providers

Use Cases

When to Create Mappings

Code Examples

JavaScript/TypeScript

Basic

Bulk Mapping

Advanced

Python

Basic

Bulk Mapping

Advanced

cURL

Company

Person

Opportunity

Response

Success Response (200 OK)

Error Responses

400 Bad Request

409 Conflict

Common Workflows

Workflow 1: Sync Kruncher with Affinity

Workflow 2: Bidirectional Sync

Workflow 3: Migration Mapping

Integration Examples

Affinity CRM Sync

Attio Integration

Best Practices

Matching Strategy

Error Handling

Performance

Data Integrity

Related Endpoints

Troubleshooting

”Mapping already exists”

Invalid Entity Type

Provider Not Supported

IDs Not Matching

Need Help?