Anti-Money Laundering (AML) API Implementation

36 min

This guide requires an API user with Compliance Management permissions. Without these permissions, the AML API calls will fail with permission errors.

This guide covers three AML implementation approaches: Manual review, Automated via SDK, and Chainalysis-driven automation. Follow the step-by-step procedures to implement AML controls in your Custody Infrastructure environment.

Introduction

Anti-Money Laundering (AML) controls are essential for detecting and preventing suspicious cryptocurrency transactions. This guide provides detailed procedures for implementing Custody Infrastructure's AML features, allowing you to screen transactions, freeze suspicious activity, and maintain regulatory compliance.

Rather than a one-time setup, AML implementation is an ongoing operational process requiring regular monitoring and action. This guide walks you through practical, step-by-step procedures for each implementation approach.

Summary

What is AML Compliance?What is AML Compliance?

Understanding the AML ProcessUnderstanding the AML Process

General Prerequisites

Implementation Approach 1: Manual

Implementation Approach 2: Automated via SDK

Implementation Approach 3: Chainalysis-driven

Monitoring and reporting frozen transactions

Endpoint Deep Dive:

Transaction Monitoring Endpoints:

getTransactionById

AML Control Endpoints:

freezeTransaction

unfreezeTransaction

exportFrozenTransactions

Chainalysis Integration Endpoints:

uploadChainalysisToken

healthCheckChainalysis

removeChainalysisToken

Webhook Endpoint (For SDK Integration):

registerWebhook

What is AML Compliance?

Anti-money laundering (AML) compliance refers to the processes and controls implemented to detect, prevent, and report suspicious cryptocurrency transactions that might indicate money laundering, terrorist financing, or other financial crimes. In the context of Custody Infrastructure, this involves:

Transaction Monitoring: Screening both incoming and outgoing transactions for suspicious patterns

Address Risk Assessment: Verifying wallet addresses against known high-risk entities

Transaction Freezing: Blocking suspicious transactions and funds pending investigation

Reporting: Documenting and reporting suspicious activity to the appropriate authorities

Understanding the AML Process

The AML process consists of two primary flows:

Inbound Transaction Flow: When funds arrive, the sending address is screened against risk databases. High-risk transactions trigger automatic or manual freezing, followed by investigation.

Outbound Transaction Flow: Before sending funds, the destination address is screened. Transactions to high-risk destinations can be automatically blocked before execution.

Your choice of implementation approach determines the level of automation in this process, but the fundamental flow remains the same.

There are three AML implementation paths available on the Custody Infrastructure platform:

Manual: Manually review and freeze/unfreeze inbound transactions using direct API calls.

Automated (SDK): Set up webhooks to automate inbound transaction screening with third-party AML providers.

Chainalysis: Implement native Chainalysis AML integration for fully automated screening of both inbound/outbound transactions

Only Chainalysis AML can perform automatic inspection of outbound transactions. Manual and Automated (SDK) AML rely on users to investigate destination addresses before sending a transaction; outbound risk management entirely relies on user diligence in these scenarios.

A detailed breakdown of these paths is found in our Anti-Money Laundering reference article.

General Prerequisites

Before implementing AML controls, ensure you have:

An API user with Compliance Management permissions

API authentication credentials

Familiarity with the usage of the Custody Infrastructure API client

For Chainalysis integration: A valid Chainalysis API token

For third-party integrations: API credentials for your chosen AML provider

Approach 1: Manual AML

In this approach, you manually review inbound transactions and explicitly freeze suspicious ones through a sequence of API calls.

Outbound manual AML compliance relies on system users to manually investigate destination addresses before transferring assets out of the organization.

Manual Inbound Transaction Handling Process Flow

When using the manual approach, the Custody Infrastructure system applies a manual review process, allowing direct oversight and risk management for inbound transactions:

By default, Custody Infrastructure sets the transaction's initial state to Neutral, making the associated funds immediately available for use. The compliance officer then reviews the transaction and decides whether to freeze it based on their analysis.

Step-by-Step Implementation Process

Use getTransactionById to retrieve the transaction details to assess for AML risks:

from ApiClient import ApiClient

api = ApiClient()
response = api.getTransactionById(
    transactionId="0x05...781f", 
    coin_type="ETH"
)

# Extract key transaction data from the response
transaction = response['FunctionData']['Transactions'][0]
txid = transaction.get('TxId')
amount = transaction.get('Amount')
sender = transaction.get('From')
recipient = transaction.get('To')
date = transaction.get('Date')

print(f"Reviewing transaction {txid} for {amount} from {sender} to {recipient}")

Evaluate the transaction based on your organization's pre-determined risk criteria.

Document your findings according to your compliance procedures. If suspicious activity is identified, proceed to step 3.

If a transaction is suspicious, use freezeTransaction to prevent the funds from being used:

from ApiClient import ApiClient

api = ApiClient()
freeze_response = api.freezeTransaction(
    txid="0x05e89fcc12a446a9f123b590823b94e9c9d4781f",
    currency="ETH",
    note="Suspicious deposit pattern from high-risk jurisdiction",
    external_id="CASE-12345"  # Optional reference to your case management system
)

if freeze_response['FunctionData']['Status'] == 'Success':
    print("Transaction successfully frozen")
else:
    print(f"Error freezing transaction: {freeze_response}")

Export the list of frozen transactions via exportFrozenTransactions to confirm it was successfully frozen:

from ApiClient import ApiClient

api = ApiClient()
verify_response = api.exportFrozenTransactions()

if 'FunctionData' in verify_response and 'Data' in verify_response['FunctionData']:
    frozen_data = verify_response['FunctionData']['Data']
    
    # Check if transaction is in the frozen list
    for currency, txids in frozen_data.items():
        if "0x05e89fcc12a446a9f123b590823b94e9c9d4781f" in txids:
            print(f"Confirmed: Transaction is frozen under {currency}")
            break
    else:
        print("Warning: Transaction not found in frozen list")

After freezing the transaction, you may conduct a thorough investigation according to your compliance procedures.

This step is performed outside the API system as part of your compliance workflow.

If you determine the transaction is legitimate, unfreeze it with unfreezeTransaction. A transaction may be unfrozen at any time.

from ApiClient import ApiClient

api = ApiClient()
unfreeze_response = api.unfreezeTransaction(
    txid="0x05e89fcc12a446a9f123b590823b94e9c9d4781f",
    currency="ETH",
    note="Transaction verified as legitimate after investigation. Source funds traced to legitimate business operations."
)

if unfreeze_response['FunctionData']['Status'] == 'Success':
    print("Transaction successfully unfrozen")
else:
    print(f"Error unfreezing transaction: {unfreeze_response}")

Export the list of frozen transactions via exportFrozenTransactions to confirm it was successfully unfrozen:

from ApiClient import ApiClient

api = ApiClient()
verify_unfreeze = api.exportFrozenTransactions()

if 'FunctionData' in verify_unfreeze and 'Data' in verify_unfreeze['FunctionData']:
    frozen_data = verify_unfreeze['FunctionData']['Data']
    
    # Check all currencies to confirm transaction is no longer frozen
    transaction_still_frozen = False
    for currency, txids in frozen_data.items():
        if "0x05e89fcc12a446a9f123b590823b94e9c9d4781f" in txids:
            transaction_still_frozen = True
            print(f"Warning: Transaction still frozen under {currency}")
            break
    
    if not transaction_still_frozen:
        print("Confirmed: Transaction is no longer frozen")

Approach 2: Automated AML via SDK

This approach uses webhooks to automatically screen inbound transactions through a third-party AML provider.

Prerequisites

Before you can automate your inbound AML process via the SDK, you must have credentials for your third-party provider of choice and instructions on how to send the provider the transaction info and what form the responses will take.

SDK Integration Process Flow

Using SDK integration with your AML provider follows this structured process flow:

Webhook Format (v2) for Inbound Transactions

When an inbound transaction is detected, Custody Infrastructure sends a webhook with this structure:

{
  "Event": "InputTransaction",
  "Timestamp": "2024-01-26T08:13:12.000Z",
  "FunctionData": {
    "TransactionId": "0xe4...46fc",
    "CoinSymbol": "ETH",
    "Amount": "689207437500000000",
    "Confirmations": "64788",
    "From": "0x6C...F455",
    "To": "0x00...611C",
    "BlockNumber": "1928374",
    "ExtraData": [
      {
        "Key": "AccountId",
        "Value": "1"
      },
      {
        "Key": "Machine",
        "Value": "MPC"
      }
    ]
  }
}

Step-by-Step Implementation Process

from ApiClient import ApiClient
from Helpers.TransactionEnums.TransactionFormatVersion import TransactionFormatVersion

api = ApiClient()
response = api.registerWebhook(
    webhook_url="https://your-server.com/aml-webhook",
    type=0,  # 0 = Input transaction webhook
    filter={
        "CoinSymbol": "ETH",  # Optional filter by currency
        "Machine": "MPC",     # Process MPC transactions
        "AccountIds": ["1"],  # Optional filter by account
        "MinAmount": "10000"  # Optional minimum amount threshold
    },
    transaction_format_version=TransactionFormatVersion.V2
)

# Store the webhook ID for future reference
webhook_id = response['FunctionData']['WebhookId']
print(f"Webhook registered successfully with ID: {webhook_id}")

Create an HTTPS endpoint on your server to receive transaction notifications from Custody Infrastructure.This requires setting up a web server with a publicly accessible HTTPS URL.

This example demonstrates using Flask in Python:

from flask import Flask, request, jsonify
import hmac
import hashlib

app = Flask(__name__)

# Configure a secret key for webhook verification
WEBHOOK_SECRET = "your-webhook-secret"

def verify_webhook_signature(request_data, signature_header):
    """
    Verify that the webhook came from GK8
    """
    expected_signature = hmac.new(
        WEBHOOK_SECRET.encode(),
        request_data,
        hashlib.sha256
    ).hexdigest()
    
    return hmac.compare_digest(expected_signature, signature_header)

@app.route('/aml-webhook', methods=['POST'])
def handle_webhook():
    # Verify webhook authenticity
    if WEBHOOK_SECRET:
        signature = request.headers.get('X-GK8-Signature')
        if not signature or not verify_webhook_signature(request.data, signature):
            return jsonify({'status': 'error', 'message': 'Invalid signature'}), 403
            
    # Process the webhook data
    if request.is_json:
        webhook_data = request.json
        print(f"Received transaction webhook: {webhook_data}")
        
        # The rest of your webhook handling logic will be implemented in step 3
        return jsonify({'status': 'success'})
    
    return jsonify({'status': 'error'}), 400

if __name__ == '__main__':
    app.run(host='0.0.0.0', port=5000, ssl_context=('cert.pem', 'key.pem'))

Configure the webhook handler to automatically forward transaction data to your AML provider for risk assessment.

import requests

# AML provider configuration
AML_PROVIDER_URL = "https://your-aml-provider.com/api/check"
AML_PROVIDER_API_KEY = "your-api-key"

def check_address_with_aml_provider(address, txid, currency, amount):
    """
    Send address to AML provider for risk assessment
    """
    try:
        response = requests.post(
            AML_PROVIDER_URL,
            headers={
                'Authorization': f'Bearer {AML_PROVIDER_API_KEY}',
                'Content-Type': 'application/json'
            },
            json={
                'address': address,
                'blockchain': currency,
                'transaction_id': txid,
                'amount': amount
            },
            timeout=10
        )
        
        if response.status_code == 200:
            return response.json()
        else:
            print(f"AML provider error: {response.status_code} - {response.text}")
            return {'error': f"AML provider returned {response.status_code}", 'risk_level': 'unknown'}
    except Exception as e:
        print(f"Exception when checking address: {e}")
        return {'error': str(e), 'risk_level': 'unknown'}

# Update your webhook handler route with this integration
@app.route('/aml-webhook', methods=['POST'])
def handle_webhook():
    # Verification code from step 2 remains here...
    
    if request.is_json:
        webhook_data = request.json
        
        # Check if this is an inbound transaction
        if webhook_data.get('Event') == 'InputTransaction':
            tx_data = webhook_data.get('FunctionData', {})
            txid = tx_data.get('TransactionId')
            currency = tx_data.get('CoinSymbol')
            source_address = tx_data.get('From')
            amount = tx_data.get('Amount')
            
            if txid and currency and source_address:
                # Check address with AML provider
                risk_data = check_address_with_aml_provider(source_address, txid, currency, amount)
                print(f"AML risk assessment for {txid}: {risk_data}")
                
                # Step 4 will handle what to do with high-risk transactions
                return jsonify({
                    'status': 'success',
                    'risk_assessment': risk_data
                })
        
        return jsonify({'status': 'success'})
    
    return jsonify({'status': 'error'}), 400

Configure your webhook handler to automatically freeze high-risk transactions.

from ApiClient import ApiClient

api = ApiClient()  # Initialize API client

def freeze_transaction_if_risky(txid, currency, risk_data):
    """Freeze transaction if risk level is high or severe"""
    risk_level = risk_data.get('risk_level', 'unknown').lower()
    
    if risk_level in ['high', 'severe']:
        try:
            print(f"High risk transaction detected ({risk_level}). Freezing {txid}...")
            freeze_response = api.freezeTransaction(
                txid=txid,
                currency=currency,
                note=f"Automatically frozen due to {risk_level} risk from AML provider",
                external_id=risk_data.get('case_id')
            )
            
            if freeze_response['FunctionData']['Status'] == 'Success':
                print(f"Successfully froze transaction {txid}")
                
                # Verify freeze by checking frozen transactions list
                verify_response = api.exportFrozenTransactions()
                frozen_data = verify_response['FunctionData']['Data']
                if currency in frozen_data and txid in frozen_data[currency]:
                    print(f"Verified: Transaction {txid} is now in frozen list")
                else:
                    print(f"Warning: Transaction {txid} not found in frozen list despite successful freeze")
                
                return True
            else:
                print(f"Error freezing transaction: {freeze_response}")
                return False
        except Exception as e:
            print(f"Exception when freezing transaction: {e}")
            return False
    
    print(f"Transaction {txid} risk level: {risk_level} - No action needed")
    return False

# Update your webhook handler with the freezing logic
@app.route('/aml-webhook', methods=['POST'])
def handle_webhook():
    # Verification code from step 2 remains here...
    
    if request.is_json:
        webhook_data = request.json
        
        # Check if this is an inbound transaction
        if webhook_data.get('Event') == 'InputTransaction':
            tx_data = webhook_data.get('FunctionData', {})
            txid = tx_data.get('TransactionId')
            currency = tx_data.get('CoinSymbol')
            source_address = tx_data.get('From')
            amount = tx_data.get('Amount')
            
            if txid and currency and source_address:
                # Check address with AML provider
                risk_data = check_address_with_aml_provider(source_address, txid, currency, amount)
                
                # Freeze if risky
                was_frozen = freeze_transaction_if_risky(txid, currency, risk_data)
                
                return jsonify({
                    'status': 'success',
                    'risk_assessment': risk_data,
                    'frozen': was_frozen
                })
        
        return jsonify({'status': 'success'})
    
    return jsonify({'status': 'error'}), 400

Add notifications to alert the compliance team about frozen transactions automatically.

import smtplib
from email.message import EmailMessage
import json

# Email configuration
EMAIL_SERVER = "smtp.yourcompany.com"
EMAIL_PORT = 587
EMAIL_USER = "[email protected]"
EMAIL_PASSWORD = "your-secure-email-password"
COMPLIANCE_TEAM_EMAIL = "[email protected]"

def send_compliance_alert(txid, currency, risk_data, transaction_details):
    """Send email alert to compliance team about a frozen transaction"""
    try:
        msg = EmailMessage()
        msg['Subject'] = f"AML ALERT: High-Risk Transaction Frozen - {currency}"
        msg['From'] = EMAIL_USER
        msg['To'] = COMPLIANCE_TEAM_EMAIL
        
        # Create email body
        body = f"""
        HIGH-RISK TRANSACTION ALERT
        
        A transaction has been automatically frozen due to high AML risk.
        
        Transaction Details:
        - Transaction ID: {txid}
        - Currency: {currency}
        - Amount: {transaction_details.get('Amount', 'Unknown')}
        - From Address: {transaction_details.get('From', 'Unknown')}
        - To Address: {transaction_details.get('To', 'Unknown')}
        - Date: {transaction_details.get('Date', 'Unknown')}
        
        Risk Assessment:
        - Risk Level: {risk_data.get('risk_level', 'Unknown')}
        - Case ID: {risk_data.get('case_id', 'N/A')}
        - Risk Indicators: {json.dumps(risk_data.get('indicators', []), indent=2)}
        
        This transaction has been automatically frozen and requires review.
        Access the compliance dashboard to review this case.
        
        --
        Automated AML System
        """
        
        msg.set_content(body)
        
        # Send email
        with smtplib.SMTP(EMAIL_SERVER, EMAIL_PORT) as server:
            server.starttls()
            server.login(EMAIL_USER, EMAIL_PASSWORD)
            server.send_message(msg)
        
        print(f"Compliance alert sent for transaction {txid}")
        return True
    except Exception as e:
        print(f"Error sending compliance alert: {e}")
        return False

# Add this to your transaction freezing function
def freeze_transaction_if_risky(txid, currency, risk_data):
    # Existing freezing code...
    
    if freeze_response['FunctionData']['Status'] == 'Success':
        # Get detailed transaction info for the alert
        tx_details_response = api.getTransactionById(txid, currency)
        if 'FunctionData' in tx_details_response and 'Transactions' in tx_details_response['FunctionData']:
            transaction_details = tx_details_response['FunctionData']['Transactions'][0]
            
            # Send alert to compliance team
            send_compliance_alert(txid, currency, risk_data, transaction_details)

Implement a process for compliance officers to review and potentially unfreeze transactions manually.

from ApiClient import ApiClient

def unfreeze_transaction_after_review(txid, currency, reason, approver_name):
    """Unfreeze a transaction after compliance review determines it's legitimate"""
    api = ApiClient()
    
    # Construct detailed note for audit trail
    note = f"OVERRIDE: Transaction unfrozen after compliance review. Reason: {reason}. Approved by: {approver_name}."
    
    # Call API to unfreeze
    response = api.unfreezeTransaction(
        txid=txid,
        currency=currency,
        note=note
    )
    
    if response['FunctionData']['Status'] == 'Success':
        print(f"Successfully unfrozen transaction {txid}")
        
        # Verify the transaction is no longer frozen
        verify_response = api.exportFrozenTransactions()
        frozen_data = verify_response['FunctionData']['Data']
        
        # Check if transaction is still in the frozen list
        if currency in frozen_data and txid in frozen_data[currency]:
            print(f"Warning: Transaction {txid} still appears in frozen list despite successful unfreeze")
            return False
        else:
            print(f"Confirmed: Transaction {txid} is no longer frozen")
            return True
    else:
        print(f"Error unfreezing transaction: {response}")
        return False

# Example usage in your application:
# unfreeze_transaction_after_review(
#     "0x123...",
#     "ETH",
#     "Customer provided proof of source of funds",
#     "Jane Smith"
# )

Implementation Approach 3: Chainalysis-driven

This approach uses the native Chainalysis integration for fully automated screening of both inbound and outbound transactions.

Chainalysis Transaction Handling Process Flows

Inbound Transaction Flow with Chainalysis

Outbound Transaction Flow with Chainalysis

Data Sent to Chainalysis

To perform risk assessments, Custody Infrastructure securely transmits the following transaction details to Chainalysis:

Asset: The cryptocurrency being transacted (e.g., BTC, ETH)

Amount: The transaction value in the smallest unit of the asset

External Address: The sender's address (for inbound) or recipient's address (for outbound)

Blockchain Hash: The transaction ID (TXID)

All data is transmitted using HTTPS encryption for secure communication.

Chainalysis Limitations

Note that Chainalysis does not analyze certain cryptocurrencies, including:

re.al

Andromeda

AVAX-P

Step-by-Step Implementation Process

Upload a valid Chainalysis API token (via uploadChainalysisToken) to enable automated screening.

from ApiClient import ApiClient

api = ApiClient()
response = api.uploadChainalysisToken(
    chainalysis_token="your-chainalysis-api-token"
)

if response['FunctionData']['Status'] == 'Success':
    print("Successfully uploaded Chainalysis token")
else:
    print(f"Error uploading Chainalysis token: {response}")

Run healthCheckChainalysis after uploading the token to verify that the Chainalysis integration is active.

from ApiClient import ApiClient

api = ApiClient()
response = api.healthCheckChainalysis()

if 'FunctionData' in response and 'ChainalysisHealthCheck' in response['FunctionData']:
    health_check = response['FunctionData']['ChainalysisHealthCheck']
    status = health_check.get('Status')
    
    if status == 0:
        print("Chainalysis connection is healthy")
    else:
        message = health_check.get('Message', {})
        title = message.get('Title', 'Unknown error')
        detail = message.get('Message', 'No details available')
        print(f"Chainalysis error: {title} - {detail}")
        
        # If token is invalid, you need to upload a new one
        if status == 1:
            print("Invalid Chainalysis token. Please obtain a new token and re-upload.")
        # If there's a general error, you may need to contact Chainalysis support
        elif status == 2:
            print("General Chainalysis connection error. Please contact Chainalysis support.")

Chainalysis automatically screens transactions, freezing suspicious ones. Set up a process to monitor these automated actions.

from ApiClient import ApiClient

api = ApiClient()
response = api.exportFrozenTransactions()

if 'FunctionData' in response and 'Data' in response['FunctionData']:
    frozen_data = response['FunctionData']['Data']
    
    # Count total frozen transactions
    total_frozen = sum(len(txids) for txids in frozen_data.values())
    print(f"Total frozen transactions: {total_frozen}")
    
    # Display frozen transactions by currency
    for currency, txids in frozen_data.items():
        if txids:  # Only display currencies with frozen transactions
            print(f"\n{currency}: {len(txids)} frozen transactions")
            for i, txid in enumerate(txids, 1):
                # Optionally get transaction details
                tx_details = api.getTransactionById(txid, currency)
                tx = tx_details['FunctionData']['Transactions'][0] if 'FunctionData' in tx_details and 'Transactions' in tx_details['FunctionData'] else {}
                
                amount = tx.get('Amount', 'Unknown')
                source = tx.get('From', 'Unknown')
                print(f"  {i}. {txid} - Amount: {amount}, From: {source}")

Create a process to review transactions frozen by Chainalysis.

from ApiClient import ApiClient

def review_frozen_transaction(txid, currency):
    """Retrieve and display details of a frozen transaction for review"""
    api = ApiClient()
    
    # Get transaction details
    tx_response = api.getTransactionById(txid, currency)
    
    if 'FunctionData' in tx_response and 'Transactions' in tx_response['FunctionData']:
        tx = tx_response['FunctionData']['Transactions'][0]
        
        print("\n===== FROZEN TRANSACTION DETAILS =====")
        print(f"Transaction ID: {tx.get('TxId', 'Unknown')}")
        print(f"Currency: {currency}")
        print(f"Amount: {tx.get('Amount', 'Unknown')}")
        print(f"From: {tx.get('From', 'Unknown')}")
        print(f"To: {tx.get('To', 'Unknown')}")
        print(f"Date: {tx.get('Date', 'Unknown')}")
        print(f"Status: {tx.get('Status', 'Unknown')}")
        print("========================================\n")
        
        # The compliance officer would then investigate this transaction
        # and make a decision to unfreeze or maintain the freeze
        
        return tx
    else:
        print(f"Error: Could not retrieve transaction details for {txid}")
        return None

Implement a process to unfreeze transactions when Chainalysis produces false positives.

from ApiClient import ApiClient

def override_chainalysis_and_unfreeze(txid, currency, reason, approver_name, case_id):
    """
    Unfreeze a transaction after review determines it was a false positive
    """
    api = ApiClient()
    
    # Construct a detailed note for audit purposes
    note = f"OVERRIDE: Chainalysis false positive. {reason} Approved by: {approver_name}. Case ID: {case_id}."
    
    # Unfreeze the transaction
    response = api.unfreezeTransaction(
        txid=txid,
        currency=currency,
        note=note
    )
    
    if response['FunctionData']['Status'] == 'Success':
        print(f"Successfully unfrozen transaction {txid}")
        
        # Verify the transaction is no longer frozen
        verify_response = api.exportFrozenTransactions()
        frozen_data = verify_response['FunctionData']['Data']
        
        # Check if transaction is still in the frozen list
        if currency in frozen_data and txid in frozen_data.get(currency, []):
            print(f"Warning: Transaction {txid} still appears in frozen list despite successful unfreeze")
            return False
        else:
            print(f"Confirmed: Transaction {txid} is no longer frozen")
            return True
    else:
        print(f"Error unfreezing transaction: {response}")
        return False

Set up a scheduled task to regularly verify that the Chainalysis integration is working correctly.

from ApiClient import ApiClient
import schedule
import time
from datetime import datetime

def chainalysis_health_check():
    """Perform regular health check on Chainalysis integration"""
    api = ApiClient()
    response = api.healthCheckChainalysis()
    
    if 'FunctionData' in response and 'ChainalysisHealthCheck' in response['FunctionData']:
        health_check = response['FunctionData']['ChainalysisHealthCheck']
        status = health_check.get('Status')
        
        if status == 0:
            print(f"[{datetime.now()}] Chainalysis connection is healthy")
            return True
        else:
            message = health_check.get('Message', {})
            title = message.get('Title', 'Unknown error')
            detail = message.get('Message', 'No details available')
            error_msg = f"[{datetime.now()}] Chainalysis error: {title} - {detail}"
            print(error_msg)
            
            # Here you would add code to alert your team
            # send_alert_email("Chainalysis Integration Error", error_msg)
            return False
    else:
        print(f"[{datetime.now()}] Error: Unexpected response format from health check")
        return False

# Schedule the health check to run every hour
schedule.every(1).hour.do(chainalysis_health_check)

# Run immediately to verify current status
chainalysis_health_check()

# In production, this would run in a background process
# while True:
#     schedule.run_pending()
#     time.sleep(60)

If you need to disable the Chainalysis integration, remove the token.

from ApiClient import ApiClient

def disable_chainalysis_integration(reason, authorized_by):
    """Remove the Chainalysis API token to disable automated AML screening"""
    api = ApiClient()
    
    print(f"WARNING: You are about to disable the Chainalysis AML integration.")
    print(f"Reason: {reason}")
    print(f"Authorized by: {authorized_by}")
    print("\nThis will disable automatic AML screening of transactions.")
    print("Your organization will need an alternative AML process in place.")
    
    # Log the action before performing it
    print(f"[{datetime.now()}] Disabling Chainalysis integration. Reason: {reason}. Authorized by: {authorized_by}")
    
    # Remove the token
    response = api.removeChainalysisToken()
    
    if response['FunctionData']['Status'] == 'Success':
        print("Successfully removed Chainalysis token")
        
        # Verify removal by checking health
        health_response = api.healthCheckChainalysis()
        if 'FunctionData' in health_response and 'ChainalysisHealthCheck' in health_response['FunctionData']:
            health_check = health_response['FunctionData']['ChainalysisHealthCheck']
            status = health_check.get('Status')
            
            if status == 1:  # Invalid token status
                print("Confirmed: Chainalysis integration is disabled")
                return True
            else:
                print(f"Warning: Unexpected health check status after removal: {status}")
                return False
        else:
            print("Warning: Unable to verify Chainalysis removal through health check")
            return True
    else:
        print(f"Error removing Chainalysis token: {response}")
        return False

Monitoring and Reporting Frozen Transactions

Regular monitoring of frozen transactions is an essential part of your AML program. Follow these steps to implement effective monitoring.

Step-by-Step Monitoring Process

First, retrieve a complete list of all transactions currently frozen in the system.

from ApiClient import ApiClient

api = ApiClient()
response = api.exportFrozenTransactions()

if 'FunctionData' in response and 'Data' in response['FunctionData']:
    frozen_data = response['FunctionData']['Data']
    
    # Count total frozen transactions
    total_frozen = sum(len(txids) for txids in frozen_data.values())
    print(f"Total frozen transactions: {total_frozen}")
    
    # Display count by currency
    for currency, txids in frozen_data.items():
        if txids:  # Only display currencies with frozen transactions
            print(f"{currency}: {len(txids)} frozen transactions")
            for txid in txids:
                print(f"  - {txid}")

For each frozen transaction, retrieve and examine its details.

from ApiClient import ApiClient

api = ApiClient()
export_response = api.exportFrozenTransactions()

if 'FunctionData' in export_response and 'Data' in export_response['FunctionData']:
    frozen_data = export_response['FunctionData']['Data']
    
    # Process each currency
    for currency, txids in frozen_data.items():
        if not txids:  # Skip currencies with no frozen transactions
            continue
            
        print(f"\nReviewing {len(txids)} frozen {currency} transactions:")
        
        # Process each transaction in this currency
        for txid in txids:
            # Get transaction details
            tx_response = api.getTransactionById(txid, currency)
            
            if 'FunctionData' in tx_response and 'Transactions' in tx_response['FunctionData']:
                tx = tx_response['FunctionData']['Transactions'][0]
                
                print(f"\n  Transaction: {txid}")
                print(f"  Amount: {tx.get('Amount', 'Unknown')}")
                print(f"  From: {tx.get('From', 'Unknown')}")
                print(f"  To: {tx.get('To', 'Unknown')}")
                print(f"  Date: {tx.get('Date', 'Unknown')}")
                
                # At this point, you would review the transaction according to your compliance procedures
            else:
                print(f"\n  Transaction: {txid} - Error retrieving details")

For each frozen transaction, make and document a compliance decision.

from ApiClient import ApiClient
import uuid
from datetime import datetime

api = ApiClient()

def document_compliance_decision(txid, currency, decision, reason, officer_name):
    """
    Document a compliance decision for a frozen transaction
    """
    # Generate a case ID if not provided
    case_id = f"AML-{uuid.uuid4().hex[:8].upper()}"
    
    # Record your decision in your compliance system
    # This is a placeholder for your internal case management system
    print(f"\nCompliance Decision:")
    print(f"Case ID: {case_id}")
    print(f"Transaction: {currency} - {txid}")
    print(f"Decision: {decision}")
    print(f"Reason: {reason}")
    print(f"Officer: {officer_name}")
    print(f"Date: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
    
    # If decision is to unfreeze, call the API
    if decision.lower() == 'unfreeze':
        note = f"UNFREEZE: {reason} | Officer: {officer_name} | Case: {case_id}"
        
        response = api.unfreezeTransaction(
            txid=txid,
            currency=currency,
            note=note
        )
        
        if response['FunctionData']['Status'] == 'Success':
            print(f"Transaction successfully unfrozen")
            return True, case_id
        else:
            print(f"Error unfreezing transaction: {response}")
            return False, case_id
    
    # If decision is to maintain freeze, document without API call
    elif decision.lower() == 'maintain':
        # Document in your system that the freeze should be maintained
        print(f"Freeze maintained per compliance decision")
        return True, case_id
    
    return False, case_id

# Example usage:
# document_compliance_decision(
#     txid="0x05e89fcc12a446a9f123b590823b94e9c9d4781f",
#     currency="ETH",
#     decision="unfreeze",  # or "maintain"
#     reason="Source of funds verified as legitimate business operation. Customer provided invoices.",
#     officer_name="John Smith"
# )

Decision Process for Frozen Transaction Review

flowchart TD
    A[Export frozen transactions] --> B[Review each transaction]
    B --> C{Investigation outcome}
    C -->|Legitimate| D[Unfreeze transaction]
    C -->|Suspicious| E[Keep frozen]
    C -->|Needs more info| F[Request more information]
    F --> B
    D --> G[Document decision]
    E --> H[File regulatory report]
    H --> G

Generating AML Reports

Basic Frozen Assets Report

The exportFrozenTransactions endpoint provides structured data that can be used to generate compliance reports.

Sample response format:

{
  "FunctionData": {
    "Data": {
      "AVAX": [],
      "BTC": [],
      "DOT": ["0x46...057b", "0x11...0272"],
      "ETH": [],
      "POL": [],
      "STRK": ["0x05...89e9"],
      "USDT_POL": []
    }
  }
}

This data can easily be transformed into CSV or other report formats:

import csv
from collections import defaultdict

def convert_to_csv(data, output_file="frozen_funds.csv"):
    """
    Converts frozen transaction data into a CSV report
    """
    frozen_data = data.get('FunctionData', {}).get('Data', {})
    transactions = []
    
    # Extract transaction data
    for currency, txids in frozen_data.items():
        for txid in txids:
            transactions.append({
                "currency": currency,
                "txid": txid
            })
    
    # Write to CSV
    with open(output_file, mode="w", newline="") as file:
        writer = csv.DictWriter(file, fieldnames=["currency", "txid"])
        writer.writeheader()
        writer.writerows(transactions)
    
    print(f"CSV report '{output_file}' has been created successfully.")

Detailed Transaction Report

For more comprehensive reporting, run getTransactionById with the results of exportFrozenTransactions to return further details on each frozen transaction:

def generate_detailed_aml_report(frozen_data, output_file="detailed_aml_report.csv"):
    """
    Generate a detailed report of all AML-frozen transactions
    """
    api = ApiClient()
    report_data = []
    
    # Process each currency and transaction
    for currency, txids in frozen_data.items():
        for txid in txids:
            # Get full transaction details
            tx_response = api.getTransactionById(txid, currency)
            
            if 'FunctionData' in tx_response and 'Transactions' in tx_response['FunctionData']:
                tx = tx_response['FunctionData']['Transactions'][0]
                
                # Add to report data
                report_data.append({
                    "currency": currency,
                    "txid": txid,
                    "amount": tx.get('Amount', ''),
                    "from_address": tx.get('From', ''),
                    "to_address": tx.get('To', ''),
                    "date": tx.get('Date', ''),
                    "status": tx.get('Status', '')
                })
    
    # Write to CSV
    with open(output_file, mode="w", newline="") as file:
        writer = csv.DictWriter(file, fieldnames=[
            "currency", "txid", "amount", "from_address", 
            "to_address", "date", "status"
        ])
        writer.writeheader()
        writer.writerows(report_data)
    
    print(f"Detailed AML report '{output_file}' created with {len(report_data)} transactions.")
    return report_data

Endpoint Deep Dive

Transaction Monitoring Endpoints

getTransactionById

Retrieve transaction details using a transaction ID. Note: This endpoint only supports Transaction IDs of input transactions.

POST

getTransactionById(transactionId="0x05e...781f", coin_type="ETH")

from ApiClient import ApiClient

api = ApiClient()
response = api.getTransactionById(
    transactionId="0x05...781f", 
    coin_type="ETH"
)

AML Control Endpoints

freezeTransaction

Freezes a transaction due to AML concerns, preventing the associated assets from being used. This endpoint can only be used by users with Compliance Management permissions. When a transaction is frozen, the assets are blocked and cannot be spent or counted in the wallet balance. The system logs all freeze actions for audit purposes, recording which compliance officer performed the action and when it occurred.

POST

freezeTransaction(txid="0x05e...781f", currency="ETH", note="Suspicious pattern of deposits", external_id="CASE-12345")

from ApiClient import ApiClient

api = ApiClient()
response = api.freezeTransaction(
    txid="0x05...781f",
    currency="ETH",
    note="Suspicious pattern of deposits",
    external_id="CASE-12345"  # Optional
)

unfreezeTransaction

Unfreezes a transaction that was previously blocked due to AML measures, making the frozen assets available for use again. This endpoint can only be used by users with Compliance Management permissions. When a transaction is unfrozen, the assets become available in the wallet balance and can be spent normally. The system logs all unfreeze actions for audit purposes, recording which compliance officer performed the action and when it occurred.

POST

unfreezeTransaction(txid="0x05e...781f", currency="ETH", note="Transaction verified as legitimate after investigation")

from ApiClient import ApiClient

api = ApiClient()
response = api.unfreezeTransaction(
    txid="0x05e89fcc12a446a9f123b590823b94e9c9d4781f",
    currency="ETH",
    note="Transaction verified as legitimate after investigation"
)

exportFrozenTransactions

Retrieves a comprehensive list of all transactions that have been frozen in the system due to AML measures. This endpoint retrieves information about all frozen transactions, including both those manually frozen by compliance officers and those automatically blocked by Chainalysis integration. The data returned can be used for audit purposes, regulatory reporting, and compliance tracking. This endpoint can only be used by users with Compliance Management permissions.

POST

exportFrozenTransactions()

from ApiClient import ApiClient

api = ApiClient()
response = api.exportFrozenTransactions()

Chainalysis Integration Endpoints

uploadChainalysisToken

Uploads or updates the Chainalysis API token used for automated AML risk assessment. This endpoint configures the system to use Chainalysis for AML verification of both inbound and outbound transactions. When enabled, Chainalysis will automatically screen transaction addresses and assign risk levels (low, medium, high, or severe). Transactions with high or severe risk are automatically blocked. This endpoint can only be used by users with administrative privileges. The token must be periodically updated to maintain the Chainalysis integration.

POST

uploadChainalysisToken(chainalysis_token="your-chainalysis-api-token")

from ApiClient import ApiClient

api = ApiClient()
response = api.uploadChainalysisToken(
    chainalysis_token="your-chainalysis-api-token"
)

healthCheckChainalysis

Verifies the connection status and health of the Chainalysis AML integration. This endpoint tests the connection to the Chainalysis API service and validates that the current API token is valid and operational. Use this function to troubleshoot Chainalysis integration issues or to confirm that automated AML screening is functioning correctly. The check verifies that transaction risk assessments can be performed successfully.

POST

healthCheckChainalysis()

from ApiClient import ApiClient

api = ApiClient()
response = api.healthCheckChainalysis()

removeChainalysisToken

Removes the Chainalysis API token from the system, disabling the Chainalysis AML integration. This endpoint deactivates the automated AML screening provided by Chainalysis. After removing the token, the system will revert to using the internal AML process for transaction monitoring. This action affects how both inbound and outbound transactions are evaluated for compliance with AML regulations. This endpoint can only be used by users with administrative privileges.

POST

removeChainalysisToken()

from ApiClient import ApiClient

api = ApiClient()
response = api.removeChainalysisToken()

Webhook Endpoint (For SDK Integration)

registerWebhook

Register a webhook with specified filters. To prepare for registering a webhook, you need to note the URL of the server you want the Custody Infrastructure system to send webhook notifications to. To secure the webhook communication, make sure to use an https address for the webhook URL. For extra security, have a random value prepared as a key and add it to the URL as follows: https://<listener_url>/random_secret_token=sadkfnherwklfnvkwlrhnjfklewjflkewjnknwklonewklfnewkonlkewnkwe Webhooks function by a combination of filters and alert events. Filters - CoinSymbol: What currency are we checking for activity? - Machine: Are we checking the "Hot" (MPC) or "Cold" (ColdVault)? - AccountId: Which account are we checking for activity? - MinAmount: For Input transactions, what is the smallest transaction you want to receive notifications for? - AlertEvent: Which statuses should trigger an alert (see below for definition of options)? - Confirmations: After how many on-chain confirmations do you want to be alerted? You can set up to ten distinct confirmation alerts for a given webhook. - Threshold: What balance amount do we want to be notified of? This works with the "Above" and "Below" alert events to send a notification whenever the main wallet's balance crosses the threshold in the indicated direction. AlertEvents - No funds available: A transaction was requested, but could not be created due to inadequate funds. - Created: A new transaction request was created. - Failed-Limits: A transaction request was attempted, but could not be created because it exceeds the user's limits. - Failed-Technical: A transaction request was attempted, but could not be created due to a technical issue. - Sent: A transaction was sent to the blockchain for confirmation. - Timeout occurred: A transaction request was automatically canceled after waiting to long for approval. - Waiting to be processed: The transaction request is waiting for approval. - Canceled: A user canceled the transaction. - Waiting for external co-signer: The transaction has been approved, but the cosigners have not worked their cryptographic magic. - Failed-Policy: A transaction request was attempted, but could not be created because it violated account policy. - Below: In conjunction with the "threshold" filter, sends an alert whenever the main wallet balance falls below the threshold value. - Above: In conjunction with the "threshold" filter, sends an alert whenever the main wallet balance rises above the threshold value.

GET

registerWebhook(self, webhook_url: str, type: int, filter: dict = {}, transaction_format_version: TransactionFormatVersion = TransactionFormatVersion.V2)

from ApiClient import ApiClient
from Helpers.TransactionEnums.TransactionFormatVersion import TransactionFormatVersion

api = ApiClient()
response = api.registerWebhook(
    webhook_url="https://your-server.com/aml-webhook",
    type=0,  # 0 = Input transaction webhook
    filter={
        "CoinSymbol": "ETH",
        "Machine": "MPC",
        "AccountIds": ["1"],
        "MinAmount": "10000"
    },
    transaction_format_version=TransactionFormatVersion.V2
)

Compliance and Audit Considerations

Audit Trail

All AML actions create an audit trail, including:

Who performed the action

When it was performed

The reason for the action (note field)

The note field in both freezeTransaction and unfreezeTransaction endpoints is mandatory. It creates a permanent record of the compliance decision and its rationale.

Compliance Officer Requirements

Compliance officers must have a role with "Compliance management" permissions, enabling them to freeze and unfreeze transactions. Without these permissions, API calls to the AML endpoints will fail with permission errors.

Reporting Recommendations

Maintain regular AML reporting:

Generate periodic reports of all frozen transactions

Document all compliance decisions and their rationale

Keep records of all AML-related transactions for the period required by your jurisdiction

Use the API endpoints to build automated reporting flows

Troubleshooting

Common AML implementation issues and their solutions are outlined below. If you encounter persistent problems, contact Customer Success team with specific error logs and transaction IDs.

Common Issues

Issue	Possible Cause	Solution
Permission errors	API user lacks Compliance Management permissions	Grant proper permissions to the API user
Chainalysis health check fails	Invalid or expired token	Generate a new token and upload it
Transaction not found in frozen list	Transaction ID incorrect or not properly frozen	Verify transaction ID and retry freeze operation
Webhook not receiving notifications	Incorrect webhook URL or network issues	Verify webhook URL is accessible and properly configured

Best Practices

Regular Testing: Periodically test your AML implementation to ensure it's working as expected

Error Handling: Implement robust error handling for all API calls

Monitoring: Set up alerts for AML-related events, especially for high-value transactions

Documentation: Keep detailed records of all AML decisions and their rationale

Token Renewal: For Chainalysis integration, establish a process to regularly renew the API token

Troubleshooting Decision Flow

flowchart TD
    A[AML Problem Detected] --> B{Issue Type?}
    B -->|Permission Error| C[Check API User Permissions]
    B -->|Chainalysis Error| D[Run healthCheckChainalysis]
    B -->|Transaction Not Frozen| E[Verify Transaction ID]
    B -->|Webhook Issue| F[Check Webhook Configuration]
    
    C --> G{Has Compliance Permission?}
    G -->|No| H[Add Compliance Management Permission]
    G -->|Yes| I[Verify API Key & Authentication]
    
    D --> J{Health Check Status?}
    J -->|Status 0| K[System Healthy - Check Other Issues]
    J -->|Status 1| L[Upload New Chainalysis Token]
    J -->|Status 2| M[Contact Customer Success Support]
    
    E --> N[Run exportFrozenTransactions]
    N --> O{Transaction in List?}
    O -->|No| P[Re-run freezeTransaction Call]
    O -->|Yes but Wrong Currency| Q[Update Currency Parameter]
    
    F --> R[Test Webhook with Sample Data]
    R --> S{Webhook Received?}
    S -->|No| T[Check URL & Network]
    S -->|Yes| U[Examine Webhook Handler Code]

Conclusion

Implementing effective AML controls is crucial for regulatory compliance and risk management. This guide has outlined three approaches to implementing AML in your Custody Infrastructure environment:

Manual AML: Gives full control through direct API calls for freeze/unfreeze operations

SDK-Automated AML: Allows integration with third-party AML providers via webhooks

Chainalysis AML: Provides fully automated screening of both inbound and outbound transactions

Each approach has its advantages depending on your organization's needs, compliance requirements, and resource availability. By following the implementation procedures described here, you can establish robust AML controls that help protect your digital asset custody operations from financial crime risks.

Remember that AML is an ongoing operational process, not a one-time setup. Regular monitoring, review, and adaptation of your AML procedures will help ensure your organization maintains effective compliance with evolving regulatory requirements.