Límites de seguridad en las integraciones con MCP

Una integración con MCP conecta un proceso de toma de decisiones controlado por IA con herramientas, recursos y prompts expuestos por servidores. Esa conexión es un límite de seguridad, no una mera comodidad de una API. Los argumentos de las herramientas generados por el modelo deben tratarse como entradas no confiables, ya que un modelo puede interpretar incorrectamente una solicitud, seguir contenido malicioso o seleccionar una herramienta con consecuencias no deseadas. Los datos devueltos por un servidor MCP también son no confiables: pueden contener información sensible, estructuras malformadas o instrucciones diseñadas para influir en el comportamiento posterior del modelo.

Un modelo de amenazas útil separa al usuario, la aplicación host, el modelo, el cliente MCP, el servidor MCP y el sistema posterior. La autenticación establece una identidad, mientras que la autorización determina qué puede hacer esa identidad. El modelo nunca debería heredar todos los permisos del proceso de la aplicación. Por ejemplo, un asistente de soporte puede leer el registro de un cliente, pero no debe recibir acceso irrestricto a la base de datos ni permiso para emitir reembolsos sin una comprobación adicional de la política.

La confianza debe asignarse a operaciones individuales, no a un servidor completo. Un servidor puede exponer una herramienta inofensiva de búsqueda de productos junto con una herramienta destructiva de eliminación de cuentas. Registra únicamente las capacidades necesarias para la aplicación, clasifícalas según su efecto y mantén las herramientas de alto impacto no disponibles a menos que el flujo de trabajo actual las requiera. La aprobación humana es adecuada cuando una acción modifica dinero, acceso, estado legal, datos de producción o una comunicación externa.

Validación y permisos de mínimo privilegio

Valida los argumentos de las herramientas inmediatamente antes de su ejecución, incluso si la definición de la herramienta MCP ya describe un esquema de entrada. Las descripciones ayudan al modelo a formular una solicitud, pero no hacen cumplir la política de la aplicación. La validación debe rechazar campos ausentes, campos inesperados, tipos incorrectos, longitudes excesivas, identificadores no válidos y valores fuera de los rangos permitidos. Las restricciones de negocio son tan importantes como las restricciones estructurales: un importe de reembolso sintácticamente válido aún puede superar el límite de aprobación del usuario actual.

El siguiente ejemplo ejecutable en Python implementa un pequeño límite de seguridad alrededor de un adaptador de herramientas MCP hipotético. Utiliza únicamente la biblioteca estándar. La función del adaptador representa la llamada que un cliente MCP real realizaría mediante su SDK o transporte. El gateway valida los argumentos, comprueba los permisos de quien realiza la llamada y exige aprobación para una operación que modifica el estado.

from dataclasses import dataclass
from typing import Any, Callable

@dataclass(frozen=True)
class Principal:
    user_id: str
    permissions: frozenset[str]

TOOL_POLICY = {
    "customer.read": {"permission": "customer:read", "mutating": False},
    "refund.create": {"permission": "refund:create", "mutating": True},
}

def validate_arguments(tool: str, arguments: dict[str, Any]) -> dict[str, Any]:
    if tool == "customer.read":
        if set(arguments) != {"customer_id"}:
            raise ValueError("customer.read requires only customer_id")
        customer_id = arguments["customer_id"]
        if not isinstance(customer_id, str) or not customer_id.startswith("cus_"):
            raise ValueError("customer_id must start with cus_")
        return {"customer_id": customer_id}

    if tool == "refund.create":
        if set(arguments) != {"payment_id", "amount_cents"}:
            raise ValueError("refund.create requires payment_id and amount_cents")
        payment_id = arguments["payment_id"]
        amount = arguments["amount_cents"]
        if not isinstance(payment_id, str) or not payment_id.startswith("pay_"):
            raise ValueError("payment_id must start with pay_")
        if isinstance(amount, bool) or not isinstance(amount, int):
            raise ValueError("amount_cents must be an integer")
        if amount < 1 or amount > 10_000:
            raise ValueError("amount_cents is outside the allowed range")
        return {"payment_id": payment_id, "amount_cents": amount}

    raise ValueError("tool is not registered")

def secure_call(
    principal: Principal,
    tool: str,
    arguments: dict[str, Any],
    approved: bool,
    adapter: Callable[[str, dict[str, Any]], dict[str, Any]],
) -> dict[str, Any]:
    policy = TOOL_POLICY.get(tool)
    if policy is None:
        raise PermissionError("tool is not allowed")
    if policy["permission"] not in principal.permissions:
        raise PermissionError("principal lacks the required permission")
    if policy["mutating"] and not approved:
        raise PermissionError("explicit approval is required")

    validated = validate_arguments(tool, arguments)
    result = adapter(tool, validated)
    if not isinstance(result, dict):
        raise ValueError("tool returned an invalid result")
    return result

def demo_adapter(tool: str, arguments: dict[str, Any]) -> dict[str, Any]:
    return {"tool": tool, "accepted": True, "arguments": arguments}

if __name__ == "__main__":
    principal = Principal(
        user_id="user_42",
        permissions=frozenset({"customer:read", "refund:create"}),
    )
    output = secure_call(
        principal,
        "refund.create",
        {"payment_id": "pay_123", "amount_cents": 2500},
        approved=True,
        adapter=demo_adapter,
    )
    print(output)

La entrada contiene una entidad principal, el nombre de la herramienta, un objeto de argumentos y la decisión de aprobación. La salida es un diccionario validado devuelto por el adaptador. Cambiar approved a false hace que la solicitud de reembolso falle antes de ejecutarse. Pasar un argumento adicional, un identificador malformado o un importe superior al límite de la política también provoca un fallo. En producción, los permisos deben proceder de un sistema de identidad confiable, y la aprobación debe representarse mediante un artefacto verificado por el servidor, no mediante un valor Booleano generado por el modelo.

Ejecución fiable de herramientas

Toda operación de MCP necesita un tiempo de espera. Sin uno, un servidor no disponible o una dependencia posterior puede consumir trabajadores indefinidamente y bloquear el flujo de trabajo del agente. Utiliza plazos separados para la conexión y la operación cuando la pila del cliente los admita, y mantén un plazo general para la solicitud del usuario. La cancelación debe propagarse desde la aplicación a través del cliente MCP para que el trabajo abandonado no siga consumiendo recursos.

Los reintentos requieren conocer los efectos secundarios. La lectura de un recurso o la ejecución de una búsqueda determinista normalmente se puede reintentar tras un fallo transitorio de transporte. La creación de un pago, reembolso, ticket o mensaje no debe reintentarse a ciegas, porque el primer intento puede haber tenido éxito aunque se haya perdido su respuesta. Para las herramientas que cambian el estado, utiliza una clave de idempotencia estable si el sistema posterior admite una. De lo contrario, reconcilia el estado de la operación antes de volver a intentarlo.

Los fallos deben clasificarse como errores de validación, autorización, tiempo de espera, transporte, servidor, límite de tasa y aplicación. El usuario puede recibir un mensaje seguro, como una notificación de que no se pudo completar la acción, mientras que la telemetría interna registra la categoría accionable. No expongas al modelo trazas de pila, direcciones de servidores, tokens de acceso ni respuestas sin procesar de sistemas posteriores. Un disyuntor puede detener temporalmente las llamadas a un servidor que falla de forma persistente, pero debe complementar los tiempos de espera y la monitorización, no sustituirlos.

Registro estructurado y trazabilidad

La observabilidad debe responder qué operación se intentó, quién la inició, si la política la permitió, cuánto tardó y por qué falló. Utiliza eventos estructurados en lugar de mensajes de formato libre. Entre los campos útiles se incluyen una marca de tiempo, un identificador de traza, un identificador de solicitud, un identificador de la entidad principal, el nombre del servidor, el nombre de la herramienta, el resultado, la categoría del error, el número de intentos y la duración. Propaga el mismo contexto de traza por la aplicación, el cliente MCP, el servidor y el servicio posterior cuando esos componentes admitan trazabilidad.

No registres de forma predeterminada los argumentos y resultados completos de las herramientas. Pueden contener registros de clientes, credenciales, contenido de documentos o prompts del modelo. Prefiere una lista permitida de metadatos seguros, como el tipo de registro, el número de resultados o el tamaño de la carga útil. Redacta los secretos en el límite de registro y protege los registros mediante controles de acceso y límites de retención. Aplicar un hash a un identificador puede facilitar la correlación, pero los identificadores predecibles aún pueden recuperarse; por eso, el hash con clave es más seguro cuando se requiere seudonimización.

Las métricas deben incluir el volumen de solicitudes, la tasa de éxito, el número de llamadas denegadas, los fallos de validación, los percentiles de latencia, los tiempos de espera agotados, el número de reintentos y la disponibilidad del servidor. Las alertas deben representar el impacto en el usuario, no ruido aislado. Por ejemplo, genera una alerta cuando la proporción de fallos de una herramienta crítica permanezca elevada durante un intervalo significativo. Los registros de auditoría de acciones sensibles deben ser duraderos e identificar la decisión de la política y las evidencias de aprobación, sin almacenar el razonamiento oculto del modelo.

Una puerta de enlace fiable y observable

El siguiente ejemplo añade tiempos de espera, reintentos conservadores, registro estructurado y redacción al límite anterior. Solo reintenta las herramientas marcadas como no mutantes. Los hilos de Python no pueden detener por la fuerza el trabajo subyacente, por lo que el propio adaptador también debe configurar plazos a nivel de transporte. El tiempo de espera de la puerta de enlace sigue evitando que el flujo de trabajo que realiza la llamada espere indefinidamente y demuestra dónde debe ubicarse el control.

import concurrent.futures
import json
import time
import uuid
from typing import Any, Callable

SAFE_TO_RETRY = {"customer.read"}

def emit_event(event: dict[str, Any]) -> None:
    print(json.dumps(event, separators=(",", ":"), sort_keys=True))

def call_with_controls(
    tool: str,
    arguments: dict[str, Any],
    adapter: Callable[[str, dict[str, Any]], dict[str, Any]],
    timeout_seconds: float = 1.0,
    max_attempts: int = 2,
) -> dict[str, Any]:
    request_id = str(uuid.uuid4())
    attempts = max_attempts if tool in SAFE_TO_RETRY else 1

    for attempt in range(1, attempts + 1):
        started = time.monotonic()
        try:
            with concurrent.futures.ThreadPoolExecutor(max_workers=1) as pool:
                future = pool.submit(adapter, tool, arguments)
                result = future.result(timeout=timeout_seconds)

            duration_ms = round((time.monotonic() - started) * 1000, 2)
            emit_event({
                "event": "mcp_tool_call",
                "request_id": request_id,
                "tool": tool,
                "attempt": attempt,
                "outcome": "success",
                "duration_ms": duration_ms,
                "argument_keys": sorted(arguments.keys()),
            })
            return result
        except concurrent.futures.TimeoutError:
            category = "timeout"
        except ConnectionError:
            category = "transport"
        except Exception:
            category = "internal"

        duration_ms = round((time.monotonic() - started) * 1000, 2)
        emit_event({
            "event": "mcp_tool_call",
            "request_id": request_id,
            "tool": tool,
            "attempt": attempt,
            "outcome": "failure",
            "error_category": category,
            "duration_ms": duration_ms,
            "argument_keys": sorted(arguments.keys()),
        })

        if attempt == attempts:
            raise RuntimeError(f"tool call failed: {category}")
        time.sleep(0.1 * attempt)

    raise RuntimeError("unreachable")

def stable_adapter(tool: str, arguments: dict[str, Any]) -> dict[str, Any]:
    return {"customer_id": arguments["customer_id"], "status": "active"}

if __name__ == "__main__":
    result = call_with_controls(
        "customer.read",
        {"customer_id": "cus_123"},
        stable_adapter,
    )
    print(json.dumps(result))

La ejecución del programa produce un evento de registro estructurado seguido del resultado del adaptador. El registro contiene los nombres de los argumentos, pero no sus valores, lo que reduce la exposición accidental de datos. Una implementación de producción debe sustituir la impresión por el exportador de telemetría de la organización, asociar un identificador de traza existente y conservar internamente la excepción tipada original. La capa orientada al modelo debe recibir un objeto de error estable y saneado que le ayude a decidir si debe preguntar al usuario, elegir otra herramienta o detenerse.

Pruebas y preparación operativa

Las pruebas de seguridad deben intentar usar herramientas desconocidas, permisos faltantes, argumentos con formato incorrecto, cadenas de texto demasiado grandes, campos inesperados, valores de aprobación falsificados y contenido malicioso devuelto por un servidor. Las pruebas de fiabilidad deben simular respuestas lentas, conexiones interrumpidas, límites de tasa, fallos parciales y resultados ambiguos en las llamadas que modifican datos. Verifica que la lógica de reintento nunca duplique un efecto secundario y que la gestión de tiempos de espera no informe erróneamente como definitivamente fallida una transacción que posiblemente se haya completado.

Las pruebas de observabilidad son igualmente importantes. Confirma que cada solicitud tenga un identificador de correlación, que las llamadas denegadas generen eventos de auditoría, que las mediciones de latencia utilicen un reloj monótono y que los valores sensibles nunca aparezcan en los registros. Genera fallos controlados y verifica que los paneles y las alertas los muestren. Una alerta que nunca se ha probado es solo una suposición sobre el pipeline de monitorización.

Por lo tanto, una integración de MCP lista para producción coloca una pasarela controlada entre la intención del modelo y las capacidades reales. Esa pasarela valida ambas direcciones, aplica una política de mínimo privilegio, obtiene aprobación para las acciones de consecuencias relevantes, establece plazos límite, reintenta únicamente operaciones seguras, sanea los fallos y emite telemetría correlacionada. Estos controles hacen que el sistema sea más seguro sin depender de que el modelo aplique las reglas de seguridad, y proporcionan las evidencias necesarias para diagnosticar fallos cuando las aplicaciones reales se comportan de forma distinta a una demostración de desarrollo.