Workable héberge son propre serveur Model Context Protocol sur https://mcp.workable.com/mcp : la question de le construire est donc réglée, vous vous connectez, vous n’en écrivez pas un. La question ouverte, c’est lesquels de ses 94 outils l’assistant de vos recruteurs a le droit de toucher. Workable a lancé le serveur le 2026-05-13 avec 38 outils et l’a étendu à 94 le 2026-07-20, et cette extension a ajouté un accès en écriture aux revues de performance, à la gestion des comptes et des permissions, et à la mise à jour des profils candidats. Le bundle d’artefacts dans apps/web/public/artifacts/mcp-server-workable-recruiting/ répond à cette question : une passerelle de moindre privilège (README.md, pyproject.toml, src/workable_gateway/policy.py, src/workable_gateway/server.py) qui relaie 33 outils, en place 13 derrière une validation humaine en deux temps et refuse les 48 autres.
Quand l’utiliser
Connectez le serveur hébergé dès que vos recruteurs travaillent déjà dans Claude sur des tâches adjacentes — brouillons d’outbound, synthèses de scorecard, points pour le hiring manager — et repassent sans arrêt dans Workable pour répondre à « à quelle étape est ce candidat », « quelles candidatures n’ont pas bougé cette semaine », « qui est dans la boucle d’entretien de cette req ». La connexion tient en une commande et ne coûte rien : Workable inclut le serveur MCP sans supplément sur tous ses plans d’abonnement.
Ajoutez la passerelle par-dessus quand la liste d’autorisation doit tenir de façon centralisée. Le settings.json d’un recruteur est appliqué par son client, sur sa machine, et il peut l’éditer. Un processus passerelle est appliqué une fois, par recruiting-ops, et le faire tourner, c’est la différence entre une politique et une préférence. La population concernée : une équipe de recrutement de cinq personnes ou plus partageant un compte Workable, dans une organisation où quelqu’un finira par demander qui a décidé que l’assistant pouvait désactiver un utilisateur.
Quand NE PAS l’utiliser
Laissez tomber la passerelle — pas le serveur — si votre client restreint déjà les outils par connecteur et que vous faites confiance à ceux qui l’utilisent. Claude Code adresse les outils MCP sous la forme mcp__<server>__<tool> et respecte permissions.deny dans settings.json. Le bundle fournit claude-code-permissions.example.json, la même politique exprimée ainsi, générée depuis le même policy.py. Elle ne coûte aucune infrastructure et c’est le bon premier geste. Ne passez à la passerelle que si vous avez besoin d’un caviardage des réponses, d’un journal d’audit central ou d’un jeton de validation lié à des arguments précis — trois choses qu’une deny list côté client ne donne pas.
Laissez tomber le workflow entier si votre compte Workable est aussi le système de référence RH, et pas seulement celui du recrutement. Le serveur de Workable couvre les salariés, les congés, le suivi du temps et tout le cycle des revues de performance depuis le même endpoint que les candidats. Un assistant branché sur ce compte atteint les contrats de travail via get_employee_documents et les enregistrements d’absence via get_timeoff_balances, sauf si quelque chose l’en empêche. Si personne n’est encore propriétaire de cette décision, faites d’abord valider la politique IA pour le recrutement.
Et laissez tomber si un seul recruteur constitue toute l’équipe. Le connecteur hébergé seul suffit à cette échelle ; l’installation de la passerelle et la revue de politique représentent environ une journée de travail pour une gouvernance que personne ne réclame encore.
Installation
Les instructions complètes sont dans apps/web/public/artifacts/mcp-server-workable-recruiting/README.md. La version courte : pip install -e ., définissez WORKABLE_ACCOUNT avec votre sous-domaine Workable, enregistrez la passerelle avec un chemin absolu et autorisez dans le navigateur au premier appel. Le serveur de Workable publie des métadonnées de serveur d’autorisation conformes à la RFC 8414 et accepte l’enregistrement dynamique de client conforme à la RFC 7591 : il n’y a donc pas de client ID à provisionner à la main ni d’API key à faire tourner.
L’étape qui compte vraiment vient avant tout cela : décider avec quel membre Workable vous vous autorisez. Chaque session MCP hérite du rôle et des affectations de poste de l’utilisateur connecté — la formulation de Workable est que l’IA ne peut lire et agir que sur les données que l’utilisateur est déjà autorisé à voir. Cela ressemble à un modèle de permissions jusqu’à ce que vous voyiez qui installe ça en premier. Les responsables recruiting-ops sont admins. Vous autoriser avec votre propre compte donne à la passerelle une portée admin et fait de la liste d’autorisation le seul mur encore debout. Créez plutôt un membre Workable dédié avec un permission set resserré ; get_permission_sets liste ceux que votre compte a définis.
Ce qu’il faut retenir hors de portée
src/workable_gateway/policy.py répartit les 94 outils en trois niveaux plus une liste de caviardage. La fonction de niveau refuse par défaut : les 37 outils que Workable a ajoutés en une seule livraison le 2026-07-20 seraient donc restés éteints jusqu’à ce qu’un humain les classe — exactement le comportement attendu d’une surface qui a grossi de 65% en neuf semaines.
48 refusés d’emblée, en six groupes avec une justification chacun. Les quatre outils de gestion des membres sautent parce qu’un agent capable d’attribuer un permission set élargit sa propre portée à la session suivante. Les quatre outils de département sautent parce que merge_department n’a pas d’inverse et que les rapports de recrutement sont découpés par département : une fusion ratée réécrit l’historique du funnel sans lever d’erreur. Les cinq outils de validation — offres, requisitions, congés — sautent parce qu’une validation est un acte d’autorité d’une personne nommée, et la déléguer efface la preuve qu’un humain a décidé. Les six outils de suivi du temps sautent parce qu’ils touchent à la paie et que bulk_create_time_entries transforme une mauvaise inférence en erreur de paie de masse. Les quinze outils de revue de performance sautent parce que submit_review est définitif ; la documentation de Workable note qu’un second envoi échoue, donc un agent qui rejoue un appel expiré est précisément le danger. Les quatorze lectures HRIS sautent parce que les documents salariés contiennent des contrats, des lettres de rémunération et des pièces de visa ou médicales.
13 derrière un portail de validation — les écritures sur candidats et requisitions, de move_candidate et disqualify_candidate jusqu’à create_requisition. Appeler l’un d’eux sans _gateway_confirm renvoie un dry run au lieu d’une écriture. Le _gateway_token de ce dry run est un hash du nom de l’outil et des arguments exacts : une validation pour « déplacer le candidat 41 vers Onsite » ne peut pas être rejouée en « déplacer le candidat 88 vers Offre ».
33 relayés directement — 32 lectures plus add_comment, la seule écriture qui soit additive, attribuable et supprimable depuis l’interface Workable. En plus de celles-là, server.py définit trois outils propres : workable_policy_report, pour qu’un appel refusé produise « c’est bloqué, faites-le dans Workable » plutôt qu’une boucle de nouvelles tentatives ; workable_pipeline_snapshot, pour les comptes par étape et les candidats à l’arrêt en un seul balayage paginé ; et workable_stage_move_review, qui résout l’étape actuelle du candidat pour que le recruteur valide un diff et non une demande.
Choix d’ingénierie
Épingler le compte plutôt que laisser le modèle choisir. Tous les outils Workable sauf get_accounts prennent un sous-domaine account, et un utilisateur ayant accès à deux comptes — une marque en production et une autre, ou un bac à sable — obtient des réponses assurées et d’apparence correcte issues du mauvais tenant. La passerelle injecte WORKABLE_ACCOUNT sur chaque appel relayé et refuse tout appel où le modèle en a mis un autre. Deux comptes, deux processus passerelle.
Un token bucket plutôt qu’un retry sur 429. Le bucket OAuth 2.0 de Workable est de 50 requêtes par tranche de 10 secondes et renvoie un HTTP 429 avec X-Rate-Limit-Reset au-delà. « Montre-moi tous les candidats de tous les postes ouverts » se déploie en get_jobs plus un get_candidates paginé par req et épuise ça en deux secondes environ, après quoi un assistant qui réessaie repart droit dans le mur. WORKABLE_RATE_PER_SEC vaut 4/s par défaut, sous le débit soutenu de 5/s, ce qui laisse de la marge à tout ce qui, dans le tenant, tient le même jeton.
Un balayage, pas un appel par étape.workable_pipeline_snapshot pagine les candidats une fois et compte les étapes à partir des lignes, plafonné par WORKABLE_PAGE_CAP (5 pages, 500 candidats). Le coût est plat que le poste ait 4 étapes ou 14, et la réponse pose page_cap_reached pour que le modèle annonce un compte partiel comme partiel.
Caviardage sur la réponse, pas seulement sur la requête. Bloquer search_employees n’empêche pas get_candidate de renvoyer un champ d’auto-déclaration que votre compte collecte pour le reporting EEO. policy.REDACT_FIELDS vide les champs par nom de clé, récursivement, parce que Workable imbrique le détail candidat et renvoie les lignes de recherche détaillée sous leurs propres clés.
Le coût réel
Le serveur coûte $0 : les communiqués de lancement et d’extension de Workable indiquent tous deux qu’il est inclus sans supplément sur tous les plans d’abonnement, les trois outils Advanced Search étant réservés aux plans Premier+ et Enterprise. C’est le chiffre intéressant, parce que Workable facture sa propre IA produit en crédits : les lots publiés aujourd’hui sont 5 000 crédits pour $600, 10 000 pour $1 000 et 50 000 pour $4 750, soit $0,095 à $0,12 le crédit. Interroger l’IA de Workable brûle des crédits. Interroger Claude via le serveur MCP brûle des tokens Anthropic et zéro crédit Workable. Pour les équipes qui paient déjà des sièges Claude, faire passer le Q&A de recrutement de l’autre côté de cette ligne est un vrai transfert, pas un match nul.
En face : environ 90 minutes pour installer la passerelle et faire les vérifications de premier lancement, et une revue de politique plus proche de trois heures parce qu’elle implique quelqu’un à qui appartient la décision sur les données RH. Le connecteur direct seul, c’est une commande et une dizaine de minutes.
Modes de défaillance
L’assistant rejoue une écriture refusée jusqu’à trouver une formulation qui passe. Garde-fou : workable_policy_report existe pour que le modèle nomme le niveau et s’arrête, et chaque message de refus renvoie vers l’interface Workable au lieu de proposer un autre outil. Testez-le — l’étape 3 du README demande à l’assistant de désactiver un membre et attend un refus, pas une tentative.
Une validation ancienne est rejouée contre d’autres arguments. Garde-fou : le _gateway_token hache les arguments, pas seulement le nom de l’outil. Modifier l’identifiant du candidat après le dry run l’invalide et force une nouvelle validation.
Le caviardage rate un champ personnalisé. La liste de champs dans policy.py est générique, et les attributs d’auto-déclaration dépendent du compte. Garde-fou : le point 1 de la liste de TODO du README consiste à récupérer vos vraies clés d’attributs avec get_account_custom_attributes et get_candidate_detailed_fields avant que tout cela touche un compte de production. Tant que ce n’est pas fait, traitez le caviardage comme non testé.
Les CV et les notes atteignent un tiers.get_candidate_files est au niveau ALLOW parce que lire des CV, c’est le métier. Cela fait transiter des données RGPD et CCPA par Anthropic. Garde-fou : la validation de la politique IA et un registre des activités de traitement qui nomme le flux — avant la mise en service du connecteur, pas après la question de quelqu’un.
L’alternative qu’il faut nommer
La comparaison évidente, c’est le schéma du workflow MCP Greenhouse, où le bundle est le serveur parce que l’éditeur n’en héberge pas. Ce n’est pas l’arbitrage ici. Construire votre propre serveur sur l’API REST de Workable revient à réimplémenter 94 endpoints et à assumer le flux OAuth pour concurrencer quelque chose de gratuit et de première main — ne le faites pas.
L’arbitrage qui mérite d’être pesé, c’est un courtier. Composio et Zapier listent tous deux des endpoints MCP Workable hébergés, et tous deux placent un second fournisseur sur le chemin, détenteur de votre jeton OAuth, à leur propre tarif à la tâche ou au siège. N’en prenez un que si vous êtes déjà standardisés dessus pour d’autres connecteurs. Sinon le classement est : serveur Workable hébergé plus deny rules côté client pour la plupart des équipes, serveur hébergé plus cette passerelle quand la liste d’autorisation doit s’appliquer là où les recruteurs ne peuvent pas la modifier. Pour situer cette ligne, voyez l’accès en écriture MCP et quand l’accorder et les serveurs MCP expliqués.
# 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()
