Ranh giới bảo mật trong các tích hợp MCP

Một tích hợp MCP kết nối quy trình ra quyết định do AI điều khiển với các tool, resource và prompt do các server cung cấp. Kết nối đó là một ranh giới bảo mật, không chỉ đơn thuần là một tiện ích API. Các đối số tool do model tạo ra phải được xem là dữ liệu đầu vào không đáng tin cậy, vì model có thể hiểu sai yêu cầu, làm theo nội dung độc hại hoặc chọn một tool gây ra hậu quả ngoài dự kiến. Dữ liệu do MCP server trả về cũng không đáng tin cậy: dữ liệu có thể chứa thông tin nhạy cảm, cấu trúc sai lệch hoặc các chỉ dẫn được thiết kế để tác động đến hành vi của model ở các bước sau.

Một mô hình mối đe dọa hữu ích sẽ phân tách người dùng, ứng dụng host, model, MCP client, MCP server và hệ thống downstream. Xác thực thiết lập danh tính, còn phân quyền quyết định danh tính đó được phép thực hiện những gì. Model không bao giờ được mặc nhiên thừa hưởng toàn bộ quyền hạn mà tiến trình ứng dụng đang có. Ví dụ, một trợ lý hỗ trợ khách hàng có thể đọc hồ sơ khách hàng nhưng không được cấp quyền truy cập cơ sở dữ liệu không giới hạn hoặc quyền hoàn tiền mà không có bước kiểm tra chính sách bổ sung.

Mức độ tin cậy nên được gán cho từng thao tác thay vì cho toàn bộ server. Một server có thể cung cấp đồng thời một tool tìm kiếm sản phẩm vô hại và một tool xóa tài khoản mang tính hủy hoại. Chỉ đăng ký các capability mà ứng dụng cần, phân loại chúng theo tác động và để các tool có ảnh hưởng lớn ở trạng thái không khả dụng trừ khi workflow hiện tại yêu cầu. Phê duyệt của con người là phù hợp khi một thao tác làm thay đổi tiền, quyền truy cập, trạng thái pháp lý, dữ liệu production hoặc một thông tin liên lạc gửi ra bên ngoài.

Xác thực và quyền hạn tối thiểu cần thiết

Hãy xác thực các đối số của tool ngay trước khi thực thi, ngay cả khi định nghĩa MCP tool đã mô tả schema đầu vào. Phần mô tả giúp model hình thành một request, nhưng không thực thi chính sách của ứng dụng. Việc xác thực nên từ chối các trường bị thiếu, trường không được phép, kiểu dữ liệu không đúng, độ dài vượt quá giới hạn, định danh không hợp lệ và các giá trị nằm ngoài phạm vi cho phép. Các ràng buộc nghiệp vụ cũng quan trọng không kém các ràng buộc cấu trúc: một khoản tiền hoàn hợp lệ về mặt cú pháp vẫn có thể vượt quá hạn mức phê duyệt của người dùng hiện tại.

Ví dụ Python có thể chạy sau đây triển khai một ranh giới bảo mật nhỏ bao quanh một MCP tool adapter giả định. Ví dụ chỉ sử dụng thư viện chuẩn. Hàm adapter đại diện cho lời gọi mà một MCP client thực tế sẽ thực hiện thông qua SDK hoặc transport của nó. Gateway xác thực các đối số, kiểm tra quyền hạn của caller và yêu cầu phê duyệt đối với một thao tác làm thay đổi trạng thái.

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)

Đầu vào gồm principal, tên tool, đối tượng đối số và quyết định phê duyệt. Đầu ra là một dictionary đã được xác thực do adapter trả về. Việc đổi approved thành false sẽ khiến request hoàn tiền thất bại trước khi được thực thi. Việc truyền thêm đối số, một định danh sai định dạng hoặc số tiền vượt quá giới hạn chính sách cũng sẽ khiến thao tác thất bại. Trong production, quyền hạn nên được lấy từ một hệ thống danh tính đáng tin cậy, và việc phê duyệt phải được biểu diễn bằng một artifact được server xác minh thay vì một giá trị Boolean do model tạo ra.

Thực thi công cụ đáng tin cậy

Mọi thao tác MCP đều cần có thời gian chờ. Nếu không có thời gian chờ, một máy chủ không khả dụng hoặc dependency ở downstream có thể chiếm dụng worker vô thời hạn và chặn quy trình làm việc của agent. Hãy sử dụng thời hạn riêng cho kết nối và thao tác khi stack client hỗ trợ, đồng thời duy trì một thời hạn tổng thể cho yêu cầu của người dùng. Việc hủy phải được truyền từ ứng dụng qua MCP client để các tác vụ đã bị bỏ không tiếp tục tiêu tốn tài nguyên.

Việc retry đòi hỏi phải hiểu rõ các side effect. Đọc một resource hoặc thực hiện một tìm kiếm xác định thường có thể được retry sau lỗi transport tạm thời. Không được retry một cách mù quáng các thao tác tạo khoản thanh toán, hoàn tiền, ticket hoặc tin nhắn, vì lần thử đầu tiên có thể đã thành công ngay cả khi phản hồi bị thất lạc. Đối với các tool làm thay đổi trạng thái, hãy sử dụng idempotency key ổn định nếu hệ thống downstream hỗ trợ. Nếu không, hãy đối soát trạng thái thao tác trước khi thử lại.

Các lỗi nên được phân loại thành lỗi validation, authorization, timeout, transport, server, rate limit và application. Người dùng có thể nhận được một thông báo an toàn, chẳng hạn như không thể hoàn tất thao tác, trong khi telemetry nội bộ ghi lại nhóm lỗi có thể xử lý. Không được cung cấp cho model stack trace, địa chỉ máy chủ, access token hoặc phản hồi thô từ downstream. Circuit breaker có thể tạm thời dừng các lệnh gọi đến một máy chủ liên tục gặp lỗi, nhưng nó nên bổ trợ cho timeout và monitoring thay vì thay thế chúng.

Logging và tracing có cấu trúc

Khả năng quan sát hệ thống phải trả lời được thao tác nào đã được thực hiện, ai khởi tạo, policy có cho phép hay không, thao tác mất bao lâu và vì sao thất bại. Hãy sử dụng các event có cấu trúc thay vì thông báo dạng tự do. Những trường hữu ích gồm timestamp, trace identifier, request identifier, principal identifier, tên máy chủ, tên tool, kết quả, nhóm lỗi, số lần thử và thời lượng. Hãy truyền cùng một trace context qua application, MCP client, server và dịch vụ downstream khi các thành phần đó hỗ trợ tracing.

Theo mặc định, không ghi log đầy đủ các tham số và kết quả của tool. Chúng có thể chứa hồ sơ khách hàng, thông tin xác thực, nội dung tài liệu hoặc prompt của model. Hãy ưu tiên allowlist gồm các metadata an toàn, chẳng hạn như loại bản ghi, số lượng kết quả hoặc kích thước payload. Hãy redacted secret tại ranh giới logging và bảo vệ log bằng các cơ chế kiểm soát quyền truy cập cùng giới hạn thời gian lưu giữ. Băm một identifier có thể hỗ trợ việc đối chiếu, nhưng các identifier dễ dự đoán vẫn có thể được khôi phục; vì vậy, keyed hashing an toàn hơn khi cần pseudonymization.

Metrics nên bao gồm lưu lượng request, tỷ lệ thành công, số lần gọi bị từ chối, lỗi validation, các percentile về độ trễ, timeout, số lần retry và khả năng sẵn sàng của server. Alert nên phản ánh tác động đến người dùng thay vì các tín hiệu nhiễu riêng lẻ. Ví dụ, hãy phát alert khi tỷ lệ lỗi của một tool quan trọng duy trì ở mức cao trong một khoảng thời gian có ý nghĩa. Audit record cho các thao tác nhạy cảm phải bền vững và cần xác định quyết định của policy cùng bằng chứng phê duyệt, nhưng không được lưu hidden reasoning của model.

Gateway đáng tin cậy và có khả năng quan sát

Ví dụ tiếp theo bổ sung timeout, retry thận trọng, logging có cấu trúc và redaction vào boundary ở phần trước. Ví dụ này chỉ retry những tool được đánh dấu là không làm thay đổi trạng thái. Thread trong Python không thể buộc dừng công việc bên dưới, vì vậy adapter cũng phải cấu hình deadline ở cấp transport. Timeout của gateway vẫn ngăn workflow gọi phải chờ vô thời hạn và minh họa nơi cơ chế kiểm soát này cần được đặt.

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

Khi chạy chương trình, kết quả là một log event có cấu trúc, sau đó là kết quả của adapter. Log ghi lại tên các tham số nhưng không ghi giá trị, nhờ đó giảm nguy cơ vô tình làm lộ dữ liệu. Một triển khai production nên thay thao tác in ra bằng telemetry exporter của tổ chức, gắn trace identifier hiện có và giữ lại typed exception ban đầu ở bên trong. Lớp hướng đến model nên nhận một error object ổn định, đã được làm sạch, giúp model quyết định nên hỏi người dùng, chọn tool khác hay dừng lại.

Kiểm thử và mức độ sẵn sàng vận hành

Các bài kiểm thử bảo mật nên thử gọi những tool không xác định, các quyền bị thiếu, tham số sai định dạng, chuỗi quá lớn, các trường không mong đợi, giá trị phê duyệt giả mạo và nội dung độc hại do máy chủ trả về. Các bài kiểm thử độ tin cậy nên mô phỏng phản hồi chậm, kết nối bị ngắt, giới hạn tốc độ, lỗi một phần và các kết quả không rõ ràng đối với những lệnh gọi làm thay đổi dữ liệu. Hãy xác minh rằng logic retry không bao giờ tạo ra tác dụng phụ trùng lặp và việc xử lý timeout không báo cáo sai một giao dịch có thể đã hoàn tất là chắc chắn thất bại.

Các bài kiểm thử khả năng quan sát cũng quan trọng không kém. Hãy xác nhận rằng mọi request đều có mã định danh tương quan, các lệnh gọi bị từ chối tạo ra sự kiện audit, phép đo độ trễ sử dụng đồng hồ đơn điệu và các giá trị nhạy cảm không bao giờ xuất hiện trong log. Hãy tạo ra các lỗi có kiểm soát và xác minh rằng dashboard cùng cảnh báo hiển thị chúng. Một cảnh báo chưa từng được kiểm thử chỉ là một giả định về pipeline giám sát.

Do đó, một tích hợp MCP sẵn sàng cho môi trường production sẽ đặt một gateway được kiểm soát giữa ý định của model và các năng lực thực tế. Gateway đó xác thực cả hai chiều, áp dụng chính sách đặc quyền tối thiểu, yêu cầu phê duyệt đối với các hành động có hậu quả đáng kể, đặt thời hạn, chỉ retry các thao tác an toàn, làm sạch thông tin lỗi và phát telemetry có mã tương quan. Những cơ chế kiểm soát này giúp hệ thống an toàn hơn mà không phụ thuộc vào việc model tự thực thi các quy tắc bảo mật, đồng thời cung cấp bằng chứng cần thiết để chẩn đoán lỗi khi các ứng dụng thực tế hoạt động khác với bản demo trong quá trình phát triển.