Security Boundaries in MCP Integrations

An MCP integration connects an AI-controlled decision process to tools, resources, and prompts exposed by servers. That connection is a security boundary, not merely an API convenience. Model-generated tool arguments must be treated as untrusted input because a model can misunderstand a request, follow malicious content, or select a tool with unintended consequences. Data returned by an MCP server is also untrusted: it may contain sensitive information, malformed structures, or instructions designed to influence later model behavior.

A useful threat model separates the user, host application, model, MCP client, MCP server, and downstream system. Authentication establishes an identity, while authorization determines what that identity may do. The model should never inherit every permission held by the application process. For example, a support assistant may read a customer record but must not receive unrestricted database access or permission to issue refunds without an additional policy check.

Trust should be assigned to individual operations rather than to an entire server. A server may expose a harmless product-search tool alongside a destructive account-deletion tool. Register only the capabilities needed for the application, classify them by effect, and keep high-impact tools unavailable unless the current workflow requires them. Human approval is appropriate when an action changes money, access, legal status, production data, or an external communication.

Validation and Least-Privilege Permissions

Validate tool arguments immediately before execution, even if the MCP tool definition already describes an input schema. Descriptions help a model form a request, but they do not enforce application policy. Validation should reject missing fields, unexpected fields, incorrect types, excessive lengths, invalid identifiers, and values outside permitted ranges. Business constraints matter as much as structural constraints: a syntactically valid refund amount can still exceed the current user's approval limit.

The following runnable Python example implements a small security boundary around a hypothetical MCP tool adapter. It uses only the standard library. The adapter function stands in for the call that a real MCP client would make through its SDK or transport. The gateway validates arguments, checks the caller's permissions, and requires approval for a state-changing operation.

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)

The input contains a principal, tool name, argument object, and approval decision. The output is a validated dictionary returned by the adapter. Changing approved to false causes the refund request to fail before execution. Passing an extra argument, a malformed identifier, or an amount above the policy limit also fails. In production, permissions should come from a trusted identity system, and approval must be represented by a server-verified artifact rather than a model-generated Boolean value.

Reliable Tool Execution

Every MCP operation needs a timeout. Without one, an unavailable server or downstream dependency can consume workers indefinitely and block the agent workflow. Use separate connection and operation deadlines when the client stack supports them, and keep an overall deadline for the user request. Cancellation should propagate from the application through the MCP client so abandoned work does not continue consuming resources.

Retries require knowledge of side effects. Reading a resource or performing a deterministic search can often be retried after a transient transport failure. Creating a payment, refund, ticket, or message must not be blindly retried because the first attempt may have succeeded even when its response was lost. For state-changing tools, use a stable idempotency key if the downstream system supports one. Otherwise, reconcile operation status before trying again.

Failures should be classified into validation, authorization, timeout, transport, server, rate-limit, and application errors. The user may receive a safe message such as an inability to complete the action, while internal telemetry records the actionable category. Do not expose stack traces, server addresses, access tokens, or raw downstream responses to the model. A circuit breaker can temporarily stop calls to a persistently failing server, but it should complement timeouts and monitoring rather than replace them.

Structured Logging and Tracing

Observability should answer which operation was attempted, who initiated it, whether policy allowed it, how long it took, and why it failed. Use structured events instead of free-form messages. Useful fields include a timestamp, trace identifier, request identifier, principal identifier, server name, tool name, outcome, error category, attempt count, and duration. Propagate the same trace context across the application, MCP client, server, and downstream service when those components support tracing.

Do not log complete tool arguments and results by default. They may contain customer records, credentials, document contents, or model prompts. Prefer an allowlist of safe metadata, such as record type, result count, or payload size. Redact secrets at the logging boundary and protect logs with access controls and retention limits. Hashing an identifier can aid correlation, but predictable identifiers may still be recoverable, so keyed hashing is safer when pseudonymization is required.

Metrics should include request volume, success rate, denied-call count, validation failures, latency percentiles, timeouts, retry counts, and server availability. Alerts should represent user impact rather than isolated noise. For example, alert when the failure ratio for a critical tool remains elevated over a meaningful window. Audit records for sensitive actions should be durable and should identify the policy decision and approval evidence without storing hidden model reasoning.

A Reliable and Observable Gateway

The next example adds timeouts, conservative retries, structured logging, and redaction to the earlier boundary. It retries only tools marked as non-mutating. Python threads cannot forcibly stop underlying work, so the adapter itself must also configure transport-level deadlines. The gateway timeout still prevents the calling workflow from waiting indefinitely and demonstrates where the control belongs.

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

Running the program produces one structured log event followed by the adapter result. The log records argument names but not values, reducing accidental data exposure. A production implementation should replace printing with the organization's telemetry exporter, attach an existing trace identifier, and preserve the original typed exception internally. The model-facing layer should receive a stable, sanitized error object that helps it decide whether to ask the user, choose another tool, or stop.

Testing and Operational Readiness

Security tests should attempt unknown tools, missing permissions, malformed arguments, oversized strings, unexpected fields, forged approval values, and malicious content returned by a server. Reliability tests should simulate slow responses, dropped connections, rate limits, partial failures, and ambiguous outcomes for mutating calls. Verify that retry logic never duplicates a side effect and that timeout handling does not misreport a possibly completed transaction as definitely failed.

Observability tests are equally important. Confirm that every request has a correlation identifier, denied calls generate audit events, latency measurements use a monotonic clock, and sensitive values never appear in logs. Generate controlled failures and verify that dashboards and alerts show them. An alert that has never been tested is only an assumption about the monitoring pipeline.

A production-ready MCP integration therefore places a controlled gateway between model intent and real capabilities. That gateway validates both directions, applies least-privilege policy, obtains approval for consequential actions, sets deadlines, retries only safe operations, sanitizes failures, and emits correlated telemetry. These controls make the system safer without relying on the model to enforce security rules, and they provide the evidence needed to diagnose failures when real applications behave differently from a development demonstration.