Lever MCP server for recruiting workflows

[project]
name = "lever-recruiting-mcp"
version = "0.1.0"
description = "Read-mostly MCP server exposing the Lever Data API to Claude / MCP-compatible clients."
readme = "README.md"
requires-python = ">=3.11"
license = { text = "MIT" }
authors = [
    { name = "ooligo authors" }
]
dependencies = [
    "mcp>=0.9.0",
    "httpx>=0.27.0",
    "pydantic>=2.7.0",
]

[project.optional-dependencies]
dev = [
    "pytest>=8.0.0",
    "pytest-asyncio>=0.23.0",
    "ruff>=0.5.0",
]

[project.scripts]
lever-recruiting-mcp = "lever_recruiting_mcp.server:main"

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

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

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

# Lever recruiting MCP server

A read-mostly MCP server exposing the Lever Data API v1 as tools to Claude Desktop, Claude Code, or any MCP-compatible client. Six read tools cover daily recruiter questions; one cautious write tool (`note_stage_stuck`) adds an internal note.

**This is a scaffold, not a runtime-tested production server.** The tool implementations are written against the Lever Data API's documented shape, but the recruiting engineer is responsible for verifying each tool against a Lever Sandbox tenant before flipping production credentials. The disclaimer is repeated in `server.py`'s module docstring.

## Lever data model (read this before wiring)

Lever's model differs from Greenhouse/Ashby in ways that matter for every tool:

- The candidate-in-pipeline object is an **Opportunity**, not an application. An Opportunity ties a **Contact** (the person) to one or more **Postings** (the jobs) and carries a single current `stage`.
- A **Posting** is the job. Postings have a `state` (`published`, `internal`, `closed`, `draft`, `pending`, `rejected`). This server treats `published` as "open."
- **Stages** are pipeline stages. An Opportunity's `stage` field is a stage **ID**, not a name — run `list_stages` first to resolve IDs to display text. `get_funnel_for_posting` does this resolution for you.
- Timestamps are **integer milliseconds since the Unix epoch**, not ISO-8601 strings. `lastInteractionAt` (any activity) and `lastAdvancedAt` (last stage advance) are the two staleness signals.
- Lever exposes the *current* stage and `lastAdvancedAt`, but has **no public stage-transition-history endpoint**. `get_opportunity_detail` returns current state + notes, not a full transition log.

## Install

```bash
cd apps/web/public/artifacts/mcp-server-lever-recruiting/
pip install -e .
# or
uv pip install -e .
```

The package exposes a `lever-recruiting-mcp` CLI entrypoint.

## Environment variables

### `LEVER_API_KEY` (required)

The Data API key from Lever → Settings → Integrations and API → API credentials. Generate a key scoped to the **minimum** resources needed:

- **Read** on `opportunities`, `postings`, `stages`, `notes`, `archive_reasons`.
- **Write** on `notes` only (required for `note_stage_stuck`).

Lever API keys are all-or-nothing on the account they belong to; there is no per-endpoint scoping the way Greenhouse Harvest offers. Compensate by using a dedicated integration user with the narrowest Lever role that still exposes the postings you need, and by keeping the key out of shared config. Wider account access silently turns the server into a higher-blast-radius surface.

### `LEVER_PERFORM_AS_USER_ID` (required)

The Lever user ID that writes will be attributed to via the `perform_as` query parameter. Find it by calling `GET https://api.lever.co/v1/users` and matching on the integration user's email.

This is required even if you only use the read tools — the server validates it at startup so writes can't slip through unattributed later.

## MCP client registration

### Claude Desktop

Add to `claude_desktop_config.json` (location varies by OS — `~/Library/Application Support/Claude/` on macOS, `%APPDATA%\Claude\` on Windows):

```json
{
  "mcpServers": {
    "lever-recruiting": {
      "command": "uv",
      "args": ["run", "lever-recruiting-mcp"],
      "cwd": "/absolute/path/to/mcp-server-lever-recruiting",
      "env": {
        "LEVER_API_KEY": "...",
        "LEVER_PERFORM_AS_USER_ID": "..."
      }
    }
  }
}
```

Restart Claude Desktop. The seven tools should appear in the tools panel.

### Claude Code

In your project's `.claude/settings.local.json`:

```json
{
  "mcpServers": {
    "lever-recruiting": {
      "command": "uv",
      "args": ["run", "lever-recruiting-mcp"],
      "cwd": "/absolute/path/to/mcp-server-lever-recruiting",
      "env": {
        "LEVER_API_KEY": "...",
        "LEVER_PERFORM_AS_USER_ID": "..."
      }
    }
  }
}
```

### Other MCP clients (Cursor, Continue, etc.)

Most accept the same `command + args + env` shape. Refer to the client's MCP documentation.

## Sanity-check invocation

Before the first real use, run the server against a Lever **Sandbox** tenant. Lever provides a separate sandbox environment (`hire.sandbox.lever.co`) with its own API host and credentials — request one from your Lever CSM or through the developer portal.

```bash
LEVER_API_KEY=sandbox_key \
LEVER_PERFORM_AS_USER_ID=sandbox_user_id \
lever-recruiting-mcp --help
```

(The CLI is the MCP stdio server; `--help` is not implemented in this scaffold. The intent is to confirm the package installed and the entrypoint resolved.)

For per-tool validation, the recommended flow is:

1. Register the server with Claude Desktop pointed at sandbox credentials.
2. Ask Claude to call each tool with known sandbox inputs and verify the responses match what you see in the sandbox Lever UI.
3. Only after every tool is verified, swap to production credentials.

## Security model

- **Auth.** Lever API key as Basic-auth username, empty password. Lever's documented pattern.
- **Writes.** Only `note_stage_stuck` mutates state. Attributed via the `perform_as` query parameter so the Lever activity feed shows the integration user, not just the API key.
- **Rate limit.** Token-bucket at 8 req/s by default (Lever steady-state ceiling is 10 req/s per API key, burst to 20). Lever separately caps application POSTs at 2 req/s — the single-call note tool stays well under. Lower the default if other systems share the key.
- **PII in MCP responses.** The server returns the data the API returns — including candidate names. The calling Claude session is downstream; the session's data-handling posture is the recruiter's responsibility. Don't paste session transcripts into shared Slack channels; don't log raw responses to your own audit table.
- **Audit log.** The server logs every tool call to stderr at INFO level with PII-stripped arguments. Recruiting engineer is responsible for capturing stderr into a durable audit log (e.g. via systemd journal, Docker log driver, or a wrapping script that tees to a file).

## Known limits — numbered TODO before production use

The scaffold is honest about what it doesn't do yet. Treat each as a numbered TODO to close before broad production use.

1. **Not runtime-tested.** Every tool needs validation against a Lever Sandbox tenant. The smoke check in this README is a credentials check, not a per-tool validation.
2. **Pagination max page count.** The async iterator caps at 50 pages per call (5,000 records at 100/page). For accounts with very large opportunity volumes, the cap needs raising or replacement with a streaming pattern.
3. **`list_postings_stalled` is O(postings × opportunities).** It walks every opportunity on every published posting to find the latest advance. On large accounts this is slow and quota-heavy — run it off-peak, or narrow by team first. A webhook-fed cache is the real fix.
4. **No stage-transition history.** Lever's Data API has no endpoint for a full stage-by-stage transition log. `get_opportunity_detail` returns current stage + `lastAdvancedAt` + notes only. Do not present its output as a complete audit trail of every stage move.
5. **No request retry beyond 429.** The 429 handler retries once after a 1-second backoff. Other 5xx errors propagate; the recruiting engineer wraps the calls if more retry resilience is needed.
6. **API-key blast radius.** Lever keys are account-scoped, not endpoint-scoped. The server cannot narrow what the key can reach; that has to be done at the Lever integration-user level. See the env-var note above.
7. **No tests.** A pytest suite for the rate limiter and the offset-pagination loop is the obvious first addition; full integration tests against a sandbox tenant are the second.

## What this server intentionally does NOT do

- **No `delete_*` tools.** Deletes happen in the Lever UI, with the audit trail that produces.
- **No opportunity-state mutations** (advance stage, archive, change owner, send email). Those are recruiter decisions and need explicit per-tool justification — adding them would compromise the read-mostly posture.
- **No bulk send / outbound email.** Outreach belongs in a sourcing tool with proper unsubscribe handling, not in an MCP read-tool surface.
- **No PII normalization** (no name-redaction in responses). The server returns what Lever returns; downstream redaction is the recruiter's responsibility.

## Adding a new tool

If you need a new tool:

1. Add a Pydantic input schema and an async implementation in `server.py`.
2. Register it in `TOOL_REGISTRY` with a clear description.
3. If it's a write tool, document the per-tool justification in the function's docstring (see `note_stage_stuck` for the template). Confirm the `perform_as` attribution flows through and stays under Lever's 2 req/s POST cap.
4. Validate against sandbox.
5. Update this README's tool list.

The scaffold's structure makes this a 30-60 minute change per tool. The discipline is in the per-tool justification step — it's the only thing that prevents the read-mostly posture from drifting.

"""Lever recruiting MCP server."""

__version__ = "0.1.0"

"""
Lever recruiting MCP server.

Exposes seven tools to MCP-compatible clients (Claude Desktop, Claude Code,
Cursor, etc.) backed by the Lever Data API v1. Six are read; one is the
cautious write (`note_stage_stuck`).

NOT runtime-tested against a live Lever tenant. The tool dispatch
implementations are written against Lever's documented Data API shape
(https://hire.lever.co/developer/documentation as of 2026-Q2), but the
production deployment requires the recruiting engineer to verify each tool
against a Lever Sandbox tenant before flipping the production credentials.

Lever data model note (differs from Greenhouse):
  - The candidate-in-pipeline object is an *Opportunity*, not an application.
    An Opportunity ties a *Contact* (the person) to one or more *Postings*
    (the jobs) and carries a single current `stage`.
  - Timestamps are integer milliseconds since the Unix epoch, NOT ISO-8601
    strings. `lastInteractionAt` (any activity) and `lastAdvancedAt` (last
    stage advance) are the two staleness signals this server uses.
  - Lever exposes the *current* stage and `lastAdvancedAt`, but has no
    public stage-transition-history endpoint. `get_opportunity_detail`
    returns current state + notes, not a full transition log — see the
    docstring there.

Security model:
  - Auth: Lever API key via Basic auth, key as username, empty password —
    Lever's documented pattern.
  - Writes: only `note_stage_stuck` mutates state; uses the `perform_as`
    query parameter (a Lever user ID) for audit attribution.
  - Rate limit: token-bucket (default 8 req/s; Lever steady-state ceiling is
    10 req/s per API key, burst to 20). Application POSTs are separately
    capped by Lever at 2 req/s — the note tool is single-call so it stays
    well under that.
  - Pagination: cursor-based via the `next` offset token in the JSON body;
    the implementations loop while `hasNext` is true.
  - Audit: every tool call logged to stderr at INFO level with tool name,
    parameters (PII-stripped), and response status. Recruiting engineer
    is responsible for capturing these into a durable audit log.
"""

from __future__ import annotations

import asyncio
import logging
import os
import time
from collections.abc import AsyncIterator
from typing import Any

import httpx
from mcp.server import Server
from mcp.types import Tool, TextContent
from pydantic import BaseModel, Field

logger = logging.getLogger(__name__)


# --- Configuration ------------------------------------------------------------

LEVER_API_BASE = "https://api.lever.co/v1"
DEFAULT_RATE_LIMIT_PER_S = 8  # Lever steady-state ceiling: 10 req/s (burst 20).
DEFAULT_TIMEOUT_S = 30.0
DEFAULT_PER_PAGE = 100


def _require_env(name: str) -> str:
    val = os.environ.get(name)
    if not val:
        raise RuntimeError(
            f"Required env var {name} not set. The Lever MCP server cannot start "
            f"without credentials. See README.md for setup."
        )
    return val


def _ms_to_epoch_s(ms: int | None) -> float | None:
    """Lever timestamps are integer ms since epoch. Convert to float seconds."""
    if ms is None:
        return None
    return ms / 1000.0


# --- Rate limiter -------------------------------------------------------------


class TokenBucket:
    """Simple token-bucket rate limiter, async-safe."""

    def __init__(self, rate: int, per_seconds: float) -> None:
        self.rate = rate
        self.per_seconds = per_seconds
        self.tokens = float(rate)
        self.last_refill = time.monotonic()
        self._lock = asyncio.Lock()

    async def acquire(self) -> None:
        async with self._lock:
            now = time.monotonic()
            elapsed = now - self.last_refill
            self.tokens = min(self.rate, self.tokens + elapsed * (self.rate / self.per_seconds))
            self.last_refill = now
            if self.tokens >= 1:
                self.tokens -= 1
                return
            wait = (1 - self.tokens) * (self.per_seconds / self.rate)
        await asyncio.sleep(wait)
        await self.acquire()


# --- Lever client -------------------------------------------------------------


class LeverClient:
    """Thin async wrapper around the Lever Data API v1."""

    def __init__(
        self,
        api_key: str,
        perform_as_user_id: str,
        rate_limit_per_s: int = DEFAULT_RATE_LIMIT_PER_S,
    ) -> None:
        self.api_key = api_key
        self.perform_as_user_id = perform_as_user_id
        self.bucket = TokenBucket(rate=rate_limit_per_s, per_seconds=1.0)
        self._client = httpx.AsyncClient(
            base_url=LEVER_API_BASE,
            auth=(api_key, ""),
            timeout=DEFAULT_TIMEOUT_S,
            headers={"User-Agent": "lever-recruiting-mcp/0.1.0"},
        )

    async def close(self) -> None:
        await self._client.aclose()

    async def _request(
        self,
        method: str,
        path: str,
        *,
        params: dict[str, Any] | None = None,
        json: dict[str, Any] | None = None,
    ) -> httpx.Response:
        await self.bucket.acquire()
        resp = await self._client.request(method, path, params=params, json=json)
        if resp.status_code == 429:
            # Lever returns 429 on rate-limit breach; Retry-After is not
            # reliably documented. Back off conservatively and retry once.
            await asyncio.sleep(1.0)
            await self.bucket.acquire()
            resp = await self._client.request(method, path, params=params, json=json)
        resp.raise_for_status()
        return resp

    async def paginate(
        self,
        path: str,
        params: dict[str, Any] | None = None,
        *,
        max_pages: int = 50,
    ) -> AsyncIterator[dict[str, Any]]:
        """
        Yield each item from a paginated Lever list endpoint.

        Lever list responses have shape {"data": [...], "next": "<token>",
        "hasNext": bool}. The `next` token is passed back as the `offset`
        query parameter on the following request.
        """
        params = dict(params or {})
        params.setdefault("limit", DEFAULT_PER_PAGE)
        offset: str | None = None
        page = 0
        while page < max_pages:
            call_params = dict(params)
            if offset:
                call_params["offset"] = offset
            resp = await self._request("GET", path, params=call_params)
            body = resp.json()
            for item in body.get("data", []):
                yield item
            if not body.get("hasNext"):
                return
            offset = body.get("next")
            if not offset:
                return
            page += 1

    async def get_one(self, path: str) -> dict[str, Any]:
        """Fetch a single resource. Lever wraps it as {"data": {...}}."""
        resp = await self._request("GET", path)
        return resp.json().get("data", {})

    async def stage_id_to_text(self) -> dict[str, str]:
        """Build a {stage_id: stage_text} map from /stages."""
        out: dict[str, str] = {}
        async for stage in self.paginate("/stages"):
            out[stage.get("id", "")] = stage.get("text", "")
        return out


# --- Pydantic schemas ---------------------------------------------------------


class ListOpportunitiesInStageInput(BaseModel):
    posting_id: str = Field(..., description="Lever posting ID")
    stage_id: str = Field(..., description="Lever stage ID (from list_stages)")
    stale_after_days: int | None = Field(
        None,
        description="Optional filter: opportunities whose last interaction was more than N days ago",
    )


class GetOpportunityDetailInput(BaseModel):
    opportunity_id: str = Field(..., description="Lever opportunity ID")


class ListPostingsOpenInput(BaseModel):
    team: str | None = Field(None, description="Optional team/department name filter")


class GetFunnelForPostingInput(BaseModel):
    posting_id: str = Field(..., description="Lever posting ID")


class ListPostingsStalledInput(BaseModel):
    stale_after_days: int = Field(
        7, description="A posting is stalled if no opportunity advanced in this many days"
    )


class SearchOpportunitiesByTagInput(BaseModel):
    tag: str = Field(..., description="Lever opportunity tag to match exactly")


class ListStagesInput(BaseModel):
    pass


class NoteStageStuckInput(BaseModel):
    opportunity_id: str = Field(..., description="Lever opportunity ID")
    note_body: str = Field(..., description="The note text. Visible internally in Lever.")


# --- Tool implementations -----------------------------------------------------


async def list_opportunities_in_stage(
    client: LeverClient, args: ListOpportunitiesInStageInput
) -> list[dict[str, Any]]:
    """Return opportunities currently in a stage on a given posting."""
    out: list[dict[str, Any]] = []
    cutoff_s = (
        time.time() - args.stale_after_days * 86400 if args.stale_after_days else None
    )
    async for opp in client.paginate(
        "/opportunities",
        params={"posting_id": args.posting_id, "stage_id": args.stage_id},
    ):
        last_interaction_s = _ms_to_epoch_s(opp.get("lastInteractionAt"))
        if cutoff_s is not None and last_interaction_s is not None:
            if last_interaction_s > cutoff_s:
                continue
        out.append(
            {
                "opportunity_id": opp.get("id"),
                "name": opp.get("name"),
                "stage_id": opp.get("stage"),
                "last_interaction_at_ms": opp.get("lastInteractionAt"),
                "last_advanced_at_ms": opp.get("lastAdvancedAt"),
                "archived": opp.get("archived"),
            }
        )
    return out


async def get_opportunity_detail(
    client: LeverClient, args: GetOpportunityDetailInput
) -> dict[str, Any]:
    """
    Return an opportunity's current state plus its notes.

    Lever exposes the *current* stage and `lastAdvancedAt`, but has NO public
    stage-transition-history endpoint. This tool returns what Lever exposes:
    current stage, staleness timestamps, tags, sources, and the notes feed.
    It does not reconstruct a full stage-by-stage transition log — Lever's
    Data API does not offer one. Do not present the output as a complete
    audit trail of every stage move.
    """
    opp = await client.get_one(f"/opportunities/{args.opportunity_id}")
    notes: list[dict[str, Any]] = []
    async for note in client.paginate(f"/opportunities/{args.opportunity_id}/notes"):
        notes.append(
            {
                "at_ms": note.get("createdAt"),
                "by_user_id": note.get("user"),
                "value": note.get("value"),
            }
        )
    return {
        "opportunity_id": opp.get("id"),
        "name": opp.get("name"),
        "current_stage_id": opp.get("stage"),
        "last_interaction_at_ms": opp.get("lastInteractionAt"),
        "last_advanced_at_ms": opp.get("lastAdvancedAt"),
        "tags": opp.get("tags"),
        "sources": opp.get("sources"),
        "posting_ids": opp.get("postings"),
        "archived": opp.get("archived"),
        "notes": notes,
    }


async def list_postings_open(
    client: LeverClient, args: ListPostingsOpenInput
) -> list[dict[str, Any]]:
    """List published (open) postings."""
    out: list[dict[str, Any]] = []
    async for posting in client.paginate("/postings", params={"state": "published"}):
        categories = posting.get("categories") or {}
        team = categories.get("team")
        if args.team and team != args.team:
            continue
        out.append(
            {
                "posting_id": posting.get("id"),
                "title": posting.get("text"),
                "state": posting.get("state"),
                "team": team,
                "department": categories.get("department"),
                "location": categories.get("location"),
                "created_at_ms": posting.get("createdAt"),
                "hiring_manager_id": posting.get("hiringManager"),
                "owner_id": posting.get("owner"),
            }
        )
    return out


async def list_stages(client: LeverClient, args: ListStagesInput) -> list[dict[str, Any]]:
    """List all pipeline stages with their IDs and display text."""
    out: list[dict[str, Any]] = []
    async for stage in client.paginate("/stages"):
        out.append({"stage_id": stage.get("id"), "text": stage.get("text")})
    return out


async def get_funnel_for_posting(
    client: LeverClient, args: GetFunnelForPostingInput
) -> dict[str, int]:
    """Return opportunity count per stage (human-readable stage names) for a posting."""
    stage_map = await client.stage_id_to_text()
    counts: dict[str, int] = {}
    async for opp in client.paginate(
        "/opportunities", params={"posting_id": args.posting_id}
    ):
        stage_id = opp.get("stage") or "unknown"
        stage_name = stage_map.get(stage_id, stage_id)
        counts[stage_name] = counts.get(stage_name, 0) + 1
    return counts


async def list_postings_stalled(
    client: LeverClient, args: ListPostingsStalledInput
) -> list[dict[str, Any]]:
    """List postings where no opportunity has advanced a stage in N days."""
    cutoff_s = time.time() - args.stale_after_days * 86400
    stalled: list[dict[str, Any]] = []
    async for posting in client.paginate("/postings", params={"state": "published"}):
        latest_advance_s = 0.0
        async for opp in client.paginate(
            "/opportunities", params={"posting_id": posting["id"]}
        ):
            advanced_s = _ms_to_epoch_s(opp.get("lastAdvancedAt"))
            if advanced_s and advanced_s > latest_advance_s:
                latest_advance_s = advanced_s
        if latest_advance_s > 0 and latest_advance_s < cutoff_s:
            stalled.append(
                {
                    "posting_id": posting.get("id"),
                    "title": posting.get("text"),
                    "days_since_advance": int((time.time() - latest_advance_s) / 86400),
                }
            )
    return stalled


async def search_opportunities_by_tag(
    client: LeverClient, args: SearchOpportunitiesByTagInput
) -> list[dict[str, Any]]:
    """Search opportunities by an exact tag match (Lever `tag` filter)."""
    out: list[dict[str, Any]] = []
    async for opp in client.paginate("/opportunities", params={"tag": args.tag}):
        out.append(
            {
                "opportunity_id": opp.get("id"),
                "name": opp.get("name"),
                "stage_id": opp.get("stage"),
                "tags": opp.get("tags"),
            }
        )
    return out


async def note_stage_stuck(
    client: LeverClient, args: NoteStageStuckInput
) -> dict[str, Any]:
    """
    Add an internal note to an opportunity. The single write tool exposed.

    Per-tool justification:
      - Required to log "Claude flagged this opportunity as stage-stuck" so
        the action is visible in the Lever activity feed and not silent.
      - No opportunity-state mutation (does not move stages, does not send
        candidate emails, does not archive, does not change the owner).
      - Attributed via the `perform_as` query parameter (a Lever user ID) so
        the Lever activity feed shows the recruiting-engineer user, not just
        the API key. Lever caps application POSTs at 2 req/s; this is one
        call so it stays well under.
    """
    body = {"value": args.note_body}
    resp = await client._request(
        "POST",
        f"/opportunities/{args.opportunity_id}/notes",
        params={"perform_as": client.perform_as_user_id},
        json=body,
    )
    return {"status": "ok", "note": resp.json().get("data", {})}


# --- MCP server wiring --------------------------------------------------------

TOOL_REGISTRY: dict[str, tuple[type[BaseModel], Any, str]] = {
    "list_opportunities_in_stage": (
        ListOpportunitiesInStageInput,
        list_opportunities_in_stage,
        "List opportunities currently in a named stage on a given posting. Optionally filter by staleness.",
    ),
    "get_opportunity_detail": (
        GetOpportunityDetailInput,
        get_opportunity_detail,
        "Return an opportunity's current stage, staleness timestamps, tags, and notes feed. Not a full stage-transition log.",
    ),
    "list_postings_open": (
        ListPostingsOpenInput,
        list_postings_open,
        "List published (open) postings. Optional team filter.",
    ),
    "list_stages": (
        ListStagesInput,
        list_stages,
        "List all pipeline stages with their IDs and display text. Needed to resolve stage_id inputs.",
    ),
    "get_funnel_for_posting": (
        GetFunnelForPostingInput,
        get_funnel_for_posting,
        "Return opportunity counts per stage (human-readable names) for a single posting.",
    ),
    "list_postings_stalled": (
        ListPostingsStalledInput,
        list_postings_stalled,
        "List postings where no opportunity has advanced a stage in N days.",
    ),
    "search_opportunities_by_tag": (
        SearchOpportunitiesByTagInput,
        search_opportunities_by_tag,
        "Search opportunities by an exact tag match.",
    ),
    "note_stage_stuck": (
        NoteStageStuckInput,
        note_stage_stuck,
        "Write tool: add an internal note to an opportunity. Audit-attributed via the perform_as user ID.",
    ),
}


def build_server() -> Server:
    server = Server("lever-recruiting-mcp")

    api_key = _require_env("LEVER_API_KEY")
    perform_as = _require_env("LEVER_PERFORM_AS_USER_ID")
    client = LeverClient(api_key=api_key, perform_as_user_id=perform_as)

    @server.list_tools()
    async def _list_tools() -> list[Tool]:
        return [
            Tool(
                name=name,
                description=desc,
                inputSchema=schema.model_json_schema(),
            )
            for name, (schema, _, desc) in TOOL_REGISTRY.items()
        ]

    @server.call_tool()
    async def _call_tool(name: str, arguments: dict[str, Any]) -> list[TextContent]:
        if name not in TOOL_REGISTRY:
            return [TextContent(type="text", text=f"Unknown tool: {name}")]
        schema, fn, _ = TOOL_REGISTRY[name]
        try:
            args = schema.model_validate(arguments)
        except Exception as exc:
            logger.warning("Tool %s called with invalid args: %s", name, exc)
            return [TextContent(type="text", text=f"Invalid arguments: {exc}")]

        # Audit: log tool call with PII-light args (drop free-text body for note tool).
        audit_args = arguments.copy()
        if name == "note_stage_stuck":
            audit_args["note_body"] = f"<{len(arguments.get('note_body', ''))} chars>"
        logger.info("Tool call: %s args=%s", name, audit_args)

        try:
            result = await fn(client, args)
        except httpx.HTTPStatusError as exc:
            logger.warning("Tool %s HTTP error: %s", name, exc)
            return [
                TextContent(
                    type="text",
                    text=f"Lever API error {exc.response.status_code}: {exc.response.text[:500]}",
                )
            ]
        except Exception as exc:
            logger.exception("Tool %s failed", name)
            return [TextContent(type="text", text=f"Tool failed: {exc}")]

        # Result returned as JSON-shaped text content; the calling Claude session parses it.
        import json
        return [TextContent(type="text", text=json.dumps(result, default=str, indent=2))]

    return server


def main() -> None:
    """Entry point for `lever-recruiting-mcp` CLI."""
    logging.basicConfig(
        level=logging.INFO,
        format="%(asctime)s %(levelname)s %(name)s %(message)s",
    )
    from mcp.server.stdio import stdio_server

    async def _run() -> None:
        server = build_server()
        async with stdio_server() as (read_stream, write_stream):
            await server.run(
                read_stream,
                write_stream,
                server.create_initialization_options(),
            )

    asyncio.run(_run())


if __name__ == "__main__":
    main()