Hit a Real Sanctions API

From demo to real — what changes

The mocked lookup_sanctions_hit tool returned in microseconds with no failure modes. Real-world tool design isn't like that. Real APIs:

• Take 100ms-2s instead of microseconds
• Sometimes return 429 (rate limited), 500 (down), or timeout
• Return data in their schema, not yours
• Have terms of service and rate limits
• Cost money or have free-tier limits
• Need credential handling

Every production MCP tool wraps a flaky upstream system. The patterns you learn here — timeout, retry, fallback, caching, schema mapping — are the patterns for tool design in compliance work.

1Install httpx~2 min

We'll use httpx for async HTTP. It's the modern requests with timeout/retry support and works cleanly in async contexts.

pip install httpx

2Replace the mock with the real call~10 min

First, get an OpenSanctions API key. Sign up at opensanctions.org/account and copy the key from your account page. The matching endpoint will return 401 Unauthorized without it.

Wire it into Claude Desktop via the server config (Claude Desktop spawns your server as a subprocess and passes env through):

{
  "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"
      }
    }
  }
}

For standalone testing with mcp dev, export it in your shell first: export OPENSANCTIONS_API_KEY=your-key-here.

Open server.py. Add these imports at the top:

import os
import httpx

Now replace the entire lookup_sanctions_hit function with this real-API version. Keep the old SDN_LIST dictionary around — we'll use it as the fallback in step 3.

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"ApiKey {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', '')}/",
    }

if sanctions["match"] is True:
    # confirmed hit — safe to read list/program/score
    ...
elif sanctions["match"] == "unknown":
    # screening was degraded — flag it, escalate, do NOT treat as clean
    ...
# else: confirmed no match, proceed

Restart Claude Desktop and paste either prompt below into a new chat. Both hit the live API; the first exercises lookup_sanctions_hit directly, the second drives the full summarize_alert chain so you can see the API result flow through to a recommendation.

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.

Expect a halt-and-escalate recommendation, a score near 1.0, ~30+ datasets (OFAC SDN, EU Consolidated, UK HMT, UN, etc.), and a source_url ending in /entities/Q7747/ or similar.

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.

Expect three flags on the response: SANCTIONS HIT (from the live API), PROHIBITED JURISDICTION: KP (from your local JURISDICTION_RISK table), and CTR THRESHOLD (amount ≥ $10K). Recommendation: halt-and-escalate.

# macOS
tail -f ~/Library/Logs/Claude/mcp-server-compliance-toolkit.log | grep -E "sanctions|tool_call"

3Add a local fallback~5 min

Real-world systems need to degrade gracefully when an upstream is down. Let's add a fallback to the local SDN_LIST we kept from the original tutorial.

Update lookup_sanctions_hit — in each of the three except branches, instead of returning "manual-screening-required", try the local fallback first:

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",
    }

Now in lookup_sanctions_hit, replace the three error returns with:

    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__}")

4Add response caching~5 min

Real APIs cost money and have rate limits. Cache identical lookups. For a session-scoped cache, just use a dict:

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)

Wire it into lookup_sanctions_hit — at the top, before any API call:

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

And before each successful return, add _cache_set(name, result):

    # 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

5Test the failure modes deliberately~8 min

The point of this exercise is the failure modes. Force each one:

logging.info(
    "sanctions_screening name=%s outcome=%s source=%s latency_ms=%d",
    name, result.get("match"), "cache" if cached else "api", elapsed_ms,
)

What you can now say in the interview

✓ Check yourself

Answer from memory before opening each one.

1. You set the timeout to 0.001 and screen a name that is not in the three-name fallback list. What does the tool return, and what should the model tell you?

   Show answer

   A result with match set to unknown, a degraded-reading flag, the timeout as the fallback reason and manual-screening-required as the action. The model should say that screening was degraded and recommend manual review. If the name were Vladimir Petrov, the fallback would return a hit and also say to re-screen when the upstream recovers. See test the failure modes.

2. After this chapter, why is "if sanctions["match"]:" in summarize_alert a bug, and what is the fix?

   Show answer

   The match field now has three states: true, false and the string "unknown". A non-empty string is truthy, so the code treats a degraded result as a hit and reads fields that do not exist. Test explicitly: handle true, handle unknown by flagging and escalating, and treat only false as clean. See replace the mock with the real call.

3. When the upstream fails, the tool recommends manual screening instead of proceeding. Why is that the right default?

   Show answer

   In compliance, ambiguity must lead to more review, not less. A silent pass on a failed check is a missed sanctions hit, while an unnecessary escalation costs a person a few minutes. See replace the mock with the real call.

4. Run the same name twice. What differs in the second response and in the log, and what does that prove?

   Show answer

   The second response carries a from-cache marker and returns faster, and the log line shows the source as cache. It proves the cache layer is serving results and that the audit log can tell fresh from cached readings. See add response caching.

5. The guide uses a one-hour cache for live screening and no cache for transaction alerts. Why do they differ, and what do you record?

   Show answer

   A fresh designation could land inside the hour, and alerts are time-sensitive, so staleness costs more there. Onboarding screening trades some freshness for cost and rate limits. Record the TTL and whether each reading was cached, so a reviewer can see what the decision rested on. See add response caching.

6. The vendor rotates its key and every call now returns a 401. What does the code do today, and what do you add so someone notices?

   Show answer

   Every call falls back to the small local list with a degraded flag, so the system keeps answering while quietly losing coverage. Add an alert on the rate of degraded readings and fallbacks, and log the failure class, so an outage becomes visible to people. See add a local fallback.

Takeaway

• A real upstream is slow, flaky, rate limited and shaped differently from your data, so every tool that wraps one needs a timeout, error handling and schema mapping.
• Return failures as structured data with an explicit unknown state, and do not raise into the agent loop.
• Failures default to more review: a failed screening says manual screening required, never proceed.
• A tri-state result forces every caller to handle the unknown branch, so test with an explicit comparison, not truthiness.
• A fallback must mark itself as a degraded reading and say to re-screen when the upstream recovers.
• Cache freshness is a policy choice, and the audit log should show which reading was used.

The tool now survives a flaky upstream; the last stretch moves the server onto HTTP and adds authentication. → Run the Server Over HTTP

Further reading

• Getting started with the API — OpenSanctions, accessed 2026. Authentication and the shape of API calls.
• The matching API — OpenSanctions, accessed 2026. The request and response format the tool maps into its own schema.
• Sanctions List Service — OFAC, accessed 2026. The primary source of the US lists that screening tools aggregate.
• Exponential Backoff And Jitter — AWS Architecture Blog, 2015. Why retries need randomized delay when an upstream is struggling.
• Tools — Model Context Protocol, 2025-06-18 revision. Why tool failures belong in the result, with an error flag.