Los servidores MCP conectan a los agentes de IA con herramientas externas, pero una credencial filtrada expone todas las herramientas detrás de ese servidor. Este tutorial implementa la rotación de credenciales, tokens de acceso con alcance y almacenamiento cifrado en reposo para que su servidor MCP permanezca seguro incluso si se filtra una sola clave. Todos los ejemplos utilizan la API de búsqueda de Scavio como recurso protegido.
Requisitos previos
- Python 3.11+ con biblioteca de criptografía instalada
- Una clave API de Scavio de https://scavio.dev
- Comprensión básica del protocolo MCP
- Un administrador de variables de entorno (direnv o dotenv)
Guia paso a paso
Paso 1: Cifrar credenciales en reposo
Almacene claves API cifradas mediante cifrado simétrico Fernet. La clave de cifrado reside en una variable de entorno, nunca en código o archivos de configuración.
Python
from cryptography.fernet import Fernet
import os
import json
from pathlib import Path
CRED_FILE = Path("mcp_credentials.enc")
def get_cipher():
key = os.environ.get("MCP_ENCRYPTION_KEY")
if not key:
raise RuntimeError("MCP_ENCRYPTION_KEY not set")
return Fernet(key.encode())
def store_credential(name: str, value: str):
cipher = get_cipher()
creds = load_all_credentials()
creds[name] = cipher.encrypt(value.encode()).decode()
CRED_FILE.write_text(json.dumps(creds, indent=2))
def load_credential(name: str) -> str:
cipher = get_cipher()
creds = load_all_credentials()
encrypted = creds.get(name)
if not encrypted:
raise KeyError(f"Credential '{name}' not found")
return cipher.decrypt(encrypted.encode()).decode()
def load_all_credentials() -> dict:
if not CRED_FILE.exists():
return {}
return json.loads(CRED_FILE.read_text())Paso 2: Implementar tokens de acceso con alcance
Genere tokens de alcance y de corta duración que limiten qué herramientas MCP puede invocar un cliente. Cada token codifica los nombres de las herramientas permitidas y una marca de tiempo de vencimiento.
Python
import hashlib
import time
import secrets
ACTIVE_TOKENS: dict[str, dict] = {}
def create_scoped_token(
allowed_tools: list[str],
ttl_seconds: int = 3600
) -> str:
token = secrets.token_urlsafe(32)
ACTIVE_TOKENS[token] = {
"tools": set(allowed_tools),
"expires": time.time() + ttl_seconds,
"created": time.time()
}
return token
def validate_token(token: str, tool_name: str) -> bool:
entry = ACTIVE_TOKENS.get(token)
if not entry:
return False
if time.time() > entry["expires"]:
del ACTIVE_TOKENS[token]
return False
return tool_name in entry["tools"]
def revoke_token(token: str):
ACTIVE_TOKENS.pop(token, None)Paso 3: Agregar rotación automática de credenciales
Gire la clave API de Scavio según un cronograma. La función de rotación vuelve a cifrar la nueva clave, actualiza la configuración del servidor MCP y registra el evento de rotación sin tiempo de inactividad.
Python
import logging
from datetime import datetime
logger = logging.getLogger("mcp_security")
ROTATION_LOG = Path("rotation_log.json")
def rotate_credential(name: str, new_value: str):
old_exists = name in load_all_credentials()
store_credential(name, new_value)
log_entry = {
"credential": name,
"rotated_at": datetime.now().isoformat(),
"action": "rotated" if old_exists else "created"
}
log_data = []
if ROTATION_LOG.exists():
log_data = json.loads(ROTATION_LOG.read_text())
log_data.append(log_entry)
ROTATION_LOG.write_text(json.dumps(log_data, indent=2))
logger.info(f"Credential '{name}' {log_entry['action']}")
def check_rotation_due(name: str, max_age_days: int = 30) -> bool:
if not ROTATION_LOG.exists():
return True
log_data = json.loads(ROTATION_LOG.read_text())
entries = [e for e in log_data if e["credential"] == name]
if not entries:
return True
last = datetime.fromisoformat(entries[-1]["rotated_at"])
age = (datetime.now() - last).days
return age >= max_age_daysPaso 4: Conecte las credenciales al controlador del servidor MCP
Cargue la credencial cifrada en el momento de la solicitud, valide el token de alcance y llame a la API de Scavio. Las credenciales nunca se mantienen en la memoria por más tiempo que una sola solicitud.
Python
import httpx
async def handle_mcp_tool_call(token: str, tool_name: str, params: dict) -> dict:
# Step 1: Validate scoped token
if not validate_token(token, tool_name):
return {"error": "Unauthorized or expired token"}
# Step 2: Load credential just-in-time
try:
api_key = load_credential("scavio_api_key")
except KeyError:
return {"error": "Credential not configured"}
# Step 3: Execute the search
if tool_name == "web_search":
async with httpx.AsyncClient(timeout=15) as client:
resp = await client.post(
"https://api.scavio.dev/api/v1/search",
headers={"x-api-key": api_key},
json={"query": params.get("query", ""), "num_results": 5}
)
resp.raise_for_status()
return {"result": resp.json()}
return {"error": f"Unknown tool: {tool_name}"}Ejemplo en Python
Python
import os
import json
import secrets
import time
import asyncio
import httpx
from pathlib import Path
from cryptography.fernet import Fernet
# --- Setup ---
# Generate encryption key once: Fernet.generate_key().decode()
# export MCP_ENCRYPTION_KEY="your-fernet-key"
CRED_FILE = Path("mcp_credentials.enc")
ACTIVE_TOKENS: dict[str, dict] = {}
def get_cipher():
return Fernet(os.environ["MCP_ENCRYPTION_KEY"].encode())
def store_credential(name: str, value: str):
cipher = get_cipher()
creds = json.loads(CRED_FILE.read_text()) if CRED_FILE.exists() else {}
creds[name] = cipher.encrypt(value.encode()).decode()
CRED_FILE.write_text(json.dumps(creds, indent=2))
def load_credential(name: str) -> str:
cipher = get_cipher()
creds = json.loads(CRED_FILE.read_text())
return cipher.decrypt(creds[name].encode()).decode()
def create_scoped_token(tools: list[str], ttl: int = 3600) -> str:
token = secrets.token_urlsafe(32)
ACTIVE_TOKENS[token] = {"tools": set(tools), "expires": time.time() + ttl}
return token
def validate_token(token: str, tool: str) -> bool:
entry = ACTIVE_TOKENS.get(token)
if not entry or time.time() > entry["expires"]:
return False
return tool in entry["tools"]
async def secure_search(token: str, query: str) -> dict:
if not validate_token(token, "web_search"):
return {"error": "Unauthorized"}
api_key = load_credential("scavio_api_key")
async with httpx.AsyncClient(timeout=15) as client:
resp = await client.post(
"https://api.scavio.dev/api/v1/search",
headers={"x-api-key": api_key},
json={"query": query, "num_results": 5}
)
resp.raise_for_status()
return resp.json()
# Usage
store_credential("scavio_api_key", "your-api-key")
token = create_scoped_token(["web_search"], ttl=1800)
result = asyncio.run(secure_search(token, "MCP server security best practices 2026"))
print(f"Results: {len(result.get('results', []))}")Ejemplo en JavaScript
JavaScript
const crypto = require("crypto");
const ALGORITHM = "aes-256-gcm";
const activeTokens = new Map();
function encrypt(text, key) {
const iv = crypto.randomBytes(16);
const cipher = crypto.createCipheriv(ALGORITHM, Buffer.from(key, "hex"), iv);
let encrypted = cipher.update(text, "utf8", "hex");
encrypted += cipher.final("hex");
const tag = cipher.getAuthTag().toString("hex");
return iv.toString("hex") + ":" + tag + ":" + encrypted;
}
function decrypt(data, key) {
const [ivHex, tagHex, encrypted] = data.split(":");
const decipher = crypto.createDecipheriv(ALGORITHM, Buffer.from(key, "hex"), Buffer.from(ivHex, "hex"));
decipher.setAuthTag(Buffer.from(tagHex, "hex"));
let decrypted = decipher.update(encrypted, "hex", "utf8");
decrypted += decipher.final("utf8");
return decrypted;
}
function createScopedToken(tools, ttlSeconds = 3600) {
const token = crypto.randomBytes(32).toString("base64url");
activeTokens.set(token, { tools: new Set(tools), expires: Date.now() + ttlSeconds * 1000 });
return token;
}
function validateToken(token, tool) {
const entry = activeTokens.get(token);
if (!entry || Date.now() > entry.expires) return false;
return entry.tools.has(tool);
}
async function secureSearch(token, query) {
if (!validateToken(token, "web_search")) throw new Error("Unauthorized");
const encKey = process.env.MCP_ENCRYPTION_KEY;
const stored = JSON.parse(require("fs").readFileSync("mcp_credentials.enc", "utf8"));
const apiKey = decrypt(stored.scavio_api_key, encKey);
const resp = await fetch("https://api.scavio.dev/api/v1/search", {
method: "POST",
headers: { "x-api-key": apiKey, "Content-Type": "application/json" },
body: JSON.stringify({ query, num_results: 5 })
});
return resp.json();
}
const token = createScopedToken(["web_search"], 1800);
secureSearch(token, "MCP server security 2026").then(r => {
console.log("Results:", (r.results || []).length);
});Salida esperada
JSON
Results: 5