Build a Compliance MCP Server

"""compliance-toolkit MCP server — three tools for AI-assisted compliance triage."""

from mcp.server.fastmcp import FastMCP
from typing import Literal

mcp = FastMCP("compliance-toolkit")


# ---------- Mock data (in production: real sources) ----------

SDN_LIST = {
    "vladimir petrov": {"list": "OFAC SDN", "added": "2024-03-15", "program": "RUSSIA-EO14024"},
    "global shadow holdings": {"list": "OFAC SDN", "added": "2023-11-02", "program": "SDGT"},
    "atlas crypto exchange": {"list": "EU Consolidated", "added": "2025-01-20", "program": "EU-CYBER"},
}

JURISDICTION_RISK = {
    "US": ("low",  "FATF compliant, mature AML regime"),
    "GB": ("low",  "FATF compliant, FCA oversight"),
    "DE": ("low",  "FATF compliant, BaFin oversight"),
    "AE": ("medium", "FATF grey list 2022-2024, recently exited"),
    "KY": ("medium", "Offshore jurisdiction, enhanced due diligence advised"),
    "RU": ("high", "Comprehensive sanctions, enhanced screening required"),
    "IR": ("prohibited", "OFAC comprehensive sanctions — do not transact"),
    "KP": ("prohibited", "OFAC comprehensive sanctions — do not transact"),
}


# ---------- Tool 1: sanctions screening ----------

@mcp.tool()
def lookup_sanctions_hit(name: str) -> dict:
    """Screen a person or entity name against active sanctions lists (OFAC SDN, EU Consolidated).

    Use this for any KYC review, transaction counterparty check, or onboarding screening.
    Returns a hit record if matched, or an explicit no-match record otherwise.
    Do NOT use this for adverse-media screening — that is a separate tool.

    Args:
        name: Full legal name or entity name to screen. Case-insensitive.
    """
    key = name.strip().lower()
    if key in SDN_LIST:
        hit = SDN_LIST[key]
        return {
            "match": True,
            "name_queried": name,
            "list": hit["list"],
            "program": hit["program"],
            "added": hit["added"],
            "recommended_action": "halt-and-escalate",
        }
    return {
        "match": False,
        "name_queried": name,
        "screened_lists": ["OFAC SDN", "EU Consolidated"],
        "recommended_action": "proceed-with-standard-cdd",
    }


# ---------- Tool 2: jurisdiction risk ----------

@mcp.tool()
def check_jurisdiction_risk(country_code: str) -> dict:
    """Return the AML risk tier for a country (ISO 3166-1 alpha-2 code).

    Tiers: 'low', 'medium', 'high', 'prohibited'.
    Prohibited jurisdictions must not be transacted with.

    Args:
        country_code: Two-letter ISO country code, e.g. 'US', 'DE', 'KP'.
    """
    code = country_code.strip().upper()
    if code not in JURISDICTION_RISK:
        return {
            "country_code": code,
            "tier": "unknown",
            "rationale": "Country code not in risk register — flag for review.",
        }
    tier, rationale = JURISDICTION_RISK[code]
    return {"country_code": code, "tier": tier, "rationale": rationale}


# ---------- Tool 3: alert summarization ----------

@mcp.tool()
def summarize_alert(
    alert_id: str,
    counterparty_name: str,
    counterparty_country: str,
    amount_usd: float,
    transaction_type: Literal["wire", "crypto-withdrawal", "crypto-deposit", "card"],
) -> dict:
    """Produce a structured triage summary for a transaction alert.

    Combines counterparty screening and jurisdiction risk into a single recommendation.
    The recommendation is advisory — a human investigator must approve any external action.
    """
    sanctions = lookup_sanctions_hit(counterparty_name)
    jurisdiction = check_jurisdiction_risk(counterparty_country)

    flags = []
    if sanctions["match"]:
        flags.append(f"SANCTIONS HIT: {sanctions['list']} ({sanctions['program']})")
    if jurisdiction["tier"] == "prohibited":
        flags.append(f"PROHIBITED JURISDICTION: {jurisdiction['country_code']}")
    if jurisdiction["tier"] == "high":
        flags.append(f"HIGH-RISK JURISDICTION: {jurisdiction['country_code']}")
    if amount_usd >= 10_000:
        flags.append(f"CTR THRESHOLD: amount ${amount_usd:,.2f} ≥ $10K")

    if any(f.startswith("SANCTIONS") or f.startswith("PROHIBITED") for f in flags):
        recommendation = "halt-and-escalate"
    elif flags:
        recommendation = "escalate-for-human-review"
    else:
        recommendation = "low-risk-suggest-dismiss"

    return {
        "alert_id": alert_id,
        "summary": (
            f"Alert {alert_id}: {transaction_type} of ${amount_usd:,.2f} with "
            f"counterparty '{counterparty_name}' ({counterparty_country})."
        ),
        "flags": flags or ["no automated flags"],
        "recommendation": recommendation,
        "requires_human_approval": True,  # always — agent never auto-acts
        "sources": {"sanctions": sanctions, "jurisdiction": jurisdiction},
    }


if __name__ == "__main__":
    mcp.run()

mcp dev server.py

{
  "alert_id": "ALT-2026-00042",
  "counterparty_name": "Vladimir Petrov",
  "counterparty_country": "RU",
  "amount_usd": 25000,
  "transaction_type": "wire"
}

{
  "mcpServers": {
    "compliance-toolkit": {
      "command": "/absolute/path/to/compliance-mcp/.venv/bin/python",
      "args": ["/absolute/path/to/compliance-mcp/server.py"]
    }
  }
}

I need to triage alert ALT-2026-00042. A wire transfer of $25,000 from a
counterparty named Vladimir Petrov in Russia. What's your recommendation?

Screen the entity 'Global Shadow Holdings' for sanctions and tell me what
action to take.

What's the AML risk tier for transactions touching the Cayman Islands?

I'm reviewing three counterparties for onboarding: ACME Corp from Germany,
Atlas Crypto Exchange from the UAE, and Joe Smith from North Korea. Screen
each and tell me what to do.