encrata 0.5.1

Python bindings for the Encrata email intelligence API

Encrata Python Library

The Encrata Python library provides convenient access to the Encrata API from applications written in Python. Look up any person by their email address — get their name, company, job title, social profiles, breach history, and more.

API Documentation

See the Encrata API docs.

Installation

pip install encrata

Requirements

• Python 3.9+
• httpx (installed automatically)

Usage

The library needs to be configured with your account's API key, available in your Encrata Dashboard.

from encrata import Encrata

client = Encrata("enc_live_...")

The client keeps a pooled HTTP connection open. Reuse a single instance for the lifetime of your application, or use it as a context manager to close the pool automatically:

with Encrata("enc_live_...") as client:
    person = client.lookup("[e-mail]")

Look up a person by email

person = client.lookup("[e-mail]")

print(person.name)        # "Elon Musk"
print(person.company)     # "Tesla"
print(person.role)        # "CEO"
print(person.location)    # "Austin, Texas"
print(person.validity)    # "valid"

Each lookup costs 1 credit. Cached results within 24 hours are served from cache.

Select specific fields

Only return the fields you need to minimize response size:

person = client.lookup("[e-mail]", fields=["name", "company", "role", "socials"])

print(person.name)             # "Satya Nadella"
print(person.socials.linkedin) # "https://linkedin.com/in/satyanadella"

Available fields: name, email, company, role, industry, location, bio, age, gender, education, phone, photo, validity, socials, breaches, registered_services, news, publications.

Validate an email address (free)

Check if an email is deliverable without using any credits:

result = client.validate("[e-mail]")

print(result.validity)  # "valid", "invalid", "disposable", or "unknown"
print(result.message)   # "Email is deliverable and valid."

Check data breaches (free)

See if an email has been exposed in known data breaches:

report = client.breaches("[e-mail]")

print(report.count)         # 3
print(report.services)      # ["Adobe", "LinkedIn", "Dropbox"]
print(report.exposed_data)  # ["email", "password", "username"]

OSINT Lookups

Enrich IPs, phone numbers, domains, and companies. Each costs 1 credit.

IP intelligence

ip = client.ip("8.8.8.8")

print(ip.location.country)  # "United States"
print(ip.asn.name)          # "GOOGLE"
print(ip.threat.malicious)  # False

Phone lookup

phone = client.phone_lookup("[phone]")

print(phone.carrier.name)        # "T-Mobile"
print(phone.country.name)        # "United States"
print(phone.validation.is_voip)  # False

Domain search

domain = client.domain_search("tesla.com")

print(domain.whois.registrar)      # "MarkMonitor Inc."
print(domain.ssl.issuer)           # "DigiCert Inc"
print(domain.threat_intel.malicious)  # False

Company search

company = client.company_search("Tesla")

print(company.profile.name)            # "Tesla, Inc."
print(company.profile.ticker)          # "TSLA"
print(company.profile.employee_count)  # 140000
for person in company.results:
    print(person.name, person.role, person.email)

Costs 1 credit per result returned.

Google dork search

search = client.google_search("site:example.com filetype:pdf")

for r in search.results:
    print(r.title, r.url)

# Free OSINT enrichment (no extra credits)
print(search.enrichment["query_type"])  # "domain"

Costs 1 credit per search; the enrichment block is free.

Dark web search

dw = client.darkweb_search("[e-mail]")

print(dw.total)  # 42
for hit in dw.results:
    print(hit.source, hit.title, hit.emails)

# Paginate (20 results per page)
page2 = client.darkweb_search("[e-mail]", offset=20)

Costs 1 credit per 100 results.

Scrape a web page

page = client.scrape("https://example.com/pricing")

print(page.status_code)        # 200
print(page.content)            # "# Pricing\n\nSimple, transparent pricing..."
print(page.metadata["title"])  # "Pricing — Example"

# Skip JS rendering for static pages
page = client.scrape("https://example.com", render_js=False)

Costs 2 credits; failed scrapes are auto-refunded.

Extract structured data

# Selector mode — JSON keyed by your CSS/XPath map
result = client.extract(
    "https://example.com/product/123",
    mode="selectors",
    selectors={"title": "h1", "price": ".price"},
)
print(result.extracted)  # {"title": "Wireless Headphones", "price": "$129.00"}

# Markdown mode (default), wait for lazy content
result = client.extract("https://example.com", wait_for=".loaded", timeout=45000)
print(result.extracted)

Costs 3 credits; failed extractions are auto-refunded.

Capture a screenshot

shot = client.screenshot("https://example.com", full_page=True, format="png")

import base64
with open("page.png", "wb") as f:
    f.write(base64.b64decode(shot.screenshot))

# Capture a single element instead
shot = client.screenshot("https://example.com", selector="#hero")

Costs 5 credits; failed captures are auto-refunded.

Face search

Match a probe image against your watchlist by public image URL:

result = client.face_search("https://example.com/photo.jpg")

print(result.matched)          # True
print(result.faces_detected)   # 1
for match in result.matches:
    print(match.name, match.probability)  # "Elon Musk" 0.97

# Tune the match threshold (0.0–1.0)
result = client.face_search("https://example.com/photo.jpg", threshold=0.9)

Costs 1 credit per search.

Bulk operations

Process many queries in a single request. Bulk endpoints stream results back over Server-Sent Events, so you can begin handling hits before the batch finishes. Each query costs 1 credit.

Bulk email lookup

Enrich up to 1,000 emails — results stream in one Person at a time:

for person in client.bulk_lookup(["[e-mail]", "[e-mail]"]):
    print(person.name, person.company)

# Limit fields to shrink each result
for person in client.bulk_lookup(emails, fields=["name", "company"]):
    print(person.name)

Bulk OSINT searches

Run up to 100 queries per call for Google, company, domain, or IP intelligence. Each collects the full stream into a single BulkSearchResponse:

res = client.bulk_google_search(["site:tesla.com", "site:microsoft.com"])
print(res.credits_used)  # 2
for item in res.results:
    print(item)

client.bulk_company_search(["Tesla", "Microsoft"])
client.bulk_domain_search(["tesla.com", "microsoft.com"])
client.bulk_ip_search(["8.8.8.8", "1.1.1.1"])

Monitoring

Set up monitors to track changes in email intelligence over time. When a person changes jobs, gets a new title, or appears in a breach — you'll know.

Create a monitor

monitor = client.create_monitor(
    "Sales Leads",
    emails=["[e-mail]", "[e-mail]"],
    frequency="weekly",
)

print(monitor.id)           # "mon_abc123..."
print(monitor.email_count)  # 2

List monitors

monitors = client.list_monitors()
for m in monitors:
    print(f"{m.name}: {m.email_count} emails, {m.status}")

Trigger a run

result = client.trigger_run(monitor.id)
print(result["run_id"])   # "run_xyz789..."
print(result["status"])   # "running"

Get run results

runs, total = client.list_runs(monitor.id)
for run in runs:
    print(f"Run {run.id}: {run.changes_detected} changes")

# Get detailed results for a run
snapshots, total = client.get_run_results(monitor.id, runs[0].id, changes_only=True)
for snap in snapshots:
    print(f"{snap.email}: changes={snap.changes}")

List all runs across monitors

all_runs, total = client.list_all_runs(limit=10)
all_results, total = client.list_all_results(changes_only=True)

These paginated endpoints also have iterator forms that fetch every page for you — see Automatic pagination below.

Contact Lists

Manage reusable target lists that can be used as data sources for monitors. Lists hold one type of target: email, phone, domain, ip, company, or darkweb.

Create a contact list

# Email list (the default type)
contact_list = client.create_contact_list(
    "Engineering Team",
    emails=["[e-mail]", "[e-mail]"],
)
print(contact_list.id, contact_list.list_type)

# A non-email list — pass `type` and use `targets`
domains = client.create_contact_list(
    "Competitor Domains",
    type="domain",
    targets=["competitor1.com", "competitor2.io"],
)

Manage list emails

# List all contact lists (optionally filter by type)
lists = client.list_contact_lists()
domain_lists = client.list_contact_lists(type="domain")

# Add targets
client.add_contact_list_emails(contact_list.id, ["[e-mail]"])

# List targets in a list
emails = client.list_contact_list_emails(contact_list.id)

# Remove targets
client.delete_contact_list_emails(contact_list.id, ["[e-mail]"])

# Delete the list
client.delete_contact_list(contact_list.id)

Use a list as a monitor source

monitor = client.create_monitor("Team Monitor", list_id=contact_list.id)

Workflows

Automate multi-step OSINT pipelines with triggers and steps. Workflows are managed on encrata.com/api/workflows — keep separate from the per-lookup endpoints above.

List, get, create, update

# List workflows (paginated → (workflows, total))
workflows, total = client.list_workflows(status="active")
for wf in workflows:
    print(wf.id, wf.name, wf.status)

# Fetch one
wf = client.get_workflow(workflows[0].id)

# Create from steps, or clone a template
wf = client.create_workflow(
    "Lead enrichment",
    trigger={"type": "webhook"},
    steps=[
        {"id": "step1", "type": "email_lookup", "config": {"field": "email"}},
        {"id": "step2", "type": "company_lookup", "config": {"field": "step1.company"}},
    ],
)

# Update (creates a new version). Returns the updated workflow;
# if the API echoes nothing, you still get a Workflow with the id set.
wf = client.update_workflow(wf.id, name="Renamed", status="paused")

Runs, templates, and secrets

runs, total = client.list_workflow_runs(workflow_id=wf.id)
run = client.get_workflow_run(runs[0].id)
for step in run.steps:
    print(step.type, step.status, step.credits_used)

templates = client.list_workflow_templates(category="enrichment")

# Secrets — referenced in webhook steps as {{secrets.NAME}} (values never returned)
secrets = client.list_workflow_secrets()
client.create_workflow_secret("SLACK_WEBHOOK_URL", "https://hooks.slack.com/...")
client.delete_workflow_secret("SLACK_WEBHOOK_URL")

API keys

Manage the keys for your account. The full key is only returned once, at creation.

keys = client.list_keys()
for k in keys:
    print(k.id, k.name, k.key_preview, k.credits_used)

new_key = client.create_key("Production")
print(new_key.key)  # store this — it is never shown again

client.revoke_key(new_key.id)                  # soft disable
client.revoke_key(new_key.id, permanent=True)  # permanent delete

Automatic pagination

List endpoints return one page at a time along with a total count. To walk an entire history without managing limit/offset yourself, use the matching iterator helpers — they fetch each subsequent page on demand as you iterate:

# Every run across all monitors
for run in client.iter_all_runs():
    print(run.id, run.status)

# Every result across all monitors (only the ones that changed)
for snapshot in client.iter_all_results(changes_only=True):
    print(snapshot.email, snapshot.has_changes)

# Scoped to a single monitor or run
for run in client.iter_runs(monitor.id):
    print(run.id)

for snapshot in client.iter_run_results(monitor.id, run.id):
    print(snapshot.email)

Pass limit to control the page size used under the hood (default 100):

for run in client.iter_all_runs(limit=250):
    ...

The async client exposes the same helpers as async iterators:

async for run in client.iter_all_runs():
    print(run.id)

Handling exceptions

from encrata import Encrata, AuthenticationError, InsufficientCreditsError

client = Encrata("enc_live_...")

try:
    person = client.lookup("[e-mail]")
except AuthenticationError:
    print("Invalid API key")
except InsufficientCreditsError:
    print("No credits remaining — top up at encrata.com/settings/billing")

Exception	Cause
AuthenticationError	Invalid or missing API key
InsufficientCreditsError	Account has 0 credits remaining
InvalidRequestError	Malformed request (e.g. invalid email)
RateLimitError	Too many requests
APIConnectionError	Network connectivity issue
APIError	Unexpected server error

Configuration options

client = Encrata(
    "enc_live_...",
    base_url="https://api.encrata.com",  # default
    timeout=30,                           # request timeout in seconds
    max_retries=3,                        # retries for transient failures
)

Transient failures (HTTP 429, 500, 502, 503, 504, timeouts, and connection errors) are retried automatically using exponential backoff with full jitter. A Retry-After response header is honored and capped at 30 seconds.

Force fresh lookup

Bypass the 24-hour cache to get the latest data:

person = client.lookup("[e-mail]", nocache=True)

Async

An async client, AsyncEncrata, exposes the same methods with async/await. It shares one connection pool and can run many lookups concurrently

import asyncio
from encrata import AsyncEncrata

async def main():
    async with AsyncEncrata("enc_live_...") as client:
        person = await client.lookup("[e-mail]")
        print(person.name)

        # Run many lookups concurrently (bounded by max_concurrency):
        people = await client.lookup_many([
            "[e-mail]",
            "[e-mail]",
            "[e-mail]",
        ])
        for p in people:
            print(p.name)

asyncio.run(main())

MCP (Model Context Protocol)

Encrata also provides an MCP server for AI agent frameworks like Claude, Cursor, and Windsurf. Add this to your MCP configuration:

{
  "mcpServers": {
    "encrata": {
      "url": "https://api.encrata.com/mcp",
      "headers": {
        "Authorization": "Bearer enc_live_..."
      }
    }
  }
}

Support

• Documentation: docs.encrata.com
• Dashboard: encrata.com
• Email: [e-mail]

License

MIT