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("elon@tesla.com")

Look up a person by email

person = client.lookup("elon@tesla.com")

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("satya@microsoft.com", 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("satya@microsoft.com")

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("sundar@google.com")

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("+14155552671")

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("user@example.com")

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("user@example.com", 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(["elon@tesla.com", "satya@microsoft.com"]):
    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=["satya@microsoft.com", "jensen@nvidia.com"],
    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=["satya@microsoft.com", "sundar@google.com"],
)
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, ["tim@apple.com"])

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

# Remove targets
client.delete_contact_list_emails(contact_list.id, ["sundar@google.com"])

# 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("satya@microsoft.com")
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("elon@tesla.com", 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("elon@tesla.com")
        print(person.name)

        # Run many lookups concurrently (bounded by max_concurrency):
        people = await client.lookup_many([
            "satya@microsoft.com",
            "sundar@google.com",
            "tim@apple.com",
        ])
        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: support@encrata.com

License

MIT