O Workable hospeda o próprio servidor Model Context Protocol em https://mcp.workable.com/mcp, então a questão de construir já está resolvida: você conecta, não escreve um. A questão em aberto é quais das 94 ferramentas dele o assistente dos seus recruiters pode tocar. O Workable lançou o servidor em 2026-05-13 com 38 ferramentas e expandiu para 94 em 2026-07-20, e essa expansão adicionou acesso de escrita a avaliações de desempenho, gestão de contas e permissões e atualização de perfis de candidatos. O bundle de artefatos em apps/web/public/artifacts/mcp-server-workable-recruiting/ é a resposta a essa questão: um gateway de menor privilégio (README.md, pyproject.toml, src/workable_gateway/policy.py, src/workable_gateway/server.py) que encaminha 33 ferramentas, coloca 13 atrás de uma aprovação humana em duas fases e recusa as outras 48.
Quando usar
Conecte o servidor hospedado assim que os recruiters já estiverem trabalhando no Claude em tarefas adjacentes — rascunhos de outbound, resumos de scorecard, updates para o hiring manager — e continuarem voltando ao Workable para responder “em que etapa está esse candidato”, “quais candidaturas não andaram esta semana”, “quem está no loop de entrevistas desta req”. A conexão é um comando só e não custa nada: o Workable inclui o MCP server sem cobrança adicional em todos os planos de assinatura.
Coloque o gateway por cima quando a allowlist precisar valer de forma centralizada. O settings.json de um recruiter é aplicado pelo cliente dele, na máquina dele, e ele consegue editar. Um processo gateway é aplicado uma vez, por recruiting-ops, e rodá-lo é a diferença entre uma política e uma preferência. A população que precisa disso é um time de recruiting de cinco pessoas ou mais compartilhando uma conta Workable, numa organização onde alguém vai acabar perguntando quem decidiu que o assistente podia desativar um usuário.
Quando NÃO usar
Pule o gateway — não o servidor — se o seu cliente já restringe ferramentas por conector e você confia em quem usa. O Claude Code identifica ferramentas MCP como mcp__<server>__<tool> e respeita permissions.deny no settings.json. O bundle traz claude-code-permissions.example.json, a mesma política expressa desse jeito, gerada a partir do mesmo policy.py. Não custa infraestrutura e é o primeiro movimento certo. Recorra ao gateway só quando precisar de redação de respostas, um log de auditoria central ou um token de aprovação preso a argumentos específicos — três coisas que uma deny list do lado do cliente não entrega.
Pule o workflow inteiro se a sua conta Workable é o sistema de registro de RH além do de contratação. O servidor do Workable cobre funcionários, folgas, controle de ponto e todo o ciclo de avaliações de desempenho a partir do mesmo endpoint dos candidatos. Um assistente ligado a essa conta alcança contratos de trabalho via get_employee_documents e registros de ausência via get_timeoff_balances a menos que algo o impeça. Se ninguém for dono dessa decisão ainda, aprove antes a política de IA para recruiting.
E pule se um recruiter sozinho é o time inteiro. O conector hospedado sozinho dá conta nessa escala; a instalação do gateway e a revisão de política dão cerca de um dia de trabalho que compra uma governança que ninguém está pedindo ainda.
Instalação
As instruções completas estão em apps/web/public/artifacts/mcp-server-workable-recruiting/README.md. A versão curta: pip install -e ., defina WORKABLE_ACCOUNT com o seu subdomínio Workable, registre o gateway com caminho absoluto e autorize no navegador na primeira chamada. O servidor do Workable publica metadados de authorization server conforme a RFC 8414 e aceita registro dinâmico de cliente conforme a RFC 7591, então não há client ID para provisionar na mão nem API key para rotacionar.
O passo que realmente importa vem antes de tudo isso: decidir com qual membro do Workable você autoriza. Cada sessão MCP herda o papel e as atribuições de vaga do usuário logado — a formulação do próprio Workable é que a IA só consegue ler e agir sobre dados que o usuário já está autorizado a ver. Parece um modelo de permissões até você notar quem instala isso primeiro. Líderes de recruiting-ops são admins. Autorizar com a sua própria conta entrega ao gateway escopo de admin e deixa a allowlist como o único muro de pé. Crie um membro Workable dedicado com um permission set estreito; get_permission_sets lista os que a sua conta tem definidos.
O que reter
O src/workable_gateway/policy.py separa as 94 ferramentas em três níveis e uma lista de redação. A função de nível nega por padrão, então as 37 ferramentas que o Workable adicionou num único release em 2026-07-20 teriam ficado no escuro até um humano classificar — que é o comportamento que você quer de uma superfície que cresceu 65% em nove semanas.
48 recusadas de saída, em seis grupos com uma justificativa cada. As quatro ferramentas de gestão de membros saem porque um agente que pode conceder um permission set consegue ampliar o próprio alcance na sessão seguinte. As quatro de departamentos saem porque merge_department não tem inversa e os relatórios de recruiting são cortados por departamento, então uma fusão errada reescreve o histórico do funil sem lançar erro. As cinco de aprovação — ofertas, requisições, folgas — saem porque aprovar é um ato de autoridade de uma pessoa com nome, e delegar apaga a evidência de que uma pessoa decidiu. As seis de controle de ponto saem porque são adjacentes à folha de pagamento e bulk_create_time_entries transforma uma inferência ruim num erro de pagamento em massa. As quinze de avaliação de desempenho saem porque submit_review é definitivo; a documentação do Workable registra que um segundo envio falha, então um agente repetindo uma chamada que deu timeout é exatamente o risco. As catorze leituras de HRIS saem porque documentos de funcionário guardam contratos, cartas de remuneração e papelada de visto ou médica.
13 atrás de um portão de aprovação — as escritas sobre candidatos e requisições, de move_candidate e disqualify_candidate até create_requisition. Chamar uma sem _gateway_confirm devolve um dry run em vez de uma escrita. O _gateway_token desse dry run é um hash do nome da ferramenta mais os argumentos exatos, então uma aprovação para “mover o candidato 41 para Onsite” não pode ser reaproveitada como “mover o candidato 88 para Oferta”.
33 encaminhadas direto — 32 leituras mais add_comment, a única escrita que é aditiva, atribuível e removível pela interface do Workable. Em cima dessas, o server.py define três ferramentas próprias: workable_policy_report, para que uma chamada recusada produza “isso está bloqueado, faça no Workable” em vez de um loop de retentativas; workable_pipeline_snapshot, para contagem por etapa e candidatos parados numa varredura paginada única; e workable_stage_move_review, que resolve a etapa atual do candidato para o recruiter aprovar um diff e não um pedido.
Decisões de engenharia
Fixar a conta em vez de deixar o modelo escolher. Toda ferramenta do Workable exceto get_accounts recebe um subdomínio account, e um usuário com acesso a duas contas — uma marca em produção e outra, ou um sandbox — recebe respostas confiantes e com cara de certas do tenant errado. O gateway injeta WORKABLE_ACCOUNT em toda chamada encaminhada e recusa qualquer uma em que o modelo tenha colocado outra coisa. Duas contas significam dois processos gateway.
Um token bucket em vez de retentar no 429. O bucket OAuth 2.0 do Workable é de 50 requisições a cada 10 segundos e devolve HTTP 429 com X-Rate-Limit-Reset acima disso. “Me mostre todos os candidatos de todas as vagas abertas” se abre em get_jobs mais um get_candidates paginado por req e esgota isso em uns dois segundos, e aí um assistente que retenta bate no mesmo muro. O WORKABLE_RATE_PER_SEC vem em 4/s por padrão, abaixo da taxa sustentada de 5/s, deixando folga para o que mais no tenant estiver usando o mesmo token.
Uma varredura, não uma chamada por etapa. O workable_pipeline_snapshot pagina candidatos uma vez e conta as etapas a partir das linhas, com teto em WORKABLE_PAGE_CAP (5 páginas, 500 candidatos). O custo é plano tendo a vaga 4 etapas ou 14, e a resposta marca page_cap_reached para o modelo reportar como parcial uma contagem parcial.
Redação na resposta, não só na requisição. Bloquear search_employees não impede o get_candidate de devolver um campo de autoidentificação que a sua conta coleta para relatórios de EEO. O policy.REDACT_FIELDS esvazia campos por nome de chave, recursivamente, porque o Workable aninha o detalhe do candidato e devolve as linhas de busca detalhada sob chaves próprias.
A realidade do custo
O servidor custa $0 — tanto o anúncio de lançamento quanto o de expansão do Workable dizem que ele está incluído sem cobrança adicional em todos os planos de assinatura, com as três ferramentas de Advanced Search restritas aos planos Premier+ e Enterprise. Esse é o número interessante, porque o Workable cobra a IA do próprio produto em créditos: os pacotes publicados hoje são 5.000 créditos por $600, 10.000 por $1.000 e 50.000 por $4.750, ou seja $0,095 a $0,12 por crédito. Perguntar para a IA do Workable queima crédito. Perguntar para o Claude via MCP server queima token da Anthropic e zero crédito do Workable. Para times que já pagam assentos de Claude, mover o Q&A de recruiting para o outro lado dessa linha é uma transferência real, não empate.
Contra isso: cerca de 90 minutos para instalar o gateway e rodar as verificações de primeira execução, e uma revisão de política que chega perto de três horas porque envolve alguém dono da decisão sobre dados de RH. O conector direto sozinho é um comando e uns dez minutos.
Modos de falha
O assistente retenta uma escrita recusada até achar uma formulação que passe. Guarda: o workable_policy_report existe para o modelo nomear o nível e parar, e toda mensagem de recusa aponta para a interface do Workable em vez de sugerir outra ferramenta. Teste — o passo 3 do README pede ao assistente para desativar um membro e espera uma recusa, não uma tentativa.
Uma aprovação velha é reaproveitada contra outros argumentos. Guarda: o _gateway_token faz hash dos argumentos, não só do nome da ferramenta. Editar o id do candidato depois do dry run invalida o token e força uma aprovação nova.
A redação pula um campo customizado. A lista de campos no policy.py é genérica e os atributos de autoidentificação variam por conta. Guarda: o item 1 da lista de TODO do README é puxar as suas chaves reais com get_account_custom_attributes e get_candidate_detailed_fields antes de isso tocar uma conta de produção. Até isso estar feito, trate a redação como não testada.
Currículos e notas chegam a um terceiro. O get_candidate_files está no nível ALLOW porque ler currículo é o trabalho. Isso roteia dados GDPR e CCPA pela Anthropic. Guarda: a aprovação da política de IA e um registro de atividades de tratamento que nomeie o fluxo — antes de o conector estar no ar, não depois de alguém perguntar.
A alternativa que vale nomear
A comparação óbvia é o padrão do workflow MCP do Greenhouse, onde o bundle é o servidor porque o fornecedor não hospeda nenhum. Não é esse o trade aqui. Construir o seu próprio servidor sobre a API REST do Workable significa reimplementar 94 endpoints e assumir o fluxo OAuth para competir com algo gratuito e de primeira parte — não faça.
O trade que vale pesar é um broker. Composio e Zapier listam os dois endpoints MCP hospedados do Workable, e os dois colocam um segundo fornecedor no caminho segurando o seu token OAuth, com preço próprio por tarefa ou por assento. Escolha um só se você já está padronizado nele para outros conectores. Fora isso, o ranking é: servidor hospedado do Workable mais deny rules do lado do cliente para a maioria dos times, e servidor hospedado mais este gateway quando a allowlist precisar ser aplicada num lugar que os recruiters não conseguem editar. Para o contexto de onde essa linha cai, veja acesso de escrita por MCP e quando conceder e MCP servers 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()
