You're a GTM engineer. Your job is building the data pipelines, integrations, and automations that power revenue operations. You don't want a point-and-click UI. You want an API that returns clean JSON, an SDK that handles retries, and documentation that doesn't make you guess.

This gtm engineer databar api mcp guide covers everything you need to integrate Databar into your GTM stack. REST API patterns, MCP server setup for AI-native workflows, Python SDK examples, common integration patterns, error handling, and rate limits. All practical. All code-first.

Bottom line up front: Databar gives you three integration paths: REST API for direct HTTP calls, Python/Node SDKs for application code, and an MCP server for AI agent workflows. All three access the same 100+ data providers with waterfall enrichment, outcome-based billing, and no annual contracts.

Databar API and MCP Server Architecture: API vs SDK vs MCP

Before you write a line of code, choose your integration pattern. Each path has different strengths.

Most GTM engineers use a combination. The SDK handles batch jobs and scheduled enrichment. The API handles real-time webhook-triggered enrichment. The MCP server lets AI agents pull enrichment data during autonomous workflows. MCP vs SDK vs API: when to use which for GTM workflows

REST API patterns

Authentication

All API requests use Bearer token authentication. Get your API key from the Databar dashboard under Settings → API Keys.

curl -X POST https://api.databar.ai/v1/enrich \
  -H "Authorization: Bearer YOUR_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{
    "domain": "stripe.com",
    "fields": ["company_name", "employee_count", "industry", "funding_total"]
  }'

curl -X POST https://api.databar.ai/v1/enrich \
  -H "Authorization: Bearer YOUR_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{
    "domain": "stripe.com",
    "fields": ["company_name", "employee_count", "industry", "funding_total"]
  }'

curl -X POST https://api.databar.ai/v1/enrich \
  -H "Authorization: Bearer YOUR_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{
    "domain": "stripe.com",
    "fields": ["company_name", "employee_count", "industry", "funding_total"]
  }'

curl -X POST https://api.databar.ai/v1/enrich \
  -H "Authorization: Bearer YOUR_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{
    "domain": "stripe.com",
    "fields": ["company_name", "employee_count", "industry", "funding_total"]
  }'

Core endpoints

The API follows RESTful conventions. Here are the endpoints you'll use most:

• POST /v1/enrich - Single record enrichment. Pass a domain, email, or company name. Returns enriched data in one call.

• POST /v1/enrich/bulk - Batch enrichment. Pass an array of records. Returns a job ID for polling.

• POST /v1/waterfall - Waterfall enrichment. Cascades through multiple providers until one returns verified data.

• GET /v1/jobs/{job_id} - Check batch job status. Returns progress and results when complete.

• GET /v1/providers - List available data providers with their capabilities and pricing.

Single enrichment request and response

// Request
POST /v1/enrich
{
  "domain": "databar.ai",
  "fields": ["company_name", "employee_count", "industry", "tech_stack", "funding_total"]
}

// Response
{
  "status": "success",
  "data": {
    "company_name": "Databar",
    "domain": "databar.ai",
    "employee_count": 25,
    "industry": "Software / SaaS",
    "tech_stack": ["React", "Node.js", "PostgreSQL", "AWS"],
    "funding_total": null,
    "enrichment_sources": ["provider_a", "provider_c"],
    "credits_used": 2
  }
}

// Request
POST /v1/enrich
{
  "domain": "databar.ai",
  "fields": ["company_name", "employee_count", "industry", "tech_stack", "funding_total"]
}

// Response
{
  "status": "success",
  "data": {
    "company_name": "Databar",
    "domain": "databar.ai",
    "employee_count": 25,
    "industry": "Software / SaaS",
    "tech_stack": ["React", "Node.js", "PostgreSQL", "AWS"],
    "funding_total": null,
    "enrichment_sources": ["provider_a", "provider_c"],
    "credits_used": 2
  }
}

// Request
POST /v1/enrich
{
  "domain": "databar.ai",
  "fields": ["company_name", "employee_count", "industry", "tech_stack", "funding_total"]
}

// Response
{
  "status": "success",
  "data": {
    "company_name": "Databar",
    "domain": "databar.ai",
    "employee_count": 25,
    "industry": "Software / SaaS",
    "tech_stack": ["React", "Node.js", "PostgreSQL", "AWS"],
    "funding_total": null,
    "enrichment_sources": ["provider_a", "provider_c"],
    "credits_used": 2
  }
}

// Request
POST /v1/enrich
{
  "domain": "databar.ai",
  "fields": ["company_name", "employee_count", "industry", "tech_stack", "funding_total"]
}

// Response
{
  "status": "success",
  "data": {
    "company_name": "Databar",
    "domain": "databar.ai",
    "employee_count": 25,
    "industry": "Software / SaaS",
    "tech_stack": ["React", "Node.js", "PostgreSQL", "AWS"],
    "funding_total": null,
    "enrichment_sources": ["provider_a", "provider_c"],
    "credits_used": 2
  }
}

Waterfall enrichment request

Waterfall is where Databar's multi-provider architecture shines. Instead of calling one provider, the API cascades through multiple sources automatically.

POST /v1/waterfall
{
  "type": "email_finder",
  "inputs": {
    "first_name": "Sarah",
    "last_name": "Chen",
    "domain": "stripe.com"
  },
  "providers": ["provider_a", "provider_b", "provider_c"],
  "verify": true
}

POST /v1/waterfall
{
  "type": "email_finder",
  "inputs": {
    "first_name": "Sarah",
    "last_name": "Chen",
    "domain": "stripe.com"
  },
  "providers": ["provider_a", "provider_b", "provider_c"],
  "verify": true
}

POST /v1/waterfall
{
  "type": "email_finder",
  "inputs": {
    "first_name": "Sarah",
    "last_name": "Chen",
    "domain": "stripe.com"
  },
  "providers": ["provider_a", "provider_b", "provider_c"],
  "verify": true
}

POST /v1/waterfall
{
  "type": "email_finder",
  "inputs": {
    "first_name": "Sarah",
    "last_name": "Chen",
    "domain": "stripe.com"
  },
  "providers": ["provider_a", "provider_b", "provider_c"],
  "verify": true
}

The verify: true flag runs email verification on the result before returning it. This is important. Finding an email is step one. Verifying it's deliverable is step two. The waterfall handles both in a single call.

Python SDK examples

The Python SDK wraps the REST API with typed responses, automatic retries, and batch helpers. Install it from PyPI:

pip install databar

pip install databar

pip install databar

pip install databar

Basic setup

from databar import Databar

client = Databar(api_key="YOUR_API_KEY")

# Single company enrichment
result = client.enrich(
    domain="stripe.com",
    fields=["company_name", "employee_count", "industry"]
)
print(result.data.company_name)  # "Stripe"
print(result.data.employee_count)  # 8000+
print(result.credits_used)  # 1

from databar import Databar

client = Databar(api_key="YOUR_API_KEY")

# Single company enrichment
result = client.enrich(
    domain="stripe.com",
    fields=["company_name", "employee_count", "industry"]
)
print(result.data.company_name)  # "Stripe"
print(result.data.employee_count)  # 8000+
print(result.credits_used)  # 1

from databar import Databar

client = Databar(api_key="YOUR_API_KEY")

# Single company enrichment
result = client.enrich(
    domain="stripe.com",
    fields=["company_name", "employee_count", "industry"]
)
print(result.data.company_name)  # "Stripe"
print(result.data.employee_count)  # 8000+
print(result.credits_used)  # 1

from databar import Databar

client = Databar(api_key="YOUR_API_KEY")

# Single company enrichment
result = client.enrich(
    domain="stripe.com",
    fields=["company_name", "employee_count", "industry"]
)
print(result.data.company_name)  # "Stripe"
print(result.data.employee_count)  # 8000+
print(result.credits_used)  # 1

Batch enrichment with progress tracking

domains = ["stripe.com", "notion.so", "figma.com", "linear.app"]

job = client.enrich_bulk(
    domains=domains,
    fields=["company_name", "employee_count", "industry", "funding_total"]
)

# Poll for completion
while not job.is_complete:
    job.refresh()
    print(f"Progress: {job.progress}%")

# Access results
for record in job.results:
    print(f"{record.company_name}: {record.employee_count} employees")

domains = ["stripe.com", "notion.so", "figma.com", "linear.app"]

job = client.enrich_bulk(
    domains=domains,
    fields=["company_name", "employee_count", "industry", "funding_total"]
)

# Poll for completion
while not job.is_complete:
    job.refresh()
    print(f"Progress: {job.progress}%")

# Access results
for record in job.results:
    print(f"{record.company_name}: {record.employee_count} employees")

domains = ["stripe.com", "notion.so", "figma.com", "linear.app"]

job = client.enrich_bulk(
    domains=domains,
    fields=["company_name", "employee_count", "industry", "funding_total"]
)

# Poll for completion
while not job.is_complete:
    job.refresh()
    print(f"Progress: {job.progress}%")

# Access results
for record in job.results:
    print(f"{record.company_name}: {record.employee_count} employees")

domains = ["stripe.com", "notion.so", "figma.com", "linear.app"]

job = client.enrich_bulk(
    domains=domains,
    fields=["company_name", "employee_count", "industry", "funding_total"]
)

# Poll for completion
while not job.is_complete:
    job.refresh()
    print(f"Progress: {job.progress}%")

# Access results
for record in job.results:
    print(f"{record.company_name}: {record.employee_count} employees")

Waterfall email enrichment

# Find and verify an email using waterfall
email_result = client.waterfall(
    type="email_finder",
    inputs={
        "first_name": "Sarah",
        "last_name": "Chen",
        "domain": "stripe.com"
    },
    verify=True
)

print(email_result.email)  # "sarah.chen@stripe.com"
print(email_result.verified)  # True
print(email_result.provider)  # Which provider found it
print(email_result.credits_used)  # Credits consumed

# Find and verify an email using waterfall
email_result = client.waterfall(
    type="email_finder",
    inputs={
        "first_name": "Sarah",
        "last_name": "Chen",
        "domain": "stripe.com"
    },
    verify=True
)

print(email_result.email)  # "sarah.chen@stripe.com"
print(email_result.verified)  # True
print(email_result.provider)  # Which provider found it
print(email_result.credits_used)  # Credits consumed

# Find and verify an email using waterfall
email_result = client.waterfall(
    type="email_finder",
    inputs={
        "first_name": "Sarah",
        "last_name": "Chen",
        "domain": "stripe.com"
    },
    verify=True
)

print(email_result.email)  # "sarah.chen@stripe.com"
print(email_result.verified)  # True
print(email_result.provider)  # Which provider found it
print(email_result.credits_used)  # Credits consumed

# Find and verify an email using waterfall
email_result = client.waterfall(
    type="email_finder",
    inputs={
        "first_name": "Sarah",
        "last_name": "Chen",
        "domain": "stripe.com"
    },
    verify=True
)

print(email_result.email)  # "sarah.chen@stripe.com"
print(email_result.verified)  # True
print(email_result.provider)  # Which provider found it
print(email_result.credits_used)  # Credits consumed

Building a CRM sync pipeline

Here's a practical pattern for enriching new CRM records automatically:

from databar import Databar
import json

client = Databar(api_key="YOUR_API_KEY")

def enrich_new_lead(lead):
    """Enrich a new lead from your CRM webhook."""

    # Step 1: Enrich company data
    company = client.enrich(
        domain=lead["company_domain"],
        fields=["employee_count", "industry", "tech_stack", "funding_total"]
    )

    # Step 2: Find and verify email if missing
    if not lead.get("email"):
        email_result = client.waterfall(
            type="email_finder",
            inputs={
                "first_name": lead["first_name"],
                "last_name": lead["last_name"],
                "domain": lead["company_domain"]
            },
            verify=True
        )
        lead["email"] = email_result.email
        lead["email_verified"] = email_result.verified

    # Step 3: Score based on enriched data
    lead["icp_score"] = calculate_icp_score(company.data)

    # Step 4: Push back to CRM
    update_crm_record(lead)

    return lead

from databar import Databar
import json

client = Databar(api_key="YOUR_API_KEY")

def enrich_new_lead(lead):
    """Enrich a new lead from your CRM webhook."""

    # Step 1: Enrich company data
    company = client.enrich(
        domain=lead["company_domain"],
        fields=["employee_count", "industry", "tech_stack", "funding_total"]
    )

    # Step 2: Find and verify email if missing
    if not lead.get("email"):
        email_result = client.waterfall(
            type="email_finder",
            inputs={
                "first_name": lead["first_name"],
                "last_name": lead["last_name"],
                "domain": lead["company_domain"]
            },
            verify=True
        )
        lead["email"] = email_result.email
        lead["email_verified"] = email_result.verified

    # Step 3: Score based on enriched data
    lead["icp_score"] = calculate_icp_score(company.data)

    # Step 4: Push back to CRM
    update_crm_record(lead)

    return lead

from databar import Databar
import json

client = Databar(api_key="YOUR_API_KEY")

def enrich_new_lead(lead):
    """Enrich a new lead from your CRM webhook."""

    # Step 1: Enrich company data
    company = client.enrich(
        domain=lead["company_domain"],
        fields=["employee_count", "industry", "tech_stack", "funding_total"]
    )

    # Step 2: Find and verify email if missing
    if not lead.get("email"):
        email_result = client.waterfall(
            type="email_finder",
            inputs={
                "first_name": lead["first_name"],
                "last_name": lead["last_name"],
                "domain": lead["company_domain"]
            },
            verify=True
        )
        lead["email"] = email_result.email
        lead["email_verified"] = email_result.verified

    # Step 3: Score based on enriched data
    lead["icp_score"] = calculate_icp_score(company.data)

    # Step 4: Push back to CRM
    update_crm_record(lead)

    return lead

from databar import Databar
import json

client = Databar(api_key="YOUR_API_KEY")

def enrich_new_lead(lead):
    """Enrich a new lead from your CRM webhook."""

    # Step 1: Enrich company data
    company = client.enrich(
        domain=lead["company_domain"],
        fields=["employee_count", "industry", "tech_stack", "funding_total"]
    )

    # Step 2: Find and verify email if missing
    if not lead.get("email"):
        email_result = client.waterfall(
            type="email_finder",
            inputs={
                "first_name": lead["first_name"],
                "last_name": lead["last_name"],
                "domain": lead["company_domain"]
            },
            verify=True
        )
        lead["email"] = email_result.email
        lead["email_verified"] = email_result.verified

    # Step 3: Score based on enriched data
    lead["icp_score"] = calculate_icp_score(company.data)

    # Step 4: Push back to CRM
    update_crm_record(lead)

    return lead

Databar API and MCP Server Setup for AI-Native Workflows

The Model Context Protocol (MCP) server lets AI agents like Claude access Databar's enrichment capabilities directly. This is the integration pattern for AI-native GTM workflows where an agent autonomously researches accounts, enriches contacts, and prepares outreach.

Why MCP matters for GTM engineers

Traditional integrations are request-response. You define the workflow in code, and the API returns data. MCP flips this. The AI agent decides what data it needs, calls the right enrichment endpoints, interprets the results, and takes the next action. Your job is setting up the server and defining the boundaries.

Setting up the Databar MCP server

Add the Databar MCP server to your configuration:

// .mcp.json
{
  "mcpServers": {
    "databar": {
      "command": "npx",
      "args": ["-y", "@databar/mcp-server"],
      "env": {
        "DATABAR_API_KEY": "YOUR_API_KEY"
      }
    }
  }
}

// .mcp.json
{
  "mcpServers": {
    "databar": {
      "command": "npx",
      "args": ["-y", "@databar/mcp-server"],
      "env": {
        "DATABAR_API_KEY": "YOUR_API_KEY"
      }
    }
  }
}

// .mcp.json
{
  "mcpServers": {
    "databar": {
      "command": "npx",
      "args": ["-y", "@databar/mcp-server"],
      "env": {
        "DATABAR_API_KEY": "YOUR_API_KEY"
      }
    }
  }
}

// .mcp.json
{
  "mcpServers": {
    "databar": {
      "command": "npx",
      "args": ["-y", "@databar/mcp-server"],
      "env": {
        "DATABAR_API_KEY": "YOUR_API_KEY"
      }
    }
  }
}

Available MCP tools

Once connected, the AI agent has access to these tools:

• search_enrichments - Find available enrichment types (company data, contact finder, email verification, etc.)

• run_enrichment - Execute a single enrichment against any provider

• run_waterfall - Execute waterfall enrichment across multiple providers

• run_bulk_enrichment - Batch enrich a list of records

• create_table - Create a table to store enrichment results

• get_table_rows - Retrieve data from stored tables

Example: AI agent researching an account

When an AI agent has access to the Databar MCP server, it can autonomously:

1. Take a company domain as input

2. Run company enrichment to get firmographic data

3. Search for decision-makers by title and department

4. Find and verify email addresses using waterfall

5. Pull technographic data to identify relevant talking points

6. Compile everything into a structured account brief

All of this happens in a single agent conversation. The agent calls the right MCP tools in sequence, handles errors, and produces a complete output. No hardcoded workflow required.

Multi-source enrichment platform: how waterfall delivers 40% higher accuracy

Common integration patterns

Pattern 1: Webhook-triggered enrichment

Trigger: New lead in CRM (HubSpot, Salesforce, Attio). Action: Enrich immediately, update CRM record, route to correct owner.

# Flask webhook handler
@app.route('/webhook/new-lead', methods=['POST'])
def handle_new_lead():
    lead = request.json

    enriched = client.enrich(
        domain=lead['company_domain'],
        fields=["employee_count", "industry", "funding_total"]
    )

    # Route based on enriched data
    headcount = enriched.data.employee_count

    if headcount in range(201, 1_000_000):
        assign_to = "enterprise_team"
    elif headcount in range(51, 201):
        assign_to = "mid_market_team"
    else:
        assign_to = "smb_team"

    update_crm(lead['id'], enriched.data, assign_to)
    return {"status": "enriched", "assigned_to": assign_to}

# Flask webhook handler
@app.route('/webhook/new-lead', methods=['POST'])
def handle_new_lead():
    lead = request.json

    enriched = client.enrich(
        domain=lead['company_domain'],
        fields=["employee_count", "industry", "funding_total"]
    )

    # Route based on enriched data
    headcount = enriched.data.employee_count

    if headcount in range(201, 1_000_000):
        assign_to = "enterprise_team"
    elif headcount in range(51, 201):
        assign_to = "mid_market_team"
    else:
        assign_to = "smb_team"

    update_crm(lead['id'], enriched.data, assign_to)
    return {"status": "enriched", "assigned_to": assign_to}

# Flask webhook handler
@app.route('/webhook/new-lead', methods=['POST'])
def handle_new_lead():
    lead = request.json

    enriched = client.enrich(
        domain=lead['company_domain'],
        fields=["employee_count", "industry", "funding_total"]
    )

    # Route based on enriched data
    headcount = enriched.data.employee_count

    if headcount in range(201, 1_000_000):
        assign_to = "enterprise_team"
    elif headcount in range(51, 201):
        assign_to = "mid_market_team"
    else:
        assign_to = "smb_team"

    update_crm(lead['id'], enriched.data, assign_to)
    return {"status": "enriched", "assigned_to": assign_to}

# Flask webhook handler
@app.route('/webhook/new-lead', methods=['POST'])
def handle_new_lead():
    lead = request.json

    enriched = client.enrich(
        domain=lead['company_domain'],
        fields=["employee_count", "industry", "funding_total"]
    )

    # Route based on enriched data
    headcount = enriched.data.employee_count

    if headcount in range(201, 1_000_000):
        assign_to = "enterprise_team"
    elif headcount in range(51, 201):
        assign_to = "mid_market_team"
    else:
        assign_to = "smb_team"

    update_crm(lead['id'], enriched.data, assign_to)
    return {"status": "enriched", "assigned_to": assign_to}

Pattern 2: Scheduled batch enrichment

Run nightly or weekly to keep your database fresh. Pull stale records, re-enrich, and update.

# Cron job: runs weekly
def weekly_enrichment_refresh():
    stale_records = get_records_older_than(days=30)

    if not stale_records:
        return

    domains = [r['domain'] for r in stale_records]
    job = client.enrich_bulk(
        domains=domains,
        fields=["employee_count", "industry", "tech_stack"]
    )

    job.wait()  # Block until complete

    updated = 0
    for result in job.results:
        if result.status == "success":
            update_crm_record(result)
            updated += 1

    log(f"Refreshed {updated}/{len(stale_records)} records")

# Cron job: runs weekly
def weekly_enrichment_refresh():
    stale_records = get_records_older_than(days=30)

    if not stale_records:
        return

    domains = [r['domain'] for r in stale_records]
    job = client.enrich_bulk(
        domains=domains,
        fields=["employee_count", "industry", "tech_stack"]
    )

    job.wait()  # Block until complete

    updated = 0
    for result in job.results:
        if result.status == "success":
            update_crm_record(result)
            updated += 1

    log(f"Refreshed {updated}/{len(stale_records)} records")

# Cron job: runs weekly
def weekly_enrichment_refresh():
    stale_records = get_records_older_than(days=30)

    if not stale_records:
        return

    domains = [r['domain'] for r in stale_records]
    job = client.enrich_bulk(
        domains=domains,
        fields=["employee_count", "industry", "tech_stack"]
    )

    job.wait()  # Block until complete

    updated = 0
    for result in job.results:
        if result.status == "success":
            update_crm_record(result)
            updated += 1

    log(f"Refreshed {updated}/{len(stale_records)} records")

# Cron job: runs weekly
def weekly_enrichment_refresh():
    stale_records = get_records_older_than(days=30)

    if not stale_records:
        return

    domains = [r['domain'] for r in stale_records]
    job = client.enrich_bulk(
        domains=domains,
        fields=["employee_count", "industry", "tech_stack"]
    )

    job.wait()  # Block until complete

    updated = 0
    for result in job.results:
        if result.status == "success":
            update_crm_record(result)
            updated += 1

    log(f"Refreshed {updated}/{len(stale_records)} records")

Pattern 3: Pre-campaign list building

Build and enrich an outbound list from scratch using company search and contact discovery.

def build_outbound_list(criteria):
    # Step 1: Find companies matching criteria
    companies = client.search_companies(
        industry=criteria['industry'],
        employee_min=criteria['min_employees'],
        employee_max=criteria['max_employees'],
        location=criteria['location']
    )

    # Step 2: Find contacts at each company
    contacts = []
    for company in companies:
        people = client.find_contacts(
            domain=company.domain,
            titles=criteria['target_titles'],
            limit=3
        )
        contacts.extend(people)

    # Step 3: Waterfall email verification
    verified_contacts = []
    for contact in contacts:
        email = client.waterfall(
            type="email_finder",
            inputs={
                "first_name": contact.first_name,
                "last_name": contact.last_name,
                "domain": contact.domain
            },
            verify=True
        )
        if email.verified:
            contact.email = email.email
            verified_contacts.append(contact)

    return verified_contacts

def build_outbound_list(criteria):
    # Step 1: Find companies matching criteria
    companies = client.search_companies(
        industry=criteria['industry'],
        employee_min=criteria['min_employees'],
        employee_max=criteria['max_employees'],
        location=criteria['location']
    )

    # Step 2: Find contacts at each company
    contacts = []
    for company in companies:
        people = client.find_contacts(
            domain=company.domain,
            titles=criteria['target_titles'],
            limit=3
        )
        contacts.extend(people)

    # Step 3: Waterfall email verification
    verified_contacts = []
    for contact in contacts:
        email = client.waterfall(
            type="email_finder",
            inputs={
                "first_name": contact.first_name,
                "last_name": contact.last_name,
                "domain": contact.domain
            },
            verify=True
        )
        if email.verified:
            contact.email = email.email
            verified_contacts.append(contact)

    return verified_contacts

def build_outbound_list(criteria):
    # Step 1: Find companies matching criteria
    companies = client.search_companies(
        industry=criteria['industry'],
        employee_min=criteria['min_employees'],
        employee_max=criteria['max_employees'],
        location=criteria['location']
    )

    # Step 2: Find contacts at each company
    contacts = []
    for company in companies:
        people = client.find_contacts(
            domain=company.domain,
            titles=criteria['target_titles'],
            limit=3
        )
        contacts.extend(people)

    # Step 3: Waterfall email verification
    verified_contacts = []
    for contact in contacts:
        email = client.waterfall(
            type="email_finder",
            inputs={
                "first_name": contact.first_name,
                "last_name": contact.last_name,
                "domain": contact.domain
            },
            verify=True
        )
        if email.verified:
            contact.email = email.email
            verified_contacts.append(contact)

    return verified_contacts

def build_outbound_list(criteria):
    # Step 1: Find companies matching criteria
    companies = client.search_companies(
        industry=criteria['industry'],
        employee_min=criteria['min_employees'],
        employee_max=criteria['max_employees'],
        location=criteria['location']
    )

    # Step 2: Find contacts at each company
    contacts = []
    for company in companies:
        people = client.find_contacts(
            domain=company.domain,
            titles=criteria['target_titles'],
            limit=3
        )
        contacts.extend(people)

    # Step 3: Waterfall email verification
    verified_contacts = []
    for contact in contacts:
        email = client.waterfall(
            type="email_finder",
            inputs={
                "first_name": contact.first_name,
                "last_name": contact.last_name,
                "domain": contact.domain
            },
            verify=True
        )
        if email.verified:
            contact.email = email.email
            verified_contacts.append(contact)

    return verified_contacts

Error handling and rate limits

Production integrations need proper error handling. Here's what to expect and how to handle it.

Rate limits

When you hit a rate limit, the API returns a 429 status code with a Retry-After header. The SDK handles this automatically with exponential backoff.

Error response format

{
  "status": "error",
  "error": {
    "code": "rate_limit_exceeded",
    "message": "Rate limit exceeded. Retry after 30 seconds.",
    "retry_after": 30
  }
}

{
  "status": "error",
  "error": {
    "code": "rate_limit_exceeded",
    "message": "Rate limit exceeded. Retry after 30 seconds.",
    "retry_after": 30
  }
}

{
  "status": "error",
  "error": {
    "code": "rate_limit_exceeded",
    "message": "Rate limit exceeded. Retry after 30 seconds.",
    "retry_after": 30
  }
}

{
  "status": "error",
  "error": {
    "code": "rate_limit_exceeded",
    "message": "Rate limit exceeded. Retry after 30 seconds.",
    "retry_after": 30
  }
}

Common error codes and handling

• 400 Bad Request: Invalid input. Check your field names and input format.

• 401 Unauthorized: Invalid or expired API key. Regenerate from dashboard.

• 402 Payment Required: Insufficient credits. Top up your account.

• 404 Not Found: No data found for the input. This is expected for some records. Handle gracefully.

• 429 Too Many Requests: Rate limited. Back off and retry.

• 500 Internal Server Error: Provider-side issue. Retry with exponential backoff.

Production-ready error handling pattern

import time
from databar import Databar, DatabarError, RateLimitError, InsufficientCreditsError

client = Databar(api_key="YOUR_API_KEY")

def safe_enrich(domain, fields, max_retries=3):
    for attempt in range(max_retries):
        try:
            return client.enrich(domain=domain, fields=fields)
        except RateLimitError as e:
            wait = e.retry_after or (2 ** attempt)
            time.sleep(wait)
        except InsufficientCreditsError:
            alert_ops_team("Databar credits exhausted")
            raise
        except DatabarError as e:
            if attempt == max_retries - 1:
                log_error(f"Enrichment failed for {domain}: {e}")
                return None
            time.sleep(2 ** attempt)
    return None

import time
from databar import Databar, DatabarError, RateLimitError, InsufficientCreditsError

client = Databar(api_key="YOUR_API_KEY")

def safe_enrich(domain, fields, max_retries=3):
    for attempt in range(max_retries):
        try:
            return client.enrich(domain=domain, fields=fields)
        except RateLimitError as e:
            wait = e.retry_after or (2 ** attempt)
            time.sleep(wait)
        except InsufficientCreditsError:
            alert_ops_team("Databar credits exhausted")
            raise
        except DatabarError as e:
            if attempt == max_retries - 1:
                log_error(f"Enrichment failed for {domain}: {e}")
                return None
            time.sleep(2 ** attempt)
    return None

import time
from databar import Databar, DatabarError, RateLimitError, InsufficientCreditsError

client = Databar(api_key="YOUR_API_KEY")

def safe_enrich(domain, fields, max_retries=3):
    for attempt in range(max_retries):
        try:
            return client.enrich(domain=domain, fields=fields)
        except RateLimitError as e:
            wait = e.retry_after or (2 ** attempt)
            time.sleep(wait)
        except InsufficientCreditsError:
            alert_ops_team("Databar credits exhausted")
            raise
        except DatabarError as e:
            if attempt == max_retries - 1:
                log_error(f"Enrichment failed for {domain}: {e}")
                return None
            time.sleep(2 ** attempt)
    return None

import time
from databar import Databar, DatabarError, RateLimitError, InsufficientCreditsError

client = Databar(api_key="YOUR_API_KEY")

def safe_enrich(domain, fields, max_retries=3):
    for attempt in range(max_retries):
        try:
            return client.enrich(domain=domain, fields=fields)
        except RateLimitError as e:
            wait = e.retry_after or (2 ** attempt)
            time.sleep(wait)
        except InsufficientCreditsError:
            alert_ops_team("Databar credits exhausted")
            raise
        except DatabarError as e:
            if attempt == max_retries - 1:
                log_error(f"Enrichment failed for {domain}: {e}")
                return None
            time.sleep(2 ** attempt)
    return None

Cost optimization tips

Databar charges per enrichment call on a credit basis. Here's how to keep costs efficient:

• Cache results. Store enrichment data locally and check cache before making API calls. Company data doesn't change daily.

• Request only needed fields. Don't pull 20 fields when you need 3. Each field may consume additional credits.

• Use batch for volume. Batch enrichment is more cost-efficient than individual calls for lists over 50 records.

• Set waterfall depth. If the first 2 providers cover 80% of your records, you might not need to cascade through all 5. Configure your waterfall depth to balance coverage vs. cost.

• Enrich progressively. Start with cheap data (company firmographics) to qualify. Only enrich expensive data (phone numbers, org charts) for accounts that pass your ICP filter.

FAQ

Can I use the API with languages other than Python?

Yes. The REST API works with any language that can make HTTP requests. Python and Node SDKs are available as convenience wrappers. For Go, Ruby, Java, or other languages, call the REST endpoints directly.

How does waterfall pricing work?

You pay only for the provider that returns data. If Provider A misses but Provider B hits, you're charged for Provider B only. Verification is an additional small credit cost. This makes waterfall cost-efficient because you're not paying for failed lookups across every provider.

What's the latency for single enrichment calls?

Single enrichment typically returns in 1-3 seconds. Waterfall enrichment takes 3-8 seconds depending on how many providers are checked. Batch enrichment runs asynchronously and completes in minutes for lists under 1,000 records.

Can the MCP server access all the same data as the REST API?

Yes. The MCP server is a wrapper around the same API. Any enrichment, waterfall, or bulk operation available via REST is also available through MCP tools. The difference is the interaction model. REST is request-response. MCP lets an AI agent decide when and how to call each tool.

How do I handle enrichment for records where no data is found?

The API returns a 200 status with null fields for records with no data. Your code should handle null values gracefully. For waterfall enrichment, a "not found" means all providers in the cascade returned empty. Log these for analysis. If your miss rate is above 20%, your ICP might not align with provider coverage.

Is there a sandbox or test environment?

The free tier serves as your test environment. You get limited credits to test all endpoints without charge. Use real domains and contacts in your tests to get accurate response formats and fill rates.

How do I monitor API usage and costs?

The Databar dashboard shows real-time credit usage, requests by endpoint, and provider-level breakdowns. You can also query your balance programmatically through the API to build usage alerts into your pipeline.

Start building with the Databar API

This gtm engineer databar api mcp guide gives you everything you need to integrate enrichment into your GTM stack. REST API for HTTP calls, Python SDK for application code, MCP server for AI-native workflows. Databar offers 100+ data providers, waterfall enrichment, and outcome-based billing with no annual contracts. Sign up, get your API key, and run your first enrichment call in under 5 minutes.

Get your Databar API key

Also interesting

• Everything you need to know about CRM enrichment

• Best CRM enrichment tools that move the revenue needle

• CRM data cleaning: automated solutions that actually work