Email finder: CLI command group, known_pattern, contract-checked routes, examples folder

CHANGELOG.md

@@ -1,5 +1,12 @@
 # Changelog
 
+## Unreleased
+
+- CLI: new `discolike email` command group wrapping the SDK email resource — `find FIRST LAST DOMAIN [--known-pattern X]`, `find-batch` (CSV file and/or repeatable `--contact "first,last,domain"`, max 500 per batch), `results BATCH_ID [--kind find|verify]`, and `job JOB_ID`, each with `--wait/--no-wait` polling.
+- SDK: `email.find` accepts `known_pattern` (sync + async), matching the platform's `POST /email/find` body. Omitted from the request when unset.
+- SDK: email routes are no longer `openapi=False` — the platform now exposes `/email/find`, `/email/find/batch`, and the poll routes in its OpenAPI spec, so `check_contract.py` validates them like every other route.
+- Examples: new `examples/` folder with runnable end-to-end scripts — `match_crm_contacts.py` (bulk-match a CRM CSV to personas with resumable checkpointing and website+email domain keys), `find_emails_from_csv.py` (batch email finding), `discover_and_enrich.py` (discover + DiscoGen enrichment). Referenced from the README.
+
 ## 0.1.2 (2026-08-19)
 
 - Packaging: both wheels now ship the MIT license text (`dist-info/licenses/LICENSE`) — it was absent from every release so far, since the only `LICENSE` sat at the repo root, outside either package root.

README.md

@@ -142,6 +142,15 @@ async def main() -> None:
 asyncio.run(main())
 ```
 
+## Examples
+
+The [`examples/`](examples/) folder has runnable scripts for common workflows — matching a CRM contact export to DiscoLike persona IDs (with checkpointing and resume), bulk-finding work emails from a CSV, and discovering companies by ICP then enriching them with DiscoGen. Each is stdlib-plus-SDK only:
+
+```bash
+export DISCOLIKE_API_KEY="dl_..."
+python examples/match_crm_contacts.py --help
+```
+
 ## CLI
 
 The same API from your terminal, with `--help` on every command:

examples/README.md

@@ -0,0 +1,17 @@
+# Examples
+
+Runnable, self-contained scripts showing how to use the DiscoLike Python SDK for common GTM workflows: matching a messy CRM export to DiscoLike contacts, finding verified work emails in bulk, and discovering plus AI-enriching target accounts. Each script is stdlib-plus-SDK only, has an argparse CLI, and is meant to be copied into your own pipeline and adapted.
+
+| Script | What it does |
+|---|---|
+| [`match_crm_contacts.py`](match_crm_contacts.py) | Match a CSV of CRM contacts to DiscoLike persona IDs via `contacts.bulk_match()`, with dual domain keys (website + email domain), resumable JSONL checkpointing, and a persona_id + match_score output CSV |
+| [`find_emails_from_csv.py`](find_emails_from_csv.py) | Find work emails for a CSV of people (first name, last name, domain) via `email.find_batch()` in chunks of 500; only status "found" bills |
+| [`discover_and_enrich.py`](discover_and_enrich.py) | Discover companies matching an ICP with `client.discover()`, then run a DiscoGen research prompt over them with `discogen.process()` and `job.wait()` |
+
+## Running
+
+```bash
+pip install discolike
+export DISCOLIKE_API_KEY="dl_..."   # create one at https://app.discolike.com/account/management/keys
+python examples/<script>.py --help
+```

examples/discover_and_enrich.py

@@ -0,0 +1,63 @@
+"""Discover companies matching an ICP, then enrich them with DiscoGen.
+
+Two calls end to end:
+
+1. ``client.discover(icp_text=..., country=..., max_records=...)`` finds
+   lookalike companies from DiscoLike's index of 80M+ business websites.
+2. ``client.discogen.process(query=..., domains=[...], web_search=True)``
+   runs an AI research prompt over the discovered domains and returns one
+   structured answer per company. ``job.wait()`` blocks until it finishes.
+
+DiscoGen runs on your own LLM provider key (BYOK) - configure one first via
+``client.llm_providers`` or in the app under Settings -> Integrations.
+
+Usage:
+    export DISCOLIKE_API_KEY="dl_..."
+    python examples/discover_and_enrich.py \
+        --icp "Cybersecurity for SMBs, managed IT services" \
+        --country US \
+        --query "What is their pricing model, and do they sell to MSPs?"
+"""
+
+from __future__ import annotations
+
+import argparse
+import json
+import sys
+
+from discolike import Discolike
+
+
+def parse_args() -> argparse.Namespace:
+    parser = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
+    parser.add_argument("--icp", required=True, help="ICP description, e.g. 'B2B SaaS for logistics'")
+    parser.add_argument("--country", action="append", help="ISO-2 country filter, repeatable (e.g. --country US)")
+    parser.add_argument("--max-records", type=int, default=10, help="How many companies to discover (default: 10)")
+    parser.add_argument("--query", required=True, help="DiscoGen research prompt to run over each company")
+    parser.add_argument("--timeout", type=float, default=1800.0, help="DiscoGen wait timeout in seconds")
+    return parser.parse_args()
+
+
+def main() -> None:
+    args = parse_args()
+    client = Discolike()
+
+    print(f"Discovering up to {args.max_records} companies for: {args.icp!r}")
+    companies = client.discover(icp_text=args.icp, country=args.country, max_records=args.max_records)
+    if not companies:
+        sys.exit("No companies found for that ICP - try broadening it.")
+    domains = [company.domain for company in companies if company.domain]
+    for company in companies:
+        print(f"  {company.domain}  {company.name or ''}  (similarity {company.similarity})")
+
+    print(f"\nRunning DiscoGen over {len(domains)} domains: {args.query!r}")
+    job = client.discogen.process(query=args.query, domains=domains, web_search=True)
+    status = job.wait(timeout=args.timeout, on_poll=lambda s: print(f"  status={s.status} progress={s.progress}%"))
+
+    print("\nEnriched results:")
+    for row in status.results or []:
+        print(json.dumps(row, ensure_ascii=False, default=str))
+
+
+if __name__ == "__main__":
+    main()

examples/find_emails_from_csv.py

@@ -0,0 +1,109 @@
+"""Find work email addresses for a CSV of people via the email finder batch API.
+
+Reads people from a CSV with ``first_name``, ``last_name``, and ``domain``
+columns (names configurable), submits them with
+``client.email.find_batch(contacts=[...])`` in chunks of up to 500, waits for
+each batch with ``batch.results()``, and writes the found emails plus status
+to an output CSV.
+
+Billing note: only results with status "found" (an SMTP-verified address) are
+billed. Catch-all domains and pattern-based guesses are returned for free.
+
+Usage:
+    export DISCOLIKE_API_KEY="dl_..."
+    python examples/find_emails_from_csv.py people.csv --output emails.csv
+"""
+
+from __future__ import annotations
+
+import argparse
+import csv
+import sys
+from pathlib import Path
+
+from discolike import Discolike
+
+MAX_CONTACTS_PER_BATCH = 500
+
+OUTPUT_FIELDS = ["first_name", "last_name", "domain", "email", "status", "is_catch_all", "error"]
+
+
+def parse_args() -> argparse.Namespace:
+    parser = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
+    parser.add_argument("input", type=Path, help="Input CSV of people")
+    parser.add_argument("--output", type=Path, default=Path("emails.csv"), help="Output CSV path")
+    parser.add_argument("--first-name-column", default="first_name", help="First-name column (default: first_name)")
+    parser.add_argument("--last-name-column", default="last_name", help="Last-name column (default: last_name)")
+    parser.add_argument("--domain-column", default="domain", help="Company-domain column (default: domain)")
+    parser.add_argument("--timeout", type=float, default=1800.0, help="Per-batch wait timeout in seconds")
+    return parser.parse_args()
+
+
+def load_contacts(path: Path, args: argparse.Namespace) -> list[dict[str, str]]:
+    contacts: list[dict[str, str]] = []
+    with path.open(newline="", encoding="utf-8-sig") as handle:
+        for row in csv.DictReader(handle):
+            first = row.get(args.first_name_column, "").strip()
+            last = row.get(args.last_name_column, "").strip()
+            domain = row.get(args.domain_column, "").strip().lower().removeprefix("www.")
+            if first and last and domain:
+                contacts.append({"first_name": first, "last_name": last, "domain": domain})
+    return contacts
+
+
+def main() -> None:
+    args = parse_args()
+    contacts = load_contacts(args.input, args)
+    if not contacts:
+        sys.exit(f"No usable rows (first name + last name + domain) found in {args.input}")
+
+    client = Discolike()
+    found = 0
+    with args.output.open("w", newline="", encoding="utf-8") as handle:
+        writer = csv.DictWriter(handle, fieldnames=OUTPUT_FIELDS)
+        writer.writeheader()
+        for start in range(0, len(contacts), MAX_CONTACTS_PER_BATCH):
+            chunk = contacts[start : start + MAX_CONTACTS_PER_BATCH]
+            print(f"Submitting batch of {len(chunk)} contacts ({start + len(chunk)}/{len(contacts)})...")
+            batch = client.email.find_batch(contacts=chunk)
+            results = batch.results(timeout=args.timeout)
+            for item in results.results:
+                output = item.result
+                if output is None:
+                    # Failed jobs carry no EnumerationOutput (so no identity),
+                    # but must not vanish from the output: keep the status and
+                    # error so the failure is visible and countable.
+                    writer.writerow(
+                        {
+                            "first_name": "",
+                            "last_name": "",
+                            "domain": "",
+                            "email": "",
+                            "status": item.status or "failed",
+                            "is_catch_all": "",
+                            "error": item.error or "",
+                        }
+                    )
+                    continue
+                match = getattr(output, "result", None)
+                email = match.email if match is not None else None
+                if output.status == "found":
+                    found += 1
+                writer.writerow(
+                    {
+                        "first_name": output.first_name,
+                        "last_name": output.last_name,
+                        "domain": output.domain,
+                        "email": email or "",
+                        "status": output.status or "",
+                        "is_catch_all": output.is_catch_all,
+                        "error": output.error or "",
+                    }
+                )
+            handle.flush()
+
+    print(f'\nDone: {found}/{len(contacts)} verified emails (status "found") -> {args.output}')
+
+
+if __name__ == "__main__":
+    main()