Ashby MCP server for Claude

[project]
name = "ashby-recruiting-mcp"
version = "0.1.0"
description = "MCP server for Ashby recruiting workflows"
readme = "README.md"
requires-python = ">=3.11"
license = { text = "MIT" }
dependencies = [
  "mcp>=1.2.0",
  "httpx>=0.27.0",
  "pydantic>=2.6.0",
]

[project.optional-dependencies]
dev = [
  "pytest>=8.0.0",
  "pytest-httpx>=0.30.0",
  "ruff>=0.4.0",
]

[build-system]
requires = ["hatchling"]
build-backend = "hatchling.build"

[tool.hatch.build.targets.wheel]
packages = ["src/ashby_recruiting_mcp"]

[tool.ruff]
line-length = 100
target-version = "py311"

# mcp-server-ashby-recruiting

An MCP server tuned for recruiting teams using Ashby. Exposes candidates, applications, jobs, and openings as Claude tools, plus two recruiting-specific helpers (`stale_candidates`, `pipeline_velocity`) and two light-write tools (`add_note`, `add_tag`) that recruiters can drive from a Claude conversation.

> **STATUS: scaffold — not runtime-tested.** The code below is structurally
> complete and follows the official `mcp` Python SDK conventions, but it
> has not been executed against a live Ashby workspace. Treat it as a
> starting point you adapt to your team's pipeline conventions, not as a
> deployable binary. Custom field IDs, stage names, source labels, and
> pipeline structure vary by workspace — re-test every helper against the
> Ashby UI before relying on it.

## What it exposes

### Object-read tools (read-only)
- `get_candidate(candidate_id)` — full candidate properties, contact info, and current applications
- `get_application(application_id)` — application record with current stage, source, and history
- `get_job(job_id)` — job record with hiring team, status, and pipeline reference
- `get_opening(opening_id)` — opening (req) record with target start and headcount

### Search tools (read-only)
- `search_candidates(query, limit?)` — name / email / company substring search across the candidate database
- `list_applications(job_id, status?)` — applications for a job, optionally filtered by status (`Active`, `Hired`, `Archived`)
- `list_jobs(status?)` — jobs in the workspace, optionally filtered by status (`Open`, `Closed`, `Draft`)

### Recruiting-specific helpers (read-only)
- `stale_candidates(days_inactive=30)` — active candidates with no application activity (stage change, note, interview event) for N days; output grouped by current stage
- `pipeline_velocity(job_id)` — average days-in-stage per stage for a job's pipeline, computed across the last 90 days of stage changes; surfaces where the funnel is stuck

### Light-write tools (recruiter-driven)
- `add_note(candidate_id, body)` — append a note to a candidate's record (visible in the Ashby UI activity feed)
- `add_tag(candidate_id, tag)` — apply a tag to a candidate (e.g. `phone-screen-passed`, `do-not-pursue-2026`)

The server **does not** expose stage advances, application archives, offer creation, or candidate deletes. The principle: Claude can ask, summarize, and document; the recruiter drives every candidate-facing change in the Ashby UI where the audit trail and approval workflow already live.

## Setup

### 1. Install

```bash
git clone <wherever you put this>
cd mcp-server-ashby-recruiting
python -m venv .venv
source .venv/bin/activate     # or .venv\Scripts\activate on Windows
pip install -e .
```

### 2. Create an Ashby API key

In Ashby: Admin → API Keys → Create API Key. Grant scopes:

- `candidatesRead` and `candidatesWrite` (write only for `add_note`, `add_tag`)
- `applicationsRead`
- `jobsRead`
- `openingsRead`
- `interviewsRead` (used by `stale_candidates` to detect interview activity)

Copy the generated key. Ashby keys are workspace-scoped and act with the privileges of an admin role — treat them like a production secret.

### 3. Configure environment

```bash
export ASHBY_API_KEY="ashby_live_..."
export ASHBY_WORKSPACE_NAME="acme"          # used in tool descriptions and logs
export ASHBY_STALE_DEFAULT_DAYS="30"        # default for stale_candidates
export ASHBY_VELOCITY_LOOKBACK_DAYS="90"    # window for pipeline_velocity
```

#### `ASHBY_API_KEY`
The generated key from Admin → API Keys. Use HTTP Basic auth: the key is the username, password is blank. The server handles this automatically.

#### `ASHBY_WORKSPACE_NAME`
A short label (no spaces) used in tool descriptions so recruiters know which workspace they are querying when multiple workspaces are wired into one Claude install. Optional — defaults to `default`.

#### `ASHBY_STALE_DEFAULT_DAYS`
Default value for the `days_inactive` parameter when the recruiter does not specify one. 30 is the sane starting point for engineering pipelines; 14 for high-volume sales pipelines.

#### `ASHBY_VELOCITY_LOOKBACK_DAYS`
Window over which `pipeline_velocity` averages stage durations. 90 days smooths out single-candidate noise without going stale across hiring-plan changes.

### 4. Register with Claude Desktop

Edit `~/Library/Application Support/Claude/claude_desktop_config.json` (macOS) or `%APPDATA%\Claude\claude_desktop_config.json` (Windows):

```json
{
  "mcpServers": {
    "ashby-recruiting": {
      "command": "python",
      "args": ["-m", "ashby_recruiting_mcp.server"],
      "env": {
        "ASHBY_API_KEY": "ashby_live_...",
        "ASHBY_WORKSPACE_NAME": "acme",
        "ASHBY_STALE_DEFAULT_DAYS": "30",
        "ASHBY_VELOCITY_LOOKBACK_DAYS": "90"
      }
    }
  }
}
```

Restart Claude Desktop. You should see ~11 tools registered under `ashby-recruiting`.

### 5. Sanity-check

Ask Claude: "Find candidates whose name starts with `Smith`." Then: "Show me stale candidates in the senior backend engineer pipeline." Compare the output against the equivalent filtered view in Ashby. Tune `ASHBY_STALE_DEFAULT_DAYS` until the result matches the recruiter's intuition for "should have heard back by now".

## Watch-outs

- **API keys bypass per-recruiter permissions.** An Ashby API key acts with full admin scope. Anyone with access to the MCP client sees every candidate the workspace contains, including senior-leadership pipelines and rejected candidates with sensitive notes. Guard: scope the MCP install to recruiting-team machines only, document the exposure with security, rotate quarterly.
- **Stage names drift across pipelines.** `pipeline_velocity` reads the pipeline definition fresh on every call so renames don't break the helper, but year-over-year comparisons get noisy when the pipeline shape changes. Guard: snapshot pipeline definitions quarterly if you care about historical trend lines.
- **Light-write tools bypass Ashby approval workflows.** `add_note` and `add_tag` write directly through the API — they do not trigger any approval that would normally fire in the UI. Guard: do not use light-write tools for status-change-equivalent tags (`hired`, `offer-extended`); reserve for descriptive tags only.
- **Rate limits are workspace-shared.** Ashby's API is rate-limited per workspace, not per key. A chatty MCP session can crowd out the candidate-engagement automation that depends on the same workspace. Guard: cap concurrent calls (the scaffold uses one client per call) and watch for 429s in the logs during heavy use.
- **Candidate data is regulated.** EU candidates fall under GDPR; California candidates under CCPA. Surfacing candidate notes through Claude potentially routes regulated data through the Anthropic API. Guard: confirm the workspace's AI policy explicitly authorizes this.

## Limits and TODOs (before production use)

- [ ] Add request-level retries with exponential backoff on 429 and 5xx (Ashby returns 429 readily under sustained load).
- [ ] Replace `str(data)` response stringification with structured JSON serialization that strips PII fields the recruiter did not ask for.
- [ ] Write integration tests against an Ashby sandbox workspace.
- [ ] Add structured logging via `python-json-logger` with candidate IDs and tool names; never log raw note bodies.
- [ ] Wire optional Sentry / OpenTelemetry export for production telemetry.
- [ ] Validate stage IDs returned by `pipeline_velocity` against the live pipeline shape on first call per session, fail loud if cached IDs no longer exist.
- [ ] Add an allow-list env var (`ASHBY_ALLOWED_JOB_IDS`) to scope MCP visibility to a subset of jobs for confidentiality-sensitive roles.
- [ ] Audit-log every light-write tool call to a separate file the recruiting-ops lead reviews weekly.

"""ashby_recruiting_mcp — MCP server for Ashby recruiting workflows."""

__version__ = "0.1.0"

"""
ashby-recruiting-mcp — MCP server tuned for recruiting workflows on Ashby.

Exposes object-read tools, search tools, two recruiting helpers
(stale_candidates, pipeline_velocity), and two light-write tools
(add_note, add_tag). Read-mostly by design — Claude asks, summarizes,
documents; the recruiter drives every candidate-facing change in the
Ashby UI.

STATUS: scaffold — not runtime-tested. Adapt the field paths, stage
names, and pipeline IDs to your workspace before use. Ashby uses POST
for almost every endpoint (yes, even reads); responses are wrapped in
{success, results} and pagination is cursor-based via the `cursor`
field returned in `moreDataAvailable` responses.

Run as: python -m ashby_recruiting_mcp.server
"""

from __future__ import annotations

import base64
import os
from collections import defaultdict
from datetime import datetime, timedelta, timezone
from typing import Any

import httpx
from mcp.server import Server
from mcp.server.stdio import stdio_server
from mcp.types import TextContent, Tool

# ----- Configuration (read from env at startup) -----

ASHBY_API_KEY = os.environ.get("ASHBY_API_KEY")
WORKSPACE_NAME = os.environ.get("ASHBY_WORKSPACE_NAME", "default")
STALE_DEFAULT_DAYS = int(os.environ.get("ASHBY_STALE_DEFAULT_DAYS", "30"))
VELOCITY_LOOKBACK_DAYS = int(os.environ.get("ASHBY_VELOCITY_LOOKBACK_DAYS", "90"))

API_BASE = "https://api.ashbyhq.com"


def require_config() -> None:
    if not ASHBY_API_KEY:
        raise RuntimeError("ASHBY_API_KEY env var is required")


def auth_headers() -> dict[str, str]:
    # Ashby auth is HTTP Basic with the API key as username and empty password.
    token = base64.b64encode(f"{ASHBY_API_KEY}:".encode()).decode()
    return {
        "Authorization": f"Basic {token}",
        "Content-Type": "application/json",
        "Accept": "application/json",
    }


# ----- Ashby HTTP helpers -----
#
# Ashby uses POST for everything (even reads). All responses are shaped
# {"success": bool, "results": ..., "errors": [...], "moreDataAvailable": bool,
#  "nextCursor": "..."}. We unwrap `results` here and surface errors loudly.


async def ashby_post(path: str, body: dict[str, Any] | None = None) -> Any:
    payload = body or {}
    async with httpx.AsyncClient(timeout=30.0) as client:
        r = await client.post(f"{API_BASE}{path}", headers=auth_headers(), json=payload)
        r.raise_for_status()
        data = r.json()
    if not data.get("success", False):
        raise RuntimeError(f"Ashby API error on {path}: {data.get('errors')}")
    return data.get("results")


async def ashby_post_paginated(
    path: str, body: dict[str, Any] | None = None, max_pages: int = 20
) -> list[Any]:
    """Drains a cursor-paginated list endpoint. Caps at max_pages to bound calls."""
    out: list[Any] = []
    cursor: str | None = None
    payload = dict(body or {})
    for _ in range(max_pages):
        if cursor:
            payload["cursor"] = cursor
        async with httpx.AsyncClient(timeout=30.0) as client:
            r = await client.post(f"{API_BASE}{path}", headers=auth_headers(), json=payload)
            r.raise_for_status()
            data = r.json()
        if not data.get("success", False):
            raise RuntimeError(f"Ashby API error on {path}: {data.get('errors')}")
        results = data.get("results") or []
        if isinstance(results, list):
            out.extend(results)
        else:
            out.append(results)
        if not data.get("moreDataAvailable"):
            break
        cursor = data.get("nextCursor")
        if not cursor:
            break
    return out


# ----- Server + tool registry -----

server = Server(f"ashby-recruiting-{WORKSPACE_NAME}")


@server.list_tools()
async def list_tools() -> list[Tool]:
    return [
        Tool(
            name="get_candidate",
            description="Fetch full candidate record (contact info + current applications).",
            inputSchema={
                "type": "object",
                "properties": {"candidate_id": {"type": "string"}},
                "required": ["candidate_id"],
            },
        ),
        Tool(
            name="get_application",
            description="Fetch an application record (current stage, source, history).",
            inputSchema={
                "type": "object",
                "properties": {"application_id": {"type": "string"}},
                "required": ["application_id"],
            },
        ),
        Tool(
            name="get_job",
            description="Fetch a job record (hiring team, status, pipeline).",
            inputSchema={
                "type": "object",
                "properties": {"job_id": {"type": "string"}},
                "required": ["job_id"],
            },
        ),
        Tool(
            name="get_opening",
            description="Fetch an opening (req) record (target start, headcount).",
            inputSchema={
                "type": "object",
                "properties": {"opening_id": {"type": "string"}},
                "required": ["opening_id"],
            },
        ),
        Tool(
            name="search_candidates",
            description="Search candidates by name / email / company substring.",
            inputSchema={
                "type": "object",
                "properties": {
                    "query": {"type": "string"},
                    "limit": {"type": "integer", "default": 25},
                },
                "required": ["query"],
            },
        ),
        Tool(
            name="list_applications",
            description=(
                "List applications for a job. Optional status filter "
                "(Active | Hired | Archived)."
            ),
            inputSchema={
                "type": "object",
                "properties": {
                    "job_id": {"type": "string"},
                    "status": {
                        "type": "string",
                        "enum": ["Active", "Hired", "Archived"],
                    },
                },
                "required": ["job_id"],
            },
        ),
        Tool(
            name="list_jobs",
            description="List jobs in the workspace. Optional status filter (Open | Closed | Draft).",
            inputSchema={
                "type": "object",
                "properties": {
                    "status": {
                        "type": "string",
                        "enum": ["Open", "Closed", "Draft"],
                    },
                },
            },
        ),
        Tool(
            name="stale_candidates",
            description=(
                "Active candidates with no application activity (stage change, note, "
                "interview event) for `days_inactive` days. Output grouped by current stage."
            ),
            inputSchema={
                "type": "object",
                "properties": {
                    "days_inactive": {
                        "type": "integer",
                        "default": STALE_DEFAULT_DAYS,
                    },
                    "job_id": {
                        "type": "string",
                        "description": "Optional — restrict to a single job's pipeline.",
                    },
                },
            },
        ),
        Tool(
            name="pipeline_velocity",
            description=(
                "Average days-in-stage per stage for a job's pipeline, computed across "
                "the configured lookback window. Surfaces where the funnel is stuck."
            ),
            inputSchema={
                "type": "object",
                "properties": {"job_id": {"type": "string"}},
                "required": ["job_id"],
            },
        ),
        Tool(
            name="add_note",
            description=(
                "Append a note to a candidate's record. Visible in the Ashby UI activity feed. "
                "Use for documenting Claude-summarized context, not for status changes."
            ),
            inputSchema={
                "type": "object",
                "properties": {
                    "candidate_id": {"type": "string"},
                    "body": {"type": "string"},
                },
                "required": ["candidate_id", "body"],
            },
        ),
        Tool(
            name="add_tag",
            description=(
                "Apply a tag to a candidate. Reserve for descriptive tags "
                "(e.g. `phone-screen-passed`); never use for status-equivalents like `hired`."
            ),
            inputSchema={
                "type": "object",
                "properties": {
                    "candidate_id": {"type": "string"},
                    "tag": {"type": "string"},
                },
                "required": ["candidate_id", "tag"],
            },
        ),
    ]


# ----- Tool dispatch -----


@server.call_tool()
async def call_tool(name: str, arguments: dict[str, Any]) -> list[TextContent]:
    require_config()

    if name == "get_candidate":
        data = await ashby_post(
            "/candidate.info", {"id": arguments["candidate_id"]}
        )
        return [TextContent(type="text", text=str(data))]

    if name == "get_application":
        data = await ashby_post(
            "/application.info", {"id": arguments["application_id"]}
        )
        return [TextContent(type="text", text=str(data))]

    if name == "get_job":
        data = await ashby_post("/job.info", {"id": arguments["job_id"]})
        return [TextContent(type="text", text=str(data))]

    if name == "get_opening":
        data = await ashby_post("/opening.info", {"openingId": arguments["opening_id"]})
        return [TextContent(type="text", text=str(data))]

    if name == "search_candidates":
        data = await ashby_post(
            "/candidate.search",
            {
                "query": arguments["query"],
                "limit": arguments.get("limit", 25),
            },
        )
        return [TextContent(type="text", text=str(data))]

    if name == "list_applications":
        body: dict[str, Any] = {"jobId": arguments["job_id"]}
        if status := arguments.get("status"):
            body["status"] = status
        results = await ashby_post_paginated("/application.list", body)
        return [TextContent(type="text", text=str(results))]

    if name == "list_jobs":
        body = {}
        if status := arguments.get("status"):
            body["status"] = status
        results = await ashby_post_paginated("/job.list", body)
        return [TextContent(type="text", text=str(results))]

    if name == "stale_candidates":
        days_inactive = arguments.get("days_inactive", STALE_DEFAULT_DAYS)
        cutoff = datetime.now(timezone.utc) - timedelta(days=days_inactive)
        # Pull active applications, optionally scoped to one job.
        list_body: dict[str, Any] = {"status": "Active"}
        if jid := arguments.get("job_id"):
            list_body["jobId"] = jid
        applications = await ashby_post_paginated("/application.list", list_body)

        stale_by_stage: dict[str, list[dict[str, Any]]] = defaultdict(list)
        for app in applications:
            # Ashby returns ISO 8601 timestamps; the field name has shifted historically.
            last_activity_str = (
                app.get("lastActivityAt")
                or app.get("updatedAt")
                or app.get("createdAt")
            )
            if not last_activity_str:
                continue
            try:
                last_activity = datetime.fromisoformat(
                    last_activity_str.replace("Z", "+00:00")
                )
            except ValueError:
                continue
            if last_activity < cutoff:
                stage_name = (app.get("currentInterviewStage") or {}).get("title") or "Unknown"
                stale_by_stage[stage_name].append(
                    {
                        "applicationId": app.get("id"),
                        "candidateId": (app.get("candidate") or {}).get("id"),
                        "candidateName": (app.get("candidate") or {}).get("name"),
                        "lastActivityAt": last_activity_str,
                        "daysInactive": (datetime.now(timezone.utc) - last_activity).days,
                    }
                )
        out = {
            "daysInactiveThreshold": days_inactive,
            "totalStale": sum(len(v) for v in stale_by_stage.values()),
            "byStage": dict(stale_by_stage),
        }
        return [TextContent(type="text", text=str(out))]

    if name == "pipeline_velocity":
        job_id = arguments["job_id"]
        cutoff = datetime.now(timezone.utc) - timedelta(days=VELOCITY_LOOKBACK_DAYS)
        # Pull every application (active + archived + hired) in the job, then
        # walk each application's interviewStageChanges to compute durations.
        applications = await ashby_post_paginated("/application.list", {"jobId": job_id})

        durations_by_stage: dict[str, list[float]] = defaultdict(list)
        for app in applications:
            changes = await ashby_post(
                "/application.interviewStageChanges",
                {"applicationId": app.get("id")},
            ) or []
            # Sort changes oldest-first.
            changes_sorted = sorted(
                changes, key=lambda c: c.get("enteredStageAt") or ""
            )
            for i, change in enumerate(changes_sorted):
                entered_str = change.get("enteredStageAt")
                if not entered_str:
                    continue
                try:
                    entered = datetime.fromisoformat(entered_str.replace("Z", "+00:00"))
                except ValueError:
                    continue
                if entered < cutoff:
                    continue
                # Stage exit is the next change's entry, or now if this is the latest.
                exit_dt = datetime.now(timezone.utc)
                if i + 1 < len(changes_sorted):
                    next_entered = changes_sorted[i + 1].get("enteredStageAt")
                    if next_entered:
                        try:
                            exit_dt = datetime.fromisoformat(
                                next_entered.replace("Z", "+00:00")
                            )
                        except ValueError:
                            pass
                duration_days = (exit_dt - entered).total_seconds() / 86400.0
                stage_name = (change.get("interviewStage") or {}).get("title") or "Unknown"
                durations_by_stage[stage_name].append(duration_days)

        velocity = {
            stage: {
                "samples": len(values),
                "avgDays": round(sum(values) / len(values), 2) if values else 0.0,
            }
            for stage, values in durations_by_stage.items()
        }
        out = {
            "jobId": job_id,
            "lookbackDays": VELOCITY_LOOKBACK_DAYS,
            "byStage": velocity,
        }
        return [TextContent(type="text", text=str(out))]

    if name == "add_note":
        result = await ashby_post(
            "/candidate.note.create",
            {
                "candidateId": arguments["candidate_id"],
                "note": arguments["body"],
            },
        )
        return [
            TextContent(
                type="text",
                text=f"Added note to candidate {arguments['candidate_id']}: {result}",
            )
        ]

    if name == "add_tag":
        result = await ashby_post(
            "/candidateTag.create",
            {
                "candidateId": arguments["candidate_id"],
                "tag": arguments["tag"],
            },
        )
        return [
            TextContent(
                type="text",
                text=f"Tagged candidate {arguments['candidate_id']} with `{arguments['tag']}`: {result}",
            )
        ]

    raise ValueError(f"Unknown tool: {name}")


# ----- Entrypoint -----


async def main() -> None:
    require_config()
    async with stdio_server() as (read, write):
        await server.run(read, write, server.create_initialization_options())


if __name__ == "__main__":
    import asyncio

    asyncio.run(main())