Les outils MCP qui accèdent aux systèmes financiers (processeurs de paiement, API bancaires, suivis de dépenses) nécessitent des garde-fous de sécurité qui empêchent les transactions non autorisées, enregistrent chaque action et imposent des limites de dépenses. Ce tutoriel ajoute une couche middleware de sécurité aux outils financiers MCP qui valide chaque opération avant exécution. Le middleware ajoute la journalisation d'audit, des plafonds de montant, une limitation de débit et une approbation humaine facultative pour les opérations de grande valeur.
Prérequis
- Python 3.9+ installé
- Compréhension de base des serveurs d'outils MCP
- Une clé API Scavio pour les vérifications de fraude basées sur la recherche
Parcours
Étape 1: Construire le middleware de sécurité
Créez un middleware qui encapsule chaque appel d'outil MCP avec validation, journalisation et limitation de débit.
Python
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')Étape 2: Ajouter la détection de fraude avec vérification par recherche
Avant de traiter les paiements vers de nouveaux destinataires, vérifiez que le destinataire existe et n'est pas signalé pour fraude à l'aide d'une recherche web.
Python
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"])}')Étape 3: Encapsuler les appels d'outils MCP avec la couche de sécurité
Créez un décorateur qui applique la protection financière à toute fonction d'outil MCP. Les opérations à haut risque nécessitent une vérification supplémentaire.
Python
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}')Exemple Python
Python
import requests, os, time, json
from collections import defaultdict
from datetime import datetime
SCAVIO_KEY = os.environ['SCAVIO_API_KEY']
daily_total = defaultdict(float)
audit_log = []
def check_payment(recipient, amount, max_amount=500, daily_limit=2000):
today = datetime.now().strftime('%Y-%m-%d')
if amount > max_amount:
return {'blocked': True, 'reason': f'Over ${max_amount} cap'}
if daily_total[today] + amount > daily_limit:
return {'blocked': True, 'reason': 'Daily limit exceeded'}
# Verify recipient
resp = requests.post('https://api.scavio.dev/api/v1/search',
headers={'x-api-key': SCAVIO_KEY, 'Content-Type': 'application/json'},
json={'query': f'{recipient} scam fraud warning', 'country_code': 'us', 'num_results': 3})
results = resp.json().get('organic_results', [])
fraud = any('scam' in r.get('snippet', '').lower() or 'fraud' in r.get('snippet', '').lower() for r in results)
if fraud:
return {'blocked': True, 'reason': 'Fraud signals detected'}
daily_total[today] += amount
audit_log.append({'time': datetime.now().isoformat(), 'recipient': recipient, 'amount': amount})
return {'blocked': False, 'audit_id': len(audit_log)}
print(check_payment('Acme Corp', 250))
print(check_payment('BigCorp', 5000))Exemple JavaScript
JavaScript
const SCAVIO_KEY = process.env.SCAVIO_API_KEY;
const dailyTotal = {};
async function checkPayment(recipient, amount, maxAmount = 500) {
if (amount > maxAmount) return { blocked: true, reason: `Over $${maxAmount} cap` };
const today = new Date().toISOString().split('T')[0];
dailyTotal[today] = (dailyTotal[today] || 0);
if (dailyTotal[today] + amount > 2000) return { blocked: true, reason: 'Daily limit' };
const resp = await fetch('https://api.scavio.dev/api/v1/search', {
method: 'POST',
headers: { 'x-api-key': SCAVIO_KEY, 'Content-Type': 'application/json' },
body: JSON.stringify({ query: `${recipient} scam fraud`, country_code: 'us', num_results: 3 })
});
const results = (await resp.json()).organic_results || [];
const fraud = results.some(r => /scam|fraud/i.test(r.snippet || ''));
if (fraud) return { blocked: true, reason: 'Fraud signals' };
dailyTotal[today] += amount;
return { blocked: false };
}
checkPayment('Acme Corp', 250).then(r => console.log(r));Sortie attendue
JSON
Financial guard initialized
Max per transaction: $500
Daily limit: $2,000
Recipient: Acme Corp
Risk level: low
Trust signals: 2
Fraud signals: 0
EXECUTED: send_payment (audit: a3b7c9d12e4f)
Result: {'status': 'sent', 'recipient': 'Acme Corp', 'amount': 250}
BLOCKED: send_payment - Amount $5000 exceeds cap $500
Result: {'error': 'Amount $5000 exceeds cap $500', 'blocked': True}