How to Secure Financial MCP Agent Tools

import time, json, hashlib
from datetime import datetime
from collections import defaultdict

class FinancialGuard:
    def __init__(self, max_amount: float = 1000, daily_limit: float = 5000,
                 rate_limit: int = 10):
        self.max_amount = max_amount
        self.daily_limit = daily_limit
        self.rate_limit = rate_limit  # calls per minute
        self.audit_log = []
        self.daily_totals = defaultdict(float)
        self.call_times = []

    def validate(self, tool_name: str, params: dict) -> dict:
        """Validate a tool call before execution."""
        # Rate limit check
        now = time.time()
        self.call_times = [t for t in self.call_times if now - t < 60]
        if len(self.call_times) >= self.rate_limit:
            return {'allowed': False, 'reason': f'Rate limit: {self.rate_limit}/min exceeded'}
        self.call_times.append(now)
        # Amount check
        amount = params.get('amount', 0)
        if amount > self.max_amount:
            return {'allowed': False, 'reason': f'Amount ${amount} exceeds cap ${self.max_amount}'}
        # Daily limit check
        today = datetime.now().strftime('%Y-%m-%d')
        if self.daily_totals[today] + amount > self.daily_limit:
            return {'allowed': False,
                    'reason': f'Daily limit ${self.daily_limit} would be exceeded'}
        # Log the operation
        entry = {
            'timestamp': datetime.now().isoformat(),
            'tool': tool_name,
            'params_hash': hashlib.sha256(json.dumps(params, sort_keys=True).encode()).hexdigest()[:12],
            'amount': amount,
            'status': 'approved'
        }
        self.audit_log.append(entry)
        self.daily_totals[today] += amount
        return {'allowed': True, 'audit_id': entry['params_hash']}

guard = FinancialGuard(max_amount=500, daily_limit=2000)
print('Financial guard initialized')
print(f'Max per transaction: $500')
print(f'Daily limit: $2,000')

import requests, os

SCAVIO_KEY = os.environ['SCAVIO_API_KEY']

def verify_recipient(name: str, domain: str = '') -> dict:
    """Search for recipient to verify legitimacy."""
    query = f'{name} {domain} company reviews' if domain else f'{name} company legitimate'
    resp = requests.post('https://api.scavio.dev/api/v1/search',
        headers={'x-api-key': SCAVIO_KEY, 'Content-Type': 'application/json'},
        json={'query': query, 'country_code': 'us', 'num_results': 5})
    results = resp.json().get('organic_results', [])
    # Check for fraud signals
    fraud_signals = []
    trust_signals = []
    for r in results:
        text = (r.get('title', '') + ' ' + r.get('snippet', '')).lower()
        if any(w in text for w in ['scam', 'fraud', 'complaint', 'warning', 'fake']):
            fraud_signals.append(r['title'][:50])
        if any(w in text for w in ['bbb', 'verified', 'trusted', 'established', 'reviews']):
            trust_signals.append(r['title'][:50])
    risk_level = 'high' if fraud_signals else 'medium' if not trust_signals else 'low'
    return {
        'recipient': name,
        'risk_level': risk_level,
        'fraud_signals': fraud_signals,
        'trust_signals': trust_signals,
        'results_found': len(results)
    }

check = verify_recipient('Acme Corp', 'acmecorp.com')
print(f'Recipient: {check["recipient"]}')
print(f'Risk level: {check["risk_level"]}')
print(f'Trust signals: {len(check["trust_signals"])}')
print(f'Fraud signals: {len(check["fraud_signals"])}')

def secured_tool(guard: FinancialGuard, require_verification: bool = False):
    def decorator(func):
        def wrapper(**params):
            tool_name = func.__name__
            # Validate with guard
            check = guard.validate(tool_name, params)
            if not check['allowed']:
                print(f'BLOCKED: {tool_name} - {check["reason"]}')
                return {'error': check['reason'], 'blocked': True}
            # Optional recipient verification
            if require_verification and 'recipient' in params:
                verify = verify_recipient(params['recipient'],
                    params.get('recipient_domain', ''))
                if verify['risk_level'] == 'high':
                    print(f'BLOCKED: High fraud risk for {params["recipient"]}')
                    return {'error': 'Recipient flagged as high risk', 'blocked': True}
            # Execute the actual tool
            result = func(**params)
            print(f'EXECUTED: {tool_name} (audit: {check["audit_id"]})')
            return result
        return wrapper
    return decorator

@secured_tool(guard, require_verification=True)
def send_payment(recipient: str, amount: float, currency: str = 'USD', **kwargs) -> dict:
    # This would call your actual payment API
    return {'status': 'sent', 'recipient': recipient, 'amount': amount}

# Test: normal payment
result = send_payment(recipient='Acme Corp', amount=250)
print(f'Result: {result}')

# Test: over limit
result = send_payment(recipient='BigCorp', amount=5000)
print(f'Result: {result}')