# Run a step with your own worker

A runnable script that authenticates, claims each kind of job, submits a result and prints what Prospex accepted or refused.

A runnable worker. It authenticates with a credential, polls for open stage jobs, claims one, produces a result for whichever stage it is, and submits it.

```bash
export PROSPEX_API_KEY=pxk_…
python scripts/prospect_connect_worker.py --project project_128 --once
```

## The script

```python
#!/usr/bin/env python3
"""An external stage worker for Prospect Connect, end to end.

Two jobs in one file. It is the worked example rendered into the ``/docs/``
reference, and it is what we actually run against a real project before
shipping a change to any of this — which is the only way an example stays
true.

It uses nothing but the documented HTTP interface: no Django, no database, no
import from the app. If it needs something the reference does not describe,
that is a gap in the reference.

    export PROSPEX_API_KEY=pxk_…
    python scripts/prospect_connect_worker.py --project project_128 --once

What it does, per pass:

1. lists the project's open stage jobs;
2. claims one, which is what entitles it to submit;
3. reads the job to get the input snapshot;
4. produces a result for whichever stage it is;
5. submits it and prints whether Prospex accepted it, or which categorical
   errors it came back with.

The three ``build_*`` functions are the part a real worker replaces. They are
written to be *honest rather than clever*: the research one cites the pages the
snapshot already names and claims nothing they do not support, because a worker
that invents evidence is exactly what the validation on the other end exists to
catch.
"""

from __future__ import annotations

import argparse
import json
import os
import sys
import time
import urllib.error
import urllib.parse
import urllib.request

DEFAULT_BASE = os.environ.get("PROSPEX_BASE_URL", "https://prospex.ch")
SCHEMA_VERSION = "2026-08-01"
TOOL_NAME = "prospex-example-worker"


class ApiError(RuntimeError):
    def __init__(self, status: int, payload: dict):
        self.status = status
        self.payload = payload
        error = payload.get("error") or {}
        super().__init__(f"HTTP {status}: {error.get('code')} — {error.get('message')}")


class Client:
    """The documented interface and nothing else."""

    def __init__(self, base: str, key: str):
        self.base = base.rstrip("/")
        self.key = key

    def _call(self, method: str, path: str, body=None, *, idempotency_key: str = ""):
        url = f"{self.base}/api/v1/prospect/{path.lstrip('/')}"
        data = json.dumps(body).encode() if body is not None else None
        request = urllib.request.Request(url, data=data, method=method)
        request.add_header("Authorization", f"Bearer {self.key}")
        request.add_header("Accept", "application/json")
        if data is not None:
            request.add_header("Content-Type", "application/json")
        if idempotency_key:
            request.add_header("Idempotency-Key", idempotency_key)
        try:
            with urllib.request.urlopen(request, timeout=30) as response:
                payload = json.loads(response.read().decode() or "{}")
                # Worth watching in a real worker: it tells you how close the
                # account is to its daily allowance before a 429 does.
                payload["_remaining"] = response.headers.get("X-RateLimit-Remaining")
                return payload
        except urllib.error.HTTPError as exc:
            raw = exc.read().decode() or "{}"
            try:
                payload = json.loads(raw)
            except json.JSONDecodeError:
                payload = {"error": {"code": "unreadable", "message": raw[:500]}}
            raise ApiError(exc.code, payload) from exc

    def get(self, path: str, **params):
        query = urllib.parse.urlencode({k: v for k, v in params.items() if v})
        return self._call("GET", f"{path}?{query}" if query else path)

    def post(self, path: str, body: dict, *, idempotency_key: str):
        return self._call("POST", path, body, idempotency_key=idempotency_key)


# --- The part a real worker replaces ----------------------------------------


def build_qualification(job: dict) -> dict:
    """Grade the candidate from the same fit document Prospex would grade.

    A real qualifier would put a model or a rule set here. This one is
    deliberately conservative: it says `needs_review` whenever the stored
    description is thin, because a confident answer about a company we know
    nothing about is worse than no answer.
    """
    snapshot = job["input"]
    document = snapshot.get("fit_document") or ""
    company = snapshot["company"]["name"]
    thin = len(document) < 80

    if thin:
        return {
            "eligibility": "needs_review",
            "fit": "needs_review",
            "priority": "low",
            "reasons": [
                f"The stored description of {company} is too short to judge fit."
            ],
            "caveats": ["Re-qualify once the company description improves."],
        }
    return {
        "eligibility": "eligible",
        "fit": "good",
        "priority": "normal",
        "reasons": [
            f"{company} matches the offering's ideal customer on the stored "
            "description.",
        ],
        "caveats": [],
    }


def build_research(job: dict) -> dict:
    """Return sourced facts, each pointing at a page that supports it.

    A real worker searches here. This one cites only what the snapshot already
    named — the Watch signals that put the company in the project — so the
    example never fabricates a URL. If there is nothing to cite it says
    `needs_review`, which is the honest answer and the one Prospex will accept.
    """
    snapshot = job["input"]
    company = snapshot["company"]["name"]
    cited = [
        row
        for row in snapshot.get("provenance") or []
        if (row.get("detail") or {}).get("source_url")
    ]
    if not cited:
        return {
            "qualification": "needs_review",
            "identity": "confirmed",
            "qualification_reason": (
                f"No public page about {company} was available to this worker, "
                "so no defensible angle could be established."
            ),
            "facts": [],
            "evidence": [],
            "caveats": ["This worker searched no sources beyond the job snapshot."],
        }

    evidence = []
    facts = []
    for index, row in enumerate(cited[:3], 1):
        key = f"E{index}"
        detail = row["detail"]
        supported = detail.get("title") or f"A public page about {company}."
        evidence.append(
            {
                "id": key,
                "url": detail["source_url"],
                "title": detail.get("title") or "Source",
                "supported_fact": supported,
                "publication_date": row.get("effective_date"),
                "interpretation": False,
            }
        )
        facts.append({"text": supported, "evidence_ids": [key], "interpretation": False})

    return {
        "qualification": "ready_for_opener",
        "identity": "confirmed",
        "qualification_reason": (
            f"Recent public activity at {company} supports a specific opening."
        ),
        "facts": facts,
        "evidence": evidence,
        "caveats": [],
    }


def build_outreach(job: dict) -> dict:
    """Write the copy, wrapping each grounded passage in its evidence tags.

    The tags are the important part. `[E1]…[/E1]` around the words a source
    backs is how Prospex derives the claim list, and a tagged draft is accepted
    where an untagged one with a missing `presentation` is not.
    """
    snapshot = job["input"]
    company = snapshot["company"]["name"]
    evidence = snapshot["evidence"]
    if not evidence:
        raise SystemExit("This outreach job carries no evidence to cite.")
    first = evidence[0]
    key = first["evidence_key"]
    claim = first["supported_fact"].rstrip(".")

    subject = f"{claim[:80]}"
    opener = (
        f"Hello — I saw that [{key}]{claim}[/{key}]. "
        f"We work with companies at that point, and I wondered whether it is "
        f"worth twenty minutes for {company}."
    )
    return {
        "language": snapshot["language"],
        "subject": subject,
        "opener": opener,
        "angle_type": "trigger",
        "angle_explanation": (
            "Opens on the most recent thing the evidence establishes, and asks "
            "for a short conversation rather than asserting a need."
        ),
        "claims": [
            {"text": claim, "evidence_ids": [key], "presentation": "fact"},
        ],
        "evidence_ids": [key],
        "caveats": [],
    }


BUILDERS = {
    "qualification": build_qualification,
    "research": build_research,
    "outreach": build_outreach,
}


# --- The loop ----------------------------------------------------------------


def handle(client: Client, project: str, summary: dict, run_id: str) -> bool:
    """Claim and complete one job. Returns whether there was one to do."""
    listing = client.get(f"projects/{project}/stage-jobs", status="open")
    jobs = listing.get("data") or []
    if not jobs:
        return False

    for entry in jobs:
        job_id = entry["id"]
        try:
            client.post(
                f"projects/{project}/stage-jobs/{job_id}/claims",
                {},
                idempotency_key=f"{run_id}:claim:{job_id}",
            )
        except ApiError as exc:
            # Another worker holds it, or it expired between the list and now.
            # Both are ordinary; move on.
            print(f"  {job_id}: not claimable ({exc.payload['error']['code']})")
            summary["skipped"] += 1
            continue

        job = client.get(f"projects/{project}/stage-jobs/{job_id}")
        stage = job["stage"]
        print(f"  {job_id}: claimed, stage={stage}")

        body = {
            "stage_job_id": job["id"],
            "input_fingerprint": job["input_fingerprint"],
            "result_schema_version": SCHEMA_VERSION,
            "external_tool": TOOL_NAME,
            "external_run_id": f"{run_id}:{job_id}",
            "result": BUILDERS[stage](job),
        }
        try:
            answer = client.post(
                f"projects/{project}/stage-jobs/{job_id}/results",
                body,
                idempotency_key=f"{run_id}:result:{job_id}",
            )
        except ApiError as exc:
            if exc.status != 422:
                raise
            errors = exc.payload["error"].get("errors") or []
            print(f"  {job_id}: REFUSED")
            for error in errors:
                print(f"      {error['code']} at {error.get('field')}: {error['detail']}")
            summary["rejected"] += 1
            continue
        print(f"  {job_id}: accepted ({answer['stage_job']['status']})")
        summary["accepted"] += 1
    return True


def main() -> int:
    parser = argparse.ArgumentParser(description=__doc__.splitlines()[0])
    parser.add_argument("--project", required=True, help="e.g. project_128")
    parser.add_argument("--base", default=DEFAULT_BASE)
    parser.add_argument("--once", action="store_true", help="One pass, then stop.")
    parser.add_argument("--interval", type=int, default=30)
    parser.add_argument(
        "--run-id",
        default=f"run-{int(time.time())}",
        help="Used to build idempotency keys, so a re-run is safe.",
    )
    args = parser.parse_args()

    key = os.environ.get("PROSPEX_API_KEY", "").strip()
    if not key:
        print("Set PROSPEX_API_KEY to a credential with the prospect:stages scope.")
        return 2

    client = Client(args.base, key)
    summary = {"accepted": 0, "rejected": 0, "skipped": 0}
    try:
        while True:
            print(f"Polling {args.project} …")
            found = handle(client, args.project, summary, args.run_id)
            if not found:
                print("  nothing open")
            if args.once:
                break
            time.sleep(args.interval)
    except KeyboardInterrupt:
        pass
    except ApiError as exc:
        print(f"Stopped: {exc}")
        return 1
    print(
        f"accepted={summary['accepted']} rejected={summary['rejected']} "
        f"skipped={summary['skipped']}"
    )
    return 0


if __name__ == "__main__":
    sys.exit(main())
```