Cuanto tiempo lleva este tutorial de cómo proteger la implementación del servidor de datos mcp?

La mayoria de los desarrolladores completan este tutorial en 15 a 30 minutos. Necesitaras una clave API de Scavio (el plan gratuito funciona) y un entorno de Python o JavaScript.

Que necesito antes de empezar?

Python 3.9+ instalado. solicita biblioteca instalada. Una clave API de Scavio de scavio.dev. Comprensión básica de los patrones del servidor MCP. Una clave API de Scavio te da 50 creditos gratuitos al registrarte.

Puedo ejecutar este tutorial con el plan gratuito?

Si. El plan gratuito incluye 50 creditos al registrarte, mas que suficiente para completar este tutorial y crear un prototipo funcional.

Con que frameworks funciona esto?

Scavio tiene un paquete nativo de LangChain (langchain-scavio), un servidor MCP y una API REST simple que funciona con cualquier cliente HTTP. Este tutorial usa the raw REST API, pero puedes adaptarlo al framework que prefieras.

Implementación segura del servidor de datos MCP (2026)

La implementación de un servidor MCP que expone herramientas de búsqueda y datos a agentes de IA requiere una seguridad adecuada. Sin validación de clave API, limitación de velocidad y registro de auditoría, su servidor es vulnerable a abusos y sobrecostos. Este tutorial agrega tres capas de seguridad a un servidor de datos MCP: autenticación de clave API con soporte de rotación, limitación de tasa por clave para evitar picos de costos y registro de auditoría estructurado para cumplimiento. Los patrones se aplican a cualquier servidor MCP que incluya la API de Scavio o puntos finales pagos similares.

Requisitos previos

Python 3.9+ instalado
solicita biblioteca instalada
Una clave API de Scavio de scavio.dev
Comprensión básica de los patrones del servidor MCP

Guia paso a paso

Paso 1: Agregar middleware de validación de clave API

Cree una capa de validación de claves que verifique las solicitudes MCP entrantes con una lista de claves autorizadas. Admite múltiples claves para diferentes clientes.

Python

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

SCAVIO_KEY = os.environ['SCAVIO_API_KEY']
H = {'x-api-key': SCAVIO_KEY, 'Content-Type': 'application/json'}

# Key management
API_KEYS = {
    'key_prod_abc123': {'name': 'production', 'rate_limit': 100, 'active': True},
    'key_dev_xyz789': {'name': 'development', 'rate_limit': 20, 'active': True},
    'key_old_retired': {'name': 'deprecated', 'rate_limit': 0, 'active': False},
}

def validate_key(api_key: str) -> dict:
    key_config = API_KEYS.get(api_key)
    if not key_config:
        return {'valid': False, 'error': 'Unknown API key'}
    if not key_config['active']:
        return {'valid': False, 'error': 'API key deactivated'}
    return {'valid': True, 'config': key_config}

# Test
print(validate_key('key_prod_abc123'))  # Valid
print(validate_key('key_old_retired'))  # Deactivated
print(validate_key('unknown_key'))  # Unknown

Paso 2: Implementar limitación de tasa por clave

Agregue un limitador de velocidad de ventana deslizante que rastree las solicitudes por clave. Bloquear solicitudes que excedan el límite configurado.

Python

class RateLimiter:
    def __init__(self, window_seconds: int = 3600):
        self.window = window_seconds
        self.requests = defaultdict(list)

    def check(self, key: str, limit: int) -> dict:
        now = time.time()
        # Clean old entries
        self.requests[key] = [t for t in self.requests[key] if now - t < self.window]
        current = len(self.requests[key])
        if current >= limit:
            return {'allowed': False, 'current': current, 'limit': limit,
                    'retry_after': int(self.requests[key][0] + self.window - now)}
        self.requests[key].append(now)
        return {'allowed': True, 'current': current + 1, 'limit': limit, 'remaining': limit - current - 1}

rate_limiter = RateLimiter()

# Test rate limiting
for i in range(5):
    result = rate_limiter.check('key_dev_xyz789', limit=3)
    print(f'Request {i+1}: {"allowed" if result["allowed"] else "BLOCKED"} ({result["current"]}/{result["limit"]})')

Paso 3: Agregar registro de auditoría

Registre cada llamada a la herramienta MCP con marca de tiempo, identidad clave, nombre de la herramienta y costo. Esto proporciona una pista de auditoría para el cumplimiento y el seguimiento de costos.

Python

AUDIT_LOG = 'mcp_audit.jsonl'

def log_request(api_key: str, tool_name: str, args: dict, result_size: int, cost: float):
    key_config = API_KEYS.get(api_key, {})
    entry = {
        'timestamp': datetime.now().isoformat(),
        'key_name': key_config.get('name', 'unknown'),
        'key_hash': hashlib.sha256(api_key.encode()).hexdigest()[:12],
        'tool': tool_name,
        'args': {k: str(v)[:50] for k, v in args.items()},
        'result_bytes': result_size,
        'cost_usd': cost,
    }
    with open(AUDIT_LOG, 'a') as f:
        f.write(json.dumps(entry) + '\n')
    return entry

# Test
entry = log_request('key_prod_abc123', 'web_search', {'query': 'test'}, 1200, 0.005)
print(f'Logged: {entry["key_name"]} called {entry["tool"]} (${entry["cost_usd"]})')

Paso 4: Conecte la seguridad al controlador de solicitudes MCP

Combine validación de claves, limitación de velocidad y registro de auditoría en un único controlador de solicitudes que englobe todas las llamadas a la herramienta MCP.

Python

def secure_mcp_handler(api_key: str, tool_name: str, args: dict) -> dict:
    # Step 1: Validate key
    key_check = validate_key(api_key)
    if not key_check['valid']:
        return {'error': key_check['error'], 'status': 401}
    config = key_check['config']
    # Step 2: Rate limit
    rate_check = rate_limiter.check(api_key, config['rate_limit'])
    if not rate_check['allowed']:
        return {'error': 'Rate limit exceeded', 'retry_after': rate_check.get('retry_after'), 'status': 429}
    # Step 3: Execute the tool
    try:
        if tool_name == 'web_search':
            resp = requests.post('https://api.scavio.dev/api/v1/search', headers=H,
                json={'query': args.get('query', ''), 'country_code': 'us', 'num_results': 5})
            result = resp.json()
            cost = 0.005
        else:
            return {'error': f'Unknown tool: {tool_name}', 'status': 400}
    except Exception as e:
        return {'error': str(e), 'status': 500}
    # Step 4: Audit log
    log_request(api_key, tool_name, args, len(json.dumps(result)), cost)
    return {'result': result, 'status': 200, 'remaining_requests': rate_check['remaining']}

# Test the full flow
result = secure_mcp_handler('key_prod_abc123', 'web_search', {'query': 'test search'})
print(f'Status: {result["status"]}, Remaining: {result.get("remaining_requests")}')

Ejemplo en Python

Python

import os, requests, time, json, hashlib
from collections import defaultdict

SCAVIO_KEY = os.environ['SCAVIO_API_KEY']
H = {'x-api-key': SCAVIO_KEY, 'Content-Type': 'application/json'}

KEYS = {'key_prod': {'limit': 100, 'active': True}, 'key_dev': {'limit': 10, 'active': True}}
requests_log = defaultdict(list)

def secure_search(api_key, query):
    if api_key not in KEYS or not KEYS[api_key]['active']:
        return {'error': 'Invalid key'}
    now = time.time()
    requests_log[api_key] = [t for t in requests_log[api_key] if now - t < 3600]
    if len(requests_log[api_key]) >= KEYS[api_key]['limit']:
        return {'error': 'Rate limited'}
    requests_log[api_key].append(now)
    resp = requests.post('https://api.scavio.dev/api/v1/search', headers=H,
        json={'query': query, 'country_code': 'us', 'num_results': 5})
    print(f'[{api_key}] {query}: {len(resp.json().get("organic_results", []))} results')
    return resp.json()

secure_search('key_prod', 'test query')
secure_search('invalid', 'test')  # Rejected

Ejemplo en JavaScript

JavaScript

const SCAVIO_KEY = process.env.SCAVIO_API_KEY;

const KEYS = { key_prod: { limit: 100, active: true }, key_dev: { limit: 10, active: true } };
const requestLog = {};

async function secureSearch(apiKey, query) {
  if (!KEYS[apiKey]?.active) return { error: 'Invalid key' };
  const now = Date.now();
  requestLog[apiKey] = (requestLog[apiKey] || []).filter(t => now - t < 3600000);
  if (requestLog[apiKey].length >= KEYS[apiKey].limit) return { error: 'Rate limited' };
  requestLog[apiKey].push(now);
  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, country_code: 'us', num_results: 5 })
  });
  const data = await resp.json();
  console.log(`[${apiKey}] ${query}: ${(data.organic_results || []).length} results`);
  return data;
}

secureSearch('key_prod', 'test').then(console.log);

Salida esperada

JSON

validate: {'valid': True, 'config': {'name': 'production', 'rate_limit': 100}}
validate: {'valid': False, 'error': 'API key deactivated'}
validate: {'valid': False, 'error': 'Unknown API key'}

Request 1: allowed (1/3)
Request 2: allowed (2/3)
Request 3: allowed (3/3)
Request 4: BLOCKED (3/3)

Logged: production called web_search ($0.005)
Status: 200, Remaining: 98

Tutoriales relacionados

Requisitos previos

Python 3.9+ instalado
solicita biblioteca instalada
Una clave API de Scavio de scavio.dev
Comprensión básica de los patrones del servidor MCP

Guia paso a paso

Paso 1: Agregar middleware de validación de clave API

Cree una capa de validación de claves que verifique las solicitudes MCP entrantes con una lista de claves autorizadas. Admite múltiples claves para diferentes clientes.

Python

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

SCAVIO_KEY = os.environ['SCAVIO_API_KEY']
H = {'x-api-key': SCAVIO_KEY, 'Content-Type': 'application/json'}

# Key management
API_KEYS = {
    'key_prod_abc123': {'name': 'production', 'rate_limit': 100, 'active': True},
    'key_dev_xyz789': {'name': 'development', 'rate_limit': 20, 'active': True},
    'key_old_retired': {'name': 'deprecated', 'rate_limit': 0, 'active': False},
}

def validate_key(api_key: str) -> dict:
    key_config = API_KEYS.get(api_key)
    if not key_config:
        return {'valid': False, 'error': 'Unknown API key'}
    if not key_config['active']:
        return {'valid': False, 'error': 'API key deactivated'}
    return {'valid': True, 'config': key_config}

# Test
print(validate_key('key_prod_abc123'))  # Valid
print(validate_key('key_old_retired'))  # Deactivated
print(validate_key('unknown_key'))  # Unknown

Paso 2: Implementar limitación de tasa por clave

Agregue un limitador de velocidad de ventana deslizante que rastree las solicitudes por clave. Bloquear solicitudes que excedan el límite configurado.

Python

class RateLimiter:
    def __init__(self, window_seconds: int = 3600):
        self.window = window_seconds
        self.requests = defaultdict(list)

    def check(self, key: str, limit: int) -> dict:
        now = time.time()
        # Clean old entries
        self.requests[key] = [t for t in self.requests[key] if now - t < self.window]
        current = len(self.requests[key])
        if current >= limit:
            return {'allowed': False, 'current': current, 'limit': limit,
                    'retry_after': int(self.requests[key][0] + self.window - now)}
        self.requests[key].append(now)
        return {'allowed': True, 'current': current + 1, 'limit': limit, 'remaining': limit - current - 1}

rate_limiter = RateLimiter()

# Test rate limiting
for i in range(5):
    result = rate_limiter.check('key_dev_xyz789', limit=3)
    print(f'Request {i+1}: {"allowed" if result["allowed"] else "BLOCKED"} ({result["current"]}/{result["limit"]})')

Paso 3: Agregar registro de auditoría

Python

AUDIT_LOG = 'mcp_audit.jsonl'

def log_request(api_key: str, tool_name: str, args: dict, result_size: int, cost: float):
    key_config = API_KEYS.get(api_key, {})
    entry = {
        'timestamp': datetime.now().isoformat(),
        'key_name': key_config.get('name', 'unknown'),
        'key_hash': hashlib.sha256(api_key.encode()).hexdigest()[:12],
        'tool': tool_name,
        'args': {k: str(v)[:50] for k, v in args.items()},
        'result_bytes': result_size,
        'cost_usd': cost,
    }
    with open(AUDIT_LOG, 'a') as f:
        f.write(json.dumps(entry) + '\n')
    return entry

# Test
entry = log_request('key_prod_abc123', 'web_search', {'query': 'test'}, 1200, 0.005)
print(f'Logged: {entry["key_name"]} called {entry["tool"]} (${entry["cost_usd"]})')

Paso 4: Conecte la seguridad al controlador de solicitudes MCP

Combine validación de claves, limitación de velocidad y registro de auditoría en un único controlador de solicitudes que englobe todas las llamadas a la herramienta MCP.

Python

def secure_mcp_handler(api_key: str, tool_name: str, args: dict) -> dict:
    # Step 1: Validate key
    key_check = validate_key(api_key)
    if not key_check['valid']:
        return {'error': key_check['error'], 'status': 401}
    config = key_check['config']
    # Step 2: Rate limit
    rate_check = rate_limiter.check(api_key, config['rate_limit'])
    if not rate_check['allowed']:
        return {'error': 'Rate limit exceeded', 'retry_after': rate_check.get('retry_after'), 'status': 429}
    # Step 3: Execute the tool
    try:
        if tool_name == 'web_search':
            resp = requests.post('https://api.scavio.dev/api/v1/search', headers=H,
                json={'query': args.get('query', ''), 'country_code': 'us', 'num_results': 5})
            result = resp.json()
            cost = 0.005
        else:
            return {'error': f'Unknown tool: {tool_name}', 'status': 400}
    except Exception as e:
        return {'error': str(e), 'status': 500}
    # Step 4: Audit log
    log_request(api_key, tool_name, args, len(json.dumps(result)), cost)
    return {'result': result, 'status': 200, 'remaining_requests': rate_check['remaining']}

# Test the full flow
result = secure_mcp_handler('key_prod_abc123', 'web_search', {'query': 'test search'})
print(f'Status: {result["status"]}, Remaining: {result.get("remaining_requests")}')

Ejemplo en Python

Python

import os, requests, time, json, hashlib
from collections import defaultdict

SCAVIO_KEY = os.environ['SCAVIO_API_KEY']
H = {'x-api-key': SCAVIO_KEY, 'Content-Type': 'application/json'}

KEYS = {'key_prod': {'limit': 100, 'active': True}, 'key_dev': {'limit': 10, 'active': True}}
requests_log = defaultdict(list)

def secure_search(api_key, query):
    if api_key not in KEYS or not KEYS[api_key]['active']:
        return {'error': 'Invalid key'}
    now = time.time()
    requests_log[api_key] = [t for t in requests_log[api_key] if now - t < 3600]
    if len(requests_log[api_key]) >= KEYS[api_key]['limit']:
        return {'error': 'Rate limited'}
    requests_log[api_key].append(now)
    resp = requests.post('https://api.scavio.dev/api/v1/search', headers=H,
        json={'query': query, 'country_code': 'us', 'num_results': 5})
    print(f'[{api_key}] {query}: {len(resp.json().get("organic_results", []))} results')
    return resp.json()

secure_search('key_prod', 'test query')
secure_search('invalid', 'test')  # Rejected

Ejemplo en JavaScript

JavaScript

const SCAVIO_KEY = process.env.SCAVIO_API_KEY;

const KEYS = { key_prod: { limit: 100, active: true }, key_dev: { limit: 10, active: true } };
const requestLog = {};

async function secureSearch(apiKey, query) {
  if (!KEYS[apiKey]?.active) return { error: 'Invalid key' };
  const now = Date.now();
  requestLog[apiKey] = (requestLog[apiKey] || []).filter(t => now - t < 3600000);
  if (requestLog[apiKey].length >= KEYS[apiKey].limit) return { error: 'Rate limited' };
  requestLog[apiKey].push(now);
  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, country_code: 'us', num_results: 5 })
  });
  const data = await resp.json();
  console.log(`[${apiKey}] ${query}: ${(data.organic_results || []).length} results`);
  return data;
}

secureSearch('key_prod', 'test').then(console.log);

Salida esperada

JSON

validate: {'valid': True, 'config': {'name': 'production', 'rate_limit': 100}}
validate: {'valid': False, 'error': 'API key deactivated'}
validate: {'valid': False, 'error': 'Unknown API key'}

Request 1: allowed (1/3)
Request 2: allowed (2/3)
Request 3: allowed (3/3)
Request 4: BLOCKED (3/3)

Logged: production called web_search ($0.005)
Status: 200, Remaining: 98

Cómo proteger la implementación del servidor de datos MCP

Requisitos previos

Guia paso a paso

Paso 1: Agregar middleware de validación de clave API

Paso 2: Implementar limitación de tasa por clave

Paso 3: Agregar registro de auditoría

Paso 4: Conecte la seguridad al controlador de solicitudes MCP

Ejemplo en Python

Ejemplo en JavaScript

Salida esperada

Tutoriales relacionados

Preguntas frecuentes

Cuanto tiempo lleva este tutorial de cómo proteger la implementación del servidor de datos mcp?

Que necesito antes de empezar?

Puedo ejecutar este tutorial con el plan gratuito?

Con que frameworks funciona esto?

Recursos relacionados

MCP Search Integration vs Direct API Integration

Las mejores API de búsqueda después de los cambios en el modo AI de Google I/O 2026

Panorama de proveedores de API de búsqueda (2026)

La mejor API de búsqueda para servidores MCP en 2026

Servidor de búsqueda personalizado MCP

Integración de API personalizada de MCP para operaciones comerciales

Empieza a construir

Cómo proteger la implementación del servidor de datos MCP

Requisitos previos

Guia paso a paso

Paso 1: Agregar middleware de validación de clave API

Paso 2: Implementar limitación de tasa por clave

Paso 3: Agregar registro de auditoría

Paso 4: Conecte la seguridad al controlador de solicitudes MCP

Ejemplo en Python

Ejemplo en JavaScript

Salida esperada

Tutoriales relacionados

Preguntas frecuentes

Cuanto tiempo lleva este tutorial de cómo proteger la implementación del servidor de datos mcp?

Que necesito antes de empezar?

Puedo ejecutar este tutorial con el plan gratuito?

Con que frameworks funciona esto?

Recursos relacionados

MCP Search Integration vs Direct API Integration

Las mejores API de búsqueda después de los cambios en el modo AI de Google I/O 2026

Panorama de proveedores de API de búsqueda (2026)

La mejor API de búsqueda para servidores MCP en 2026

Servidor de búsqueda personalizado MCP

Integración de API personalizada de MCP para operaciones comerciales

Empieza a construir