MCP server exposing Attio records and lists to Claude

[project]
name = "attio-revops-mcp"
version = "0.1.0"
description = "Read-mostly MCP server for Attio CRM records, lists, and gated attribute writes"
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/attio_revops_mcp"]

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

# mcp-server-attio-revops

A read-mostly MCP server over the [Attio](https://attio.com) REST API. Gives Claude four read tools — object discovery, record query, single-record fetch, list-entry query — and exactly one gated write, `update_record_attribute`, which is off by default and allowlisted per attribute. Built so a RevOps team can ask "which companies in the Q3 pipeline list have no owner set?" in chat without granting an agent the run of the CRM.

> **STATUS: scaffold — not runtime-tested.** The code follows the official `mcp` Python SDK conventions and the endpoint paths, scopes, and parameters track the public Attio API docs (docs.attio.com) as of July 2026, but it has not been executed against a live Attio workspace. Attribute slugs are workspace-specific and object configuration varies; verify against your own workspace before relying on it.

## Read this first: Attio ships an official hosted MCP server

Attio hosts its own MCP server at `https://mcp.attio.com/mcp`. It authenticates over OAuth (no key to store or rotate), exposes 30+ tools across nine areas — records, lists, comments, notes, tasks, meetings, emails, workspace, reporting — plus an SQL tool, auto-approves reads, and prompts for confirmation on writes. It is the right default for most teams, and it is less work than this.

Run this scaffold instead when one of the following is true:

- **You need a service-account identity, not a user identity.** The hosted server grants the signed-in user's own permissions. If a shared agent should see less than any individual human does, a workspace API key with a chosen scope set is the only way to express that.
- **You need to narrow the tool surface.** 30+ tools including SQL and semantic email search is a wide grant for an agent that only answers pipeline questions. Here you get five, and `ATTIO_ALLOWED_OBJECTS` bounds even the reads.
- **You need writes allowlisted per attribute.** Confirmation prompts depend on a human reading them. `ATTIO_WRITABLE_ATTRIBUTES` does not.
- **You need your own audit log.** A local process logs to your infrastructure.

If none of those apply, use the hosted server.

## What it exposes

### Reads

- `list_objects()` — `GET /v2/objects`. Returns each object's `api_slug`, nouns, and whether it is inside your allowlist. Call this before guessing a slug; Attio object and attribute slugs are per-workspace.
- `query_records(object, filter?, sorts?, limit=25, offset=0, attributes?)` — `POST /v2/objects/{object}/records/query`. The object must be in `ATTIO_ALLOWED_OBJECTS`. `limit` is clamped to 100 (Attio's own default is 500). Pass `attributes` to keep only the columns you care about.
- `get_record(object, record_id)` — `GET /v2/objects/{object}/records/{record_id}`. Returns the slimmed attribute map plus `web_url` so a human can open the record.
- `query_list_entries(list, filter?, sorts?, limit=25, offset=0)` — `POST /v2/lists/{list}/entries/query`. Lists are Attio's pipeline surface; use this rather than querying the parent object for stage questions.

### Write (gated)

- `update_record_attribute(object, record_id, attribute, value, justification)` — `PATCH /v2/objects/{object}/records/{record_id}`. Requires `ATTIO_ALLOW_WRITES=true`, a `{object}.{attribute}` entry in `ATTIO_WRITABLE_ATTRIBUTES`, and a justification of at least 10 characters. One attribute, one record, per call.

There is no delete tool, no bulk update, no SQL tool, and no `PUT` path. `PUT` is what overwrites and removes multiselect values; only `PATCH` is wired, and `PATCH` prepends. This server cannot erase an existing multiselect value.

## Setup

### 1. Install

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

### 2. Create an Attio API key

In Attio: **Workspace settings → Developers → create an integration**, then generate an access token for it. Scopes are chosen at creation and cannot be edited afterwards — to change them, create a new integration.

Grant the minimum for the tools you want:

| Tool | Scopes |
|---|---|
| `list_objects` | `object_configuration:read` |
| `query_records`, `get_record` | `record_permission:read`, `object_configuration:read` |
| `query_list_entries` | `list_entry:read`, `list_configuration:read` |
| `update_record_attribute` | `record_permission:read-write`, `object_configuration:read` |

If writes stay off — the default — do not grant `record_permission:read-write`. A read-only token means the write tool cannot fire even if someone flips `ATTIO_ALLOW_WRITES` by accident.

### 3. Configure environment

#### `ATTIO_API_KEY` (required)

The access token from step 2. Sent as `Authorization: Bearer <token>`. Store it in your OS keychain or secret manager, not in a dotfile that syncs.

#### `ATTIO_BASE_URL` (optional)

Defaults to `https://api.attio.com/v2`. Override only to point at a proxy.

#### `ATTIO_ALLOWED_OBJECTS` (recommended)

Comma-separated object `api_slug` values the agent may read. Defaults to `companies,people,deals`. Run `list_objects` first to see what your workspace actually has — custom objects holding contract terms, compensation, or investor notes are common in Attio, and they should not be in this list. An empty value disables the check; do not ship that.

#### `ATTIO_ALLOW_WRITES` (default `false`)

Master switch for `update_record_attribute`. Attio has no undo API. Leave it off unless you have decided, deliberately, that chat-driven CRM writes are acceptable.

#### `ATTIO_WRITABLE_ATTRIBUTES` (required if writes are on)

Comma-separated `object_slug.attribute_slug` pairs, e.g. `companies.lifecycle_stage,deals.owner`. Anything not listed is refused. Empty means no write is permitted regardless of `ATTIO_ALLOW_WRITES`.

### 4. Register with Claude

Claude Desktop — edit `claude_desktop_config.json` (macOS: `~/Library/Application Support/Claude/`; Windows: `%APPDATA%\Claude\`):

```json
{
  "mcpServers": {
    "attio-revops": {
      "command": "/absolute/path/to/mcp-server-attio-revops/.venv/bin/python",
      "args": ["-m", "attio_revops_mcp.server"],
      "env": {
        "ATTIO_API_KEY": "your-token-here",
        "ATTIO_ALLOWED_OBJECTS": "companies,people,deals",
        "ATTIO_ALLOW_WRITES": "false"
      }
    }
  }
}
```

Claude Code — from the repo root:

```bash
claude mcp add attio-revops -- /absolute/path/to/.venv/bin/python -m attio_revops_mcp.server
```

Then set the environment variables in the shell Claude Code inherits, or add them to the generated config.

### 5. Sanity check

Restart Claude, then ask, in order:

1. **"List the objects in my Attio workspace."** Exercises `list_objects` and confirms auth. A `403` here means the token lacks `object_configuration:read`.
2. **"Query 5 companies and show me their name and domain."** Exercises `query_records`, the object allowlist, and the slimming layer. If `values` comes back with attribute slugs you do not recognize, that is the workspace's real schema — note the slugs you care about.
3. **"Set the lifecycle stage on company X to Customer."** With writes off, this must refuse and name `ATTIO_ALLOW_WRITES`. If it succeeds, your config is not what you think it is.

## Security model

- **The token is a workspace credential, not a user credential.** Everything the agent reads is what the token's scopes allow, independent of who is chatting. Scope it down, and treat the key as production infrastructure.
- **Records reach the model as text.** Every field returned by `query_records` — names, emails, deal values, notes stored as attributes — enters the Claude conversation. `ATTIO_ALLOWED_OBJECTS` and the `attributes` parameter are the controls that keep that set small. If any object is off-limits for a third-party LLM, it must not be in the allowlist.
- **Writes are three-gated:** the env flag, the per-attribute allowlist, and the mandatory justification. The justification is for the audit trail, not the enforcement — the first two gates are what actually stop a write.
- **`PATCH` only, by construction.** Multiselect values can be added, never removed.
- **The credential never reaches the model.** It lives in the server process; Claude sees tool names and results.

## Known limits

None of the following are wired. Address them before any unattended or multi-user deployment:

1. **No retry or backoff on 429.** The error is surfaced with `Retry-After` and an explanation of Attio's score-based query limits, but nothing retries. Add exponential backoff if an agent will loop.
2. **No pagination loop.** `offset` is exposed; walking pages is left to the caller. This is deliberate — an automatic loop is how a "quick question" turns into ten thousand records of PII in the context window.
3. **No audit log.** Tool calls are not written anywhere. Wrap `call_tool` with structured logging to your own sink before this serves more than one person.
4. **`_simplify_value` probes rather than dispatches.** It checks common payload keys in order instead of switching on `attribute_type`. Rare attribute types fall through to a stripped object. Replace it with an explicit type map once you know which types your workspace uses.
5. **No tests.** `pytest` and `pytest-httpx` are in the dev extras and nothing uses them. Record fixtures from your workspace and pin the slimming behavior first.
6. **Single-record writes only.** Intentional, but it means a 40-record cleanup is 40 calls. Do bulk work with a script and Attio's own API, reviewed as a diff — not from chat.

"""attio_revops_mcp — read-mostly MCP server for Attio CRM records, lists, and gated writes."""

__version__ = "0.1.0"

"""
attio-revops-mcp — a read-mostly MCP server over the Attio REST API.

Exposes schema discovery, record query, single-record fetch, and list-entry query
as reads, plus one gated write (update_record_attribute). Every read is bounded by
an object allowlist and a page-size cap; the write is off unless ATTIO_ALLOW_WRITES
is set, restricted to an explicit attribute allowlist, and requires a justification.

This exists alongside Attio's own hosted MCP server at https://mcp.attio.com/mcp.
The hosted server authenticates the individual user over OAuth and grants that
user's full permissions across 30+ tools. This scaffold instead runs on a workspace
API key whose scope set you choose, and narrows what an agent can reach to the
objects and attributes you name. Use the hosted server when you want breadth; use
this when you want a small, auditable surface.

STATUS: scaffold — not runtime-tested. Endpoint paths, scopes, and parameters track
the public Attio API docs (docs.attio.com) as of 2026-07. Attribute slugs are
workspace-specific; verify against your own workspace before relying on it.

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

from __future__ import annotations

import json
import os
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) -----

ATTIO_API_KEY = os.environ.get("ATTIO_API_KEY")
ATTIO_BASE_URL = os.environ.get("ATTIO_BASE_URL", "https://api.attio.com/v2").rstrip("/")

# Objects the agent may touch at all, by api_slug. Attio workspaces routinely carry
# custom objects holding contract terms, comp data, or investor notes that have no
# business reaching an LLM. Empty means "no restriction", which you should not ship.
ATTIO_ALLOWED_OBJECTS = [
    s.strip() for s in os.environ.get("ATTIO_ALLOWED_OBJECTS", "companies,people,deals").split(",") if s.strip()
]

# Writes are off unless explicitly enabled. Attio has no undo API — a wrong value
# written from chat is repaired by hand, record by record.
ATTIO_ALLOW_WRITES = os.environ.get("ATTIO_ALLOW_WRITES", "false").lower() == "true"

# Attributes the write tool may set, as "object_slug.attribute_slug" pairs. The
# hosted server can update any attribute the signed-in user can; this list is the
# reason to run your own.
ATTIO_WRITABLE_ATTRIBUTES = [
    s.strip() for s in os.environ.get("ATTIO_WRITABLE_ATTRIBUTES", "").split(",") if s.strip()
]

# Attio's query endpoints default to limit=500. That is a large payload of personal
# data to hand a model for a question that usually wants ten rows.
MAX_LIMIT = 100
DEFAULT_LIMIT = 25


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


def auth_headers() -> dict[str, str]:
    # Attio authenticates with a standard bearer token, whether the credential is a
    # workspace API key or an OAuth access token.
    return {
        "Authorization": f"Bearer {ATTIO_API_KEY}",
        "Content-Type": "application/json",
    }


def clamp_limit(value: Any) -> int:
    try:
        n = int(value)
    except (TypeError, ValueError):
        return DEFAULT_LIMIT
    return max(1, min(n, MAX_LIMIT))


def check_object(slug: str) -> str:
    if ATTIO_ALLOWED_OBJECTS and slug not in ATTIO_ALLOWED_OBJECTS:
        raise PermissionError(
            f"Object {slug!r} is not in ATTIO_ALLOWED_OBJECTS "
            f"({', '.join(ATTIO_ALLOWED_OBJECTS)}). Add it deliberately if the agent should read it."
        )
    return slug


# ----- Attio REST helpers -----


async def attio_request(method: str, path: str, *, json_body: dict[str, Any] | None = None) -> dict[str, Any]:
    async with httpx.AsyncClient(timeout=30.0) as client:
        r = await client.request(
            method, f"{ATTIO_BASE_URL}{path}", headers=auth_headers(), json=json_body
        )
        _raise_for_attio(r)
        return r.json() if r.content else {}


def _raise_for_attio(r: httpx.Response) -> None:
    if r.status_code == 403:
        raise PermissionError(
            "Attio returned 403. The token is missing a scope this call needs. Reads need "
            "record_permission:read, object_configuration:read, list_entry:read, and "
            "list_configuration:read; the write tool additionally needs "
            "record_permission:read-write. Scopes are fixed when the key is created — "
            "generate a new one in workspace settings rather than editing this one."
        )
    if r.status_code == 429:
        retry_after = r.headers.get("Retry-After", "unknown")
        raise RuntimeError(
            f"Attio returned 429 (rate limit); Retry-After: {retry_after}s. The record and "
            "list query endpoints price each request by complexity — sorts, filters, and the "
            "object's total record count all raise the score, and scores are summed over a "
            "10-second sliding window. Narrow the filter or drop the sort and retry."
        )
    r.raise_for_status()


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

server = Server("attio-revops")


@server.list_tools()
async def list_tools() -> list[Tool]:
    return [
        Tool(
            name="list_objects",
            description=(
                "List the objects configured in the workspace (GET /v2/objects) so you can "
                "discover api_slug values before querying. Attribute slugs are workspace-"
                "specific; never guess one. Read-only."
            ),
            inputSchema={"type": "object", "properties": {}},
        ),
        Tool(
            name="query_records",
            description=(
                "Query records of one object (POST /v2/objects/{object}/records/query). "
                "Read-only. Pass an Attio filter object and optional sorts. Results are "
                "slimmed to the currently-active value per attribute. Capped at 100 rows "
                "per call; defaults to 25."
            ),
            inputSchema={
                "type": "object",
                "properties": {
                    "object": {
                        "type": "string",
                        "description": "Object api_slug or UUID, e.g. 'companies'.",
                    },
                    "filter": {
                        "type": "object",
                        "description": "Attio filter object, e.g. {'name': {'$contains': 'Acme'}}.",
                    },
                    "sorts": {
                        "type": "array",
                        "items": {"type": "object"},
                        "description": "Each entry takes direction, attribute, and optional field.",
                    },
                    "limit": {"type": "integer", "default": DEFAULT_LIMIT},
                    "offset": {"type": "integer", "default": 0},
                    "attributes": {
                        "type": "array",
                        "items": {"type": "string"},
                        "description": "Attribute slugs to keep in the response. Omit for all.",
                    },
                },
                "required": ["object"],
            },
        ),
        Tool(
            name="get_record",
            description=(
                "Fetch one record by id (GET /v2/objects/{object}/records/{record_id}). "
                "Read-only. Returns the slimmed attribute map plus the record's web_url so "
                "a human can open it in Attio."
            ),
            inputSchema={
                "type": "object",
                "properties": {
                    "object": {"type": "string"},
                    "record_id": {"type": "string", "description": "Record UUID."},
                },
                "required": ["object", "record_id"],
            },
        ),
        Tool(
            name="query_list_entries",
            description=(
                "Query entries on a list (POST /v2/lists/{list}/entries/query). Read-only. "
                "A list is Attio's pipeline surface — use this for 'what is in stage X' "
                "questions rather than querying the parent object. Capped at 100 entries."
            ),
            inputSchema={
                "type": "object",
                "properties": {
                    "list": {"type": "string", "description": "List api_slug or UUID."},
                    "filter": {"type": "object"},
                    "sorts": {"type": "array", "items": {"type": "object"}},
                    "limit": {"type": "integer", "default": DEFAULT_LIMIT},
                    "offset": {"type": "integer", "default": 0},
                },
                "required": ["list"],
            },
        ),
        Tool(
            name="update_record_attribute",
            description=(
                "Set ONE attribute on ONE record (PATCH /v2/objects/{object}/records/"
                "{record_id}). A write. Disabled unless ATTIO_ALLOW_WRITES=true, restricted "
                "to ATTIO_WRITABLE_ATTRIBUTES, and requires a justification of at least 10 "
                "characters. PATCH is used deliberately: it prepends to multiselect values "
                "rather than replacing them, so this tool cannot erase existing values."
            ),
            inputSchema={
                "type": "object",
                "properties": {
                    "object": {"type": "string"},
                    "record_id": {"type": "string"},
                    "attribute": {
                        "type": "string",
                        "description": "Attribute api_slug, e.g. 'owner' or 'lifecycle_stage'.",
                    },
                    "value": {
                        "description": "Scalar for single-value attributes, array for multiselect."
                    },
                    "justification": {"type": "string", "minLength": 10},
                },
                "required": ["object", "record_id", "attribute", "value", "justification"],
            },
        ),
    ]


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


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

    if name == "list_objects":
        data = await attio_request("GET", "/objects")
        rows = [
            {
                "api_slug": o.get("api_slug"),
                "singular_noun": o.get("singular_noun"),
                "plural_noun": o.get("plural_noun"),
                "readable": (not ATTIO_ALLOWED_OBJECTS) or o.get("api_slug") in ATTIO_ALLOWED_OBJECTS,
            }
            for o in data.get("data", [])
        ]
        return [TextContent(type="text", text=json.dumps({"objects": rows}, indent=2))]

    if name == "query_records":
        obj = check_object(arguments["object"])
        body: dict[str, Any] = {
            "limit": clamp_limit(arguments.get("limit", DEFAULT_LIMIT)),
            "offset": int(arguments.get("offset", 0)),
        }
        if v := arguments.get("filter"):
            body["filter"] = v
        if v := arguments.get("sorts"):
            body["sorts"] = v
        data = await attio_request("POST", f"/objects/{obj}/records/query", json_body=body)
        keep = arguments.get("attributes")
        rows = [_slim_record(rec, keep) for rec in data.get("data", [])]
        return [
            TextContent(
                type="text",
                text=json.dumps({"object": obj, "returned": len(rows), "records": rows}, indent=2),
            )
        ]

    if name == "get_record":
        obj = check_object(arguments["object"])
        record_id = arguments["record_id"]
        data = await attio_request("GET", f"/objects/{obj}/records/{record_id}")
        return [TextContent(type="text", text=json.dumps(_slim_record(data.get("data", {}), None), indent=2))]

    if name == "query_list_entries":
        list_ref = arguments["list"]
        body = {
            "limit": clamp_limit(arguments.get("limit", DEFAULT_LIMIT)),
            "offset": int(arguments.get("offset", 0)),
        }
        if v := arguments.get("filter"):
            body["filter"] = v
        if v := arguments.get("sorts"):
            body["sorts"] = v
        data = await attio_request("POST", f"/lists/{list_ref}/entries/query", json_body=body)
        rows = [_slim_entry(e) for e in data.get("data", [])]
        return [
            TextContent(
                type="text",
                text=json.dumps({"list": list_ref, "returned": len(rows), "entries": rows}, indent=2),
            )
        ]

    if name == "update_record_attribute":
        justification = (arguments.get("justification") or "").strip()
        if len(justification) < 10:
            raise ValueError("justification is mandatory and must be at least 10 characters.")
        if not ATTIO_ALLOW_WRITES:
            raise PermissionError(
                "update_record_attribute is disabled. Set ATTIO_ALLOW_WRITES=true to allow "
                "chat-driven CRM writes, and list the permitted attributes in "
                "ATTIO_WRITABLE_ATTRIBUTES."
            )
        obj = check_object(arguments["object"])
        attribute = arguments["attribute"]
        qualified = f"{obj}.{attribute}"
        if qualified not in ATTIO_WRITABLE_ATTRIBUTES:
            raise PermissionError(
                f"{qualified!r} is not in ATTIO_WRITABLE_ATTRIBUTES "
                f"({', '.join(ATTIO_WRITABLE_ATTRIBUTES) or 'empty'}). Writes are allowlisted "
                "per attribute, not per object."
            )
        record_id = arguments["record_id"]
        body = {"data": {"values": {attribute: arguments["value"]}}}
        data = await attio_request(
            "PATCH", f"/objects/{obj}/records/{record_id}", json_body=body
        )
        web_url = (data.get("data") or {}).get("web_url")
        return [
            TextContent(
                type="text",
                text=(
                    f"Set {qualified} on record {record_id} ({justification!r}). "
                    f"Multiselect values were prepended, not replaced. Open in Attio: {web_url}"
                ),
            )
        ]

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


# ----- Response slimming (keep model payloads tractable) -----


def _slim_record(rec: dict[str, Any], keep: list[str] | None) -> dict[str, Any]:
    """Reduce Attio's value-history shape to one current value per attribute.

    Attio returns every attribute as an array of value objects carrying active_from,
    active_until, created_by_actor, and type-specific fields — the full history, not
    just the present. Handing that to a model multiplies token cost several times over
    for information nobody asked for. We keep the entries where active_until is null.
    """
    values = rec.get("values", {}) or {}
    out: dict[str, Any] = {}
    for slug, entries in values.items():
        if keep and slug not in keep:
            continue
        if not isinstance(entries, list):
            continue
        current = [e for e in entries if isinstance(e, dict) and e.get("active_until") is None]
        simplified = [_simplify_value(e) for e in current]
        if not simplified:
            continue
        out[slug] = simplified[0] if len(simplified) == 1 else simplified
    ids = rec.get("id", {}) or {}
    return {
        "record_id": ids.get("record_id"),
        "web_url": rec.get("web_url"),
        "created_at": rec.get("created_at"),
        "values": out,
    }


def _simplify_value(entry: dict[str, Any]) -> Any:
    """Pull the human-meaningful field out of one Attio value object.

    Attio's value shape is discriminated by `attribute_type`, and each type puts its
    payload under a different key. Rather than enumerate every type, we probe the
    common carriers in order and fall back to the stripped object.
    """
    for key in (
        "value",
        "full_name",
        "email_address",
        "phone_number",
        "domain",
        "status",
        "option",
        "target_record_id",
        "referenced_actor_name",
        "currency_value",
    ):
        if key in entry and entry[key] is not None:
            v = entry[key]
            if isinstance(v, dict):
                return v.get("title") or v.get("name") or v
            return v
    return {k: v for k, v in entry.items() if k not in ("active_from", "active_until", "created_by_actor")}


def _slim_entry(entry: dict[str, Any]) -> dict[str, Any]:
    ids = entry.get("id", {}) or {}
    parent = entry.get("parent_record_id")
    values = entry.get("entry_values", entry.get("values", {})) or {}
    out = {}
    for slug, entries in values.items():
        if isinstance(entries, list) and entries:
            current = [e for e in entries if isinstance(e, dict) and e.get("active_until") is None]
            if current:
                out[slug] = _simplify_value(current[0])
    return {
        "entry_id": ids.get("entry_id"),
        "parent_record_id": parent,
        "parent_object": entry.get("parent_object"),
        "created_at": entry.get("created_at"),
        "values": out,
    }


# ----- 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())