> ## Documentation Index
> Fetch the complete documentation index at: https://docs.oathnet.org/llms.txt
> Use this file to discover all available pages before exploring further.

# Python SDK

> Official Python client library for OathNet API

## Installation

```bash theme={null}
pip install oathnet
```

**Requirements:** Python 3.9+

**Source:** [oathnet/oathnet-py](https://github.com/oathnet/oathnet-py)

## Quick Start

```python theme={null}
from oathnet import OathNetClient

# Initialize client with API key
client = OathNetClient(api_key="your-api-key")

# Search breaches
result = client.search.breach("user@example.com")
print(f"Found {result.data.results_found} results")

for record in result.data.results:
    print(record.email, "password_present=", bool(record.password))
```

## Authentication

```python theme={null}
import os
from oathnet import OathNetClient

# Explicit API key (required)
client = OathNetClient(api_key="your-api-key")

# From environment variable
client = OathNetClient(api_key=os.environ["OATHNET_API_KEY"])
```

## Search Service

### Breach Search

```python theme={null}
# Basic search
result = client.search.breach("user@example.com")

# With filters
result = client.search.breach(
    "user@example.com",
    dbnames="linkedin_2012,adobe_2013"
)

# Access results
print(f"Total: {result.data.results_found}")
print(f"Shown: {result.data.results_shown}")

for record in result.data.results:
    print(f"Email: {record.email}")
    print(f"Password present: {bool(record.password)}")
    print(f"Database: {record.dbname}")
    print("---")

# Pagination with cursor
if result.data.cursor:
    next_page = client.search.breach(
        "user@example.com",
        cursor=result.data.cursor
    )
```

### Initialize Session

Search sessions group related lookups for the same query. Pass `search_type`
when you already know what the query is; otherwise OathNet will detect it.
Reuse the returned `session.id` as `search_id` on follow-up searches.

```python theme={null}
# Create a search session for quota optimization.
result = client.search.init_session("user@example.com", search_type="email")

session_id = result.data.session.id
print(f"Session ID: {session_id}")
print(f"Status: {result.data.session.status}")
print(f"Search Type: {result.data.session.search_type}")
print(f"Expires: {result.data.session.expires_at}")

if result.data.summary:
    print(f"Available services: {result.data.summary.available_services}")

# Reuse the session ID on current V2 search calls.
breach_results = client.breach.search_breach_v2(
    q="user@example.com",
    search_id=session_id,
    page_size=25,
)
print(f"Breach items: {len(breach_results.data.items)}")
```

## AI Filters

Translate natural-language prompts into reusable V2 structured filters.

```python theme={null}
ai_filter = client.ai_filter.create(
    query="US gmail users with LinkedIn records after 2020",
    index="breach",
)

print(ai_filter.filter_id)
print(ai_filter.filter)

context = client.ai_filter.get_context(ai_filter.filter_id)
print(context.index_type)
print(context.query)
```

Use `ai_filter.filter_id` with V2 search, export, bulk-search, or scanner
`query_config` flows when you want OathNet to reuse the generated filter
context. Use `client.ai_filter.create(..., filter_id=...)` to refine an
existing context.

CLI commands are also available:

```bash theme={null}
oathnet ai-filter create --index breach --query "US gmail users"
oathnet ai-filter context 0123456789abcdef01234567
```

## V2 Breach Search

Use V2 breach search when you need fielded filters, structured filters, or
autocomplete-driven workflows. GET is best for simple filters that fit in query
parameters. POST is best when the filter tree comes from AI filters, saved UI
state, or a complex manual builder.

```python theme={null}
# Simple pivots stay readable and shareable
result = client.breach.search(
    q="user@example.com",
    dbname=["linkedin_2012"],
    fields=["email", "password", "dbname"],
    page_size=25,
)

# The same method accepts structured filters
filtered = client.breach.search(
    filter={
        "and": [
            {"field": "email_domain", "operator": "eq", "value": "example.com"},
            {"field": "country", "operator": "eq", "value": "US"},
        ]
    },
    fields=["email", "username", "dbname"],
)

# Use autocomplete before building a narrow query or scanner query_config
values = client.breach.autocomplete_values(
    field="email_domain",
    q="example",
    limit=10,
    include_info=True,
)
dbnames = client.breach.autocomplete_dbnames(q="link", limit=10)
field_coverage = client.breach.autocomplete_fields(
    field="discord_id",
    limit=10,
)
```

## V2 Stealer Search

```python theme={null}
# Basic search
result = client.stealer.search("user@example.com")

# Advanced filtering
result = client.stealer.search(
    "user@example.com",
    domain=["google.com", "facebook.com"],
    has_log_id=True,
    page_size=50
)

# Access items
for item in result.data.items:
    print(f"URL: {item.url}")
    print(f"Username: {item.username}")
    print(f"Password present: {bool(item.password)}")
    print(f"Log ID: {item.log_id}")
    print("---")

# Pagination
if result.data.next_cursor:
    next_page = client.stealer.search(
        "user@example.com",
        cursor=result.data.next_cursor
    )

# Subdomain extraction with optional live checks and session reuse
result = client.stealer.subdomain(
    domain="example.com",
    q="mail",
    alive=True,
    search_id="sess_0123456789abcdef",
)
print(f"Found {result.data.count} subdomains")
for sub in result.data.subdomains:
    print(f"  {sub}")
```

## Investigation And Phonebook

Use investigation when one query should fan out across credentials, victims,
files, properties, and related credentials. Use phonebook when you want domain
host and email intelligence before deciding which pivots to run next.

```python theme={null}
# Structured filters use the same search method
filtered = client.stealer.search(
    filter={"field": "domain", "operator": "eq", "value": "example.com"},
    fields=["log_id", "url_str", "username"],
    view="enriched",
)

investigation = client.stealer.investigate(
    q="example.com",
    scope="all",
    include=["credentials", "victims", "files"],
    page_size=25,
    view="enriched",
)

investigation_from_body = client.stealer.investigate(
    q="example.com",
    scope="all",
    include=["credentials"],
    filters={"credentials": {"domain": "example.com"}},
)

for item in investigation.data.sections.credentials.items:
    print(item.log_id, item.username)

phonebook = client.stealer.phonebook(
    domain="example.com",
    alive=True,
)
print(len(phonebook.data.subdomains), phonebook.data.email_count)
```

## Victims

Victim search returns stealer-log profiles. Use the `log_id` from a selected
victim to inspect its file tree, fetch one raw file, or download the whole
archive. Passing the `search_id` from `init_session` keeps these follow-up
requests tied to the same search workflow.

```python theme={null}
# Search victims
result = client.victims.search("user@example.com")

for victim in result.data.items:
    print(f"Log ID: {victim.log_id}")
    print(f"Users: {victim.device_users}")
    print(f"IPs: {victim.device_ips}")
    print(f"Documents: {victim.total_docs}")
    print("---")

# Use the same method when the pivot comes from an AI/manual structured filter
victims = client.victims.search(
    filter={"field": "service", "operator": "eq", "value": "discord"},
    view="enriched",
)

# Search victim properties globally, then inspect one selected log
properties = client.victims.search_properties(
    q="example.com",
    service="discord",
    confidence="high",
    exclude_cookie_evidence=True,
)
properties_from_body = client.victims.search_properties(
    q="example.com",
    service="discord",
    confidence=["high", "medium"],
)
print(len(properties.data.items), len(properties_from_body.data.items))

details = client.victims.get_properties(
    "log_id_here",
    service="discord",
)
print(len(details.data.items))

summary = client.victims.get_summary("log_id_here")
cookies = client.victims.get_cookies(
    "log_id_here",
    domain="example.com",
    status="active",
)

# Raw cookie-domain inspection returns copyable text
cookie_text = client.victims.inspect_cookie_domain(
    "log_id_here",
    domain="example.com",
)
print(f"Cookie export characters: {len(cookie_text.copy_text or '')}")

# Get victim manifest (file tree) for the selected log
search_id = "sess_0123456789abcdef"
log_id = "log_id_here"
manifest = client.victims.get_manifest(log_id, search_id=search_id)
print(f"Root: {manifest.data.victim_tree.name}")

for child in (manifest.data.victim_tree.children or [])[:10]:
    print(f"  {child.name} ({child.type})")

# Get one file from the manifest. The SDK returns decoded text.
file_content = client.victims.get_file(
    log_id,
    "Browser Passwords/file 1.txt",
    search_id=search_id,
)
print(f"Downloaded file characters: {len(file_content.data.content or '')}")

# Download the full victim archive as a ZIP
client.victims.download_archive(
    log_id,
    "./victim_archive.zip",
    search_id=search_id,
)
```

## File Search

File search runs an asynchronous scan across selected stealer-log files. Use
metadata search first when you only need file names and IDs. Create a file-search
job when you need to scan file contents with a literal, regex, or wildcard
expression.

```python theme={null}
# Search victim file metadata before fetching any file bytes
files = client.file_search.search_metadata(
    q="cookies",
    kind="cookies",
    page_size=25,
)
for file in files.data.items:
    print(file.log_id, file.file_id, file.path)

# Create a file search job
result = client.file_search.create(
    expression="password",
    log_ids=["log_id_1", "log_id_2"],
    search_mode="literal"  # literal, regex, or wildcard
)
job_id = result.data.job_id
print(f"Job ID: {job_id}")

# Check job status
status = client.file_search.get_status(job_id)
print(f"Status: {status.data.status}")
if status.data.summary:
    print(f"Files scanned: {status.data.summary.files_scanned}")
    print(f"Matches: {status.data.summary.matches}")

# Wait for completion and get results
result = client.file_search.search(
    expression="password",
    log_ids=["log_id_1", "log_id_2"],
    search_mode="literal",
    timeout=300  # seconds
)

if result.data.matches:
    for match in result.data.matches:
        print(f"File: {match.file_name}")
        print(f"Log ID: {match.log_id}")
        print(f"Match: {match.match_text}")
        print("---")
```

## Exports

Exports are asynchronous jobs for large result sets. Create a job with the
service and structured `query_config` you want, poll the job status, then
download the completed file. Use `docs` for credential-style stealer exports,
`victims` for victim/profile exports, and `breach` for breach-record exports.

```python theme={null}
# List recent export jobs before creating a duplicate
exports = client.exports.list_exports_v2(page=1, page_size=20)
for job in exports.results:
    print(job.job_id, job.status)

# Create an export job from a structured breach query
result = client.exports.create_export_v2(
    export_type="breach",
    service="breach",
    limit=100,
    format="jsonl",  # json, jsonl, csv, txt, or html
    fields=["email", "password", "dbname"],
    query_config={
        "filter": {
            "field": "email_domain",
            "operator": "eq",
            "value": "example.com",
        }
    },
)
job_id = result.data.job_id
print(f"Export Job ID: {job_id}")

# Check export status
status = client.exports.get_export_v2(job_id)
print(f"Status: {status.data.status}")
if status.data.progress:
    print(f"Progress: {status.data.progress.percent}%")

# Wait for completion and download
result = client.exports.wait_for_completion(job_id, timeout=600)
if result.data.status == "completed":
    # Download the export file
    client.exports.download_export_v2(job_id, "./export.jsonl")
```

## Bulk Search

Bulk search is for many input terms or a saved structured filter that should run
as one background job. Create the job, poll it, and download the result file
when it completes.

```python theme={null}
job = client.bulk_search.create(
    service="breach",
    terms=["alice@example.com", "bob@example.com"],
    format="jsonl",
    query_config={
        "filter": {
            "field": "email_domain",
            "operator": "eq",
            "value": "example.com",
        }
    },
    fields=["email", "password", "dbname"],
)

job_id = job.data.job_id
status = client.bulk_search.get_status(job_id)

jobs = client.bulk_search.list(page=1, page_size=20)
for item in jobs.results:
    print(item.job_id, item.status)

completed = client.bulk_search.wait_for_completion(job_id, timeout=600)
if completed.data.status == "completed":
    client.bulk_search.download(job_id, "./bulk-results.jsonl")
```

## Scanners

Scanners monitor newly indexed breach or stealer data and send notifications.
Use regular search endpoints for historical investigation; scanners only watch
new data after their baseline.

```python theme={null}
# Check scanner quota before creating a new monitor
quota = client.scanners.get_quota()
print(f"Remaining scanners: {quota.remaining}")

query_config = {
    "filter": {
        "and": [
            {"field": "email_domain", "operator": "eq", "value": "example.com"},
            {"field": "dbname", "operator": "in", "value": ["twitter.com", "linkedin.com"]},
        ]
    }
}

# Validate webhook or Discord delivery before saving the scanner.
# A 200 response can still have success=False when the remote endpoint rejects
# the test delivery, so inspect the returned object.
test = client.scanners.test_delivery(
    scanner_type="breach",
    query_config=query_config,
    notification_type="webhook",
    webhook_url="https://alerts.example.com/oathnet",
    webhook_security_mode="signed_json",
)
if not test.success:
    print(f"Delivery test failed: {test.message}")

scanner = client.scanners.create(
    name="Example breach monitor",
    scanner_type="breach",
    query_config=query_config,
    notification_type="webhook",
    webhook_url="https://alerts.example.com/oathnet",
    webhook_security_mode="signed_json",
    notify_on_zero_results=False,
)

security = client.scanners.get_webhook_security(scanner.uid)
if security.data:
    print(security.data.verification_method)

client.scanners.update(scanner.uid, name="Example breach monitor v2")
client.scanners.trigger(scanner.uid)

runs = client.scanners.list_runs(scanner.uid, limit=10)
for run in runs:
    print(f"{run.uid}: {run.status} ({run.results_count} results)")

client.scanners.pause(scanner.uid)
client.scanners.resume(scanner.uid)
```

Use `client.scanners.delete(scanner_uid)` when you want to permanently remove a
scanner and its run history.

Webhook receivers should verify `signed_json` or `signed_encrypted` deliveries
against the exact raw request body before parsing JSON. See [Scanners](/guides/scanners#verifying-webhook-signatures)
for the HMAC and encryption contract.

## OSINT Lookups

```python theme={null}
# IP Info
result = client.osint.ip_info("8.8.8.8")
print(f"Location: {result.data.city}, {result.data.country}")
print(f"ISP: {result.data.isp}")
print(f"Proxy: {result.data.proxy}")

# Steam
result = client.osint.steam("76561198012345678")
print(f"Username: {result.data.username}")
print(f"Avatar: {result.data.avatar}")

# Xbox
result = client.osint.xbox("GamerTag123")
print(f"Username: {result.data.username}")
print(f"Avatar: {result.data.avatar}")

# Discord User Info
result = client.osint.discord_userinfo("123456789012345678")
print(f"Username: {result.data.username}")
print(f"Global Name: {result.data.global_name}")
print(f"Created: {result.data.creation_date}")

# Discord Username History
result = client.osint.discord_username_history("123456789012345678")
for entry in result.data.history:
    if entry.name and entry.time:
        print(f"{entry.name[0]} at {entry.time[0]}")

# Roblox User Info
result = client.osint.roblox_userinfo(user_id="123456789")
# Or by username:
result = client.osint.roblox_userinfo(username="PlayerName")
print(f"Username: {result.data.username}")
print(f"Display Name: {result.data.display_name}")

# Holehe - Email account detection
result = client.osint.holehe("user@example.com")
print(f"Found on {len(result.data.domains)} services:")
for domain in result.data.domains:
    print(f"  {domain}")

# GHunt - Google account lookup
result = client.osint.ghunt("user@gmail.com")
profile = (result.data.data or {}).get("profile", {}) if result.data else {}
if profile:
    print(f"Name: {profile.get('Name')}")

# Subdomain extraction
result = client.osint.extract_subdomain("example.com", is_alive=True)
for sub in result.data.subdomains:
    if isinstance(sub, str):
        print(sub)
    else:
        print(sub.get("subdomain", sub))

# Minecraft username history
result = client.osint.minecraft_history("PlayerName")
for entry in result.data.history:
    print(f"{entry.username} - {entry.changed_at}")
```

## Utility Service

```python theme={null}
# Database name autocomplete
result = client.utility.dbname_autocomplete("link")
for name in result:
    print(name)  # linkedin_2012, linkedin_2021, etc.
```

## Error Handling

```python theme={null}
from oathnet.exceptions import (
    OathNetError,
    AuthenticationError,
    NotFoundError,
    ValidationError,
    RateLimitError,
    QuotaExceededError,
    ServiceUnavailableError
)

try:
    result = client.search.breach("user@example.com")

except AuthenticationError:
    print("Invalid API key")

except QuotaExceededError:
    print("Daily quota exceeded")

except RateLimitError as e:
    print(f"Rate limited. Retry after {e.retry_after} seconds")

except NotFoundError as e:
    print(f"Resource not found: {e.message}")

except ValidationError as e:
    print(f"Invalid input: {e.message}")

except ServiceUnavailableError:
    print("Server error. Try again later.")

except OathNetError as e:
    print(f"API error: {e.message}")
```

## Configuration

```python theme={null}
from oathnet import OathNetClient

client = OathNetClient(
    api_key="your-api-key",
    base_url="https://oathnet.org/api",  # Custom base URL
    timeout=30.0                          # Request timeout in seconds
)
```

## Context Manager

```python theme={null}
# Use context manager for automatic cleanup
with OathNetClient(api_key="your-api-key") as client:
    result = client.search.breach("user@example.com")
    print(result.data.results_found)
# Connection automatically closed
```

<CardGroup cols={2}>
  <Card title="JavaScript SDK" icon="js" href="/sdks/javascript">
    Next: JavaScript SDK
  </Card>

  <Card title="API Reference" icon="code" href="/api-reference/overview">
    Full API documentation
  </Card>
</CardGroup>