Serveur MCP exposant Salesforce en lecture et écriture à Claude

[project]
name = "salesforce-revops-mcp"
version = "0.1.0"
description = "MCP server for Salesforce revenue-operations workflows"
readme = "README.md"
requires-python = ">=3.11"
license = { text = "MIT" }
dependencies = [
  "mcp>=1.2.0",
  "httpx>=0.27.0",
  "pydantic>=2.6.0",
  "simple-salesforce>=1.12.6",
]

[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/salesforce_revops_mcp"]

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

# mcp-server-salesforce-revops

An MCP server tuned for revenue-operations teams using Salesforce. Exposes account, opportunity, contact, and lead reads, a SELECT-only SOQL endpoint, three RevOps helpers (`pipeline_by_stage`, `stale_opps`, `at_risk_commits`), and two audit-aware light writes (`add_note`, `update_field`). Designed to make Claude useful for "what's actually in the forecast" conversations without handing it the keys to delete or bulk-mutate the org.

> **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 Salesforce org. Treat it as a starting point you adapt to your org's object model, picklist values, and custom audit object name. Stage labels, owner role hierarchy, and the audit object schema vary by org.

## What it exposes

### Object reads (read-only)
- `get_account(account_id)` — full Account fields
- `get_opportunity(opp_id)` — full Opportunity fields + owner
- `get_contact(contact_id)` — full Contact fields
- `get_lead(lead_id)` — full Lead fields

### SOQL (read-only)
- `query(soql, bypass_sharing=False)` — runs the SOQL string against the REST `/query` endpoint. **Refuses any statement that is not a single `SELECT`.** Any `INSERT`, `UPDATE`, `DELETE`, `UPSERT`, `MERGE`, or DML keyword raises before a request is made. `bypass_sharing` defaults to `False`; when `True`, the helper appends nothing — sharing rules apply via the running user. The flag is reserved for future tooling-API integration and is rejected today with a clear error.

### RevOps helpers (read-only)
- `pipeline_by_stage(close_date_window_days=90, owner_id?)` — open opportunities closing in the window, aggregated by stage. Optionally filtered to one owner.
- `stale_opps(days_in_stage_threshold=30)` — open opportunities whose `LastStageChangeDate` is older than the threshold.
- `at_risk_commits(quarter_end_date)` — Commit-stage opportunities whose `LastActivityDate` is more than 14 days ago **or** whose `CloseDate` is within 14 days of `quarter_end_date` and have no future-dated `Event` on the account.

### Audit-aware light writes
- `add_note(object_type, object_id, body)` — creates a `ContentNote` and links it via `ContentDocumentLink` to the parent record.
- `update_field(object_type, object_id, field_name, new_value, justification)` — single-field update with **mandatory `justification` parameter** (rejected if blank or shorter than 10 chars). Writes a `Cleanup_Audit__c` row first (object name, record id, field, old value, new value, justification, who, when), then performs the field update. If the audit insert fails, the update never runs.

The server **does not** expose `delete_*` tools, bulk DML, or stage-transition shortcuts. All bulk reads cap at 200 records per request to stay inside REST's per-call envelope and to give callers a predictable token budget.

## Setup

### 1. Install

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

### 2. Create a Salesforce Connected App (OAuth)

In Salesforce Setup: App Manager → New Connected App.

- **API (Enable OAuth Settings):** check.
- **Callback URL:** `http://localhost:1717/callback` (or any URL your OAuth helper accepts — the token is what we keep, not the redirect).
- **Scopes:** `api`, `refresh_token, offline_access`. Add `chatter_api` only if you also want to post Chatter feed items (out of scope for this scaffold).
- **Require Secret for Refresh Token Flow:** check.
- Save, wait 10 minutes for propagation, copy the **Consumer Key** and **Consumer Secret**.

Then run a one-time OAuth web-server flow to obtain a refresh token. The scaffold expects you to bring your own access token (refresh handling is on the TODO list, see below). For local development you can use the `sfdx auth:web:login` CLI to mint one and `sfdx force:org:display --verbose` to read it out.

**Auth choice for this scaffold.** We read a long-lived `SFDC_ACCESS_TOKEN` from env. Refresh-token rotation is documented as a TODO rather than implemented, because every team's secret-store choice (Vault, AWS Secrets Manager, 1Password, plain env) differs. The fields are split out so swapping in a refresh-flow client is a one-file change in `server.py`.

### 3. Configure environment

```bash
export SFDC_INSTANCE_URL="https://yourdomain.my.salesforce.com"
export SFDC_ACCESS_TOKEN="00D...!ARQAQ..."
export SFDC_API_VERSION="v60.0"                 # optional, default v60.0
export SFDC_AUDIT_OBJECT="Cleanup_Audit__c"     # custom object for write audits
export SFDC_COMMIT_STAGE_NAME="Commit"          # picklist label for commit-stage opps
```

The `Cleanup_Audit__c` object must exist with at least these custom fields: `Object_Name__c` (text), `Record_Id__c` (text 18), `Field_Name__c` (text), `Old_Value__c` (long text), `New_Value__c` (long text), `Justification__c` (long text), `Performed_By__c` (text). If you use a different object name, set `SFDC_AUDIT_OBJECT` accordingly.

### 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": {
    "salesforce-revops": {
      "command": "python",
      "args": ["-m", "salesforce_revops_mcp.server"],
      "env": {
        "SFDC_INSTANCE_URL": "https://yourdomain.my.salesforce.com",
        "SFDC_ACCESS_TOKEN": "00D...!ARQAQ...",
        "SFDC_API_VERSION": "v60.0",
        "SFDC_AUDIT_OBJECT": "Cleanup_Audit__c",
        "SFDC_COMMIT_STAGE_NAME": "Commit"
      }
    }
  }
}
```

Restart Claude Desktop. You should see ~10 tools registered under `salesforce-revops`.

### 5. Sanity-check

Ask Claude: "Show me pipeline by stage for the next ninety days." Compare the per-stage totals against the equivalent Pipeline report in Salesforce. Then run `update_field` against a sandbox opportunity with a short justification and confirm both the field changed **and** a `Cleanup_Audit__c` row was written.

## Watch-outs

- **Connected App scope discipline.** A Connected App with `api` scope can read every object the running user can read. Create a dedicated integration user with a profile that exposes only the objects this server needs (Account, Opportunity, Contact, Lead, the audit object), and assign field-level permissions narrowly. Document this with your security team. **Guard:** integration-user profile reviewed quarterly; record the review date in the audit object.
- **Governor limits on bulk reads.** A naive `query("SELECT Id FROM Opportunity")` against a 500K-row org will paginate forever and exhaust your daily API quota. **Guard:** every helper caps `LIMIT` at 200; the `query` tool also injects `LIMIT 200` if the SOQL has none.
- **FLS bypass risk.** Salesforce's REST `/query` endpoint does **not** enforce field-level security unless you explicitly ask for `WITH SECURITY_ENFORCED`. **Guard:** the `query` tool appends `WITH SECURITY_ENFORCED` when missing, so a user without read on a field gets a clear error rather than silent data leakage.
- **Audit log gap on writes.** If the audit insert succeeds but the field update fails (or vice versa), you have a recorded change with no actual change (or a change with no record). **Guard:** the scaffold writes the audit row first; if the field update raises, the audit row is left in place tagged as `Failed__c=true` (you will need to add this field if you want to use the flag — TODO listed below). Reconcile via a weekly report.
- **OAuth token refresh failure.** Long-lived access tokens expire; a 401 on a Friday afternoon is the worst time to discover this. **Guard:** front the server with a token-refresh sidecar (or implement the refresh flow per the TODO). Fail loud on 401, do not silently retry.

## Limits and TODOs (before production use)

- [ ] Implement OAuth refresh-token flow so the server self-heals on 401, instead of crashing.
- [ ] Add request-level retries with exponential backoff (Salesforce returns 503 under maintenance windows).
- [ ] Write integration tests against a Salesforce sandbox org (Trailhead Playground works).
- [ ] Add structured logging via `python-json-logger`; emit one JSON line per tool call with name, arguments hash, duration, status.
- [ ] Wire optional Sentry / OpenTelemetry export.
- [ ] Add `Failed__c` boolean to `Cleanup_Audit__c` and flip it true if the post-audit field update raises.
- [ ] Validate `SFDC_AUDIT_OBJECT` and `SFDC_COMMIT_STAGE_NAME` against the org on first run; fail loud if either does not exist.
- [ ] Replace the long-lived token env var with a secret-store lookup (Vault, AWS Secrets Manager, 1Password CLI).
- [ ] Add a per-tool `--dry-run` flag that returns the SOQL/DML payload without executing.

"""salesforce_revops_mcp — MCP server for Salesforce revenue-operations workflows."""

__version__ = "0.1.0"

"""
salesforce-revops-mcp — MCP server tuned for revenue-operations workflows on Salesforce.

Exposes object reads (Account, Opportunity, Contact, Lead), a SELECT-only SOQL
endpoint, three RevOps helpers (pipeline_by_stage, stale_opps, at_risk_commits),
and two audit-aware light writes (add_note, update_field). Read-mostly by design;
every write requires a justification and lands in a custom Cleanup_Audit__c row
before the field is touched.

STATUS: scaffold — not runtime-tested. Adapt the audit object name, picklist
labels, and field-level permissions to your org before use.

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

from __future__ import annotations

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

SFDC_INSTANCE_URL = os.environ.get("SFDC_INSTANCE_URL", "").rstrip("/")
SFDC_ACCESS_TOKEN = os.environ.get("SFDC_ACCESS_TOKEN")
SFDC_API_VERSION = os.environ.get("SFDC_API_VERSION", "v60.0")
SFDC_AUDIT_OBJECT = os.environ.get("SFDC_AUDIT_OBJECT", "Cleanup_Audit__c")
SFDC_COMMIT_STAGE_NAME = os.environ.get("SFDC_COMMIT_STAGE_NAME", "Commit")

# Hard cap for any single REST call we make. Keeps us inside Salesforce's
# 200-record per-batch envelope on bulk endpoints and gives callers a
# predictable token budget on read tools.
MAX_RECORDS_PER_REQUEST = 200

# DML keywords we refuse in the read-only SOQL endpoint. SOQL itself is
# read-only — there is no UPDATE/DELETE/INSERT in the language — but we
# explicitly refuse strings that look like SOSL or apex anonymous to make
# the intent loud.
DML_KEYWORDS = (
    "INSERT",
    "UPDATE",
    "DELETE",
    "UPSERT",
    "MERGE",
    "FIND",  # SOSL
    "EXEC",  # apex anonymous-ish
)


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


def auth_headers() -> dict[str, str]:
    return {
        "Authorization": f"Bearer {SFDC_ACCESS_TOKEN}",
        "Content-Type": "application/json",
    }


# ----- Salesforce REST helpers -----


async def sf_get(path: str, params: dict[str, Any] | None = None) -> dict[str, Any]:
    async with httpx.AsyncClient(timeout=30.0) as client:
        r = await client.get(
            f"{SFDC_INSTANCE_URL}{path}", headers=auth_headers(), params=params
        )
        r.raise_for_status()
        return r.json()


async def sf_post(path: str, body: dict[str, Any]) -> dict[str, Any]:
    async with httpx.AsyncClient(timeout=30.0) as client:
        r = await client.post(
            f"{SFDC_INSTANCE_URL}{path}", headers=auth_headers(), json=body
        )
        r.raise_for_status()
        return r.json() if r.content else {}


async def sf_patch(path: str, body: dict[str, Any]) -> None:
    async with httpx.AsyncClient(timeout=30.0) as client:
        r = await client.patch(
            f"{SFDC_INSTANCE_URL}{path}", headers=auth_headers(), json=body
        )
        r.raise_for_status()


async def sf_query(soql: str) -> dict[str, Any]:
    return await sf_get(f"/services/data/{SFDC_API_VERSION}/query", {"q": soql})


# ----- SOQL safety -----


def harden_soql(raw: str) -> str:
    """
    Validate a user-provided SOQL string.

    - Strip leading/trailing whitespace.
    - Refuse non-SELECT or DML-flavoured input.
    - Inject `LIMIT MAX_RECORDS_PER_REQUEST` if no LIMIT clause present.
    - Inject `WITH SECURITY_ENFORCED` if missing, so FLS errors surface
      instead of silent partial reads.
    """
    soql = raw.strip().rstrip(";").strip()
    if not soql:
        raise ValueError("SOQL string is empty.")
    upper = soql.upper()
    if not upper.startswith("SELECT"):
        raise ValueError("Only SELECT statements are allowed.")
    for kw in DML_KEYWORDS:
        # match whole-word; avoid flagging fields like "Description"
        if re.search(rf"\b{kw}\b", upper):
            raise ValueError(f"Refusing SOQL containing forbidden keyword: {kw}")

    if "WITH SECURITY_ENFORCED" not in upper:
        # Insert just before LIMIT/ORDER BY if present, else append.
        m = re.search(r"\b(LIMIT|ORDER\s+BY|GROUP\s+BY)\b", soql, re.IGNORECASE)
        if m:
            idx = m.start()
            soql = soql[:idx].rstrip() + " WITH SECURITY_ENFORCED " + soql[idx:]
        else:
            soql = soql + " WITH SECURITY_ENFORCED"

    if not re.search(r"\bLIMIT\b", soql, re.IGNORECASE):
        soql = f"{soql} LIMIT {MAX_RECORDS_PER_REQUEST}"
    return soql


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

server = Server("salesforce-revops")


@server.list_tools()
async def list_tools() -> list[Tool]:
    return [
        Tool(
            name="get_account",
            description="Fetch full fields for an Account by Id.",
            inputSchema={
                "type": "object",
                "properties": {"account_id": {"type": "string"}},
                "required": ["account_id"],
            },
        ),
        Tool(
            name="get_opportunity",
            description="Fetch full fields + owner for an Opportunity by Id.",
            inputSchema={
                "type": "object",
                "properties": {"opp_id": {"type": "string"}},
                "required": ["opp_id"],
            },
        ),
        Tool(
            name="get_contact",
            description="Fetch full fields for a Contact by Id.",
            inputSchema={
                "type": "object",
                "properties": {"contact_id": {"type": "string"}},
                "required": ["contact_id"],
            },
        ),
        Tool(
            name="get_lead",
            description="Fetch full fields for a Lead by Id.",
            inputSchema={
                "type": "object",
                "properties": {"lead_id": {"type": "string"}},
                "required": ["lead_id"],
            },
        ),
        Tool(
            name="query",
            description=(
                "Run a read-only SOQL SELECT statement. Refuses INSERT/UPDATE/"
                "DELETE/UPSERT/MERGE. Auto-injects WITH SECURITY_ENFORCED and "
                "caps LIMIT at 200 if missing. bypass_sharing must be False."
            ),
            inputSchema={
                "type": "object",
                "properties": {
                    "soql": {"type": "string"},
                    "bypass_sharing": {"type": "boolean", "default": False},
                },
                "required": ["soql"],
            },
        ),
        Tool(
            name="pipeline_by_stage",
            description=(
                "Open Opportunities closing within close_date_window_days, aggregated "
                "by StageName. Optional owner_id filter."
            ),
            inputSchema={
                "type": "object",
                "properties": {
                    "close_date_window_days": {"type": "integer", "default": 90},
                    "owner_id": {"type": "string"},
                },
            },
        ),
        Tool(
            name="stale_opps",
            description=(
                "Open Opportunities whose LastStageChangeDate is older than "
                "days_in_stage_threshold."
            ),
            inputSchema={
                "type": "object",
                "properties": {
                    "days_in_stage_threshold": {"type": "integer", "default": 30}
                },
            },
        ),
        Tool(
            name="at_risk_commits",
            description=(
                "Commit-stage Opportunities whose LastActivityDate > 14 days ago, "
                "or whose CloseDate is within 14 days of quarter_end_date."
            ),
            inputSchema={
                "type": "object",
                "properties": {
                    "quarter_end_date": {
                        "type": "string",
                        "description": "ISO date, e.g. 2026-06-30",
                    }
                },
                "required": ["quarter_end_date"],
            },
        ),
        Tool(
            name="add_note",
            description=(
                "Create a ContentNote and link it to a parent record via "
                "ContentDocumentLink."
            ),
            inputSchema={
                "type": "object",
                "properties": {
                    "object_type": {
                        "type": "string",
                        "enum": ["Account", "Opportunity", "Contact", "Lead"],
                    },
                    "object_id": {"type": "string"},
                    "body": {"type": "string"},
                },
                "required": ["object_type", "object_id", "body"],
            },
        ),
        Tool(
            name="update_field",
            description=(
                "Update a single field on a single record. Justification is "
                "mandatory and must be at least 10 chars; an audit row is written "
                "to Cleanup_Audit__c (or SFDC_AUDIT_OBJECT) before the update."
            ),
            inputSchema={
                "type": "object",
                "properties": {
                    "object_type": {
                        "type": "string",
                        "enum": ["Account", "Opportunity", "Contact", "Lead"],
                    },
                    "object_id": {"type": "string"},
                    "field_name": {"type": "string"},
                    "new_value": {
                        "description": "New value (string, number, or boolean).",
                    },
                    "justification": {
                        "type": "string",
                        "minLength": 10,
                    },
                },
                "required": [
                    "object_type",
                    "object_id",
                    "field_name",
                    "new_value",
                    "justification",
                ],
            },
        ),
    ]


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


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

    if name == "get_account":
        data = await sf_get(
            f"/services/data/{SFDC_API_VERSION}/sobjects/Account/{arguments['account_id']}"
        )
        return [TextContent(type="text", text=str(data))]

    if name == "get_opportunity":
        data = await sf_get(
            f"/services/data/{SFDC_API_VERSION}/sobjects/Opportunity/{arguments['opp_id']}"
        )
        return [TextContent(type="text", text=str(data))]

    if name == "get_contact":
        data = await sf_get(
            f"/services/data/{SFDC_API_VERSION}/sobjects/Contact/{arguments['contact_id']}"
        )
        return [TextContent(type="text", text=str(data))]

    if name == "get_lead":
        data = await sf_get(
            f"/services/data/{SFDC_API_VERSION}/sobjects/Lead/{arguments['lead_id']}"
        )
        return [TextContent(type="text", text=str(data))]

    if name == "query":
        if arguments.get("bypass_sharing", False):
            raise ValueError(
                "bypass_sharing=True is not supported in this scaffold; "
                "use a Tooling-API client and a documented justification."
            )
        soql = harden_soql(arguments["soql"])
        result = await sf_query(soql)
        # Cap returned record count even if Salesforce paginated.
        records = result.get("records", [])[:MAX_RECORDS_PER_REQUEST]
        return [
            TextContent(
                type="text",
                text=str({"totalSize": result.get("totalSize"), "records": records}),
            )
        ]

    if name == "pipeline_by_stage":
        window = arguments.get("close_date_window_days", 90)
        cutoff = (datetime.now(timezone.utc) + timedelta(days=window)).date().isoformat()
        owner_clause = ""
        if owner := arguments.get("owner_id"):
            # Single-quote escape: SOQL strings use single quotes and \\' escapes them.
            safe = owner.replace("'", "\\'")
            owner_clause = f" AND OwnerId = '{safe}'"
        soql = (
            "SELECT StageName, COUNT(Id) opps, SUM(Amount) total "
            "FROM Opportunity "
            f"WHERE IsClosed = false AND CloseDate <= {cutoff}{owner_clause} "
            "GROUP BY StageName"
        )
        soql = harden_soql(soql)
        result = await sf_query(soql)
        return [TextContent(type="text", text=str(result))]

    if name == "stale_opps":
        threshold = arguments.get("days_in_stage_threshold", 30)
        cutoff = (
            datetime.now(timezone.utc) - timedelta(days=threshold)
        ).date().isoformat()
        soql = (
            "SELECT Id, Name, StageName, Amount, OwnerId, LastStageChangeDate "
            "FROM Opportunity "
            f"WHERE IsClosed = false AND LastStageChangeDate <= {cutoff} "
            "ORDER BY LastStageChangeDate ASC"
        )
        soql = harden_soql(soql)
        result = await sf_query(soql)
        return [TextContent(type="text", text=str(result))]

    if name == "at_risk_commits":
        # Validate the supplied quarter_end_date to fail before hitting the API.
        try:
            qe = datetime.fromisoformat(arguments["quarter_end_date"]).date()
        except ValueError as exc:
            raise ValueError("quarter_end_date must be ISO format YYYY-MM-DD") from exc

        activity_cutoff = (
            datetime.now(timezone.utc) - timedelta(days=14)
        ).date().isoformat()
        close_window_start = (qe - timedelta(days=14)).isoformat()
        close_window_end = qe.isoformat()
        # Single-quote escape on the picklist label.
        commit = SFDC_COMMIT_STAGE_NAME.replace("'", "\\'")
        soql = (
            "SELECT Id, Name, AccountId, Amount, CloseDate, OwnerId, "
            "LastActivityDate, StageName "
            "FROM Opportunity "
            f"WHERE IsClosed = false AND StageName = '{commit}' AND "
            f"(LastActivityDate <= {activity_cutoff} OR "
            f"(CloseDate >= {close_window_start} AND CloseDate <= {close_window_end})) "
            "ORDER BY CloseDate ASC"
        )
        soql = harden_soql(soql)
        result = await sf_query(soql)
        return [TextContent(type="text", text=str(result))]

    if name == "add_note":
        # Step 1: create the ContentNote (Title + Content, base64-encoded).
        import base64

        body_b64 = base64.b64encode(arguments["body"].encode("utf-8")).decode("ascii")
        note = await sf_post(
            f"/services/data/{SFDC_API_VERSION}/sobjects/ContentNote",
            {
                "Title": f"Claude note on {arguments['object_type']} {arguments['object_id']}",
                "Content": body_b64,
            },
        )
        # Step 2: link via ContentDocumentLink.
        await sf_post(
            f"/services/data/{SFDC_API_VERSION}/sobjects/ContentDocumentLink",
            {
                "ContentDocumentId": note["id"],
                "LinkedEntityId": arguments["object_id"],
                "ShareType": "V",  # Viewer
                "Visibility": "AllUsers",
            },
        )
        return [
            TextContent(
                type="text",
                text=(
                    f"Added note {note['id']} to "
                    f"{arguments['object_type']} {arguments['object_id']}"
                ),
            )
        ]

    if name == "update_field":
        justification = (arguments.get("justification") or "").strip()
        if len(justification) < 10:
            raise ValueError(
                "justification is mandatory and must be at least 10 characters."
            )

        object_type = arguments["object_type"]
        object_id = arguments["object_id"]
        field_name = arguments["field_name"]
        new_value = arguments["new_value"]

        # Step 1: read the current value so we can store old_value in the audit row.
        try:
            current = await sf_get(
                f"/services/data/{SFDC_API_VERSION}/sobjects/{object_type}/{object_id}",
                {"fields": field_name},
            )
            old_value = current.get(field_name)
        except httpx.HTTPStatusError:
            old_value = None

        # Step 2: write the audit row FIRST. If this fails, do not mutate.
        await sf_post(
            f"/services/data/{SFDC_API_VERSION}/sobjects/{SFDC_AUDIT_OBJECT}",
            {
                "Object_Name__c": object_type,
                "Record_Id__c": object_id,
                "Field_Name__c": field_name,
                "Old_Value__c": str(old_value) if old_value is not None else "",
                "New_Value__c": str(new_value),
                "Justification__c": justification,
                "Performed_By__c": "claude-mcp",
            },
        )

        # Step 3: perform the field update.
        await sf_patch(
            f"/services/data/{SFDC_API_VERSION}/sobjects/{object_type}/{object_id}",
            {field_name: new_value},
        )
        return [
            TextContent(
                type="text",
                text=(
                    f"Updated {object_type} {object_id}.{field_name} "
                    f"(old={old_value!r}, new={new_value!r}); audit row written."
                ),
            )
        ]

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