Les plateformes d'agents en tant que service ont besoin d'une couche de recherche partagée qui suit l'utilisation des crédits par locataire, applique des limites de débit et fonctionne avec MCP. Ce tutoriel construit une couche de service de recherche qui enveloppe l'API Scavio avec un suivi des crédits multi-locataire, une limitation de débit par locataire et des définitions d'outils compatibles MCP. Chaque recherche sous-jacente coûte 0,005 $, et vous pouvez appliquer une majoration pour vos locataires.
Prérequis
- Python 3.8+
- bibliothèque requests
- Une clé API Scavio provenant de scavio.dev
- Compréhension de base de l'architecture multi-locataire
Parcours
Étape 1: Définir le système de crédits du locataire
Créer un système de suivi des crédits pour l'utilisation de la recherche multi-locataire.
Python
import os, requests, json, time
from collections import defaultdict
API_KEY = os.environ['SCAVIO_API_KEY']
SH = {'x-api-key': API_KEY, 'Content-Type': 'application/json'}
class TenantCredits:
def __init__(self):
self.credits = {} # tenant_id -> remaining credits
self.usage = defaultdict(list) # tenant_id -> [{timestamp, query, cost}]
def add_tenant(self, tenant_id, credits):
self.credits[tenant_id] = credits
print(f'Tenant {tenant_id}: {credits} credits allocated')
def use_credit(self, tenant_id, amount=1):
if tenant_id not in self.credits:
return False, 'Tenant not found'
if self.credits[tenant_id] < amount:
return False, 'Insufficient credits'
self.credits[tenant_id] -= amount
self.usage[tenant_id].append({'time': time.time(), 'cost': amount})
return True, f'{self.credits[tenant_id]} credits remaining'
def get_usage(self, tenant_id):
return {'remaining': self.credits.get(tenant_id, 0),
'used': len(self.usage.get(tenant_id, [])),
'total_cost': sum(u['cost'] for u in self.usage.get(tenant_id, []))}
credits = TenantCredits()
credits.add_tenant('tenant_a', 1000)
credits.add_tenant('tenant_b', 500)
print(credits.get_usage('tenant_a'))Étape 2: Construire l'encapsuleur de recherche avec limitation de débit
Créer une fonction de recherche avec limitation de débit par locataire et vérification des crédits.
Python
class SearchLayer:
def __init__(self):
self.credits = TenantCredits()
self.rate_limits = {} # tenant_id -> {max_per_minute, window_start, count}
def set_rate_limit(self, tenant_id, max_per_minute):
self.rate_limits[tenant_id] = {'max': max_per_minute, 'start': time.time(), 'count': 0}
def check_rate_limit(self, tenant_id):
if tenant_id not in self.rate_limits:
return True
rl = self.rate_limits[tenant_id]
now = time.time()
if now - rl['start'] > 60:
rl['start'] = now
rl['count'] = 0
if rl['count'] >= rl['max']:
return False
rl['count'] += 1
return True
def search(self, tenant_id, query, platform=None):
if not self.check_rate_limit(tenant_id):
return {'error': 'Rate limit exceeded', 'retry_after': 60}
ok, msg = self.credits.use_credit(tenant_id)
if not ok:
return {'error': msg}
body = {'query': query, 'country_code': 'us'}
if platform: body['platform'] = platform
data = requests.post('https://api.scavio.dev/api/v1/search',
headers=SH, json=body).json()
return {'results': data.get('organic_results', [])[:5],
'credits_remaining': self.credits.credits.get(tenant_id, 0),
'tenant': tenant_id}
layer = SearchLayer()
layer.credits.add_tenant('demo', 100)
layer.set_rate_limit('demo', 30) # 30 searches/minute
result = layer.search('demo', 'best serp api 2026')
print(f'Results: {len(result.get("results", []))}, Credits: {result.get("credits_remaining")}')Étape 3: Ajouter des définitions d'outils compatibles MCP
Définir des outils de recherche au format MCP pour l'intégration des agents.
Python
def mcp_tool_definitions():
"""Return MCP-compatible tool definitions for the search layer."""
return [
{
'name': 'search',
'description': 'Search the web for current information. Supports platforms: google, reddit, youtube, amazon, walmart.',
'inputSchema': {
'type': 'object',
'properties': {
'query': {'type': 'string', 'description': 'Search query'},
'platform': {'type': 'string', 'enum': ['google', 'reddit', 'youtube', 'amazon', 'walmart'],
'description': 'Search platform (default: google)'}
},
'required': ['query']
}
},
{
'name': 'usage',
'description': 'Check remaining search credits and usage stats.',
'inputSchema': {'type': 'object', 'properties': {}}
}
]
def handle_mcp_call(tenant_id, tool_name, args):
"""Handle an MCP tool call from an agent."""
if tool_name == 'search':
return layer.search(tenant_id, args['query'], args.get('platform'))
elif tool_name == 'usage':
return layer.credits.get_usage(tenant_id)
return {'error': f'Unknown tool: {tool_name}'}
tools = mcp_tool_definitions()
print(f'MCP tools available: {[t["name"] for t in tools]}')
result = handle_mcp_call('demo', 'search', {'query': 'python tutorial', 'platform': 'reddit'})
print(f'MCP search result: {len(result.get("results", []))} results')Étape 4: Générer des rapports d'utilisation des locataires
Construire des rapports d'utilisation pour la facturation et la surveillance.
Python
def tenant_report(tenant_id):
usage = layer.credits.get_usage(tenant_id)
print(f'\n=== Tenant Report: {tenant_id} ===')
print(f' Credits remaining: {usage["remaining"]}')
print(f' Searches used: {usage["used"]}')
print(f' Total cost (internal): ${usage["total_cost"] * 0.005:.3f}')
# Calculate billing at markup
markup = 2.0 # 2x markup
billed = usage['total_cost'] * 0.005 * markup
print(f' Billed to tenant (2x markup): ${billed:.3f}')
print(f' Margin: ${billed - usage["total_cost"] * 0.005:.3f}')
def platform_report():
print(f'\n=== Platform Usage Report ===')
total_searches = 0
total_revenue = 0
for tenant_id in layer.credits.credits:
usage = layer.credits.get_usage(tenant_id)
total_searches += usage['used']
revenue = usage['total_cost'] * 0.005 * 2 # 2x markup
total_revenue += revenue
print(f' {tenant_id}: {usage["used"]} searches, ${revenue:.3f} billed')
internal_cost = total_searches * 0.005
print(f'\n Total searches: {total_searches}')
print(f' Internal cost: ${internal_cost:.3f}')
print(f' Total revenue: ${total_revenue:.3f}')
print(f' Profit: ${total_revenue - internal_cost:.3f}')
# Simulate some usage
for i in range(5):
layer.search('demo', f'test query {i}')
tenant_report('demo')
platform_report()Exemple Python
Python
import os, requests
SH = {'x-api-key': os.environ['SCAVIO_API_KEY'], 'Content-Type': 'application/json'}
credits = {'tenant_a': 100}
def tenant_search(tenant_id, query):
if credits.get(tenant_id, 0) <= 0:
return {'error': 'No credits'}
data = requests.post('https://api.scavio.dev/api/v1/search',
headers=SH, json={'query': query, 'country_code': 'us'}).json()
credits[tenant_id] -= 1
print(f'[{tenant_id}] "{query}": {len(data.get("organic_results", []))} results ({credits[tenant_id]} credits left)')
return data.get('organic_results', [])[:3]
tenant_search('tenant_a', 'serp api')Exemple JavaScript
JavaScript
const SH = { 'x-api-key': process.env.SCAVIO_API_KEY, 'Content-Type': 'application/json' };
const credits = { tenant_a: 100 };
async function tenantSearch(tenantId, query) {
if ((credits[tenantId] || 0) <= 0) return { error: 'No credits' };
const data = await fetch('https://api.scavio.dev/api/v1/search', {
method: 'POST', headers: SH,
body: JSON.stringify({ query, country_code: 'us' })
}).then(r => r.json());
credits[tenantId]--;
console.log(`[${tenantId}] "${query}": ${(data.organic_results||[]).length} results (${credits[tenantId]} left)`);
return (data.organic_results || []).slice(0, 3);
}
await tenantSearch('tenant_a', 'serp api');Sortie attendue
JSON
Tenant demo: 100 credits allocated
Results: 5, Credits: 99
MCP tools available: ['search', 'usage']
MCP search result: 5 results
=== Tenant Report: demo ===
Credits remaining: 93
Searches used: 7
Total cost (internal): $0.035
Billed to tenant (2x markup): $0.070
Margin: $0.035
=== Platform Usage Report ===
demo: 7 searches, $0.070 billed
Total searches: 7
Internal cost: $0.035
Profit: $0.035