Hit a Real Sanctions API

pip install httpx

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

import os
import httpx

OPENSANCTIONS_BASE = "https://api.opensanctions.org"
OPENSANCTIONS_API_KEY = os.environ.get("OPENSANCTIONS_API_KEY", "")


@mcp.tool()
def lookup_sanctions_hit(name: str) -> dict:
    """Screen a person or entity name against the live OpenSanctions database
    (OFAC SDN, EU Consolidated, UN, and ~250 other sanction sources).

    Returns a hit record with match details if found, or an explicit no-match
    record otherwise. Use for KYC, transaction counterparty, or onboarding screening.

    Args:
        name: Full legal name or entity name to screen. Case-insensitive.
    """
    payload = {
        "queries": {
            "q1": {
                "schema": "Thing",  # Person | Organization | Thing (Thing matches both)
                "properties": {"name": [name]},
            }
        }
    }
    try:
        response = httpx.post(
            f"{OPENSANCTIONS_BASE}/match/sanctions",
            json=payload,
            headers={"Authorization": f"Bearer {OPENSANCTIONS_API_KEY}"},
            timeout=5.0,
        )
        response.raise_for_status()
        results = response.json().get("responses", {}).get("q1", {}).get("results", [])
    except httpx.TimeoutException:
        return {
            "match": "unknown",
            "name_queried": name,
            "error": "OpenSanctions API timed out after 5s — escalate to manual review",
            "recommended_action": "manual-screening-required",
        }
    except httpx.HTTPStatusError as e:
        return {
            "match": "unknown",
            "name_queried": name,
            "error": f"OpenSanctions API returned {e.response.status_code}",
            "recommended_action": "manual-screening-required",
        }
    except Exception as e:
        return {
            "match": "unknown",
            "name_queried": name,
            "error": f"Sanctions screening failed: {type(e).__name__}",
            "recommended_action": "manual-screening-required",
        }

    if not results:
        return {
            "match": False,
            "name_queried": name,
            "screened_sources": "OpenSanctions aggregated (250+ sources)",
            "recommended_action": "proceed-with-standard-cdd",
        }

    top = results[0]
    return {
        "match": True,
        "name_queried": name,
        "matched_entity": top.get("caption"),
        "score": top.get("score"),
        "schema": top.get("schema"),
        "datasets": top.get("datasets", []),
        "first_seen": top.get("first_seen"),
        "recommended_action": "halt-and-escalate",
        "source_url": f"https://www.opensanctions.org/entities/{top.get('id', '')}/",
    }

Screen the individual "Vladimir Putin" for sanctions. Tell me which lists
he appears on, the match score, and the recommended action. Include the
OpenSanctions source URL in your answer.

Triage alert ALT-2026-00099: a $75,000 wire transfer to counterparty
"Kim Jong Un" in North Korea (KP). Walk me through the screening result,
every flag that fired, and the recommended action.

def _local_fallback(name: str, reason: str) -> dict:
    """Fallback to the small embedded SDN list when the upstream API is unavailable.

    NOT a substitute for the full sanctions check — clearly marks the response
    as a degraded reading so investigators know to re-screen when upstream recovers.
    """
    key = name.strip().lower()
    if key in SDN_LIST:
        hit = SDN_LIST[key]
        return {
            "match": True,
            "name_queried": name,
            "degraded_reading": True,
            "fallback_reason": reason,
            "list": hit["list"],
            "program": hit["program"],
            "added": hit["added"],
            "recommended_action": "halt-and-escalate-AND-re-screen-when-upstream-recovers",
        }
    return {
        "match": "unknown",
        "name_queried": name,
        "degraded_reading": True,
        "fallback_reason": reason,
        "screened_sources": "embedded fallback only (3 names)",
        "recommended_action": "manual-screening-required",
    }

    except httpx.TimeoutException:
        return _local_fallback(name, "OpenSanctions API timed out after 5s")
    except httpx.HTTPStatusError as e:
        return _local_fallback(name, f"OpenSanctions API returned {e.response.status_code}")
    except Exception as e:
        return _local_fallback(name, f"Sanctions screening failed: {type(e).__name__}")

from time import time

_CACHE: dict[str, tuple[float, dict]] = {}
_CACHE_TTL_SECONDS = 3600  # 1 hour — adjust for compliance freshness requirements


def _cache_get(name: str) -> dict | None:
    key = name.strip().lower()
    if key not in _CACHE:
        return None
    ts, value = _CACHE[key]
    if time() - ts > _CACHE_TTL_SECONDS:
        del _CACHE[key]
        return None
    return value


def _cache_set(name: str, value: dict) -> None:
    _CACHE[name.strip().lower()] = (time(), value)

    cached = _cache_get(name)
    if cached is not None:
        return {**cached, "from_cache": True}

    # Replace the no-match return:
    no_match = {
        "match": False,
        "name_queried": name,
        "screened_sources": "OpenSanctions aggregated (250+ sources)",
        "recommended_action": "proceed-with-standard-cdd",
    }
    _cache_set(name, no_match)
    return no_match

    # And the hit return:
    hit = { ... existing dict ... }
    _cache_set(name, hit)
    return hit