diff --git a/CHANGELOG.md b/CHANGELOG.md
index dcf0388..3a2754b 100644
--- a/CHANGELOG.md
+++ b/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.
diff --git a/README.md b/README.md
index 8d6d1ee..9475c05 100644
--- a/README.md
+++ b/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:
diff --git a/examples/README.md b/examples/README.md
new file mode 100644
index 0000000..20e96da
--- /dev/null
+++ b/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
+```
diff --git a/examples/discover_and_enrich.py b/examples/discover_and_enrich.py
new file mode 100644
index 0000000..e86b4cf
--- /dev/null
+++ b/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()
diff --git a/examples/find_emails_from_csv.py b/examples/find_emails_from_csv.py
new file mode 100644
index 0000000..6924d5b
--- /dev/null
+++ b/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()
diff --git a/examples/match_crm_contacts.py b/examples/match_crm_contacts.py
new file mode 100644
index 0000000..457b8d4
--- /dev/null
+++ b/examples/match_crm_contacts.py
@@ -0,0 +1,235 @@
+"""Match a CSV of CRM contacts to DiscoLike persona IDs via bulk-match.
+
+Reads contacts from a CSV (column names are configurable), submits them to
+``client.contacts.bulk_match()`` in chunks, and writes an output CSV with the
+matched ``persona_id`` and ``match_score`` per row.
+
+Lessons baked in from production runs:
+
+- **Two domain keys beat one.** For every row we derive a domain from BOTH the
+  website column AND the email address (skipping free-mail providers such as
+  gmail.com), submit a query per key, and keep the best-scoring hit. This
+  measurably improves the match rate on messy CRM exports.
+- **Resumable checkpointing.** Every completed chunk is appended to a JSONL
+  checkpoint, so an interrupted run picks up where it left off. On a chunk
+  failure or timeout the rows are deliberately NOT checkpointed as no-match --
+  they stay pending, so simply rerunning the script retries them.
+
+Country values must be ISO-3166-1 alpha-2 codes (e.g. "US", "DE"); anything
+else is omitted, because ``person_country`` is a hard filter server-side.
+
+Usage:
+    export DISCOLIKE_API_KEY="dl_..."
+    python examples/match_crm_contacts.py contacts.csv --output matched.csv
+"""
+
+from __future__ import annotations
+
+import argparse
+import csv
+import json
+import sys
+from pathlib import Path
+from typing import Any
+from urllib.parse import urlparse
+
+from discolike import Discolike
+
+MAX_QUERIES_PER_CALL = 500  # bulk_match accepts 1-500 queries per request
+
+FREE_MAIL_DOMAINS = frozenset(
+    {
+        "gmail.com",
+        "googlemail.com",
+        "yahoo.com",
+        "yahoo.co.uk",
+        "hotmail.com",
+        "hotmail.co.uk",
+        "outlook.com",
+        "live.com",
+        "msn.com",
+        "aol.com",
+        "icloud.com",
+        "me.com",
+        "gmx.com",
+        "gmx.de",
+        "proton.me",
+        "protonmail.com",
+        "web.de",
+        "mail.com",
+    }
+)
+
+
+def parse_args() -> argparse.Namespace:
+    parser = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
+    parser.add_argument("input", type=Path, help="Input CSV of CRM contacts")
+    parser.add_argument("--output", type=Path, default=Path("matched.csv"), help="Output CSV path")
+    parser.add_argument(
+        "--checkpoint", type=Path, default=None, help="Checkpoint JSONL path (default: <output>.checkpoint.jsonl)"
+    )
+    parser.add_argument("--name-column", default="name", help="Full-name column (default: name)")
+    parser.add_argument("--email-column", default="email", help="Email column (default: email)")
+    parser.add_argument("--website-column", default="website", help="Website/domain column (default: website)")
+    parser.add_argument("--country-column", default="country", help="ISO-2 country column (default: country)")
+    parser.add_argument("--company-column", default="company", help="Company-name column, a soft matching signal")
+    parser.add_argument("--chunk-size", type=int, default=MAX_QUERIES_PER_CALL, help="Queries per bulk-match call")
+    parser.add_argument("--timeout", type=float, default=1800.0, help="Per-chunk wait timeout in seconds")
+    return parser.parse_args()
+
+
+def website_domain(value: str) -> str | None:
+    """'https://www.example.com/about' -> 'example.com'."""
+    value = value.strip()
+    if not value:
+        return None
+    if "//" not in value:
+        value = "//" + value
+    host = (urlparse(value).netloc or urlparse(value).path).lower().split("/")[0].strip()
+    return host.removeprefix("www.") or None
+
+
+def email_domain(value: str) -> str | None:
+    """'jane@acme.com' -> 'acme.com', skipping free-mail providers."""
+    value = value.strip().lower()
+    if "@" not in value:
+        return None
+    host = value.rsplit("@", 1)[1].strip().strip(".").removeprefix("www.")
+    if not host or host in FREE_MAIL_DOMAINS:
+        return None
+    return host
+
+
+def build_queries(row_id: int, row: dict[str, str], args: argparse.Namespace) -> list[dict[str, Any]]:
+    """One query per distinct domain key (website + email domain), tagged with the row id.
+
+    The ``input:row`` key is echoed back untouched in each result's ``query``,
+    which is how results are joined back to input rows.
+    """
+    name = row.get(args.name_column, "").strip()
+    email = row.get(args.email_column, "").strip()
+    if not name and not email:
+        return []  # bulk-match needs at least a name or an email
+
+    base: dict[str, Any] = {"input:row": row_id}
+    if name:
+        base["name"] = name
+    if email:
+        base["email"] = email
+    company = row.get(args.company_column, "").strip()
+    if company:
+        base["company_name"] = company
+    country = row.get(args.country_column, "").strip()
+    if len(country) == 2:
+        base["person_country"] = country.upper()
+
+    domains = {d for d in (website_domain(row.get(args.website_column, "")), email_domain(email)) if d}
+    if not domains:
+        return [base]
+    return [{**base, "domain": d} for d in sorted(domains)]
+
+
+def load_checkpoint(path: Path) -> dict[int, dict[str, Any] | None]:
+    done: dict[int, dict[str, Any] | None] = {}
+    if path.is_file():
+        for line in path.read_text(encoding="utf-8").splitlines():
+            if line.strip():
+                record = json.loads(line)
+                done[record["row"]] = record["result"]
+    return done
+
+
+def best_hits(results: list[Any], row_ids: list[int]) -> dict[int, dict[str, Any] | None]:
+    """Union the hits across each row's queries and keep the highest-scoring one."""
+    hits: dict[int, dict[str, Any] | None] = dict.fromkeys(row_ids)
+    for item in results or []:
+        if not isinstance(item, dict):
+            continue
+        row_id = (item.get("query") or {}).get("input:row")
+        if row_id not in hits:
+            continue
+        for match in item.get("matches") or []:
+            if match.get("persona_id") is None:
+                continue
+            score = float(match.get("match_score") or 0.0)
+            current = hits[row_id]
+            if current is None or score > float(current.get("match_score") or 0.0):
+                hits[row_id] = {"persona_id": match["persona_id"], "match_score": score}
+    return hits
+
+
+def main() -> None:
+    args = parse_args()
+    checkpoint_path = args.checkpoint or args.output.with_suffix(".checkpoint.jsonl")
+
+    with args.input.open(newline="", encoding="utf-8-sig") as handle:
+        reader = csv.DictReader(handle)
+        fieldnames = list(reader.fieldnames or [])
+        rows = list(reader)
+    if not rows:
+        sys.exit(f"No data rows found in {args.input}")
+
+    done = load_checkpoint(checkpoint_path)
+    if done:
+        print(f"Resuming: {len(done)} rows already checkpointed in {checkpoint_path}")
+
+    pending = [(i, row) for i, row in enumerate(rows) if i not in done]
+    chunk_size = min(args.chunk_size, MAX_QUERIES_PER_CALL)
+
+    # Pack rows into chunks by QUERY count, not row count: a row expands to up
+    # to two queries (website domain + email domain), so chunking 500 rows at a
+    # time could send up to 1,000 queries and blow the per-call limit.
+    chunks: list[list[tuple[int, list[dict[str, Any]]]]] = []
+    current: list[tuple[int, list[dict[str, Any]]]] = []
+    query_count = 0
+    for row_id, row in pending:
+        row_queries = build_queries(row_id, row, args)
+        if current and query_count + len(row_queries) > chunk_size:
+            chunks.append(current)
+            current, query_count = [], 0
+        current.append((row_id, row_queries))
+        query_count += len(row_queries)
+    if current:
+        chunks.append(current)
+    print(f"{len(rows)} rows total, {len(pending)} to match now in {len(chunks)} chunk(s) of <= {chunk_size} queries")
+
+    client = Discolike()
+    failed_chunks = 0
+    with checkpoint_path.open("a", encoding="utf-8") as checkpoint:
+        for index, chunk in enumerate(chunks):
+            queries = [query for _, row_queries in chunk for query in row_queries]
+            row_ids = [row_id for row_id, _ in chunk]
+            label = f"chunk {index + 1}/{len(chunks)}"
+            if not queries:
+                hits: dict[int, dict[str, Any] | None] = dict.fromkeys(row_ids)
+            else:
+                try:
+                    job = client.contacts.bulk_match(queries=queries, limit=1)
+                    status = job.wait(timeout=args.timeout)
+                except Exception as exc:  # leave the chunk un-checkpointed so a rerun retries it
+                    failed_chunks += 1
+                    print(f"  ! {label} failed ({exc}); rows left pending, rerun to retry")
+                    continue
+                hits = best_hits(status.results, row_ids)
+            for row_id, result in hits.items():
+                checkpoint.write(json.dumps({"row": row_id, "result": result}) + "\n")
+            checkpoint.flush()
+            done.update(hits)
+            matched = sum(1 for value in hits.values() if value)
+            print(f"  {label}: {matched}/{len(chunk)} matched")
+
+    extra = ["persona_id", "match_score"]
+    with args.output.open("w", newline="", encoding="utf-8") as handle:
+        writer = csv.DictWriter(handle, fieldnames=fieldnames + extra, extrasaction="ignore")
+        writer.writeheader()
+        for i, row in enumerate(rows):
+            writer.writerow({**row, **(done.get(i) or {})})
+
+    matched_total = sum(1 for value in done.values() if value)
+    print(f"\nDone: {matched_total}/{len(rows)} matched -> {args.output}")
+    if failed_chunks:
+        print(f"{failed_chunks} chunk(s) failed and were not checkpointed; rerun this command to retry them.")
+
+
+if __name__ == "__main__":
+    main()
diff --git a/packages/discolike-cli/src/discolike_cli/email.py b/packages/discolike-cli/src/discolike_cli/email.py
new file mode 100644
index 0000000..321ef09
--- /dev/null
+++ b/packages/discolike-cli/src/discolike_cli/email.py
@@ -0,0 +1,166 @@
+from __future__ import annotations
+
+import csv
+import pathlib
+import sys
+
+import typer
+
+from discolike._email import EmailBatch
+from discolike._email import EmailBatchResults
+from discolike._email import EmailJobResult
+from discolike._exceptions import JobTimeoutError
+from discolike_cli._output import emit
+from discolike_cli._output import handle_errors
+
+DEFAULT_WAIT_TIMEOUT_SECONDS = 900.0
+MAX_BATCH_CONTACTS = 500
+EMAIL_KINDS = ("find", "verify")
+CSV_COLUMNS = ("first_name", "last_name", "domain")
+
+FORMAT_HELP = "Output format: json or table (table auto-selected on a TTY; falls back to JSON for non-tabular data)."
+WAIT_HELP = "Block until the job finishes, streaming progress to stderr."
+TIMEOUT_HELP = "Max seconds to wait with --wait."
+KIND_HELP = "Batch kind: find or verify (verify batches are created by the DiscoLike app)."
+
+app = typer.Typer(
+    help=(
+        "Find work email addresses: submit single or batch find jobs, poll them, and fetch results. "
+        "Only proven addresses bill; catch-all and pattern guesses are free."
+    )
+)
+
+
+def _job_status_to_stderr(status: EmailJobResult) -> None:
+    sys.stderr.write(f"status: {status.status}\n")
+
+
+def _batch_progress_to_stderr(results: EmailBatchResults) -> None:
+    sys.stderr.write(f"progress: {results.completed}/{results.total} completed, {results.failed} failed\n")
+
+
+def _parse_contact(value: str) -> dict[str, str]:
+    parts = [part.strip() for part in value.split(",")]
+    if len(parts) != len(CSV_COLUMNS) or not all(parts):
+        raise typer.BadParameter(f'--contact must be "first_name,last_name,domain", got {value!r}')
+    return dict(zip(CSV_COLUMNS, parts, strict=True))
+
+
+def _read_contacts_file(contacts_file: pathlib.Path) -> list[dict[str, str]]:
+    with contacts_file.open(newline="") as handle:
+        reader = csv.DictReader(handle)
+        missing = set(CSV_COLUMNS) - set(reader.fieldnames or [])
+        if missing:
+            raise typer.BadParameter(f"--contacts-file is missing required CSV columns: {', '.join(sorted(missing))}")
+        return [{column: (row.get(column) or "").strip() for column in CSV_COLUMNS} for row in reader]
+
+
+def _fetch_batch_snapshot(batch: EmailBatch) -> EmailBatchResults:
+    """Fetch the batch results once, whether or not the batch has finished."""
+    snapshot: dict[str, EmailBatchResults] = {}
+    try:
+        return batch.results(timeout=0, on_poll=lambda results: snapshot.update(latest=results))
+    except JobTimeoutError:
+        return snapshot["latest"]
+
+
+@app.command("find")
+@handle_errors
+def find_command(
+    ctx: typer.Context,
+    first_name: str = typer.Argument(..., help="First name of the person."),
+    last_name: str = typer.Argument(..., help="Last name of the person."),
+    domain: str = typer.Argument(..., help="Company domain to search, e.g. acme.com."),
+    known_pattern: str | None = typer.Option(
+        None, "--known-pattern", help="Known email local-part pattern for this domain, e.g. first.last."
+    ),
+    wait: bool = typer.Option(False, "--wait/--no-wait", help=WAIT_HELP),
+    timeout: float = typer.Option(DEFAULT_WAIT_TIMEOUT_SECONDS, "--timeout", help=TIMEOUT_HELP),
+    fmt: str | None = typer.Option(None, "--format", help=FORMAT_HELP),
+) -> None:
+    """Submit a single email find job (async); only a proven address bills."""
+    from discolike_cli.main import get_client
+
+    job = get_client(ctx).email.find(
+        first_name=first_name, last_name=last_name, domain=domain, known_pattern=known_pattern
+    )
+    if not wait:
+        emit({"job_id": job.job_id, "hint": f"poll with: discolike email job {job.job_id}"})
+        return
+    emit(job.wait(timeout=timeout, on_poll=_job_status_to_stderr), fmt=fmt)
+
+
+@app.command("find-batch")
+@handle_errors
+def find_batch_command(
+    ctx: typer.Context,
+    contacts_file: pathlib.Path | None = typer.Option(
+        None,
+        "--contacts-file",
+        help="Path to a CSV file with first_name,last_name,domain columns (max 500 contacts per batch).",
+    ),
+    contact: list[str] | None = typer.Option(
+        None, "--contact", help='Inline contact as "first_name,last_name,domain" (repeatable).'
+    ),
+    wait: bool = typer.Option(False, "--wait/--no-wait", help=WAIT_HELP),
+    timeout: float = typer.Option(DEFAULT_WAIT_TIMEOUT_SECONDS, "--timeout", help=TIMEOUT_HELP),
+    fmt: str | None = typer.Option(None, "--format", help=FORMAT_HELP),
+) -> None:
+    """Submit an email find batch from a CSV file and/or inline contacts (async)."""
+    from discolike_cli.main import get_client
+
+    contacts: list[dict[str, str]] = []
+    if contacts_file is not None:
+        contacts.extend(_read_contacts_file(contacts_file))
+    contacts.extend(_parse_contact(value) for value in contact or [])
+    if not contacts:
+        raise typer.BadParameter("provide --contacts-file and/or at least one --contact")
+    if len(contacts) > MAX_BATCH_CONTACTS:
+        raise typer.BadParameter(f"a batch holds at most {MAX_BATCH_CONTACTS} contacts, got {len(contacts)}")
+
+    batch = get_client(ctx).email.find_batch(contacts=contacts)
+    if not wait:
+        emit({"batch_id": batch.batch_id, "hint": f"fetch with: discolike email results {batch.batch_id}"})
+        return
+    emit(batch.results(timeout=timeout, on_poll=_batch_progress_to_stderr), fmt=fmt)
+
+
+@app.command("results")
+@handle_errors
+def results_command(
+    ctx: typer.Context,
+    batch_id: str = typer.Argument(..., help="Batch ID returned by `discolike email find-batch`."),
+    kind: str = typer.Option("find", "--kind", help=KIND_HELP),
+    wait: bool = typer.Option(False, "--wait/--no-wait", help=WAIT_HELP),
+    timeout: float = typer.Option(DEFAULT_WAIT_TIMEOUT_SECONDS, "--timeout", help=TIMEOUT_HELP),
+    fmt: str | None = typer.Option(None, "--format", help=FORMAT_HELP),
+) -> None:
+    """Fetch results for an email find or verify batch."""
+    from discolike_cli.main import get_client
+
+    if kind not in EMAIL_KINDS:
+        raise typer.BadParameter(f"--kind must be one of: {', '.join(EMAIL_KINDS)}")
+    batch = get_client(ctx).email.batch(batch_id, kind=kind)  # type: ignore[arg-type]
+    if wait:
+        emit(batch.results(timeout=timeout, on_poll=_batch_progress_to_stderr), fmt=fmt)
+        return
+    emit(_fetch_batch_snapshot(batch), fmt=fmt)
+
+
+@app.command("job")
+@handle_errors
+def job_command(
+    ctx: typer.Context,
+    job_id: str = typer.Argument(..., help="Job ID returned by `discolike email find`."),
+    wait: bool = typer.Option(False, "--wait/--no-wait", help=WAIT_HELP),
+    timeout: float = typer.Option(DEFAULT_WAIT_TIMEOUT_SECONDS, "--timeout", help=TIMEOUT_HELP),
+    fmt: str | None = typer.Option(None, "--format", help=FORMAT_HELP),
+) -> None:
+    """Poll a single email find job: print its current status, or block with --wait."""
+    from discolike_cli.main import get_client
+
+    job = get_client(ctx).email.job(job_id)
+    if wait:
+        emit(job.wait(timeout=timeout, on_poll=_job_status_to_stderr), fmt=fmt)
+        return
+    emit(job.status(), fmt=fmt)
diff --git a/packages/discolike-cli/src/discolike_cli/main.py b/packages/discolike-cli/src/discolike_cli/main.py
index 92561a6..92fe359 100644
--- a/packages/discolike-cli/src/discolike_cli/main.py
+++ b/packages/discolike-cli/src/discolike_cli/main.py
@@ -13,6 +13,7 @@
 from discolike_cli import contacts
 from discolike_cli import discogen
 from discolike_cli import discover
+from discolike_cli import email
 from discolike_cli import enrich
 from discolike_cli import match
 from discolike_cli import providers
@@ -57,6 +58,7 @@ def get_client(ctx: typer.Context) -> Discolike:
 app.add_typer(company.app, name="company")
 app.add_typer(contacts.app, name="contacts")
 app.add_typer(discogen.app, name="discogen")
+app.add_typer(email.app, name="email")
 app.add_typer(queries.app, name="queries")
 app.add_typer(account.app, name="account")
 app.add_typer(providers.search_providers_app, name="search-providers")
diff --git a/packages/discolike-cli/tests/test_email_cli.py b/packages/discolike-cli/tests/test_email_cli.py
new file mode 100644
index 0000000..a4426b1
--- /dev/null
+++ b/packages/discolike-cli/tests/test_email_cli.py
@@ -0,0 +1,283 @@
+from __future__ import annotations
+
+import json
+import pathlib
+import time
+from collections.abc import Callable
+
+import httpx
+import pytest
+from typer.testing import CliRunner
+
+from discolike_cli.main import app
+from discolike_testkit import Handler
+
+runner = CliRunner()
+
+
+@pytest.fixture(autouse=True)
+def no_sleep(monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.setattr(time, "sleep", lambda seconds: None)
+
+
+FOUND_RESULT = {
+    "first_name": "Jane",
+    "last_name": "Doe",
+    "domain": "acme.com",
+    "status": "found",
+    "result": {"email": "jane.doe@acme.com", "pattern": "{first}.{last}", "valid": True},
+}
+
+
+def test_email_find_without_wait_prints_job_hint(install_build_client: Callable[[Handler], None]) -> None:
+    captured: dict[str, httpx.Request] = {}
+
+    def handler(request: httpx.Request) -> httpx.Response:
+        captured["request"] = request
+        return httpx.Response(200, json={"job_id": "ej-1"})
+
+    install_build_client(handler)
+    result = runner.invoke(app, ["email", "find", "Jane", "Doe", "acme.com"])
+    assert result.exit_code == 0, result.output
+    request = captured["request"]
+    assert request.url.path == "/v1/email/find"
+    assert json.loads(request.content) == {"first_name": "Jane", "last_name": "Doe", "domain": "acme.com"}
+    payload = json.loads(result.stdout)
+    assert payload["job_id"] == "ej-1"
+    assert "discolike email job ej-1" in payload["hint"]
+
+
+def test_email_find_with_wait_polls_to_completion(install_build_client: Callable[[Handler], None]) -> None:
+    statuses = iter(
+        [
+            httpx.Response(200, json={"job_id": "ej-2", "status": "processing"}),
+            httpx.Response(200, json={"job_id": "ej-2", "status": "completed", "result": FOUND_RESULT}),
+        ]
+    )
+
+    def handler(request: httpx.Request) -> httpx.Response:
+        if request.url.path == "/v1/email/find":
+            return httpx.Response(200, json={"job_id": "ej-2"})
+        assert request.url.path == "/v1/email/jobs/ej-2"
+        return next(statuses)
+
+    install_build_client(handler)
+    result = runner.invoke(app, ["email", "find", "Jane", "Doe", "acme.com", "--wait", "--timeout", "5"])
+    assert result.exit_code == 0, result.output
+    payload = json.loads(result.stdout)
+    assert payload["status"] == "found"
+    assert payload["result"]["email"] == "jane.doe@acme.com"
+    assert "status: processing" in result.stderr
+
+
+def test_email_find_batch_from_csv_file(
+    tmp_path: pathlib.Path, install_build_client: Callable[[Handler], None]
+) -> None:
+    contacts_file = tmp_path / "contacts.csv"
+    contacts_file.write_text("first_name,last_name,domain\nJane,Doe,acme.com\nJohn,Smith,beta.com\n")
+    captured: dict[str, httpx.Request] = {}
+
+    def handler(request: httpx.Request) -> httpx.Response:
+        captured["request"] = request
+        return httpx.Response(200, json={"batch_id": "eb-1"})
+
+    install_build_client(handler)
+    result = runner.invoke(app, ["email", "find-batch", "--contacts-file", str(contacts_file)])
+    assert result.exit_code == 0, result.output
+    request = captured["request"]
+    assert request.url.path == "/v1/email/find/batch"
+    assert json.loads(request.content) == {
+        "requests": [
+            {"first_name": "Jane", "last_name": "Doe", "domain": "acme.com"},
+            {"first_name": "John", "last_name": "Smith", "domain": "beta.com"},
+        ]
+    }
+    payload = json.loads(result.stdout)
+    assert payload["batch_id"] == "eb-1"
+    assert "discolike email results eb-1" in payload["hint"]
+
+
+def test_email_find_batch_from_inline_contacts(install_build_client: Callable[[Handler], None]) -> None:
+    captured: dict[str, httpx.Request] = {}
+
+    def handler(request: httpx.Request) -> httpx.Response:
+        captured["request"] = request
+        return httpx.Response(200, json={"batch_id": "eb-2"})
+
+    install_build_client(handler)
+    result = runner.invoke(
+        app,
+        ["email", "find-batch", "--contact", "Jane,Doe,acme.com", "--contact", "John, Smith, beta.com"],
+    )
+    assert result.exit_code == 0, result.output
+    body = json.loads(captured["request"].content)
+    assert body["requests"] == [
+        {"first_name": "Jane", "last_name": "Doe", "domain": "acme.com"},
+        {"first_name": "John", "last_name": "Smith", "domain": "beta.com"},
+    ]
+
+
+def test_email_find_batch_with_wait_polls_results(install_build_client: Callable[[Handler], None]) -> None:
+    results_pages = iter(
+        [
+            httpx.Response(200, json={"batch_id": "eb-3", "total": 1, "completed": 0, "failed": 0, "results": []}),
+            httpx.Response(
+                200,
+                json={
+                    "batch_id": "eb-3",
+                    "total": 1,
+                    "completed": 1,
+                    "failed": 0,
+                    "results": [{"job_id": "ej-9", "status": "completed", "result": FOUND_RESULT}],
+                },
+            ),
+        ]
+    )
+
+    def handler(request: httpx.Request) -> httpx.Response:
+        if request.url.path == "/v1/email/find/batch":
+            return httpx.Response(200, json={"batch_id": "eb-3"})
+        assert request.url.path == "/v1/email/batch/eb-3/results"
+        return next(results_pages)
+
+    install_build_client(handler)
+    result = runner.invoke(
+        app,
+        ["email", "find-batch", "--contact", "Jane,Doe,acme.com", "--wait", "--timeout", "5"],
+    )
+    assert result.exit_code == 0, result.output
+    payload = json.loads(result.stdout)
+    assert payload["completed"] == 1
+    assert payload["results"][0]["result"]["result"]["email"] == "jane.doe@acme.com"
+    assert "progress: 0/1 completed, 0 failed" in result.stderr
+
+
+@pytest.mark.parametrize(
+    "args",
+    [
+        [],  # neither source given
+        ["--contact", "Jane,Doe"],  # too few fields
+        ["--contact", "Jane,,acme.com"],  # empty field
+    ],
+)
+def test_email_find_batch_bad_contacts_exit_2(args: list[str], install_build_client: Callable[[Handler], None]) -> None:
+    def handler(request: httpx.Request) -> httpx.Response:
+        return httpx.Response(200, json={"batch_id": "eb-4"})
+
+    install_build_client(handler)
+    result = runner.invoke(app, ["email", "find-batch", *args])
+    assert result.exit_code == 2
+
+
+def test_email_find_batch_missing_csv_columns_exits_2(
+    tmp_path: pathlib.Path, install_build_client: Callable[[Handler], None]
+) -> None:
+    contacts_file = tmp_path / "contacts.csv"
+    contacts_file.write_text("first,last,site\nJane,Doe,acme.com\n")
+
+    def handler(request: httpx.Request) -> httpx.Response:
+        return httpx.Response(200, json={"batch_id": "eb-5"})
+
+    install_build_client(handler)
+    result = runner.invoke(app, ["email", "find-batch", "--contacts-file", str(contacts_file)])
+    assert result.exit_code == 2
+    assert "domain" in result.output
+
+
+def test_email_find_batch_over_500_contacts_exits_2(
+    tmp_path: pathlib.Path, install_build_client: Callable[[Handler], None]
+) -> None:
+    rows = "\n".join(f"Jane{i},Doe,acme.com" for i in range(501))
+    contacts_file = tmp_path / "contacts.csv"
+    contacts_file.write_text(f"first_name,last_name,domain\n{rows}\n")
+
+    def handler(request: httpx.Request) -> httpx.Response:
+        return httpx.Response(200, json={"batch_id": "eb-6"})
+
+    install_build_client(handler)
+    result = runner.invoke(app, ["email", "find-batch", "--contacts-file", str(contacts_file)])
+    assert result.exit_code == 2
+    assert "500" in result.output
+
+
+def test_email_results_without_wait_returns_partial_snapshot(
+    install_build_client: Callable[[Handler], None],
+) -> None:
+    captured: dict[str, httpx.Request] = {}
+
+    def handler(request: httpx.Request) -> httpx.Response:
+        captured["request"] = request
+        return httpx.Response(
+            200,
+            json={"batch_id": "eb-7", "total": 2, "completed": 1, "failed": 0, "results": [{"status": "completed"}]},
+        )
+
+    install_build_client(handler)
+    result = runner.invoke(app, ["email", "results", "eb-7"])
+    assert result.exit_code == 0, result.output
+    assert captured["request"].url.path == "/v1/email/batch/eb-7/results"
+    payload = json.loads(result.stdout)
+    assert payload["batch_id"] == "eb-7"
+    assert payload["completed"] == 1
+
+
+def test_email_results_verify_kind_decodes_validation_output(
+    install_build_client: Callable[[Handler], None],
+) -> None:
+    def handler(request: httpx.Request) -> httpx.Response:
+        assert request.url.path == "/v1/email/batch/eb-8/results"
+        return httpx.Response(
+            200,
+            json={
+                "batch_id": "eb-8",
+                "total": 1,
+                "completed": 1,
+                "failed": 0,
+                "results": [
+                    {
+                        "job_id": "ej-8",
+                        "status": "completed",
+                        "result": {"email": "jane@acme.com", "status": "valid", "reason": "deliverable"},
+                    }
+                ],
+            },
+        )
+
+    install_build_client(handler)
+    result = runner.invoke(app, ["email", "results", "eb-8", "--kind", "verify"])
+    assert result.exit_code == 0, result.output
+    payload = json.loads(result.stdout)
+    assert payload["results"][0]["result"]["reason"] == "deliverable"
+
+
+def test_email_results_invalid_kind_exits_2(install_build_client: Callable[[Handler], None]) -> None:
+    def handler(request: httpx.Request) -> httpx.Response:
+        return httpx.Response(200, json={})
+
+    install_build_client(handler)
+    result = runner.invoke(app, ["email", "results", "eb-9", "--kind", "bogus"])
+    assert result.exit_code == 2
+
+
+def test_email_job_prints_current_status(install_build_client: Callable[[Handler], None]) -> None:
+    def handler(request: httpx.Request) -> httpx.Response:
+        assert request.url.path == "/v1/email/jobs/ej-5"
+        return httpx.Response(200, json={"job_id": "ej-5", "status": "processing"})
+
+    install_build_client(handler)
+    result = runner.invoke(app, ["email", "job", "ej-5"])
+    assert result.exit_code == 0, result.output
+    payload = json.loads(result.stdout)
+    assert payload["job_id"] == "ej-5"
+    assert payload["status"] == "processing"
+
+
+def test_email_job_unauthorized_exits_3(install_build_client: Callable[[Handler], None]) -> None:
+    def handler(request: httpx.Request) -> httpx.Response:
+        return httpx.Response(401, json={"detail": "invalid key"})
+
+    install_build_client(handler)
+    result = runner.invoke(app, ["email", "job", "ej-6"])
+    assert result.exit_code == 3
+    payload = json.loads(result.stderr)
+    assert payload["error"] == "AuthenticationError"
diff --git a/packages/discolike/src/discolike/resources/email.py b/packages/discolike/src/discolike/resources/email.py
index 78563a4..2f37ca2 100644
--- a/packages/discolike/src/discolike/resources/email.py
+++ b/packages/discolike/src/discolike/resources/email.py
@@ -31,13 +31,13 @@
 
 
 class EmailResource(SyncAPIResource):
-    @api_route("POST", "/email/find", openapi=False)
-    def find(self, *, first_name: str, last_name: str, domain: str) -> EmailJob:
+    @api_route("POST", "/email/find")
+    def find(self, *, first_name: str, last_name: str, domain: str, known_pattern: str | None = None) -> EmailJob:
         body = {k: v for k, v in locals().items() if k != "self"}
         response = self._transport.request("POST", "/email/find", json_body=drop_none(body))
         return EmailJob(self._transport, job_id=response.json()["job_id"], kind="find")
 
-    @api_route("POST", "/email/find/batch", openapi=False, ignore_params=("contacts",))
+    @api_route("POST", "/email/find/batch", ignore_params=("contacts",))
     def find_batch(self, *, contacts: list[dict[str, str]]) -> EmailBatch:
         response = self._transport.request("POST", "/email/find/batch", json_body={"requests": contacts})
         return EmailBatch(self._transport, batch_id=response.json()["batch_id"], kind="find")
@@ -50,13 +50,15 @@ def job(self, job_id: str) -> EmailJob:
 
 
 class AsyncEmailResource(AsyncAPIResource):
-    @api_route("POST", "/email/find", openapi=False)
-    async def find(self, *, first_name: str, last_name: str, domain: str) -> AsyncEmailJob:
+    @api_route("POST", "/email/find")
+    async def find(
+        self, *, first_name: str, last_name: str, domain: str, known_pattern: str | None = None
+    ) -> AsyncEmailJob:
         body = {k: v for k, v in locals().items() if k != "self"}
         response = await self._transport.request("POST", "/email/find", json_body=drop_none(body))
         return AsyncEmailJob(self._transport, job_id=response.json()["job_id"], kind="find")
 
-    @api_route("POST", "/email/find/batch", openapi=False, ignore_params=("contacts",))
+    @api_route("POST", "/email/find/batch", ignore_params=("contacts",))
     async def find_batch(self, *, contacts: list[dict[str, str]]) -> AsyncEmailBatch:
         response = await self._transport.request("POST", "/email/find/batch", json_body={"requests": contacts})
         return AsyncEmailBatch(self._transport, batch_id=response.json()["batch_id"], kind="find")
diff --git a/packages/discolike/tests/test_email.py b/packages/discolike/tests/test_email.py
index 7bbf398..20095e1 100644
--- a/packages/discolike/tests/test_email.py
+++ b/packages/discolike/tests/test_email.py
@@ -241,6 +241,26 @@ def handler(request: httpx.Request) -> httpx.Response:
     assert output.result.email == "grace@navy.mil"
 
 
+def test_find_sends_known_pattern_and_omits_it_when_unset(make_client: ClientFactory) -> None:
+    bodies: list[dict] = []
+
+    def handler(request: httpx.Request) -> httpx.Response:
+        bodies.append(json.loads(request.content))
+        return httpx.Response(202, json={"job_id": "j-kp", "status": "queued"})
+
+    with make_client(handler) as client:
+        client.email.find(first_name="Grace", last_name="Hopper", domain="navy.mil", known_pattern="first.last")
+        client.email.find(first_name="Grace", last_name="Hopper", domain="navy.mil")
+
+    assert bodies[0] == {
+        "first_name": "Grace",
+        "last_name": "Hopper",
+        "domain": "navy.mil",
+        "known_pattern": "first.last",
+    }
+    assert bodies[1] == {"first_name": "Grace", "last_name": "Hopper", "domain": "navy.mil"}
+
+
 def test_find_wait_raises_on_failed_job(make_client: ClientFactory) -> None:
     def handler(request: httpx.Request) -> httpx.Response:
         if request.method == "POST":
@@ -370,8 +390,8 @@ def test_route_metadata_stamped() -> None:
     from discolike.resources._base import get_discolike_route
     from discolike.resources.email import EmailResource
 
-    assert get_discolike_route(EmailResource.find) == ("POST", "/email/find", False, ())
-    assert get_discolike_route(EmailResource.find_batch) == ("POST", "/email/find/batch", False, ("contacts",))
+    assert get_discolike_route(EmailResource.find) == ("POST", "/email/find", True, ())
+    assert get_discolike_route(EmailResource.find_batch) == ("POST", "/email/find/batch", True, ("contacts",))
     assert get_discolike_route(EmailResource.job) is None
     assert get_discolike_route(EmailResource.batch) is None