How to Automate Vendor Due Diligence Screening with SanctionShield AI API

pip install httpx python-dotenv asyncio

# .env
SANCTIONSHIELD_API_KEY=YOUR_API_KEY
DATABASE_URL=postgresql://user:password@localhost/vendors

import httpx
import asyncio
import os
from datetime import datetime
from typing import Optional
 
SANCTIONSHIELD_BASE_URL = "https://apivult.com/api/sanctionshield"
 
async def screen_entity(
    name: str,
    country: Optional[str] = None,
    dob: Optional[str] = None,
    entity_type: str = "organization"
) -> dict:
    """
    Screen an entity against all major sanctions lists.
    Returns screening result with match score and list sources.
    """
    async with httpx.AsyncClient(timeout=10.0) as client:
        response = await client.post(
            f"{SANCTIONSHIELD_BASE_URL}/screen",
            headers={
                "X-RapidAPI-Key": os.getenv("SANCTIONSHIELD_API_KEY"),
                "Content-Type": "application/json"
            },
            json={
                "name": name,
                "country": country,
                "date_of_birth": dob,
                "entity_type": entity_type,
                "lists": ["OFAC_SDN", "EU_CONSOLIDATED", "UN_SECURITY_COUNCIL", "HM_TREASURY"],
                "fuzzy_match": True,
                "match_threshold": 0.85
            }
        )
        response.raise_for_status()
        return response.json()
 
 
async def screen_vendor_with_ubos(vendor: dict) -> dict:
    """
    Screen a vendor entity plus all its ultimate beneficial owners.
    Returns aggregated risk assessment.
    """
    tasks = [
        screen_entity(
            name=vendor["legal_name"],
            country=vendor.get("country"),
            entity_type="organization"
        )
    ]
 
    # Screen each UBO
    for ubo in vendor.get("ubos", []):
        tasks.append(
            screen_entity(
                name=ubo["full_name"],
                country=ubo.get("nationality"),
                dob=ubo.get("date_of_birth"),
                entity_type="individual"
            )
        )
 
    results = await asyncio.gather(*tasks)
 
    # Aggregate: any match = blocked
    any_match = any(r.get("match_found", False) for r in results)
    highest_score = max(r.get("match_score", 0) for r in results)
 
    matches = [r for r in results if r.get("match_found")]
 
    return {
        "vendor_id": vendor["id"],
        "vendor_name": vendor["legal_name"],
        "screening_date": datetime.utcnow().isoformat(),
        "status": "BLOCKED" if any_match else "APPROVED",
        "highest_match_score": highest_score,
        "matches": matches,
        "screened_entities": len(tasks)
    }

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from typing import List, Optional
 
app = FastAPI()
 
 
class UBO(BaseModel):
    full_name: str
    nationality: Optional[str] = None
    date_of_birth: Optional[str] = None
    ownership_percentage: float
 
 
class VendorOnboardingRequest(BaseModel):
    legal_name: str
    country: str
    tax_id: Optional[str] = None
    ubos: List[UBO] = []
 
 
@app.post("/api/vendors/onboard")
async def onboard_vendor(request: VendorOnboardingRequest):
    vendor_data = {
        "id": generate_vendor_id(),
        "legal_name": request.legal_name,
        "country": request.country,
        "ubos": [u.dict() for u in request.ubos]
    }
 
    # Screen before storing
    screening_result = await screen_vendor_with_ubos(vendor_data)
 
    if screening_result["status"] == "BLOCKED":
        # Log the attempt but do not onboard
        log_blocked_vendor(vendor_data, screening_result)
        raise HTTPException(
            status_code=403,
            detail={
                "error": "vendor_blocked",
                "reason": "Sanctions match detected during screening",
                "case_reference": generate_case_id(),
                "compliance_contact": "[email protected]"
            }
        )
 
    # Safe to onboard
    vendor_id = save_vendor_to_database(vendor_data, screening_result)
 
    return {
        "vendor_id": vendor_id,
        "status": "approved",
        "screening_passed": True,
        "screening_date": screening_result["screening_date"]
    }

import asyncio
from datetime import datetime, timedelta
 
async def daily_rescreening_job():
    """
    Re-screen all active vendors that haven't been screened in 24 hours.
    Run this as a daily cron job.
    """
    vendors = get_vendors_needing_rescreening(hours_since_last_check=24)
 
    print(f"[{datetime.utcnow()}] Re-screening {len(vendors)} vendors...")
 
    blocked_vendors = []
 
    for vendor in vendors:
        result = await screen_vendor_with_ubos(vendor)
 
        update_vendor_screening_record(vendor["id"], result)
 
        if result["status"] == "BLOCKED":
            # This vendor was previously clean — now matched
            blocked_vendors.append({
                "vendor": vendor,
                "match_details": result["matches"]
            })
 
            # Suspend pending payments immediately
            suspend_vendor_payments(vendor["id"])
 
            # Notify compliance team
            send_compliance_alert(
                subject=f"URGENT: Vendor {vendor['legal_name']} now matches sanctions list",
                details=result
            )
 
    print(f"Re-screening complete. {len(blocked_vendors)} new matches found.")
    return blocked_vendors
 
 
def get_vendors_needing_rescreening(hours_since_last_check: int) -> list:
    cutoff = datetime.utcnow() - timedelta(hours=hours_since_last_check)
    # Query your database for vendors where last_screened_at < cutoff
    # Return vendor records with UBO data
    return []  # Replace with actual DB query
 
 
# Schedule with cron or APScheduler
if __name__ == "__main__":
    asyncio.run(daily_rescreening_job())

async def pre_payment_screen(
    vendor_id: str,
    payment_amount: float,
    currency: str
) -> dict:
    """
    Final gate check before authorizing payment.
    Screens vendor in real time regardless of scheduled re-screening status.
    """
    vendor = get_vendor_by_id(vendor_id)
 
    if not vendor:
        return {"authorized": False, "reason": "vendor_not_found"}
 
    # Always re-screen for high-value payments
    if payment_amount > 10000:
        result = await screen_vendor_with_ubos(vendor)
 
        if result["status"] == "BLOCKED":
            return {
                "authorized": False,
                "reason": "sanctions_match",
                "case_reference": generate_case_id(),
                "screening_result": result
            }
 
    # Check cached screening status for lower-value payments
    cached_result = get_latest_screening_result(vendor_id)
    if cached_result and cached_result["status"] == "BLOCKED":
        return {
            "authorized": False,
            "reason": "vendor_suspended",
            "screening_date": cached_result["screening_date"]
        }
 
    return {
        "authorized": True,
        "vendor_id": vendor_id,
        "amount": payment_amount,
        "currency": currency
    }

def generate_compliance_report(period_start: str, period_end: str) -> dict:
    """
    Generate a compliance report for the specified period.
    Required for OFAC compliance documentation.
    """
    screenings = get_screenings_in_period(period_start, period_end)
 
    return {
        "report_period": {"start": period_start, "end": period_end},
        "generated_at": datetime.utcnow().isoformat(),
        "summary": {
            "total_screenings": len(screenings),
            "approved": sum(1 for s in screenings if s["status"] == "APPROVED"),
            "blocked": sum(1 for s in screenings if s["status"] == "BLOCKED"),
            "lists_checked": ["OFAC_SDN", "EU_CONSOLIDATED", "UN_SECURITY_COUNCIL", "HM_TREASURY"]
        },
        "blocked_vendors": [s for s in screenings if s["status"] == "BLOCKED"],
        "methodology": "AI-powered fuzzy matching against consolidated watchlists via SanctionShield AI API"
    }