Cómo crear una capa de búsqueda para agentes como servicio

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'))

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")}')

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')

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()