Real-Time AML Sanctions Screening in Python: A Complete Integration Guide

pip install requests python-dotenv pydantic

# .env
SANCTIONSHIELD_API_KEY=YOUR_API_KEY

import os
import requests
from dotenv import load_dotenv
 
load_dotenv()
 
API_KEY = os.getenv("SANCTIONSHIELD_API_KEY")
BASE_URL = "https://apivult.com/sanctionshield/v1"
 
HEADERS = {
    "X-RapidAPI-Key": API_KEY,
    "Content-Type": "application/json",
}

from pydantic import BaseModel
from typing import Optional
 
class IndividualScreenRequest(BaseModel):
    full_name: str
    date_of_birth: Optional[str] = None  # ISO 8601: YYYY-MM-DD
    nationality: Optional[str] = None     # ISO 3166-1 alpha-2
    id_number: Optional[str] = None
    fuzzy_threshold: float = 0.85         # 0.0–1.0, higher = stricter matching
 
def screen_individual(person: IndividualScreenRequest) -> dict:
    """Screen an individual against all sanctions lists."""
    response = requests.post(
        f"{BASE_URL}/screen/individual",
        headers=HEADERS,
        json=person.dict(exclude_none=True),
        timeout=10,
    )
    response.raise_for_status()
    return response.json()
 
# Example: Screen a new customer at signup
result = screen_individual(IndividualScreenRequest(
    full_name="John Smith",
    date_of_birth="1985-04-12",
    nationality="US",
    fuzzy_threshold=0.88,
))
 
print(result)

{
  "screening_id": "scr_7x9k2m3p",
  "status": "clear",
  "match_count": 0,
  "lists_checked": ["OFAC_SDN", "OFAC_NON_SDN", "UN_CONSOLIDATED", "EU_CONSOLIDATED", "UK_HMT"],
  "screened_at": "2026-03-31T09:14:22Z",
  "processing_ms": 87
}

class EntityScreenRequest(BaseModel):
    entity_name: str
    country: Optional[str] = None
    registration_number: Optional[str] = None
    aliases: Optional[list[str]] = None
    fuzzy_threshold: float = 0.85
 
def screen_entity(entity: EntityScreenRequest) -> dict:
    """Screen a business entity against sanctions lists."""
    response = requests.post(
        f"{BASE_URL}/screen/entity",
        headers=HEADERS,
        json=entity.dict(exclude_none=True),
        timeout=10,
    )
    response.raise_for_status()
    return response.json()
 
# Example: Screen a counterparty before onboarding
result = screen_entity(EntityScreenRequest(
    entity_name="Acme Trading LLC",
    country="AE",
    aliases=["Acme Trade", "ACME LLC"],
    fuzzy_threshold=0.90,  # stricter for entities
))

{
  "screening_id": "scr_3p8n1q4r",
  "status": "review_required",
  "match_count": 1,
  "matches": [
    {
      "list": "OFAC_SDN",
      "matched_name": "ACME TRADING COMPANY",
      "similarity_score": 0.91,
      "entry_id": "SDN-17842",
      "designation_date": "2024-11-03",
      "program": "IRAN",
      "additional_info": "Vessel operator linked to petroleum shipments"
    }
  ],
  "recommendation": "manual_review",
  "screened_at": "2026-03-31T09:15:44Z"
}

from enum import Enum
 
class ScreeningStatus(str, Enum):
    CLEAR = "clear"
    REVIEW_REQUIRED = "review_required"
    CONFIRMED_MATCH = "confirmed_match"
 
def process_screening_result(result: dict, entity_id: str) -> str:
    """
    Apply business rules based on screening result.
    Returns the action taken: 'approved', 'flagged', 'blocked'
    """
    status = result.get("status")
 
    if status == ScreeningStatus.CLEAR:
        log_screening(entity_id, result, action="approved")
        return "approved"
 
    elif status == ScreeningStatus.REVIEW_REQUIRED:
        # Queue for compliance officer review
        create_review_task(entity_id, result)
        log_screening(entity_id, result, action="flagged")
        return "flagged"
 
    elif status == ScreeningStatus.CONFIRMED_MATCH:
        # Auto-block — no human override possible for confirmed hits
        block_entity(entity_id, result)
        notify_compliance_team(entity_id, result)
        log_screening(entity_id, result, action="blocked")
        return "blocked"
 
    else:
        raise ValueError(f"Unknown screening status: {status}")
 
def log_screening(entity_id: str, result: dict, action: str):
    """Persist screening record for audit trail."""
    record = {
        "entity_id": entity_id,
        "screening_id": result["screening_id"],
        "action": action,
        "screened_at": result["screened_at"],
        "lists_checked": result["lists_checked"],
        "match_count": result["match_count"],
    }
    # Save to your database / audit log system
    print(f"[AUDIT] {record}")

def screen_payment(payment: dict) -> dict:
    """
    Screen all parties in a payment transaction simultaneously.
    payment: { sender, recipient, amount, currency, reference }
    """
    # Screen both sender and recipient in parallel
    import concurrent.futures
 
    parties = [
        {"role": "sender", "data": payment["sender"]},
        {"role": "recipient", "data": payment["recipient"]},
    ]
 
    results = {}
    with concurrent.futures.ThreadPoolExecutor(max_workers=2) as executor:
        futures = {
            executor.submit(
                screen_individual,
                IndividualScreenRequest(**party["data"])
            ): party["role"]
            for party in parties
        }
        for future in concurrent.futures.as_completed(futures):
            role = futures[future]
            results[role] = future.result()
 
    # Payment is clear only if ALL parties are clear
    all_clear = all(r["status"] == "clear" for r in results.values())
 
    return {
        "payment_id": payment.get("id"),
        "overall_status": "clear" if all_clear else "blocked",
        "party_results": results,
    }
 
# Example usage
payment = {
    "id": "txn_abc123",
    "sender": {"full_name": "Alice Johnson", "nationality": "GB"},
    "recipient": {"full_name": "Bob Martinez", "nationality": "MX"},
    "amount": 15000,
    "currency": "USD",
}
 
result = screen_payment(payment)
print(result)

def batch_screen_customers(customer_list: list[dict]) -> list[dict]:
    """
    Batch screen a list of customers.
    Handles rate limiting with exponential backoff.
    """
    import time
    results = []
 
    for i, customer in enumerate(customer_list):
        try:
            result = screen_individual(IndividualScreenRequest(**customer))
            results.append({
                "customer_id": customer.get("id"),
                **result
            })
        except requests.exceptions.HTTPError as e:
            if e.response.status_code == 429:  # rate limit
                retry_after = int(e.response.headers.get("Retry-After", 1))
                time.sleep(retry_after)
                # retry once
                result = screen_individual(IndividualScreenRequest(**customer))
                results.append({"customer_id": customer.get("id"), **result})
            else:
                raise
 
        # Progress logging
        if (i + 1) % 100 == 0:
            print(f"Screened {i + 1}/{len(customer_list)} customers")
 
    return results
 
# Generate compliance report
def generate_screening_report(results: list[dict]) -> dict:
    total = len(results)
    clear = sum(1 for r in results if r["status"] == "clear")
    review = sum(1 for r in results if r["status"] == "review_required")
    blocked = sum(1 for r in results if r["status"] == "confirmed_match")
 
    return {
        "total_screened": total,
        "clear": clear,
        "review_required": review,
        "confirmed_matches": blocked,
        "clear_rate": f"{(clear / total * 100):.1f}%",
        "generated_at": datetime.utcnow().isoformat() + "Z",
    }