Pull Vitally account health into Claude via MCP for CS conversations

[project]
name = "vitally-cs-mcp"
version = "0.1.0"
description = "MCP server for Customer Success workflows on Vitally"
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/vitally_cs_mcp"]

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

# mcp-server-vitally-cs

An MCP server for Customer Success teams using Vitally. Exposes account reads, health score breakdowns, and conversation history — three tools designed for CSMs who want to pull Vitally context into Claude conversations without leaving the chat surface.

> **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 Vitally tenant. Adapt the subdomain, field names, and any custom trait keys to your org before use.

## What it exposes

### Account reads (read-only)
- `get_account_health(account_id)` — fetches a single account plus its full health score breakdown via the `/accounts/:id/healthScores` endpoint
- `list_at_risk_accounts(health_threshold?, limit?)` — pages through accounts, filters to those whose `healthScore` is at or below the threshold, returns a ranked list
- `get_account_conversations(account_id, limit?)` — fetches the most recent conversations linked to an account, ordered newest-first

The server does **not** expose write endpoints. No note creation, no conversation mutations, no trait updates. Read-only by design so CSMs can drop this into Claude Desktop without a security review of write scope.

## Setup

### 1. Install

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

### 2. Locate your Vitally API key

In Vitally: **Settings → Integrations → API**. Generate or copy an existing API key. Vitally authenticates via HTTP Basic Auth with the API key as the username and an empty password. The `Authorization` header value is `Basic <base64(apikey:)>`.

The server handles this encoding for you — you only need to supply the raw key as `VITALLY_API_KEY`.

### 3. Find your subdomain

Your Vitally base URL is `https://<subdomain>.rest.vitally.io`. The subdomain is the prefix of your Vitally workspace URL (e.g. if you log in at `acme.vitally.io`, your subdomain is `acme`). EU tenants use `rest.vitally-eu.io` — set `VITALLY_BASE_URL` accordingly.

### 4. Configure environment

```bash
export VITALLY_API_KEY="your_api_key_here"
export VITALLY_SUBDOMAIN="acme"                          # your workspace subdomain
export VITALLY_BASE_URL="https://acme.rest.vitally.io"   # full base URL; auto-built from subdomain if omitted
export VITALLY_HEALTH_THRESHOLD="50"                     # default red/orange threshold for list_at_risk_accounts
export VITALLY_PAGE_LIMIT="100"                          # max records per page (Vitally caps at 100)
```

**`VITALLY_API_KEY`** — required. Raw API key from Vitally Settings → Integrations → API. Never commit this to source control.

**`VITALLY_SUBDOMAIN`** — required unless `VITALLY_BASE_URL` is set explicitly. Used to construct `https://{subdomain}.rest.vitally.io`.

**`VITALLY_BASE_URL`** — optional override. Set this explicitly for EU tenants: `https://rest.vitally-eu.io`. If set, it takes precedence over `VITALLY_SUBDOMAIN`.

**`VITALLY_HEALTH_THRESHOLD`** — optional, default `50`. Accounts whose overall health score is ≤ this value are returned by `list_at_risk_accounts`. Vitally health scores run 0–100; typical "red" band is 0–33, "orange" 34–66, "green" 67–100 — adjust to match your org's thresholds.

**`VITALLY_PAGE_LIMIT`** — optional, default `100`. Vitally's REST API caps list responses at 100 records per page. The scaffold respects this ceiling.

### 5. Register with Claude Desktop

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

```json
{
  "mcpServers": {
    "vitally-cs": {
      "command": "python",
      "args": ["-m", "vitally_cs_mcp.server"],
      "env": {
        "VITALLY_API_KEY": "your_api_key_here",
        "VITALLY_SUBDOMAIN": "acme",
        "VITALLY_HEALTH_THRESHOLD": "50"
      }
    }
  }
}
```

Restart Claude Desktop. You should see 3 tools registered under `vitally-cs`.

### 6. Register with Claude Code

Add to your project or user `settings.json` under `mcpServers`:

```json
{
  "mcpServers": {
    "vitally-cs": {
      "command": "python",
      "args": ["-m", "vitally_cs_mcp.server"],
      "env": {
        "VITALLY_API_KEY": "your_api_key_here",
        "VITALLY_SUBDOMAIN": "acme"
      }
    }
  }
}
```

### 7. Sanity check

Ask Claude: "Show me the health score for account ID `<any real account id>`." The response should include an overall health score and a breakdown of individual health components. Then ask: "List my at-risk accounts with a threshold of 40." Verify the returned accounts actually have health scores at or below 40 in the Vitally UI.

## Security model

- **Read-only by design.** The three tools call only `GET` endpoints. No write operations are included. This means there is nothing to scope-limit beyond read access to account, health, and conversation data — the entire risk surface is disclosure, not mutation.
- **API key scope.** Vitally API keys give read access to all objects the generating admin can see. There is no per-object scope at the API key level. Create a dedicated Vitally admin user for this integration, give it view-only access to the accounts and conversations your CSMs need, and generate the key from that user. Do not use a full-admin key.
- **Data in Claude's context window.** Account names, health scores, conversation subjects, and message bodies pass through Claude's context. Review your org's AI/data policy before enabling this for accounts that hold PII or contractually sensitive information. The server does not log requests or responses; what Claude does with the data depends on your Claude plan's data-retention settings.
- **Key storage.** The scaffold reads `VITALLY_API_KEY` from env. For team deployments, store the key in your secret manager (Vault, AWS Secrets Manager, 1Password CLI) and inject it at process start, rather than hardcoding it in the Claude Desktop config JSON.

## Known limits and TODOs (before production use)

1. [ ] Add OAuth or service-account key rotation — the current scaffold uses a static API key; implement a refresh path or a key-rotation reminder cron.
2. [ ] Implement cursor-based pagination in `list_at_risk_accounts` — the current scaffold fetches one page (up to 100 accounts) and filters locally; for orgs with >100 accounts, add a loop over the `next` cursor until exhausted or limit reached.
3. [ ] Add structured logging via `python-json-logger` — emit one JSON line per tool call with name, arguments hash, duration ms, and HTTP status code.
4. [ ] Wire optional Sentry / OpenTelemetry export for latency and error tracking.
5. [ ] Add integration tests against a Vitally sandbox or staging tenant.
6. [ ] Validate `VITALLY_SUBDOMAIN` at startup against the Vitally API (a lightweight `GET /resources/accounts?limit=1` can confirm the key and subdomain are valid before accepting tool calls).
7. [ ] Support EU data center (`rest.vitally-eu.io`) configuration with a clearer env var — currently handled via `VITALLY_BASE_URL` override, but a `VITALLY_REGION=eu` flag would be friendlier.
8. [ ] Add `list_at_risk_accounts` sort by health score ascending so the worst accounts surface first without the caller needing to sort client-side.

"""vitally_cs_mcp — MCP server for Customer Success workflows on Vitally."""

__version__ = "0.1.0"

"""
vitally-cs-mcp — MCP server for Customer Success workflows on Vitally.

Exposes three read-only tools:
  - get_account_health(account_id)         — single account + health score breakdown
  - list_at_risk_accounts(threshold, limit) — paginated list filtered by health score
  - get_account_conversations(account_id, limit) — recent conversations for an account

All tools are read-only; no write operations are included. Authentication uses
Vitally's HTTP Basic Auth with the API key as the username (empty password).

STATUS: scaffold — not runtime-tested. Adapt the subdomain, field names, and any
custom trait keys to your org before use.

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

from __future__ import annotations

import base64
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) -----

VITALLY_API_KEY = os.environ.get("VITALLY_API_KEY", "")
VITALLY_SUBDOMAIN = os.environ.get("VITALLY_SUBDOMAIN", "")

# VITALLY_BASE_URL can be set explicitly (useful for EU: https://rest.vitally-eu.io)
# or auto-built from VITALLY_SUBDOMAIN.
_base_url_env = os.environ.get("VITALLY_BASE_URL", "").rstrip("/")
VITALLY_BASE_URL: str = _base_url_env or (
    f"https://{VITALLY_SUBDOMAIN}.rest.vitally.io" if VITALLY_SUBDOMAIN else ""
)

# Default health score threshold for list_at_risk_accounts.
# Vitally scores run 0–100; typical red band is 0–33, orange 34–66, green 67–100.
VITALLY_HEALTH_THRESHOLD: int = int(os.environ.get("VITALLY_HEALTH_THRESHOLD", "50"))

# Vitally caps list responses at 100 records per page.
VITALLY_PAGE_LIMIT: int = min(int(os.environ.get("VITALLY_PAGE_LIMIT", "100")), 100)


def require_config() -> None:
    if not VITALLY_API_KEY:
        raise RuntimeError("VITALLY_API_KEY env var is required")
    if not VITALLY_BASE_URL:
        raise RuntimeError(
            "Either VITALLY_SUBDOMAIN or VITALLY_BASE_URL env var is required"
        )


def auth_headers() -> dict[str, str]:
    """
    Vitally uses HTTP Basic Auth with the API key as the username and an empty password.
    The Authorization header value is: Basic base64("<apikey>:")
    """
    token = base64.b64encode(f"{VITALLY_API_KEY}:".encode()).decode("ascii")
    return {
        "Authorization": f"Basic {token}",
        "Content-Type": "application/json",
    }


# ----- Vitally REST helpers -----


async def vitally_get(path: str, params: dict[str, Any] | None = None) -> dict[str, Any]:
    """
    Make an authenticated GET request to the Vitally REST API.
    Path should start with /resources/...
    """
    url = f"{VITALLY_BASE_URL}{path}"
    async with httpx.AsyncClient(timeout=30.0) as client:
        r = await client.get(url, headers=auth_headers(), params=params)
        r.raise_for_status()
        return r.json()


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

server = Server("vitally-cs")


@server.list_tools()
async def list_tools() -> list[Tool]:
    return [
        Tool(
            name="get_account_health",
            description=(
                "Fetch a Vitally account by ID and its full health score breakdown. "
                "Returns the account record and the healthScores array, which includes "
                "each health component name, score, and status."
            ),
            inputSchema={
                "type": "object",
                "properties": {
                    "account_id": {
                        "type": "string",
                        "description": (
                            "Vitally account ID or externalId. "
                            "Use the Vitally-assigned UUID or the externalId your system sent."
                        ),
                    }
                },
                "required": ["account_id"],
            },
        ),
        Tool(
            name="list_at_risk_accounts",
            description=(
                "List accounts whose overall health score is at or below a threshold. "
                "Fetches one page of accounts (up to the limit), filters by health score, "
                "and returns matches sorted by health score ascending (worst first). "
                "For orgs with >100 accounts, only the first page is scanned — "
                "see the TODO in the README for full pagination."
            ),
            inputSchema={
                "type": "object",
                "properties": {
                    "health_threshold": {
                        "type": "integer",
                        "description": (
                            "Accounts with a health score at or below this value are returned. "
                            "Vitally scores run 0–100. Defaults to the VITALLY_HEALTH_THRESHOLD "
                            "env var (default 50)."
                        ),
                    },
                    "limit": {
                        "type": "integer",
                        "description": "Max accounts to return in the response (after filtering). Default 20.",
                        "default": 20,
                    },
                },
            },
        ),
        Tool(
            name="get_account_conversations",
            description=(
                "Fetch recent conversations linked to a Vitally account. "
                "Returns conversations in newest-first order with subject, message count, "
                "and traits. Useful for reviewing recent CSM activity before a call."
            ),
            inputSchema={
                "type": "object",
                "properties": {
                    "account_id": {
                        "type": "string",
                        "description": "Vitally account ID or externalId.",
                    },
                    "limit": {
                        "type": "integer",
                        "description": "Max conversations to return. Default 10, max 100.",
                        "default": 10,
                    },
                },
                "required": ["account_id"],
            },
        ),
    ]


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


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

    # ------------------------------------------------------------------
    # get_account_health
    # Fetches /resources/accounts/:id and /resources/accounts/:id/healthScores
    # in two requests, then merges the results.
    # ------------------------------------------------------------------
    if name == "get_account_health":
        account_id = arguments["account_id"]

        # Fetch the base account record.
        account = await vitally_get(f"/resources/accounts/{account_id}")

        # Fetch the health score breakdown.
        # Response: { "results": [ { "name": "...", "score": 72, "healthStatus": "healthy", ... }, ... ] }
        health_data = await vitally_get(f"/resources/accounts/{account_id}/healthScores")
        health_scores = health_data.get("results", [])

        result = {
            "account": {
                "id": account.get("id"),
                "externalId": account.get("externalId"),
                "name": account.get("name"),
                "healthScore": account.get("healthScore"),
                "traits": account.get("traits", {}),
            },
            "healthScores": health_scores,
        }
        return [TextContent(type="text", text=str(result))]

    # ------------------------------------------------------------------
    # list_at_risk_accounts
    # Fetches one page of accounts, filters to those at or below threshold.
    # Returns sorted by healthScore ascending (worst first).
    # ------------------------------------------------------------------
    if name == "list_at_risk_accounts":
        threshold = arguments.get("health_threshold", VITALLY_HEALTH_THRESHOLD)
        return_limit = min(int(arguments.get("limit", 20)), 100)

        # Fetch one full page of active accounts.
        # Vitally's list-accounts endpoint: GET /resources/accounts?limit=100&status=active
        page = await vitally_get(
            "/resources/accounts",
            params={"limit": VITALLY_PAGE_LIMIT, "status": "active"},
        )
        accounts = page.get("results", [])

        # Filter: include only accounts where healthScore is defined and <= threshold.
        # healthScore may be None if Vitally hasn't computed it yet (new accounts, missing data).
        at_risk = [
            {
                "id": a.get("id"),
                "externalId": a.get("externalId"),
                "name": a.get("name"),
                "healthScore": a.get("healthScore"),
                "traits": a.get("traits", {}),
            }
            for a in accounts
            if a.get("healthScore") is not None and a["healthScore"] <= threshold
        ]

        # Sort by healthScore ascending so the worst accounts surface first.
        at_risk.sort(key=lambda a: a["healthScore"])

        # Trim to the requested return limit.
        at_risk = at_risk[:return_limit]

        result = {
            "threshold": threshold,
            "total_scanned": len(accounts),
            "at_risk_count": len(at_risk),
            "note": (
                "Only the first page of accounts was scanned. "
                "See README TODO #2 to add full cursor pagination."
            )
            if page.get("next")
            else "All active accounts were scanned.",
            "accounts": at_risk,
        }
        return [TextContent(type="text", text=str(result))]

    # ------------------------------------------------------------------
    # get_account_conversations
    # Fetches /resources/accounts/:id/conversations?limit=N
    # Conversations are returned by Vitally in newest-first order by default.
    # ------------------------------------------------------------------
    if name == "get_account_conversations":
        account_id = arguments["account_id"]
        limit = min(int(arguments.get("limit", 10)), 100)

        data = await vitally_get(
            f"/resources/accounts/{account_id}/conversations",
            params={"limit": limit},
        )
        conversations = data.get("results", [])

        # Trim each conversation to the fields most useful in a Claude context window.
        trimmed = [
            {
                "id": c.get("id"),
                "externalId": c.get("externalId"),
                "subject": c.get("subject"),
                "messageCount": len(c.get("messages", [])),
                "firstMessage": (c.get("messages") or [{}])[0].get("body", ""),
                "traits": c.get("traits", {}),
                "createdAt": c.get("createdAt"),
                "updatedAt": c.get("updatedAt"),
            }
            for c in conversations
        ]

        result = {
            "account_id": account_id,
            "conversation_count": len(trimmed),
            "conversations": trimmed,
        }
        return [TextContent(type="text", text=str(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())