Workable hosts its own Model Context Protocol server at https://mcp.workable.com/mcp, so the build question is settled — you connect, you do not write one. The open question is which of its 94 tools your recruiters’ assistant gets to touch. Workable launched the server on 2026-05-13 with 38 tools and expanded it to 94 on 2026-07-20, and that expansion added write access across performance reviews, account and permissions management, and candidate profile updates. The artifact bundle at apps/web/public/artifacts/mcp-server-workable-recruiting/ is the answer to that question: a least-privilege gateway (README.md, pyproject.toml, src/workable_gateway/policy.py, src/workable_gateway/server.py) that forwards 33 tools, gates 13 behind a two-phase human approval, and refuses the other 48.
When to use
Connect the hosted server directly the moment recruiters are already working in Claude on adjacent tasks — outreach drafts, scorecard summaries, hiring-manager updates — and keep bouncing back into Workable to answer “what stage is this candidate at”, “which applications have not moved this week”, “who is on the interview loop for this req”. The connection is one command and costs nothing: Workable includes the MCP server at no added charge on every subscription plan.
Add the gateway on top when the allowlist has to hold centrally. A recruiter’s own settings.json is enforced by their client, on their laptop, and they can edit it. A gateway process is enforced once, by recruiting-ops, and running it is the difference between a policy and a preference. The population that needs this is a recruiting team of five or more sharing one Workable account, in an org where somebody will eventually ask who decided the assistant could deactivate a user.
When NOT to use
Skip the gateway — not the server — if your client already restricts tools per connector and you trust the people using it. Claude Code matches MCP tools as mcp__<server>__<tool> and honours permissions.deny in settings.json. The bundle ships claude-code-permissions.example.json, the same policy expressed that way, generated from the same policy.py. It costs no infrastructure and it is the right first move. Reach for the gateway only when you need response redaction, a central audit log, or an approval token bound to specific arguments — three things a client-side deny list does not give you.
Skip the whole workflow if your Workable account is the system of record for HR as well as hiring. Workable’s server covers employees, time off, time tracking, and the full performance-review lifecycle from the same endpoint as candidates. An assistant wired into that account reaches employment contracts through get_employee_documents and absence records through get_timeoff_balances unless something stops it. If nobody owns that decision yet, get the recruiting AI policy signed off first.
And skip it if a single recruiter is the whole team. The hosted connector alone is fine at that scale; the gateway’s install and policy review is roughly a day of work that buys governance nobody is asking for yet.
Setup
Full instructions are in apps/web/public/artifacts/mcp-server-workable-recruiting/README.md. The short version: pip install -e ., set WORKABLE_ACCOUNT to your Workable subdomain, register the gateway with an absolute path, and authorize in the browser on first call. 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 and no API key to rotate.
The step that actually matters comes before any of that: decide which Workable member you authorize as. Every MCP session inherits the signed-in user’s role and job assignments — Workable’s own wording is that the AI can only read and act on data the user is already authorized to see. That sounds like a permission model until you notice who installs this first. Recruiting-ops leads are admins. Authorizing as yourself hands the gateway admin scope and makes the allowlist the only wall standing. Create a dedicated Workable member with a narrowed permission set instead; get_permission_sets lists what your account has defined.
What to withhold
src/workable_gateway/policy.py sorts all 94 tools into three tiers and a redaction list. The tier function is default-deny, so the 37 tools Workable added in one release on 2026-07-20 would have stayed dark until a human classified them — which is the behaviour you want from a surface that grew by 65% in nine weeks.
48 refused outright, in six groups with one rationale each. The four member-management tools go because an agent that can grant a permission set can widen its own reach on the next session. The four department tools go because merge_department has no inverse and recruiting reports are cut by department, so a bad merge rewrites funnel history without an error. The five approval tools — offers, requisitions, time off — go because an approval is an act of authority by a named person, and delegating it erases the evidence that a person decided. The six time-tracking tools go because they are payroll-adjacent and bulk_create_time_entries turns one bad inference into a bulk pay error. The fifteen performance-review tools go because submit_review is final; Workable’s docs note a second submit fails, so an agent retrying a timed-out call is the exact hazard. The fourteen HRIS reads go because employee documents hold contracts, comp letters, and visa or medical paperwork.
13 behind an approval gate — the candidate and requisition writes, from move_candidate and disqualify_candidate through create_requisition. Calling one without _gateway_confirm returns a dry run instead of a write. 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”.
33 forwarded directly — 32 reads plus add_comment, the one write that is additive, attributable, and removable in the Workable UI. On top of those, server.py defines three tools of its own: workable_policy_report so a refused call produces “that is blocked, do it in Workable” rather than a retry loop, workable_pipeline_snapshot for stage counts and stalled candidates in one paged sweep, and workable_stage_move_review, which resolves the candidate’s current stage so the recruiter approves a diff rather than a request.
Engineering choices
Account pinning over model choice. Every Workable tool except get_accounts takes an account subdomain, and a user with access to two accounts — a production brand and a second one, or a sandbox — gets confident, correct-looking answers from the wrong tenant. The gateway injects WORKABLE_ACCOUNT on every forwarded call and rejects any call where the model supplied something else. Two accounts means two gateway processes.
A token bucket rather than retry-on-429. Workable’s OAuth 2.0 bucket is 50 requests per 10 seconds and returns HTTP 429 with X-Rate-Limit-Reset above it. “Show me every candidate across all open roles” fans out into get_jobs plus a paged get_candidates per req and clears that in about two seconds, after which an assistant that retries walks straight back into the wall. WORKABLE_RATE_PER_SEC defaults to 4/s, under the 5/s sustained rate, leaving headroom for whatever else in the tenant holds the same token.
One sweep, not one call per stage.workable_pipeline_snapshot pages candidates once and counts stages from the rows, capped at WORKABLE_PAGE_CAP (5 pages, 500 candidates). Cost is flat whether the job has 4 stages or 14, and the response sets page_cap_reached so the model reports a partial count as partial.
Redaction on the response, not just the request. Blocking search_employees does not stop get_candidate returning a self-identification field your account collects for EEO reporting. policy.REDACT_FIELDS blanks fields by key name, recursively, because Workable nests candidate detail and returns detailed-search rows under their own keys.
Cost reality
The server is $0 — Workable’s launch and expansion announcements both state it is included at no added cost across all subscription plans, with the three Advanced Search tools restricted to Premier+ and Enterprise. That is the interesting number, because Workable meters its own in-product AI in credits: the published bundles today are 5,000 credits for $600, 10,000 for $1,000, and 50,000 for $4,750, which is $0.095 to $0.12 per credit. Asking Workable’s AI burns credits. Asking Claude through the MCP server burns Anthropic tokens and zero Workable credits. For teams already paying for Claude seats, moving recruiter Q&A across that line is a real transfer, not a wash.
Against that: about 90 minutes for the gateway install and first-run checks, and a policy review that runs closer to three hours because it involves someone who owns the HR-data decision. The direct connector alone is one command and about ten minutes.
Failure modes
The assistant retries a refused write until it finds a phrasing that works. Guard: workable_policy_report exists so the model can name the tier and stop, and every refusal string points at the Workable UI instead of suggesting an alternative tool. Test it — README step 3 asks the assistant to deactivate a member and expects a refusal, not an attempt.
A stale approval gets replayed against different arguments. Guard: the _gateway_token hashes the arguments, not just the tool name. Editing the candidate id after the dry run invalidates it and forces a fresh approval.
Redaction misses a custom field. The field list in policy.py is generic, and self-identification attributes are account-specific. Guard: item 1 in the README’s TODO list is pulling your real attribute keys with get_account_custom_attributes and get_candidate_detailed_fields before this touches a production account. Until that is done, treat the redaction as untested.
Résumés and notes reach a third party.get_candidate_files is in the ALLOW tier because reading résumés is the job. That routes GDPR and CCPA data through Anthropic. Guard: the AI policy sign-off, and a records-of-processing entry that names the flow — before the connector is live, not after somebody asks.
The alternative worth naming
The obvious comparison is the Greenhouse MCP workflow pattern, where the bundle is the server because the vendor does not host one. That is not the trade here. Building your own server over Workable’s REST API means re-implementing 94 endpoints and owning the OAuth flow to compete with something free and first-party — do not.
The trade worth weighing is a broker. Composio and Zapier both list hosted Workable MCP endpoints, and both put a second vendor in the path holding your OAuth token, on their own per-task or per-seat pricing. Pick one only if you are already standardized on it for other connectors. Otherwise the ranking is: hosted Workable server plus client-side deny rules for most teams, hosted server plus this gateway when the allowlist has to be enforced somewhere the recruiters cannot edit. For background on where that line sits, see MCP write access, and when to grant it and MCP servers explained.
# 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()
