Workable aloja su propio servidor Model Context Protocol en https://mcp.workable.com/mcp, así que la pregunta de construirlo ya está resuelta: te conectas, no escribes uno. La pregunta abierta es cuáles de sus 94 herramientas puede tocar el asistente de tus recruiters. Workable lanzó el servidor el 2026-05-13 con 38 herramientas y lo amplió a 94 el 2026-07-20, y esa ampliación agregó acceso de escritura a evaluaciones de desempeño, gestión de cuentas y permisos, y actualización de perfiles de candidatos. El bundle de artefactos en apps/web/public/artifacts/mcp-server-workable-recruiting/ es la respuesta a esa pregunta: un gateway de mínimo privilegio (README.md, pyproject.toml, src/workable_gateway/policy.py, src/workable_gateway/server.py) que reenvía 33 herramientas, pone 13 detrás de una aprobación humana en dos fases y rechaza las otras 48.
Cuándo usarlo
Conecta el servidor alojado en cuanto los recruiters ya estén trabajando en Claude en tareas adyacentes — borradores de outbound, resúmenes de scorecards, actualizaciones para el hiring manager — y sigan saltando de vuelta a Workable para responder “en qué etapa está este candidato”, “qué postulaciones no se movieron esta semana”, “quién está en el loop de entrevistas de este req”. La conexión es un solo comando y no cuesta nada: Workable incluye el servidor MCP sin cargo adicional en todos sus planes de suscripción.
Suma el gateway encima cuando la lista de permitidos tenga que sostenerse de forma centralizada. El settings.json de un recruiter lo aplica su cliente, en su computadora, y él lo puede editar. Un proceso gateway se aplica una vez, desde recruiting-ops, y ejecutarlo es la diferencia entre una política y una preferencia. La población que lo necesita es un equipo de recruiting de cinco personas o más compartiendo una cuenta de Workable, en una organización donde alguien va a preguntar quién decidió que el asistente pudiera desactivar a un usuario.
Cuándo NO usarlo
Sáltate el gateway — no el servidor — si tu cliente ya restringe herramientas por conector y confías en quienes lo usan. Claude Code identifica las herramientas MCP como mcp__<server>__<tool> y respeta permissions.deny en settings.json. El bundle incluye claude-code-permissions.example.json, la misma política expresada así, generada desde el mismo policy.py. No cuesta infraestructura y es el primer movimiento correcto. Recurre al gateway solo cuando necesites redacción de respuestas, un log de auditoría central o un token de aprobación atado a argumentos específicos — tres cosas que una deny list del lado del cliente no te da.
Sáltate el workflow completo si tu cuenta de Workable es el sistema de registro de RR. HH. además del de contratación. El servidor de Workable cubre empleados, ausencias, control horario y todo el ciclo de evaluaciones de desempeño desde el mismo endpoint que los candidatos. Un asistente conectado a esa cuenta alcanza contratos laborales vía get_employee_documents y registros de ausencias vía get_timeoff_balances salvo que algo lo detenga. Si nadie es dueño todavía de esa decisión, aprueba primero la política de IA para recruiting.
Y sáltatelo si un solo recruiter es todo el equipo. El conector alojado solo alcanza a esa escala; la instalación del gateway y la revisión de la política son aproximadamente un día de trabajo que compra una gobernanza que nadie está pidiendo todavía.
Instalación
Las instrucciones completas están en apps/web/public/artifacts/mcp-server-workable-recruiting/README.md. La versión corta: pip install -e ., define WORKABLE_ACCOUNT con tu subdominio de Workable, registra el gateway con una ruta absoluta y autoriza en el navegador en la primera llamada. El servidor de Workable publica metadatos de servidor de autorización según RFC 8414 y acepta registro dinámico de clientes según RFC 7591, así que no hay client ID que aprovisionar a mano ni API key que rotar.
El paso que de verdad importa va antes de todo eso: decidir con qué miembro de Workable te autorizas. Cada sesión MCP hereda el rol y las asignaciones de puestos del usuario que inició sesión — la formulación de Workable es que la IA solo puede leer y actuar sobre datos que el usuario ya está autorizado a ver. Suena a modelo de permisos hasta que notas quién instala esto primero. Los líderes de recruiting-ops son admins. Autorizarte con tu propia cuenta le entrega al gateway alcance de admin y deja la lista de permitidos como único muro en pie. Crea en su lugar un miembro de Workable dedicado con un permission set acotado; get_permission_sets lista los que tu cuenta tiene definidos.
Qué retener
src/workable_gateway/policy.py clasifica las 94 herramientas en tres niveles y una lista de redacción. La función de nivel deniega por defecto, así que las 37 herramientas que Workable agregó en un solo release el 2026-07-20 se habrían quedado a oscuras hasta que un humano las clasificara — que es el comportamiento que quieres de una superficie que creció 65% en nueve semanas.
48 rechazadas de plano, en seis grupos con una justificación cada uno. Las cuatro herramientas de gestión de miembros se van porque un agente que puede otorgar un permission set puede ampliar su propio alcance en la sesión siguiente. Las cuatro de departamentos se van porque merge_department no tiene inversa y los reportes de recruiting se cortan por departamento, así que una fusión mal hecha reescribe el histórico del funnel sin lanzar un error. Las cinco de aprobación — ofertas, requisiciones, ausencias — se van porque una aprobación es un acto de autoridad de una persona con nombre, y delegarla borra la evidencia de que una persona decidió. Las seis de control horario se van porque son adyacentes a la nómina y bulk_create_time_entries convierte una inferencia mala en un error de pago masivo. Las quince de evaluaciones de desempeño se van porque submit_review es definitivo; la documentación de Workable señala que un segundo envío falla, así que un agente reintentando una llamada que expiró es exactamente el riesgo. Las catorce lecturas de HRIS se van porque los documentos de empleados guardan contratos, cartas de compensación y papeles de visa o médicos.
13 detrás de una compuerta de aprobación — las escrituras sobre candidatos y requisiciones, desde move_candidate y disqualify_candidate hasta create_requisition. Llamar a una sin _gateway_confirm devuelve un dry run en vez de una escritura. El _gateway_token de ese dry run es un hash del nombre de la herramienta más los argumentos exactos, así que una aprobación para “mover al candidato 41 a Onsite” no se puede reutilizar como “mover al candidato 88 a Oferta”.
33 reenviadas directamente — 32 lecturas más add_comment, la única escritura que es aditiva, atribuible y eliminable desde la interfaz de Workable. Encima de esas, server.py define tres herramientas propias: workable_policy_report, para que una llamada rechazada produzca “eso está bloqueado, hazlo en Workable” en lugar de un bucle de reintentos; workable_pipeline_snapshot, para conteos por etapa y candidatos estancados en un solo barrido paginado; y workable_stage_move_review, que resuelve la etapa actual del candidato para que el recruiter apruebe un diff y no una solicitud.
Decisiones de ingeniería
Fijar la cuenta en vez de dejar que el modelo elija. Toda herramienta de Workable salvo get_accounts recibe un subdominio account, y un usuario con acceso a dos cuentas — una marca productiva y otra, o un sandbox — obtiene respuestas seguras y con pinta de correctas del tenant equivocado. El gateway inyecta WORKABLE_ACCOUNT en cada llamada reenviada y rechaza cualquiera donde el modelo haya puesto otra cosa. Dos cuentas significan dos procesos gateway.
Un token bucket en lugar de reintentar ante un 429. El bucket OAuth 2.0 de Workable es de 50 solicitudes cada 10 segundos y devuelve HTTP 429 con X-Rate-Limit-Reset por encima de eso. “Muéstrame todos los candidatos de todos los puestos abiertos” se abre en get_jobs más un get_candidates paginado por req y agota eso en unos dos segundos, tras lo cual un asistente que reintenta se estrella contra el mismo muro. WORKABLE_RATE_PER_SEC tiene 4/s por defecto, bajo la tasa sostenida de 5/s, dejando margen para lo demás en el tenant que use el mismo token.
Un barrido, no una llamada por etapa.workable_pipeline_snapshot pagina candidatos una vez y cuenta las etapas desde las filas, con tope en WORKABLE_PAGE_CAP (5 páginas, 500 candidatos). El costo es plano tenga el puesto 4 etapas o 14, y la respuesta marca page_cap_reached para que el modelo reporte como parcial un conteo parcial.
Redacción sobre la respuesta, no solo sobre la petición. Bloquear search_employees no impide que get_candidate devuelva un campo de autoidentificación que tu cuenta recolecta para reportes de EEO. policy.REDACT_FIELDS vacía campos por nombre de clave, de forma recursiva, porque Workable anida el detalle del candidato y devuelve las filas de búsqueda detallada bajo sus propias claves.
La realidad del costo
El servidor cuesta $0 — tanto el anuncio de lanzamiento como el de ampliación de Workable indican que está incluido sin cargo adicional en todos los planes de suscripción, con las tres herramientas de Advanced Search restringidas a los planes Premier+ y Enterprise. Ese es el número interesante, porque Workable cobra su propia IA de producto en créditos: los paquetes publicados hoy son 5.000 créditos por $600, 10.000 por $1.000 y 50.000 por $4.750, es decir entre $0,095 y $0,12 por crédito. Preguntarle a la IA de Workable quema créditos. Preguntarle a Claude a través del servidor MCP quema tokens de Anthropic y cero créditos de Workable. Para equipos que ya pagan asientos de Claude, mover el Q&A de recruiting a través de esa línea es una transferencia real, no un empate.
En contra: unos 90 minutos para instalar el gateway y hacer las verificaciones de primera ejecución, y una revisión de política que se acerca a tres horas porque involucra a alguien dueño de la decisión sobre datos de RR. HH. El conector directo por sí solo es un comando y unos diez minutos.
Modos de falla
El asistente reintenta una escritura rechazada hasta encontrar una formulación que funcione. Guarda: workable_policy_report existe para que el modelo nombre el nivel y se detenga, y cada mensaje de rechazo apunta a la interfaz de Workable en vez de sugerir otra herramienta. Pruébalo — el paso 3 del README le pide al asistente desactivar a un miembro y espera un rechazo, no un intento.
Una aprobación vieja se reutiliza contra otros argumentos. Guarda: el _gateway_token hashea los argumentos, no solo el nombre de la herramienta. Editar el id del candidato después del dry run lo invalida y fuerza una aprobación nueva.
La redacción se salta un campo personalizado. La lista de campos en policy.py es genérica y los atributos de autoidentificación dependen de cada cuenta. Guarda: el punto 1 de la lista de TODO del README es extraer tus claves reales con get_account_custom_attributes y get_candidate_detailed_fields antes de que esto toque una cuenta productiva. Hasta que eso esté hecho, trata la redacción como no probada.
CV y notas llegan a un tercero.get_candidate_files está en el nivel ALLOW porque leer CV es el trabajo. Eso enruta datos GDPR y CCPA a través de Anthropic. Guarda: la aprobación de la política de IA y un registro de actividades de tratamiento que nombre el flujo — antes de que el conector esté vivo, no después de que alguien pregunte.
La alternativa que vale nombrar
La comparación obvia es el patrón del workflow MCP de Greenhouse, donde el bundle es el servidor porque el proveedor no aloja ninguno. Ese no es el trade aquí. Construir tu propio servidor sobre la API REST de Workable significa reimplementar 94 endpoints y hacerte cargo del flujo OAuth para competir con algo gratis y de primera parte — no lo hagas.
El trade que sí vale sopesar es un bróker. Composio y Zapier listan ambos endpoints MCP alojados de Workable, y ambos meten un segundo proveedor en la ruta sosteniendo tu token OAuth, con su propio precio por tarea o por asiento. Elige uno solo si ya estás estandarizado en él para otros conectores. Si no, el ranking es: servidor alojado de Workable más deny rules del lado del cliente para la mayoría de los equipos, y servidor alojado más este gateway cuando la lista de permitidos tenga que aplicarse en un lugar que los recruiters no puedan editar. Para el contexto de dónde cae esa línea, mira acceso de escritura por MCP y cuándo concederlo y servidores MCP explicados.
# 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()
