Enriching CRM contacts with search API data adds recent news, tech stack signals, and funding context that static databases miss. Sales teams close faster when they know a prospect just raised a round, adopted a new tool, or appeared in industry news -- but manual research per contact does not scale. The Scavio search API lets you programmatically query for company-specific information and extract structured signals from the results. This tutorial builds a Python enrichment pipeline that takes a CSV of contacts, runs targeted searches for each company, parses the results for key signals, and writes the enriched data back to a CSV ready for CRM import.
Prerequisites
- Python 3.8 or higher installed
- requests and csv (standard library) available
- A Scavio API key from scavio.dev
- A CSV file with contact data including company names
Walkthrough
Step 1: Load contacts from CSV
Read the input CSV containing at minimum a company_name column. The script will enrich each row with additional columns derived from search results.
Python
import csv
from pathlib import Path
def load_contacts(filepath: str) -> list[dict]:
with open(filepath, 'r', encoding='utf-8') as f:
reader = csv.DictReader(f)
return list(reader)
contacts = load_contacts('contacts.csv')
print(f'Loaded {len(contacts)} contacts')Step 2: Run targeted search queries per company
For each company, run multiple targeted queries to find recent news, funding info, and tech stack signals. Each query is tailored to surface a specific enrichment signal.
Python
import requests
import os
API_KEY = os.environ.get('SCAVIO_API_KEY', 'your_scavio_api_key')
ENDPOINT = 'https://api.scavio.dev/api/v1/search'
def search_company(company: str, query_suffix: str) -> dict:
response = requests.post(
ENDPOINT,
headers={'x-api-key': API_KEY},
json={'query': f'{company} {query_suffix}', 'country_code': 'us'}
)
response.raise_for_status()
return response.json()
SIGNAL_QUERIES = {
'recent_news': 'news 2026',
'funding': 'funding round raised',
'tech_stack': 'tech stack tools uses',
'hiring': 'hiring jobs open positions'
}Step 3: Parse search results into structured signals
Extract the top organic result snippets for each signal query and condense them into a single-line enrichment field. This avoids raw HTML and gives clean text for CRM notes.
Python
def extract_signal(data: dict, max_snippets: int = 3) -> str:
results = data.get('organic_results', [])[:max_snippets]
snippets = []
for r in results:
snippet = r.get('snippet', '')
if snippet:
snippets.append(snippet.strip())
return ' | '.join(snippets) if snippets else 'No data found'
def enrich_contact(contact: dict) -> dict:
company = contact.get('company_name', '')
if not company:
return contact
for signal_name, query_suffix in SIGNAL_QUERIES.items():
data = search_company(company, query_suffix)
contact[signal_name] = extract_signal(data)
return contactStep 4: Run enrichment and write results to CSV
Iterate through all contacts, enrich each one, and write the results to a new CSV file with the additional signal columns. Add rate limiting to stay within API credit budget.
Python
import time
def run_enrichment(input_file: str, output_file: str):
contacts = load_contacts(input_file)
enriched = []
for i, contact in enumerate(contacts):
print(f'Enriching {i + 1}/{len(contacts)}: {contact.get("company_name", "unknown")}')
enriched.append(enrich_contact(contact))
time.sleep(1) # rate limit: 4 queries per contact
fieldnames = list(enriched[0].keys())
with open(output_file, 'w', newline='', encoding='utf-8') as f:
writer = csv.DictWriter(f, fieldnames=fieldnames)
writer.writeheader()
writer.writerows(enriched)
print(f'Wrote {len(enriched)} enriched contacts to {output_file}')
run_enrichment('contacts.csv', 'contacts_enriched.csv')Python Example
Python
import os
import csv
import time
import requests
API_KEY = os.environ.get('SCAVIO_API_KEY', 'your_scavio_api_key')
ENDPOINT = 'https://api.scavio.dev/api/v1/search'
SIGNAL_QUERIES = {
'recent_news': 'news 2026',
'funding': 'funding round raised',
'tech_stack': 'tech stack tools uses',
'hiring': 'hiring jobs open positions'
}
def search_company(company: str, query_suffix: str) -> dict:
response = requests.post(
ENDPOINT,
headers={'x-api-key': API_KEY},
json={'query': f'{company} {query_suffix}', 'country_code': 'us'}
)
response.raise_for_status()
return response.json()
def extract_signal(data: dict, max_snippets: int = 3) -> str:
results = data.get('organic_results', [])[:max_snippets]
snippets = [r.get('snippet', '').strip() for r in results if r.get('snippet')]
return ' | '.join(snippets) if snippets else 'No data found'
def load_contacts(filepath: str) -> list[dict]:
with open(filepath, 'r', encoding='utf-8') as f:
return list(csv.DictReader(f))
def enrich_contact(contact: dict) -> dict:
company = contact.get('company_name', '')
if not company:
return contact
for signal_name, query_suffix in SIGNAL_QUERIES.items():
data = search_company(company, query_suffix)
contact[signal_name] = extract_signal(data)
return contact
def run_enrichment(input_file: str, output_file: str):
contacts = load_contacts(input_file)
enriched = []
for i, contact in enumerate(contacts):
print(f'Enriching {i + 1}/{len(contacts)}: {contact.get("company_name", "unknown")}')
enriched.append(enrich_contact(contact))
time.sleep(1)
fieldnames = list(enriched[0].keys())
with open(output_file, 'w', newline='', encoding='utf-8') as f:
writer = csv.DictWriter(f, fieldnames=fieldnames)
writer.writeheader()
writer.writerows(enriched)
print(f'Wrote {len(enriched)} enriched contacts to {output_file}')
if __name__ == '__main__':
run_enrichment('contacts.csv', 'contacts_enriched.csv')JavaScript Example
JavaScript
const API_KEY = process.env.SCAVIO_API_KEY || 'your_scavio_api_key';
const ENDPOINT = 'https://api.scavio.dev/api/v1/search';
const fs = require('fs');
const SIGNAL_QUERIES = {
recent_news: 'news 2026',
funding: 'funding round raised',
tech_stack: 'tech stack tools uses',
hiring: 'hiring jobs open positions'
};
async function searchCompany(company, querySuffix) {
const response = await fetch(ENDPOINT, {
method: 'POST',
headers: { 'x-api-key': API_KEY, 'Content-Type': 'application/json' },
body: JSON.stringify({ query: company + ' ' + querySuffix, country_code: 'us' })
});
if (!response.ok) throw new Error('HTTP ' + response.status);
return response.json();
}
function extractSignal(data) {
const results = (data.organic_results || []).slice(0, 3);
const snippets = results.map(r => (r.snippet || '').trim()).filter(Boolean);
return snippets.length ? snippets.join(' | ') : 'No data found';
}
function parseCSV(text) {
const lines = text.trim().split('\n');
const headers = lines[0].split(',').map(h => h.trim());
return lines.slice(1).map(line => {
const values = line.split(',');
return Object.fromEntries(headers.map((h, i) => [h, (values[i] || '').trim()]));
});
}
async function main() {
const contacts = parseCSV(fs.readFileSync('contacts.csv', 'utf-8'));
const enriched = [];
for (let i = 0; i < contacts.length; i++) {
const contact = contacts[i];
const company = contact.company_name || '';
console.log('Enriching ' + (i + 1) + '/' + contacts.length + ': ' + company);
if (company) {
for (const [signal, suffix] of Object.entries(SIGNAL_QUERIES)) {
const data = await searchCompany(company, suffix);
contact[signal] = extractSignal(data);
}
}
enriched.push(contact);
await new Promise(r => setTimeout(r, 1000));
}
const headers = Object.keys(enriched[0]);
const csv = [headers.join(','), ...enriched.map(r => headers.map(h => '"' + (r[h] || '') + '"').join(','))].join('\n');
fs.writeFileSync('contacts_enriched.csv', csv);
console.log('Wrote ' + enriched.length + ' enriched contacts');
}
main().catch(console.error);Expected Output
JSON
Enriching 1/3: Acme Corp
Enriching 2/3: TechStart Inc
Enriching 3/3: DataFlow Labs
Wrote 3 enriched contacts to contacts_enriched.csv
-- contacts_enriched.csv sample row --
company_name,email,recent_news,funding,tech_stack,hiring
Acme Corp,john@acme.com,"Acme Corp announces Q1 revenue growth...","Acme Corp raises $45M Series B...","Acme Corp uses React, PostgreSQL, and AWS...","Acme Corp hiring 15 engineers..."