MCP tools that access financial systems (payment processors, banking APIs, expense trackers) need security guardrails that prevent unauthorized transactions, log every action, and enforce spending limits. This tutorial adds a security middleware layer to MCP financial tools that validates every operation before execution. The middleware adds audit logging, amount caps, rate limiting, and optional human approval for high-value operations.
Prerequisites
- Python 3.9+ installed
- Basic understanding of MCP tool servers
- A Scavio API key for search-based fraud checks
Walkthrough
Step 1: Build the security middleware
Create a middleware that wraps every MCP tool call with validation, logging, and rate limiting.
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')Step 2: Add fraud detection with search verification
Before processing payments to new recipients, verify the recipient exists and is not flagged for fraud using web search.
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"])}')Step 3: Wrap MCP tool calls with the security layer
Create a decorator that applies the financial guard to any MCP tool function. High-risk operations require additional verification.
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}')Python Example
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))JavaScript Example
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));Expected Output
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}