How to Automate Email Prospecting with Python and Easy Email Finder

From Manual to Automated Prospecting

Most sales teams spend hours each week manually searching for leads and copying emails into spreadsheets. With Python and the Easy Email Finder API, you can automate the entire process and reclaim that time for actual selling.

In this tutorial, we will build a production-ready prospecting script that runs on a schedule, finds new leads, filters out duplicates, and exports results to both CSV and JSON formats.

Setting Up the Project

mkdir email-prospector && cd email-prospector
python -m venv venv
source venv/bin/activate  # Windows: venv\Scripts\activate
pip install requests schedule python-dotenv

Create a .env file for your API key:

EEF_API_KEY=eef_live_your_api_key_here

The Prospecting Engine

import os
import json
import csv
import hashlib
import requests
import schedule
import time
from datetime import datetime
from dotenv import load_dotenv

load_dotenv()

API_KEY = os.getenv("EEF_API_KEY")
BASE = "https://easyemailfinder.com/api/v1"
HEADERS = {
    "Authorization": f"Bearer {API_KEY}",
    "Content-Type": "application/json"
}
SEEN_FILE = "seen_emails.json"

def load_seen_emails() -> set:
    """Load previously discovered emails to avoid duplicates."""
    if os.path.exists(SEEN_FILE):
        with open(SEEN_FILE, "r") as f:
            return set(json.load(f))
    return set()

def save_seen_emails(seen: set):
    with open(SEEN_FILE, "w") as f:
        json.dump(list(seen), f)

def search_and_enrich(query: str, location: str, mode: str = "local") -> list:
    """Use the combined search-and-enrich endpoint for efficiency."""
    resp = requests.post(f"{BASE}/search-and-enrich", headers=HEADERS, json={
        "query": query,
        "location": location,
        "mode": mode
    })
    resp.raise_for_status()
    return resp.json().get("results", [])

def filter_quality_leads(leads: list) -> list:
    """Filter for leads with emails and quality indicators."""
    quality = []
    for lead in leads:
        if not lead.get("email"):
            continue
        # Skip generic emails
        email = lead["email"].lower()
        if any(prefix in email for prefix in ["info@", "noreply@", "admin@", "support@"]):
            continue
        quality.append(lead)
    return quality

def export_csv(leads: list, filename: str):
    keys = ["email", "name", "website", "phone", "address"]
    with open(filename, "a", newline="") as f:
        writer = csv.DictWriter(f, fieldnames=keys, extrasaction="ignore")
        if f.tell() == 0:
            writer.writeheader()
        writer.writerows(leads)

def export_json(leads: list, filename: str):
    existing = []
    if os.path.exists(filename):
        with open(filename, "r") as f:
            existing = json.load(f)
    existing.extend(leads)
    with open(filename, "w") as f:
        json.dump(existing, f, indent=2)

def prospect(targets: list):
    """Run a prospecting cycle across all targets."""
    seen = load_seen_emails()
    new_leads = []
    timestamp = datetime.now().strftime("%Y%m%d")

    for target in targets:
        query = target["query"]
        location = target["location"]
        mode = target.get("mode", "local")
        print(f"[{datetime.now()}] Searching: {query} in {location}")

        try:
            leads = search_and_enrich(query, location, mode)
            quality = filter_quality_leads(leads)

            for lead in quality:
                email_hash = hashlib.md5(lead["email"].lower().encode()).hexdigest()
                if email_hash not in seen:
                    seen.add(email_hash)
                    new_leads.append(lead)
                    print(f"  NEW: {lead['email']}")
        except Exception as e:
            print(f"  Error: {e}")

        time.sleep(2)  # Respect rate limits

    if new_leads:
        export_csv(new_leads, f"leads_{timestamp}.csv")
        export_json(new_leads, f"leads_{timestamp}.json")
        print(f"\nExported {len(new_leads)} new leads")

    save_seen_emails(seen)
    print(f"Total unique emails tracked: {len(seen)}")

# Define your prospecting targets
TARGETS = [
    {"query": "dentists", "location": "Austin, TX"},
    {"query": "dentists", "location": "Dallas, TX"},
    {"query": "chiropractors", "location": "Austin, TX"},
    {"query": "web design agencies", "location": "New York, NY", "mode": "digital"},
]

if __name__ == "__main__":
    # Run immediately, then every 24 hours
    prospect(TARGETS)
    schedule.every(24).hours.do(prospect, TARGETS)
    while True:
        schedule.run_pending()
        time.sleep(60)

Key Design Decisions

Deduplication with hashing: We store MD5 hashes of previously seen emails rather than the emails themselves. This keeps the dedup file small and avoids storing PII on disk unnecessarily.

Quality filtering: The filter_quality_leads function removes generic email addresses like info@ and noreply@. You can customize this to match your ideal customer profile.

The search-and-enrich endpoint: Instead of calling /search and /enrich separately, we use the combined /search-and-enrich endpoint. This is more efficient and reduces the number of API calls. Learn more in our search-and-enrich deep dive.

Running as a Cron Job

For production use, you might prefer a cron job over the schedule library. Add this to your crontab to run daily at 6 AM:

# Run prospector daily at 6 AM
0 6 * * * cd /path/to/email-prospector && /path/to/venv/bin/python prospector.py --once

Cost Tracking

The /search-and-enrich endpoint costs 1 credit per result. With 4 targets averaging 15 results each, a daily run costs about 60 credits or $15. Over a month, that is $450 for a continuously updated lead pipeline. See our cost optimization guide for ways to reduce this further.

Next Steps

Consider adding Slack notifications for new leads, pushing results directly to your CRM, or building a simple dashboard to track lead volume over time. The API documentation covers all available endpoints and response formats. For CRM integration patterns, see our CRM integration guide.