AI Agents Integration Guide

What You'll Need:

• API Bearer Token (stored in WALLET_BEARER_TOKEN_SECRET)
• Base URL: https://your-domain.com/api
• Wallet address to verify

Function Schema

{
  "name": "check_wallet_sanctions",
  "description": "Check if a cryptocurrency wallet address is sanctioned, flagged, or associated with illicit activity. Returns sanctions status, balance information, and risk assessment.",
  "parameters": {
    "type": "object",
    "properties": {
      "address": {
        "type": "string",
        "description": "The cryptocurrency wallet address to verify (Ethereum, Bitcoin, Solana, XRP, or Base)"
      },
      "protocol": {
        "type": "string",
        "enum": ["eth", "sol", "btc", "xrp", "base"],
        "description": "The blockchain protocol. Defaults to 'eth' if not specified."
      },
      "includeDeBank": {
        "type": "boolean",
        "description": "Include detailed DeFi portfolio analysis and transaction history. Default: true"
      }
    },
    "required": ["address"]
  }
}

Python Implementation

import openai
import requests
import json

# Configure OpenAI
openai.api_key = "your-openai-api-key"

# Your API configuration
API_BASE_URL = "https://your-domain.com/api"
API_TOKEN = "your-wallet-api-token"

def check_wallet_sanctions(address: str, protocol: str = "eth", includeDeBank: bool = True):
    """Function that OpenAI will call"""
    url = f"{API_BASE_URL}/wallet/check"
    headers = {
        "Authorization": f"Bearer {API_TOKEN}",
        "Content-Type": "application/json"
    }
    payload = {
        "address": address,
        "protocol": protocol,
        "includeDeBank": includeDeBank
    }
    
    response = requests.post(url, json=payload, headers=headers)
    response.raise_for_status()
    return response.json()

# Define the function for OpenAI
functions = [
    {
        "name": "check_wallet_sanctions",
        "description": "Check if a cryptocurrency wallet is sanctioned or flagged",
        "parameters": {
            "type": "object",
            "properties": {
                "address": {"type": "string", "description": "Wallet address"},
                "protocol": {"type": "string", "enum": ["eth", "sol", "btc", "xrp", "base"]},
                "includeDeBank": {"type": "boolean"}
            },
            "required": ["address"]
        }
    }
]

# Example conversation
messages = [
    {"role": "user", "content": "Is wallet 0x742d35Cc6634C0532925a3b844Bc9e7595f0bEb safe to transact with?"}
]

response = openai.ChatCompletion.create(
    model="gpt-4",
    messages=messages,
    functions=functions,
    function_call="auto"
)

# Handle function call
if response.choices[0].message.get("function_call"):
    function_name = response.choices[0].message["function_call"]["name"]
    function_args = json.loads(response.choices[0].message["function_call"]["arguments"])
    
    # Execute the function
    if function_name == "check_wallet_sanctions":
        result = check_wallet_sanctions(**function_args)
        
        # Send result back to OpenAI
        messages.append(response.choices[0].message)
        messages.append({
            "role": "function",
            "name": function_name,
            "content": json.dumps(result)
        })
        
        final_response = openai.ChatCompletion.create(
            model="gpt-4",
            messages=messages
        )
        
        print(final_response.choices[0].message["content"])

Tool Schema

{
  "name": "check_wallet_sanctions",
  "description": "Verify if a cryptocurrency wallet address is sanctioned, flagged for illicit activity, or associated with high-risk entities. Returns comprehensive sanctions screening results, balance information across multiple chains, and transaction risk analysis.",
  "input_schema": {
    "type": "object",
    "properties": {
      "address": {
        "type": "string",
        "description": "The cryptocurrency wallet address to verify. Supports Ethereum (0x...), Bitcoin (1... or bc1...), Solana (base58), XRP (r...), and Base addresses."
      },
      "protocol": {
        "type": "string",
        "enum": ["eth", "sol", "btc", "xrp", "base"],
        "description": "The blockchain protocol. Use 'eth' for Ethereum, 'sol' for Solana, 'btc' for Bitcoin, 'xrp' for Ripple, 'base' for Base chain. Defaults to 'eth'."
      },
      "includeDeBank": {
        "type": "boolean",
        "description": "Whether to include detailed DeFi portfolio analysis and transaction history. Set to false for faster responses. Default: true"
      }
    },
    "required": ["address"]
  }
}

Python Implementation

import anthropic
import requests
import json

# Configure Anthropic
client = anthropic.Anthropic(api_key="your-anthropic-api-key")

# Your API configuration
API_BASE_URL = "https://your-domain.com/api"
API_TOKEN = "your-wallet-api-token"

def check_wallet_sanctions(address: str, protocol: str = "eth", includeDeBank: bool = True):
    """Function that Claude will call"""
    url = f"{API_BASE_URL}/wallet/check"
    headers = {
        "Authorization": f"Bearer {API_TOKEN}",
        "Content-Type": "application/json"
    }
    payload = {
        "address": address,
        "protocol": protocol,
        "includeDeBank": includeDeBank
    }
    
    response = requests.post(url, json=payload, headers=headers)
    response.raise_for_status()
    return response.json()

# Define the tool
tools = [
    {
        "name": "check_wallet_sanctions",
        "description": "Verify if a cryptocurrency wallet is sanctioned or flagged",
        "input_schema": {
            "type": "object",
            "properties": {
                "address": {"type": "string", "description": "Wallet address"},
                "protocol": {"type": "string", "enum": ["eth", "sol", "btc", "xrp", "base"]},
                "includeDeBank": {"type": "boolean"}
            },
            "required": ["address"]
        }
    }
]

# Example conversation
message = client.messages.create(
    model="claude-3-5-sonnet-20241022",
    max_tokens=1024,
    tools=tools,
    messages=[
        {"role": "user", "content": "Check if wallet 0x742d35Cc6634C0532925a3b844Bc9e7595f0bEb is safe"}
    ]
)

# Handle tool use
if message.stop_reason == "tool_use":
    tool_use = next(block for block in message.content if block.type == "tool_use")
    tool_name = tool_use.name
    tool_input = tool_use.input
    
    # Execute the tool
    if tool_name == "check_wallet_sanctions":
        result = check_wallet_sanctions(**tool_input)
        
        # Send result back to Claude
        response = client.messages.create(
            model="claude-3-5-sonnet-20241022",
            max_tokens=1024,
            tools=tools,
            messages=[
                {"role": "user", "content": "Check if wallet 0x742d35Cc6634C0532925a3b844Bc9e7595f0bEb is safe"},
                {"role": "assistant", "content": message.content},
                {
                    "role": "user",
                    "content": [
                        {
                            "type": "tool_result",
                            "tool_use_id": tool_use.id,
                            "content": json.dumps(result)
                        }
                    ]
                }
            ]
        )
        
        print(response.content[0].text)

Python Implementation

from langchain.tools import BaseTool
from langchain.agents import initialize_agent, AgentType
from langchain.chat_models import ChatOpenAI
import requests
from typing import Optional
from pydantic import BaseModel, Field

class WalletCheckInput(BaseModel):
    address: str = Field(description="The cryptocurrency wallet address to verify")
    protocol: str = Field(default="eth", description="Blockchain protocol: eth, sol, btc, xrp, or base")
    includeDeBank: bool = Field(default=True, description="Include DeFi portfolio analysis")

class WalletSanctionsTool(BaseTool):
    name = "check_wallet_sanctions"
    description = """
    Check if a cryptocurrency wallet is sanctioned or flagged for illicit activity.
    Returns sanctions status, balance information, and risk assessment.
    Use this when you need to verify wallet safety or compliance.
    """
    args_schema = WalletCheckInput
    
    def _run(self, address: str, protocol: str = "eth", includeDeBank: bool = True) -> dict:
        """Execute the wallet check"""
        url = "https://your-domain.com/api/wallet/check"
        headers = {
            "Authorization": f"Bearer {API_TOKEN}",
            "Content-Type": "application/json"
        }
        payload = {
            "address": address,
            "protocol": protocol,
            "includeDeBank": includeDeBank
        }
        
        response = requests.post(url, json=payload, headers=headers)
        response.raise_for_status()
        return response.json()
    
    async def _arun(self, address: str, protocol: str = "eth", includeDeBank: bool = True) -> dict:
        """Async version"""
        raise NotImplementedError("Async not implemented")

# Initialize the agent
llm = ChatOpenAI(model="gpt-4", temperature=0)
tools = [WalletSanctionsTool()]

agent = initialize_agent(
    tools=tools,
    llm=llm,
    agent=AgentType.OPENAI_FUNCTIONS,
    verbose=True
)

# Use the agent
result = agent.run("Is wallet 0x742d35Cc6634C0532925a3b844Bc9e7595f0bEb safe to transact with?")
print(result)

JavaScript/TypeScript

async function checkWallet(address: string, protocol: string = 'eth') {
  const response = await fetch('https://your-domain.com/api/wallet/check', {
    method: 'POST',
    headers: {
      'Authorization': 'Bearer ' + process.env.WALLET_API_TOKEN,
      'Content-Type': 'application/json'
    },
    body: JSON.stringify({ address, protocol })
  });
  
  if (!response.ok) {
    throw new Error('API error: ' + response.status);
  }
  
  return await response.json();
}

// Usage
const result = await checkWallet('0x742d35Cc6634C0532925a3b844Bc9e7595f0bEb');
console.log('Sanctioned:', result.data.overallSanctioned);
console.log('Balance:', result.data.debank?.totalUsdValue);

Python

import requests
import os

def check_wallet(address: str, protocol: str = "eth", include_debank: bool = True) -> dict:
    """Check wallet sanctions and balances"""
    
    url = "https://your-domain.com/api/wallet/check"
    headers = {
        "Authorization": "Bearer " + os.getenv('WALLET_API_TOKEN'),
        "Content-Type": "application/json"
    }
    payload = {
        "address": address,
        "protocol": protocol,
        "includeDeBank": include_debank
    }
    
    response = requests.post(url, json=payload, headers=headers)
    response.raise_for_status()
    
    return response.json()

# Usage
result = check_wallet("0x742d35Cc6634C0532925a3b844Bc9e7595f0bEb")
print("Sanctioned:", result["data"]["overallSanctioned"])
if result["data"].get("debank"):
    balance = result["data"]["debank"]["totalUsdValue"]
    print("Balance: $" + str(balance))

cURL

curl -X POST https://your-domain.com/api/wallet/check \
  -H "Authorization: Bearer YOUR_API_TOKEN" \
  -H "Content-Type: application/json" \
  -d '{
    "address": "0x742d35Cc6634C0532925a3b844Bc9e7595f0bEb",
    "protocol": "eth",
    "includeDeBank": true
  }'

Error Response Format

{
  "success": false,
  "error": {
    "code": "INVALID_ADDRESS",
    "message": "Invalid wallet address format",
    "details": "Address must be a valid Ethereum address starting with 0x"
  }
}

Common Error Codes

Python Error Handling Example

def check_wallet_safe(address: str, protocol: str = "eth"):
    """Check wallet with comprehensive error handling"""
    try:
        response = requests.post(
            "https://your-domain.com/api/wallet/check",
            headers={
                "Authorization": "Bearer " + os.getenv('WALLET_API_TOKEN'),
                "Content-Type": "application/json"
            },
            json={"address": address, "protocol": protocol},
            timeout=30
        )
        
        if response.status_code == 401:
            return {"error": "Invalid API token"}
        elif response.status_code == 429:
            return {"error": "Rate limit exceeded, please try again later"}
        elif response.status_code >= 500:
            return {"error": "Service temporarily unavailable"}
        
        response.raise_for_status()
        return response.json()
        
    except requests.exceptions.Timeout:
        return {"error": "Request timeout, please try again"}
    except requests.exceptions.RequestException as e:
        return {"error": f"Network error: {str(e)}"}