Metadata-Version: 2.4
Name: encrata
Version: 0.5.1
Summary: Python bindings for the Encrata email intelligence API
Project-URL: Homepage, https://encrata.com
Project-URL: Documentation, https://docs.encrata.com
Project-URL: Repository, https://github.com/Encratahq/encrata-python
Project-URL: Changelog, https://github.com/Encratahq/encrata-python/blob/main/CHANGELOG.md
Author-email: Encrata <support@encrata.com>
License-Expression: MIT
License-File: LICENSE
Keywords: api,breach,email,encrata,enrichment,intelligence,lookup,osint,validation
Classifier: Development Status :: 4 - Beta
Classifier: Intended Audience :: Developers
Classifier: License :: OSI Approved :: MIT License
Classifier: Operating System :: OS Independent
Classifier: Programming Language :: Python
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.9
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Topic :: Software Development :: Libraries :: Python Modules
Classifier: Typing :: Typed
Requires-Python: >=3.9
Requires-Dist: httpx>=0.23
Provides-Extra: test
Requires-Dist: pytest; extra == 'test'
Description-Content-Type: text/markdown

# Encrata Python Library

[![PyPI version](https://img.shields.io/pypi/v/encrata.svg)](https://pypi.org/project/encrata/)
[![Python](https://img.shields.io/pypi/pyversions/encrata.svg)](https://pypi.org/project/encrata/)
[![License: MIT](https://img.shields.io/badge/License-MIT-blue.svg)](https://opensource.org/licenses/MIT)

The Encrata Python library provides convenient access to the [Encrata API](https://docs.encrata.com) 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](https://docs.encrata.com).

## Installation

```bash
pip install encrata
```

### Requirements

- Python 3.9+
- [`httpx`](https://www.python-httpx.org/) (installed automatically)

## Usage

The library needs to be configured with your account's API key, available in your [Encrata Dashboard](https://encrata.com/settings/api-keys).

```python
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:

```python
with Encrata("enc_live_...") as client:
    person = client.lookup("elon@tesla.com")
```

### Look up a person by email

```python
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:

```python
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:

```python
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:

```python
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

```python
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

```python
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

```python
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

```python
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

```python
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

```python
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

```python
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

```python
# 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

```python
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:

```python
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:

```python
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`:

```python
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

```python
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

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

### Trigger a run

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

### Get run results

```python
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

```python
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](#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

```python
# 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

```python
# 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

```python
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

```python
# 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

```python
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.

```python
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:

```python
# 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`):

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

The async client exposes the same helpers as async iterators:

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

## Handling exceptions

```python
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

```python
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:

```python
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

```python
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:

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

## Support

- Documentation: [docs.encrata.com](https://docs.encrata.com)
- Dashboard: [encrata.com](https://encrata.com)
- Email: support@encrata.com

## License

MIT
