Workable betreibt seinen eigenen Model-Context-Protocol-Server unter https://mcp.workable.com/mcp, damit ist die Bau-Frage erledigt: Sie verbinden sich, Sie schreiben keinen. Offen ist, welche seiner 94 Tools der Assistent Ihrer Recruiter anfassen darf. Workable hat den Server am 2026-05-13 mit 38 Tools gestartet und ihn am 2026-07-20 auf 94 erweitert, und diese Erweiterung brachte Schreibzugriff auf Performance Reviews, Konto- und Rechteverwaltung sowie Kandidatenprofile. Das Artefakt-Bundle unter apps/web/public/artifacts/mcp-server-workable-recruiting/ ist die Antwort auf diese Frage: ein Gateway nach dem Least-Privilege-Prinzip (README.md, pyproject.toml, src/workable_gateway/policy.py, src/workable_gateway/server.py), das 33 Tools durchreicht, 13 hinter eine zweistufige menschliche Freigabe stellt und die übrigen 48 ablehnt.
Wann Sie es einsetzen
Verbinden Sie den gehosteten Server, sobald Ihre Recruiter ohnehin in Claude an angrenzenden Aufgaben arbeiten — Outbound-Entwürfe, Scorecard-Zusammenfassungen, Updates an den Hiring Manager — und trotzdem ständig zurück in Workable springen, um “in welcher Stage ist dieser Kandidat”, “welche Bewerbungen haben sich diese Woche nicht bewegt”, “wer sitzt im Interview-Loop dieser Req” zu beantworten. Die Verbindung ist ein einziger Befehl und kostet nichts: Workable enthält den MCP-Server ohne Aufpreis in allen Abo-Plänen.
Setzen Sie das Gateway obendrauf, wenn die Allowlist zentral halten muss. Die settings.json eines Recruiters wird von seinem Client durchgesetzt, auf seinem Rechner, und er kann sie ändern. Ein Gateway-Prozess wird einmal durchgesetzt, von Recruiting-Ops, und ihn zu betreiben ist der Unterschied zwischen einer Richtlinie und einer Präferenz. Die passende Größe ist ein Recruiting-Team ab fünf Personen an einem gemeinsamen Workable-Konto, in einer Organisation, in der irgendwann jemand fragt, wer entschieden hat, dass der Assistent Nutzer deaktivieren darf.
Wann Sie es NICHT einsetzen
Lassen Sie das Gateway weg — nicht den Server — wenn Ihr Client Tools bereits pro Connector einschränkt und Sie den Nutzern vertrauen. Claude Code adressiert MCP-Tools als mcp__<server>__<tool> und beachtet permissions.deny in der settings.json. Das Bundle liefert claude-code-permissions.example.json, dieselbe Policy in dieser Form, erzeugt aus derselben policy.py. Das kostet keine Infrastruktur und ist der richtige erste Schritt. Zum Gateway greifen Sie erst, wenn Sie Redaktion auf den Antworten, ein zentrales Audit-Log oder ein an konkrete Argumente gebundenes Freigabe-Token brauchen — drei Dinge, die eine clientseitige Deny-List nicht liefert.
Lassen Sie den ganzen Workflow weg, wenn Ihr Workable-Konto neben der Einstellung auch das führende HR-System ist. Der Server von Workable deckt Mitarbeitende, Abwesenheiten, Zeiterfassung und den kompletten Performance-Review-Zyklus über denselben Endpoint ab wie Kandidaten. Ein an dieses Konto angeschlossener Assistent erreicht Arbeitsverträge über get_employee_documents und Abwesenheitsdaten über get_timeoff_balances, sofern ihn nichts stoppt. Wenn diese Entscheidung noch niemandem gehört, lassen Sie zuerst die KI-Richtlinie fürs Recruiting freigeben.
Und lassen Sie es, wenn ein einzelner Recruiter das ganze Team ist. Der gehostete Connector allein reicht in dieser Größenordnung; Installation und Policy-Review des Gateways sind rund ein Arbeitstag für eine Governance, nach der noch niemand fragt.
Einrichtung
Die vollständige Anleitung steht in apps/web/public/artifacts/mcp-server-workable-recruiting/README.md. Die Kurzfassung: pip install -e ., WORKABLE_ACCOUNT auf Ihre Workable-Subdomain setzen, das Gateway mit absolutem Pfad registrieren und beim ersten Aufruf im Browser autorisieren. Der Server von Workable veröffentlicht Authorization-Server-Metadaten nach RFC 8414 und akzeptiert dynamische Client-Registrierung nach RFC 7591 — es gibt also keine Client-ID von Hand und keinen API-Key zu rotieren.
Der Schritt, auf den es wirklich ankommt, kommt davor: entscheiden, als welches Workable-Mitglied Sie autorisieren. Jede MCP-Session erbt Rolle und Job-Zuweisungen des angemeldeten Nutzers — Workables eigene Formulierung lautet, die KI könne nur Daten lesen und bearbeiten, die der Nutzer ohnehin sehen darf. Das klingt nach einem Rechtemodell, bis Sie sehen, wer das zuerst installiert. Recruiting-Ops-Leads sind Admins. Autorisieren Sie mit Ihrem eigenen Konto, bekommt das Gateway Admin-Scope und die Allowlist ist die einzige verbleibende Mauer. Legen Sie stattdessen ein eigenes Workable-Mitglied mit engem Permission Set an; get_permission_sets listet die in Ihrem Konto definierten auf.
Was Sie zurückhalten
src/workable_gateway/policy.py sortiert alle 94 Tools in drei Stufen plus eine Redaktionsliste. Die Stufenfunktion verweigert per Default, deshalb wären die 37 Tools, die Workable in einem einzigen Release am 2026-07-20 hinzugefügt hat, dunkel geblieben, bis ein Mensch sie einordnet — genau das Verhalten, das Sie von einer Oberfläche wollen, die in neun Wochen um 65% gewachsen ist.
48 werden rundheraus abgelehnt, in sechs Gruppen mit je einer Begründung. Die vier Tools zur Mitgliederverwaltung fliegen, weil ein Agent, der ein Permission Set vergeben kann, in der nächsten Session seine eigene Reichweite erweitert. Die vier Abteilungs-Tools fliegen, weil merge_department keine Umkehrung hat und Recruiting-Reports nach Abteilung geschnitten werden — eine falsche Zusammenführung schreibt die Funnel-Historie ohne Fehlermeldung um. Die fünf Freigabe-Tools — Angebote, Requisitions, Abwesenheiten — fliegen, weil eine Freigabe ein Autoritätsakt einer namentlich benannten Person ist und die Delegation den Beleg dafür löscht, dass ein Mensch entschieden hat. Die sechs Zeiterfassungs-Tools fliegen, weil sie an der Lohnabrechnung hängen und bulk_create_time_entries aus einer schlechten Schlussfolgerung einen Massen-Lohnfehler macht. Die fünfzehn Performance-Review-Tools fliegen, weil submit_review endgültig ist; Workables Dokumentation hält fest, dass ein zweiter Submit fehlschlägt — ein Agent, der einen abgelaufenen Aufruf wiederholt, ist damit exakt das Risiko. Die vierzehn HRIS-Lesezugriffe fliegen, weil Mitarbeitendendokumente Verträge, Vergütungsschreiben und Visa- oder Gesundheitsunterlagen enthalten.
13 hinter einem Freigabe-Gate — die Schreibzugriffe auf Kandidaten und Requisitions, von move_candidate und disqualify_candidate bis create_requisition. Ein Aufruf ohne _gateway_confirm liefert einen Dry Run statt eines Schreibvorgangs. Das _gateway_token aus diesem Dry Run ist ein Hash aus Tool-Name plus den exakten Argumenten — eine Freigabe für “Kandidat 41 auf Onsite” lässt sich also nicht als “Kandidat 88 auf Angebot” wiederverwenden.
33 werden direkt durchgereicht — 32 Lesezugriffe plus add_comment, der einzige Schreibvorgang, der additiv, zurechenbar und in der Workable-Oberfläche entfernbar ist. Darüber hinaus definiert server.py drei eigene Tools: workable_policy_report, damit ein abgelehnter Aufruf “das ist gesperrt, machen Sie es in Workable” ergibt statt einer Wiederholungsschleife; workable_pipeline_snapshot für Stage-Zählungen und liegengebliebene Kandidaten in einem einzigen paginierten Durchlauf; und workable_stage_move_review, das die aktuelle Stage des Kandidaten auflöst, damit der Recruiter ein Diff freigibt und keine Anfrage.
Technische Entscheidungen
Konto festnageln statt das Modell wählen lassen. Jedes Workable-Tool außer get_accounts nimmt eine account-Subdomain entgegen, und ein Nutzer mit Zugriff auf zwei Konten — eine produktive Marke und eine zweite, oder eine Sandbox — bekommt selbstsichere, richtig aussehende Antworten aus dem falschen Tenant. Das Gateway injiziert WORKABLE_ACCOUNT in jeden durchgereichten Aufruf und lehnt jeden ab, in dem das Modell etwas anderes gesetzt hat. Zwei Konten heißen zwei Gateway-Prozesse.
Token-Bucket statt Retry auf 429. Workables OAuth-2.0-Bucket liegt bei 50 Requests pro 10 Sekunden und liefert darüber HTTP 429 mit X-Rate-Limit-Reset. “Zeig mir alle Kandidaten über alle offenen Stellen” fächert sich in get_jobs plus ein paginiertes get_candidates je Req auf und räumt das in etwa zwei Sekunden ab — ein Assistent, der es erneut versucht, läuft direkt in dieselbe Mauer. WORKABLE_RATE_PER_SEC steht per Default auf 4/s, unter der Dauerrate von 5/s, und lässt Luft für alles andere im Tenant, das denselben Token hält.
Ein Durchlauf, kein Aufruf pro Stage.workable_pipeline_snapshot paginiert Kandidaten einmal und zählt die Stages aus den Zeilen, gedeckelt durch WORKABLE_PAGE_CAP (5 Seiten, 500 Kandidaten). Die Kosten sind flach, ob die Stelle 4 oder 14 Stages hat, und die Antwort setzt page_cap_reached, damit das Modell eine Teilzählung als Teilzählung meldet.
Redaktion auf der Antwort, nicht nur auf der Anfrage.search_employees zu sperren hindert get_candidate nicht daran, ein Selbstauskunftsfeld zurückzugeben, das Ihr Konto für EEO-Reporting erhebt. policy.REDACT_FIELDS leert Felder rekursiv nach Schlüsselnamen, weil Workable die Kandidatendetails verschachtelt und die Zeilen der Detailsuche unter eigenen Schlüsseln liefert.
Was es tatsächlich kostet
Der Server kostet $0 — sowohl die Launch- als auch die Erweiterungsmitteilung von Workable nennen ihn ohne Aufpreis in allen Abo-Plänen enthalten, wobei die drei Advanced-Search-Tools auf Premier+ und Enterprise beschränkt sind. Das ist die interessante Zahl, denn Workable rechnet seine eigene Produkt-KI in Credits ab: die heute veröffentlichten Pakete sind 5.000 Credits für $600, 10.000 für $1.000 und 50.000 für $4.750, also $0,095 bis $0,12 je Credit. Die KI von Workable zu fragen verbrennt Credits. Claude über den MCP-Server zu fragen verbrennt Anthropic-Tokens und null Workable-Credits. Für Teams, die ohnehin Claude-Plätze zahlen, ist das Verschieben der Recruiting-Fragen über diese Linie eine echte Verlagerung, kein Nullsummenspiel.
Dagegen stehen: rund 90 Minuten für Gateway-Installation und die Erstprüfungen, und ein Policy-Review, das eher bei drei Stunden landet, weil jemand beteiligt ist, dem die Entscheidung über HR-Daten gehört. Der direkte Connector allein ist ein Befehl und etwa zehn Minuten.
Fehlermodi
Der Assistent wiederholt einen abgelehnten Schreibvorgang, bis eine Formulierung durchkommt. Absicherung: workable_policy_report existiert, damit das Modell die Stufe benennt und aufhört, und jede Ablehnungsmeldung verweist auf die Workable-Oberfläche, statt ein Ersatz-Tool vorzuschlagen. Testen Sie es — Schritt 3 im README lässt den Assistenten ein Mitglied deaktivieren und erwartet eine Ablehnung, keinen Versuch.
Eine alte Freigabe wird gegen andere Argumente wiederverwendet. Absicherung: das _gateway_token hasht die Argumente, nicht nur den Tool-Namen. Wird die Kandidaten-ID nach dem Dry Run geändert, verfällt es und erzwingt eine neue Freigabe.
Die Redaktion übersieht ein eigenes Feld. Die Feldliste in policy.py ist generisch, und Selbstauskunftsattribute sind kontospezifisch. Absicherung: Punkt 1 der TODO-Liste im README ist, Ihre echten Attributschlüssel mit get_account_custom_attributes und get_candidate_detailed_fields zu ziehen, bevor das ein Produktivkonto berührt. Bis dahin gilt die Redaktion als ungetestet.
Lebensläufe und Notizen erreichen einen Dritten.get_candidate_files liegt in der ALLOW-Stufe, weil Lebensläufe zu lesen die Arbeit ist. Damit laufen DSGVO- und CCPA-Daten über Anthropic. Absicherung: die Freigabe der KI-Richtlinie und ein Verzeichnis von Verarbeitungstätigkeiten, das den Datenfluss benennt — bevor der Connector live geht, nicht nachdem jemand fragt.
Die Alternative, die man nennen sollte
Der naheliegende Vergleich ist das Muster des Greenhouse-MCP-Workflows, wo das Bundle der Server ist, weil der Anbieter keinen betreibt. Das ist hier nicht der Handel. Einen eigenen Server über Workables REST-API zu bauen heißt, 94 Endpoints nachzubauen und den OAuth-Flow selbst zu betreiben, um gegen etwas Kostenloses aus erster Hand anzutreten — tun Sie es nicht.
Abzuwägen ist ein Broker. Composio und Zapier listen beide gehostete Workable-MCP-Endpoints, und beide setzen einen zweiten Anbieter in den Pfad, der Ihr OAuth-Token hält, zu eigenen Preisen pro Task oder pro Platz. Nehmen Sie einen davon nur, wenn Sie ohnehin darauf standardisiert sind. Sonst lautet das Ranking: gehosteter Workable-Server plus clientseitige Deny-Rules für die meisten Teams, gehosteter Server plus dieses Gateway, wenn die Allowlist an einer Stelle greifen muss, die Recruiter nicht bearbeiten können. Zum Hintergrund, wo diese Linie verläuft, siehe MCP-Schreibzugriff und wann man ihn gewährt und MCP-Server erklärt.
# mcp-server-workable-recruiting
A least-privilege MCP gateway that sits between Claude and Workable's hosted MCP server. Workable's server exposes 94 tools; this one forwards 33 of them, puts 13 more behind a two-phase human approval, and refuses the remaining 48 outright. It also pins the Workable account, caps the call rate, and strips EEO and compensation fields out of every response before they reach model context.
> **STATUS: scaffold — not runtime-tested.** The code follows the official `mcp` Python SDK conventions, and the endpoint, transport, OAuth discovery behaviour, tool names, and the `account` parameter rule all track Workable's published MCP documentation (`workable.readme.io/reference/workable-mcp-server`) as of 2026-08-23. It has not been executed against a live Workable account. Response field names in particular are account-specific. Verify against your own account before trusting the redaction list.
## Read this first: you might not need this
Workable hosts the server itself at `https://mcp.workable.com/mcp`. It is included at no added cost on every Workable subscription plan, it authenticates over OAuth2 with no key to store or rotate, and every session is scoped to the signed-in user's own role and job assignments. Connecting to it directly takes one command:
```bash
claude mcp add workable --transport http https://mcp.workable.com/mcp
```
If your Claude client can restrict tools per connector — Claude Code can, through `permissions.deny` in `settings.json` — do that instead of running this gateway. `claude-code-permissions.example.json` in this bundle is the same policy expressed that way, generated from the same source file, and it costs zero infrastructure.
Run this gateway when at least one of these is true:
- **The allowlist has to hold centrally, not per laptop.** A client-side settings file is enforced by each recruiter's client. A gateway is enforced once, by you, and a recruiter who edits their own `settings.json` does not widen it.
- **You need responses redacted, not just tools blocked.** Blocking `search_employees` does not stop `get_candidate` returning a self-identification field your account happens to collect. Only a response-side filter does.
- **You need your own audit log.** The gateway logs every forwarded call to your infrastructure, including the ones it refused.
- **You need two-phase approval on writes, not a client-side prompt.** A confirmation dialog depends on a human reading it. A token bound to the exact arguments does not.
## Why 48 tools are refused
Workable launched the server on 2026-05-13 with 38 tools and expanded it to 94 on 2026-07-20. The July release added read *and* write access across performance reviews, account and permissions management, and candidate profile updates. That is a wide grant for an assistant that answers pipeline questions, and the hosted server's own scoping does not narrow it — it inherits whatever the signed-in human can do. Recruiting-ops leads, who install this first, are usually admins.
The refusals are grouped in `src/workable_gateway/policy.py`, one set per rationale:
| Group | Tools | Why |
|---|---|---|
| `DENY_IDENTITY` | 4 | `invite_member`, `update_member`, `enable_member`, `delete_member`. An agent that can grant a permission set can widen its own reach on the next session. |
| `DENY_ORG_STRUCTURE` | 4 | `merge_department` has no inverse, and recruiting reports are cut by department. A bad merge rewrites funnel history silently. |
| `DENY_APPROVALS` | 5 | Offer, requisition, and time-off approvals are acts of authority by a named person. Delegating them erases the evidence that a person decided. |
| `DENY_TIME_TRACKING` | 6 | Payroll-adjacent. `bulk_create_time_entries` turns one bad inference into a bulk pay error. |
| `DENY_PERFORMANCE` | 15 | `submit_review` is final — Workable's docs note a second submit fails — and `sign_review` is an attestation. The reads go with them: review content is manager-confidential and has no recruiting use. |
| `DENY_HRIS_READS` | 14 | Employee documents hold contracts, comp letters, and visa or medical paperwork. Time-off records are absence data. |
The tier function is default-deny. Workable added 37 tools in a single release; anything that appears upstream after this file was written stays dark until a human classifies it.
## What it exposes
**33 forwarded directly** — 32 reads plus `add_comment`, the one write that is additive, attributable, and removable in the Workable UI. The reads cover jobs (9), candidate records and activity (6), offers and requisitions (3), members and permission sets (2), pipeline and account config (3), org context (2), advanced candidate search (3), and remaining context (3), plus `get_accounts`.
**13 behind the approval gate** — the candidate and requisition writes: `move_candidate`, `disqualify_candidate`, `revert_disqualification`, `relocate_candidate`, `copy_candidate`, `create_candidate`, `create_talent_pool_candidate`, `update_candidate`, `update_candidate_tags`, `upsert_candidate_rating`, `add_review`, `create_requisition`, `update_requisition`. Calling one without `_gateway_confirm` returns a dry run. The `_gateway_token` in that dry run is a hash of the tool name plus the exact arguments, so an approval for "move candidate 41 to Onsite" cannot be replayed as "move candidate 88 to Offer".
**3 gateway-native tools**, defined in `src/workable_gateway/server.py`:
- `workable_policy_report(include_withheld?)` — what this assistant can and cannot reach, with the tier for each tool and the calls used so far against the process ceiling. Point the model at this when a call is refused, so the recruiter gets "that is blocked, do it in Workable" instead of a retry loop.
- `workable_pipeline_snapshot(shortcode, stalled_after_days=14)` — job title, stage list, candidate count per stage, and the candidates with no activity for N days. One paged sweep capped at `WORKABLE_PAGE_CAP` calls, rather than one call per stage: the cost is the same whether the job has 4 stages or 14.
- `workable_stage_move_review(candidate_id, target_stage, reason, confirm?, dry_run_token?)` — the richer path for the most common write. The dry run resolves the candidate's current stage so the recruiter approves a diff, not a request. On confirm it writes the reason to the activity feed with `add_comment` first, then calls `move_candidate`, so the audit trail exists even if the move fails.
## Setup
### 1. Install
```bash
cd mcp-server-workable-recruiting
python -m venv .venv
source .venv/bin/activate # .venv\Scripts\activate on Windows
pip install -e .
```
### 2. Choose the Workable identity you connect as
Do this before the first OAuth run, because the browser sign-in decides the ceiling for everything below. The hosted server grants the authenticated user's permissions, so signing in as yourself gives the gateway your access. Create a dedicated Workable member for it and assign a narrowed permission set — `get_permission_sets` lists what your account has defined. The gateway's allowlist is then a second wall, not the only one.
### 3. Set the environment variables
**`WORKABLE_ACCOUNT`** (required). Your Workable subdomain. Every tool except `get_accounts` takes an `account` parameter, and this is that value. Find it in the host of your Workable URL — for `https://acme.workable.com` it is `acme` — or run the hosted server's `get_accounts` once and read the subdomain it returns. The gateway injects this on every forwarded call and rejects any call where the model supplied a different one.
**`WORKABLE_MCP_URL`** (default `https://mcp.workable.com/mcp`). Only change this if Workable publishes a regional endpoint.
**`WORKABLE_TOKEN_PATH`** (default `~/.workable-gateway.json`). Where the OAuth client registration and refresh token are written, mode 0600. On a shared host, put it somewhere only the gateway's service user can read — this file is a live credential.
**`WORKABLE_OAUTH_CALLBACK_PORT`** (default `8765`). The localhost port the one-shot redirect listener binds during authorization. Change it if something else owns 8765. If you see a redirect-URI mismatch on first connect, this is the value that has to agree with what got registered.
**`WORKABLE_RATE_PER_SEC`** (default `4`). Workable's OAuth 2.0 rate bucket is 50 requests per 10 seconds — 5/s sustained — and returns HTTP 429 with `X-Rate-Limit-Reset` above it. The default leaves headroom for whatever else in your tenant holds the same token.
**`WORKABLE_MAX_CALLS_PER_PROCESS`** (default `400`). Hard ceiling per gateway process. A single chat turn that needs hundreds of upstream calls is a report, not a conversation; the ceiling makes that visible instead of letting it drain the rate budget.
**`WORKABLE_PAGE_CAP`** (default `5`). Maximum pages `workable_pipeline_snapshot` drains, at 100 candidates per page. 500 candidates covers a normal req; the response sets `page_cap_reached` when it does not, so the model can say so rather than quietly reporting a partial count.
**`WORKABLE_LOG_LEVEL`** (default `INFO`).
### 4. Register the gateway with your client
Claude Code:
```bash
claude mcp add workable-gateway -- /absolute/path/to/.venv/bin/workable-gateway
```
Claude Desktop — `claude_desktop_config.json`:
```json
{
"mcpServers": {
"workable-gateway": {
"command": "/absolute/path/to/.venv/bin/workable-gateway",
"env": {
"WORKABLE_ACCOUNT": "acme",
"WORKABLE_RATE_PER_SEC": "4",
"WORKABLE_MAX_CALLS_PER_PROCESS": "400"
}
}
}
}
```
Use absolute paths. Claude Desktop does not run a login shell, so `workable-gateway` on your `PATH` is not on its `PATH`.
### 5. Authorize
The first tool call opens a browser to Workable's authorization page. Sign in as the identity from step 2 and approve. The registration and refresh token land in `WORKABLE_TOKEN_PATH`; later runs do not prompt. Workable's server advertises RFC 8414 authorization-server metadata and accepts RFC 7591 dynamic client registration, so there is no client ID to provision by hand.
## First-run verification
Run these four in order. Each proves one wall works before you let a recruiter near it.
1. **Policy loads.** Ask: *"Run workable_policy_report with include_withheld."* Expect `upstream_tool_count: 94`, `exposed_count: 46`, `withheld_count: 48`. If `upstream_tool_count` is higher than 94, Workable shipped new tools — they are already dark by default-deny, and classifying them is your next task, not an emergency.
2. **Reads work and the account is pinned.** Ask: *"Search Workable for jobs matching 'engineer'."* You should get results. Then check the log line for the forwarded call and confirm `account` matches `WORKABLE_ACCOUNT`.
3. **The deny wall holds.** Ask: *"Deactivate the Workable member for jane@example.com."* Expect a refusal naming `delete_member` and pointing at the Workable UI — not an attempt, and not a hedge.
4. **The approval gate holds.** Ask: *"Move candidate `<id>` to the Onsite stage because the phone screen went well."* Expect a dry run with the current stage, the target stage, and a `dry_run_token` — and no move. Confirm in Workable that the candidate did not move. Then approve and re-check.
Only step 4 writes anything. Do all four against a test job with a fake candidate first.
## Security model
- **The token is a live Workable credential.** It grants whatever the authorizing member can do — read *and* write. Treat `WORKABLE_TOKEN_PATH` as you would an API key. Revoke by removing the connector from the authorizing member's Workable account.
- **The gateway's allowlist is defence in depth, not the boundary.** The boundary is the permission set on the Workable member you authorized as. Anyone who can reach the gateway's stdio can reach every tool in the ALLOW tier; anyone who can edit `policy.py` can reach all 94. Deploy it where recruiters can use it and cannot edit it.
- **Candidate data reaches Anthropic.** Résumés from `get_candidate_files`, notes, and activity feeds enter model context. EU candidates are GDPR data subjects and California candidates are CCPA data subjects. Get the AI policy signed off before this touches a live account, not after.
- **Redaction is name-based and account-specific.** `policy.REDACT_FIELDS` blanks fields by key name, recursively. If your account collects self-identification under a custom attribute with a different key, it is not covered until you add it. Confirm the real names with `get_account_custom_attributes` and `get_candidate_detailed_fields`.
- **Advanced candidate search is plan-gated.** Workable restricts the Advanced Search tools to Premier+ and Enterprise plans. On lower plans those three tools are in the ALLOW tier but will not appear upstream, which is correct — the gateway advertises the intersection of policy and what Workable actually serves.
## Limits and TODOs
Before this runs against a production account:
1. **Verify the redaction field names.** The list in `policy.py` is generic. Pull your account's real attribute keys and replace it. This is the single highest-value item here.
2. **Add a persistent audit log.** Calls currently go to Python `logging` at INFO. Write them to durable storage with the tool name, tier, arguments hash, the authorizing member, and a timestamp — that record is what makes the deployment defensible to a works council or an auditor.
3. **Handle 429 explicitly.** The token bucket avoids the limit; it does not react to one. Read `X-Rate-Limit-Reset` from the upstream error and back off to it instead of retrying blind.
4. **Confirm the write tools' argument names.** `move_candidate` and `add_comment` are called in `handle_stage_move_review` with the argument shapes in this scaffold. Read the live `inputSchema` from `list_tools` and align.
5. **Reconnect on upstream drop.** `Upstream.connect` runs once at startup. A dropped session currently kills the process rather than re-authorizing.
6. **Decide the HRIS profile separately.** `DENY_HRIS_READS` is right for recruiters and wrong for People Ops. Build a second `Policy` instance and a second gateway process rather than widening this one.
7. **Pin the dependency versions.** `pyproject.toml` uses lower bounds. Lock them before deploying.
## Files
```
mcp-server-workable-recruiting/
├── README.md
├── pyproject.toml
├── claude-code-permissions.example.json # same policy, no gateway
└── src/workable_gateway/
├── __init__.py
├── policy.py # the tiers — edit this file
└── server.py # stdio server, upstream client, gates
```
"""Tool policy for the Workable MCP gateway.
Workable's hosted server exposed 94 tools as of 2026-07-20. This module decides
which of them reach the model, and which of the survivors need a human to say yes
before they run.
Three tiers plus a redaction list:
DENY never forwarded, never listed. Identity, org structure, approvals,
payroll records, HRIS reads, and the irreversible review writes.
CONFIRM forwarded only after a dry-run the human approved. See
server.workable_stage_move_review.
ALLOW forwarded as-is. Reads plus one additive write (add_comment).
REDACT applies to every forwarded response: named fields are stripped
before the payload reaches model context.
Edit RECRUITER_PROFILE for your own org. The assignments below are the
recruiter / recruiting-ops profile: 33 of 94 tools exposed, 61 withheld.
"""
from __future__ import annotations
from dataclasses import dataclass, field
from enum import Enum
class Tier(str, Enum):
ALLOW = "allow"
CONFIRM = "confirm"
DENY = "deny"
# ---------------------------------------------------------------------------
# DENY - the surface an assistant never gets, on any profile.
# ---------------------------------------------------------------------------
# Members: an agent that can grant a permission set can widen its own reach on
# the next session, because the hosted server inherits the signed-in user's role.
DENY_IDENTITY = {
"invite_member",
"update_member",
"enable_member",
"delete_member",
}
# Departments: merge_department has no inverse. Recruiting reporting is cut by
# department, so a bad merge silently rewrites every historical funnel report.
DENY_ORG_STRUCTURE = {
"create_department",
"update_department",
"delete_department",
"merge_department",
}
# Approvals are an act of authority by a named human. Delegating them to an
# assistant destroys the only evidence that a person made the decision.
DENY_APPROVALS = {
"approve_offer",
"reject_offer",
"approve_requisition",
"reject_requisition",
"update_timeoff_approval",
}
# Payroll-adjacent. A wrong or duplicated time entry becomes a pay error, and
# bulk_create_time_entries makes that a bulk pay error.
DENY_TIME_TRACKING = {
"list_time_entries",
"create_time_entry",
"clock_in",
"clock_out",
"bulk_create_time_entries",
"update_time_entry",
}
# submit_review is final - Workable's docs note a second submit fails - and
# sign_review is an attestation. An agent retrying a timed-out call must not be
# able to reach either. The reads go too: review content is manager-confidential
# and has no recruiting use.
DENY_PERFORMANCE = {
"get_review_cycle_templates",
"get_review_cycle_template",
"create_review_cycle_template",
"get_review_cycles",
"get_review_cycle",
"list_review_tasks",
"get_review_form",
"update_review_form",
"mark_review_task_ready",
"get_review",
"submit_review",
"share_review",
"sign_review",
"get_review_aggregate",
"list_review_cycle_answers",
}
# HR reads that are not recruiting reads. Employee documents hold contracts, comp
# letters, and visa or medical paperwork. Time-off records are absence data. The
# profile-update feed is a change log over personal data.
DENY_HRIS_READS = {
"get_employees",
"get_employee",
"get_employee_documents",
"get_employee_fields",
"get_employee_filter_options",
"search_employees",
"get_profile_update_fields",
"get_profile_update_filter_options",
"search_profile_updates",
"get_timeoff_requests",
"get_timeoff_balances",
"get_timeoff_categories",
"create_timeoff_request",
"get_work_schedules",
}
DENY: set[str] = (
DENY_IDENTITY
| DENY_ORG_STRUCTURE
| DENY_APPROVALS
| DENY_TIME_TRACKING
| DENY_PERFORMANCE
| DENY_HRIS_READS
)
# ---------------------------------------------------------------------------
# CONFIRM - reachable, but each call needs an explicit human yes first.
# ---------------------------------------------------------------------------
CONFIRM: set[str] = {
"move_candidate",
"disqualify_candidate",
"revert_disqualification",
"relocate_candidate",
"copy_candidate",
"create_candidate",
"create_talent_pool_candidate",
"update_candidate",
"update_candidate_tags",
"upsert_candidate_rating",
"add_review",
"create_requisition",
"update_requisition",
}
# ---------------------------------------------------------------------------
# ALLOW - the recruiter profile. 33 tools: 32 reads plus add_comment.
# ---------------------------------------------------------------------------
ALLOW: set[str] = {
# Accounts. get_accounts is the only tool that takes no account parameter.
"get_accounts",
# Jobs (9)
"get_jobs",
"search_jobs",
"get_job",
"get_job_activities",
"get_job_application_form",
"get_job_custom_attributes",
"get_job_members",
"get_job_recruiters",
"get_job_stages",
# Candidate reads (6)
"get_candidates",
"get_candidate",
"get_candidate_activities",
"get_candidate_activity",
"get_candidate_offer",
"get_candidate_files",
# Offers, requisitions, members - read only (5)
"get_offer",
"get_requisitions",
"get_requisition",
"get_members",
"get_permission_sets",
# Pipeline and account config (3)
"get_stages",
"get_disqualification_reasons",
"get_account_custom_attributes",
# Org context (2)
"get_orgchart",
"get_departments",
# Advanced search over candidates - Premier+ and Enterprise plans only (3)
"get_candidate_detailed_fields",
"get_candidate_detailed_filter_options",
"search_candidates_detailed",
# Remaining context reads (3)
"get_legal_entities",
"get_events",
"get_event",
# The one additive write. Appends to the candidate activity feed: visible to
# the recruiter, attributable, and removable in the Workable UI.
"add_comment",
}
# ---------------------------------------------------------------------------
# REDACT - response fields stripped before the payload enters model context.
# ---------------------------------------------------------------------------
#
# Workable candidate records can carry self-identification data collected for
# EEO/OFCCP reporting. That data has a lawful purpose and a hiring conversation
# is not it. Field names vary by account: confirm yours with
# get_account_custom_attributes and get_candidate_detailed_fields, then edit.
REDACT_FIELDS: set[str] = {
"ethnicity",
"race",
"gender",
"veteran_status",
"disability_status",
"date_of_birth",
"national_id",
"social_security_number",
"salary",
"current_salary",
"salary_expectations",
}
@dataclass(frozen=True)
class Policy:
"""Resolved policy for one gateway process."""
allow: set[str] = field(default_factory=lambda: set(ALLOW))
confirm: set[str] = field(default_factory=lambda: set(CONFIRM))
deny: set[str] = field(default_factory=lambda: set(DENY))
redact_fields: set[str] = field(default_factory=lambda: set(REDACT_FIELDS))
def tier(self, tool_name: str) -> Tier:
if tool_name in self.deny:
return Tier.DENY
if tool_name in self.confirm:
return Tier.CONFIRM
if tool_name in self.allow:
return Tier.ALLOW
# Default-deny. Workable added 37 tools in a single release on
# 2026-07-20; anything that appears upstream after this file was written
# stays dark until a human classifies it.
return Tier.DENY
def is_exposed(self, tool_name: str) -> bool:
return self.tier(tool_name) in (Tier.ALLOW, Tier.CONFIRM)
RECRUITER_PROFILE = Policy()
"""Least-privilege MCP gateway in front of Workable's hosted MCP server.
Claude talks to this process over stdio. This process talks to
https://mcp.workable.com/mcp over Streamable HTTP with OAuth. Between the two it
applies four rules:
1. Tool allowlist. Only tools that policy.RECRUITER_PROFILE marks ALLOW or
CONFIRM are advertised or forwarded. Everything else - including any tool
Workable ships after this file was written - is dark.
2. Account pinning. Every upstream tool except get_accounts takes an `account`
subdomain. The model never chooses it; WORKABLE_ACCOUNT does.
3. Rate budget. A token bucket at WORKABLE_RATE_PER_SEC plus a per-process call
ceiling, so one broad question cannot burn the tenant's API budget.
4. Field redaction. policy.REDACT_FIELDS are stripped from every response
before the payload enters model context.
On top of the forwarded set it defines three tools of its own:
workable_policy_report, workable_pipeline_snapshot, workable_stage_move_review.
STATUS: scaffold. Not runtime-tested against a live Workable account. See
README.md, "Limits and TODOs".
"""
from __future__ import annotations
import asyncio
import json
import logging
import os
import time
from contextlib import AsyncExitStack
from typing import Any
import mcp.types as types
from mcp import ClientSession
from mcp.client.auth import OAuthClientProvider, TokenStorage
from mcp.client.streamable_http import streamablehttp_client
from mcp.server import NotificationOptions, Server
from mcp.server.models import InitializationOptions
from mcp.server.stdio import stdio_server
from mcp.shared.auth import OAuthClientInformationFull, OAuthClientMetadata, OAuthToken
from .policy import RECRUITER_PROFILE, Policy, Tier
LOG = logging.getLogger("workable_gateway")
UPSTREAM_URL = os.environ.get("WORKABLE_MCP_URL", "https://mcp.workable.com/mcp")
ACCOUNT = os.environ.get("WORKABLE_ACCOUNT", "")
TOKEN_PATH = os.environ.get("WORKABLE_TOKEN_PATH", os.path.expanduser("~/.workable-gateway.json"))
CALLBACK_PORT = int(os.environ.get("WORKABLE_OAUTH_CALLBACK_PORT", "8765"))
RATE_PER_SEC = float(os.environ.get("WORKABLE_RATE_PER_SEC", "4"))
MAX_CALLS = int(os.environ.get("WORKABLE_MAX_CALLS_PER_PROCESS", "400"))
PAGE_CAP = int(os.environ.get("WORKABLE_PAGE_CAP", "5"))
POLICY: Policy = RECRUITER_PROFILE
# ---------------------------------------------------------------------------
# Rate budget
# ---------------------------------------------------------------------------
class TokenBucket:
"""Workable's OAuth bucket is 50 requests per 10 seconds (5/s sustained).
The gateway runs at 4/s so a burst from a fan-out question leaves headroom for
whatever else in the tenant is holding the same token. Exceeding the bucket
upstream returns HTTP 429, and an assistant that retries walks straight back
into it - hence a hard ceiling, not just a delay.
"""
def __init__(self, rate_per_sec: float, capacity: float | None = None) -> None:
self.rate = rate_per_sec
self.capacity = capacity if capacity is not None else max(rate_per_sec, 1.0)
self.tokens = self.capacity
self.updated = time.monotonic()
self._lock = asyncio.Lock()
async def take(self) -> None:
async with self._lock:
while True:
now = time.monotonic()
self.tokens = min(self.capacity, self.tokens + (now - self.updated) * self.rate)
self.updated = now
if self.tokens >= 1.0:
self.tokens -= 1.0
return
await asyncio.sleep((1.0 - self.tokens) / self.rate)
# ---------------------------------------------------------------------------
# OAuth token storage
# ---------------------------------------------------------------------------
class FileTokenStorage(TokenStorage):
"""Persists the OAuth client registration and tokens to one 0600 file.
Workable's server advertises RFC 8414 metadata and accepts RFC 7591 dynamic
client registration, so there is no client ID to provision by hand. The first
run opens a browser; later runs reuse what lands here.
"""
def __init__(self, path: str) -> None:
self.path = path
def _read(self) -> dict[str, Any]:
if not os.path.exists(self.path):
return {}
with open(self.path, encoding="utf-8") as handle:
return json.load(handle)
def _write(self, data: dict[str, Any]) -> None:
with open(self.path, "w", encoding="utf-8") as handle:
json.dump(data, handle)
os.chmod(self.path, 0o600)
async def get_tokens(self) -> OAuthToken | None:
raw = self._read().get("tokens")
return OAuthToken.model_validate(raw) if raw else None
async def set_tokens(self, tokens: OAuthToken) -> None:
data = self._read()
data["tokens"] = tokens.model_dump(mode="json", exclude_none=True)
self._write(data)
async def get_client_info(self) -> OAuthClientInformationFull | None:
raw = self._read().get("client")
return OAuthClientInformationFull.model_validate(raw) if raw else None
async def set_client_info(self, info: OAuthClientInformationFull) -> None:
data = self._read()
data["client"] = info.model_dump(mode="json", exclude_none=True)
self._write(data)
# ---------------------------------------------------------------------------
# Redaction
# ---------------------------------------------------------------------------
def redact(value: Any, fields: set[str]) -> Any:
"""Walk a decoded JSON payload and blank every key named in `fields`.
Recursive rather than top-level: Workable nests candidate detail under
`candidate`, and detailed search returns rows under `results`, so a shallow
pass would miss most of what matters.
"""
if isinstance(value, dict):
out: dict[str, Any] = {}
for key, item in value.items():
if key.lower() in fields:
out[key] = "[redacted by gateway policy]"
else:
out[key] = redact(item, fields)
return out
if isinstance(value, list):
return [redact(item, fields) for item in value]
return value
def redact_content(blocks: list[types.ContentBlock], fields: set[str]) -> list[types.ContentBlock]:
out: list[types.ContentBlock] = []
for block in blocks:
if isinstance(block, types.TextContent):
try:
parsed = json.loads(block.text)
except (json.JSONDecodeError, TypeError):
out.append(block)
continue
out.append(
types.TextContent(type="text", text=json.dumps(redact(parsed, fields), indent=2))
)
else:
out.append(block)
return out
# ---------------------------------------------------------------------------
# Upstream client
# ---------------------------------------------------------------------------
class Upstream:
"""One long-lived authenticated session against mcp.workable.com."""
def __init__(self) -> None:
self.session: ClientSession | None = None
self.tools: dict[str, types.Tool] = {}
self.bucket = TokenBucket(RATE_PER_SEC)
self.calls = 0
self._stack = AsyncExitStack()
async def connect(self) -> None:
auth = OAuthClientProvider(
server_url=UPSTREAM_URL,
client_metadata=OAuthClientMetadata(
client_name="ooligo Workable gateway",
redirect_uris=[f"http://localhost:{CALLBACK_PORT}/callback"],
grant_types=["authorization_code", "refresh_token"],
response_types=["code"],
),
storage=FileTokenStorage(TOKEN_PATH),
redirect_handler=_open_browser,
callback_handler=_await_callback,
)
read, write, _ = await self._stack.enter_async_context(
streamablehttp_client(UPSTREAM_URL, auth=auth)
)
self.session = await self._stack.enter_async_context(ClientSession(read, write))
await self.session.initialize()
listed = await self.session.list_tools()
self.tools = {tool.name: tool for tool in listed.tools}
exposed = [name for name in self.tools if POLICY.is_exposed(name)]
LOG.info(
"upstream advertises %d tools; policy exposes %d, withholds %d",
len(self.tools),
len(exposed),
len(self.tools) - len(exposed),
)
async def close(self) -> None:
await self._stack.aclose()
async def call(self, name: str, arguments: dict[str, Any]) -> types.CallToolResult:
if self.session is None:
raise RuntimeError("upstream session not connected")
if self.calls >= MAX_CALLS:
raise RuntimeError(
f"gateway call ceiling reached ({MAX_CALLS}). Restart the server if this was "
"a legitimate workload, or narrow the question - a single request that needs "
"hundreds of upstream calls is usually a report, not a chat turn."
)
# Account pinning. Rule 2: the model does not get to pick the tenant.
if name != "get_accounts":
supplied = arguments.get("account")
if supplied and supplied != ACCOUNT:
raise ValueError(
f"tool {name} was called with account={supplied!r}; this gateway is pinned "
f"to {ACCOUNT!r}. Run a second gateway process for the other account."
)
arguments = {**arguments, "account": ACCOUNT}
await self.bucket.take()
self.calls += 1
return await self.session.call_tool(name, arguments)
async def _open_browser(url: str) -> None:
import webbrowser
LOG.info("opening browser for Workable authorization")
webbrowser.open(url)
async def _await_callback() -> tuple[str, str | None]:
"""Block until the OAuth redirect lands on localhost.
Kept deliberately small: a single-request HTTP listener on CALLBACK_PORT.
Swap for your own handler if the machine already runs something there.
"""
from http.server import BaseHTTPRequestHandler, HTTPServer
from urllib.parse import parse_qs, urlparse
captured: dict[str, str] = {}
class Handler(BaseHTTPRequestHandler):
def do_GET(self) -> None: # noqa: N802 - stdlib naming
params = parse_qs(urlparse(self.path).query)
captured["code"] = params.get("code", [""])[0]
captured["state"] = params.get("state", [""])[0]
self.send_response(200)
self.send_header("Content-Type", "text/plain")
self.end_headers()
self.wfile.write(b"Workable authorization received. Close this tab.")
def log_message(self, *args: Any) -> None:
return
server = HTTPServer(("localhost", CALLBACK_PORT), Handler)
await asyncio.get_running_loop().run_in_executor(None, server.handle_request)
server.server_close()
return captured.get("code", ""), captured.get("state") or None
# ---------------------------------------------------------------------------
# Gateway-native tools
# ---------------------------------------------------------------------------
GATEWAY_TOOLS = [
types.Tool(
name="workable_policy_report",
description=(
"Report which Workable MCP tools this gateway exposes and which it withholds, "
"with the tier for each. Call this when the user asks what the assistant can or "
"cannot do in Workable, or when a tool call was refused."
),
inputSchema={
"type": "object",
"properties": {
"include_withheld": {
"type": "boolean",
"description": "List every withheld tool name, not just the count.",
"default": False,
}
},
"additionalProperties": False,
},
),
types.Tool(
name="workable_pipeline_snapshot",
description=(
"One-call pipeline summary for a job: stage list, candidate count per stage, and "
"the candidates with no activity for N days. Use this instead of chaining get_job, "
"get_job_stages and get_candidates, which costs four or more upstream calls."
),
inputSchema={
"type": "object",
"properties": {
"shortcode": {
"type": "string",
"description": "Workable job shortcode. Get it from search_jobs.",
},
"stalled_after_days": {
"type": "integer",
"description": "Flag candidates with no activity for this many days.",
"default": 14,
"minimum": 1,
"maximum": 365,
},
},
"required": ["shortcode"],
"additionalProperties": False,
},
),
types.Tool(
name="workable_stage_move_review",
description=(
"Two-phase stage move. Called without confirm, it validates the target stage and "
"returns the exact change for a human to approve. Called with confirm=true and the "
"token from that dry run, it performs move_candidate. The only path to a stage "
"change through this gateway."
),
inputSchema={
"type": "object",
"properties": {
"candidate_id": {"type": "string", "description": "Workable candidate id."},
"target_stage": {
"type": "string",
"description": "Exact stage name from get_job_stages.",
},
"reason": {
"type": "string",
"description": "Why the candidate is moving. Written to the activity feed.",
"minLength": 10,
},
"confirm": {
"type": "boolean",
"description": "Set true only after a human approved the dry run.",
"default": False,
},
"dry_run_token": {
"type": "string",
"description": "The token returned by the dry run. Required when confirm is true.",
},
},
"required": ["candidate_id", "target_stage", "reason"],
"additionalProperties": False,
},
),
]
def _text(payload: Any) -> list[types.ContentBlock]:
return [types.TextContent(type="text", text=json.dumps(payload, indent=2, default=str))]
def _first_json(result: types.CallToolResult) -> Any:
for block in result.content:
if isinstance(block, types.TextContent):
try:
return json.loads(block.text)
except (json.JSONDecodeError, TypeError):
continue
return None
async def handle_policy_report(up: Upstream, args: dict[str, Any]) -> list[types.ContentBlock]:
exposed: dict[str, str] = {}
withheld: list[str] = []
for name in sorted(up.tools):
tier = POLICY.tier(name)
if tier is Tier.DENY:
withheld.append(name)
else:
exposed[name] = tier.value
payload: dict[str, Any] = {
"upstream_tool_count": len(up.tools),
"exposed_count": len(exposed),
"withheld_count": len(withheld),
"exposed": exposed,
"redacted_response_fields": sorted(POLICY.redact_fields),
"account": ACCOUNT,
"calls_used_this_process": up.calls,
"call_ceiling": MAX_CALLS,
}
if args.get("include_withheld"):
payload["withheld"] = withheld
return _text(payload)
async def handle_pipeline_snapshot(up: Upstream, args: dict[str, Any]) -> list[types.ContentBlock]:
shortcode = args["shortcode"]
stalled_after = int(args.get("stalled_after_days", 14))
job = _first_json(await up.call("get_job", {"shortcode": shortcode}))
stages = _first_json(await up.call("get_job_stages", {"shortcode": shortcode})) or {}
stage_names = [s.get("name") for s in stages.get("stages", []) if s.get("name")]
# One paged sweep, not one call per stage. Stage counts come from the rows,
# which keeps the cost at PAGE_CAP calls regardless of how many stages exist.
rows: list[dict[str, Any]] = []
since_id: str | None = None
for _ in range(PAGE_CAP):
params: dict[str, Any] = {"shortcode": shortcode, "limit": 100}
if since_id:
params["since_id"] = since_id
page = _first_json(await up.call("get_candidates", params)) or {}
batch = page.get("candidates", [])
rows.extend(batch)
if len(batch) < 100:
break
since_id = batch[-1].get("id")
cutoff = time.time() - stalled_after * 86400
per_stage: dict[str, int] = {name: 0 for name in stage_names}
stalled: list[dict[str, Any]] = []
for row in rows:
stage = row.get("stage") or "unknown"
per_stage[stage] = per_stage.get(stage, 0) + 1
updated = row.get("updated_at") or row.get("created_at")
ts = _parse_ts(updated)
if ts is not None and ts < cutoff:
stalled.append(
{"id": row.get("id"), "name": row.get("name"), "stage": stage, "last_activity": updated}
)
return redact_content(
_text(
{
"job": {
"shortcode": shortcode,
"title": (job or {}).get("title"),
"state": (job or {}).get("state"),
},
"stages": stage_names,
"candidates_scanned": len(rows),
"page_cap_reached": len(rows) >= PAGE_CAP * 100,
"per_stage": per_stage,
"stalled_after_days": stalled_after,
"stalled": stalled[:50],
"stalled_total": len(stalled),
}
),
POLICY.redact_fields,
)
def _parse_ts(value: Any) -> float | None:
if not isinstance(value, str):
return None
from datetime import datetime
try:
return datetime.fromisoformat(value.replace("Z", "+00:00")).timestamp()
except ValueError:
return None
def _dry_run_token(tool: str, arguments: dict[str, Any]) -> str:
"""Bind an approval to the exact call the human saw.
Hashing the arguments, not just the tool name, is the point: an approval for
"move candidate 41 to Onsite" must not authorize "move candidate 88 to
Offer". Any edit to the arguments invalidates the token and forces a fresh
dry run.
"""
import hashlib
payload = json.dumps(
{k: v for k, v in arguments.items() if not k.startswith("_gateway")},
sort_keys=True,
default=str,
)
return hashlib.sha256(f"{tool}|{ACCOUNT}|{payload}".encode()).hexdigest()[:16]
async def handle_stage_move_review(up: Upstream, args: dict[str, Any]) -> list[types.ContentBlock]:
candidate_id = args["candidate_id"]
target_stage = args["target_stage"]
reason = args["reason"]
token = _dry_run_token(
"move_candidate", {"id": candidate_id, "target_stage": target_stage, "reason": reason}
)
if not args.get("confirm"):
current = _first_json(await up.call("get_candidate", {"id": candidate_id})) or {}
candidate = current.get("candidate", current)
return redact_content(
_text(
{
"phase": "dry_run",
"candidate": {
"id": candidate_id,
"name": candidate.get("name"),
"job": (candidate.get("job") or {}).get("title"),
"current_stage": candidate.get("stage"),
},
"target_stage": target_stage,
"reason": reason,
"dry_run_token": token,
"next_step": (
"Show this to the recruiter. If they approve, call again with "
"confirm=true and this dry_run_token. Do not confirm on your own."
),
}
),
POLICY.redact_fields,
)
if args.get("dry_run_token") != token:
raise ValueError(
"dry_run_token does not match this candidate and target stage. Run the dry run "
"again and have a human approve the result before confirming."
)
await up.call("add_comment", {"id": candidate_id, "comment": {"body": f"Stage move: {reason}"}})
moved = await up.call("move_candidate", {"id": candidate_id, "target_stage": target_stage})
return redact_content(
_text({"phase": "committed", "candidate_id": candidate_id, "target_stage": target_stage,
"upstream": _first_json(moved)}),
POLICY.redact_fields,
)
GATEWAY_HANDLERS = {
"workable_policy_report": handle_policy_report,
"workable_pipeline_snapshot": handle_pipeline_snapshot,
"workable_stage_move_review": handle_stage_move_review,
}
# ---------------------------------------------------------------------------
# Server wiring
# ---------------------------------------------------------------------------
CONFIRM_NOTE = (
" GATEWAY POLICY: this tool writes to Workable and needs a human approval. Call it first "
"without _gateway_confirm to get a dry run describing the change, show that to the user, "
"and only after they approve call again with _gateway_confirm=true and the _gateway_token "
"from the dry run. Never approve on the user's behalf."
)
CONFIRM_ARGS = {
"_gateway_confirm": {
"type": "boolean",
"description": "True only after a human approved the dry run.",
"default": False,
},
"_gateway_token": {
"type": "string",
"description": "The _gateway_token returned by the dry run for these exact arguments.",
},
}
def _with_confirm_gate(tool: types.Tool) -> types.Tool:
"""Advertise a CONFIRM-tier tool with its approval parameters attached."""
schema = json.loads(json.dumps(tool.inputSchema))
schema.setdefault("type", "object")
schema.setdefault("properties", {})
schema["properties"].update(CONFIRM_ARGS)
# Upstream schemas can be strict; the gateway adds two properties to them.
schema["additionalProperties"] = True
return types.Tool(
name=tool.name,
description=(tool.description or "") + CONFIRM_NOTE,
inputSchema=schema,
)
def build_server(up: Upstream) -> Server:
server = Server("workable-gateway")
@server.list_tools()
async def list_tools() -> list[types.Tool]:
forwarded: list[types.Tool] = []
for name, tool in sorted(up.tools.items()):
tier = POLICY.tier(name)
if tier is Tier.ALLOW:
forwarded.append(tool)
elif tier is Tier.CONFIRM:
forwarded.append(_with_confirm_gate(tool))
return GATEWAY_TOOLS + forwarded
@server.call_tool()
async def call_tool(name: str, arguments: dict[str, Any]) -> list[types.ContentBlock]:
args = dict(arguments or {})
if name in GATEWAY_HANDLERS:
return await GATEWAY_HANDLERS[name](up, args)
tier = POLICY.tier(name)
if tier is Tier.DENY:
raise ValueError(
f"{name} is withheld by gateway policy. Call workable_policy_report for the "
"list of tools this assistant can reach, and do the rest in the Workable UI."
)
if tier is Tier.CONFIRM:
token = _dry_run_token(name, args)
if not args.pop("_gateway_confirm", False):
args.pop("_gateway_token", None)
return _text(
{
"phase": "dry_run",
"tool": name,
"arguments": args,
"account": ACCOUNT,
"_gateway_token": token,
"next_step": (
"Show this to the user verbatim. If they approve, call the same "
"tool again with identical arguments plus _gateway_confirm=true "
"and this _gateway_token."
),
}
)
if args.pop("_gateway_token", None) != token:
raise ValueError(
f"_gateway_token does not match the arguments passed to {name}. The "
"arguments changed after the dry run, so the approval no longer applies. "
"Run the dry run again and have the user approve the new version."
)
result = await up.call(name, args)
return redact_content(list(result.content), POLICY.redact_fields)
return server
async def run() -> None:
logging.basicConfig(level=os.environ.get("WORKABLE_LOG_LEVEL", "INFO"), stream=None)
if not ACCOUNT:
raise SystemExit(
"WORKABLE_ACCOUNT is required. It is your Workable subdomain - the value "
"get_accounts returns, and the one every other tool takes."
)
up = Upstream()
await up.connect()
try:
server = build_server(up)
async with stdio_server() as (read, write):
await server.run(
read,
write,
InitializationOptions(
server_name="workable-gateway",
server_version="0.1.0",
capabilities=server.get_capabilities(
notification_options=NotificationOptions(),
experimental_capabilities={},
),
),
)
finally:
await up.close()
def main() -> None:
asyncio.run(run())
if __name__ == "__main__":
main()
