Tự xây lớp adapter MCP bằng Flask: khai báo công cụ, định tuyến và gỡ lỗi

Khi kết nối môi trường trình duyệt với AI Agent, có một vấn đề không thể tránh: công cụ tiêu chuẩn chỉ cung cấp các thao tác nguyên thủy, còn hành động nghiệp vụ phải tự ghép lại.

Lấy một ví dụ cụ thể. Để hoàn thành một lượt thu thập, quy trình thực tế là “mở một môi trường theo khu vực, gắn đường ra mạng, truy cập một lần để làm nóng và xác nhận đường ra hoạt động”, rồi mới giao cho Agent xử lý. Một MCP Server có sẵn thường chỉ cung cấp các công cụ một bước như mở môi trường, điều hướng, nhấp và chụp ảnh màn hình. Không có ai ghép chuỗi logic đó thay bạn. Mỗi lần Agent đều phải sắp xếp lại từ đầu, vừa chậm vừa phải tự quyết định cách xử lý nếu bất kỳ bước trung gian nào thất bại.

Đó chính là vai trò của lớp adapter: giữ logic kết hợp, kiểm tra và trạng thái ở phía bạn, còn bên ngoài chỉ lộ ra một hành động nghiệp vụ.

Cấu trúc tối thiểu có thể chạy

Một lớp adapter chạy được chỉ cần ba phần: khai báo công cụ, định tuyến yêu cầu và thực thi kèm kết quả trả về. Với Flask, tất cả có thể nằm trong một tiến trình.

Trước hết là khai báo công cụ, vì phần này quyết định Agent có thể nhìn thấy gì.

# adapter/tools.py
TOOLS = [
    {
        "name": "prepare_environment",
        "description": "Chuẩn bị một môi trường khả dụng cho khu vực và trả về ID môi trường khi hoàn tất",
        "inputSchema": {
            "type": "object",
            "properties": {
                "region": {"type": "string", "description": "Khu vực đường ra, ví dụ US-CA"},
                "purpose": {"type": "string", "description": "Nhãn mục đích dùng cho tái sử dụng và thống kê hạn ngạch"},
                "timeout": {"type": "integer", "minimum": 10, "maximum": 120, "default": 60},
            },
            "required": ["region"],
            "additionalProperties": False,
        },
    },
    {
        "name": "open_page",
        "description": "Mở trang trong môi trường được chỉ định và chờ đến khi có thể tương tác",
        "inputSchema": {
            "type": "object",
            "properties": {
                "env_id": {"type": "string"},
                "url": {"type": "string"},
            },
            "required": ["env_id", "url"],
            "additionalProperties": False,
        },
    },
]

Router phân phối các phương thức JSON-RPC. Client sẽ hỏi tools/list trước, sau đó gọi tools/call theo tên.

# adapter/app.py
from flask import Flask, request, jsonify
from adapter.tools import TOOLS
from adapter.runner import run_tool

app = Flask(__name__)

@app.post("/mcp")
def mcp():
    req = request.get_json(force=True)
    method, rid = req.get("method"), req.get("id")

    if method == "tools/list":
        return jsonify({"jsonrpc": "2.0", "id": rid, "result": {"tools": TOOLS}})

    if method == "tools/call":
        name = req["params"]["name"]
        args = req["params"].get("arguments", {})
        return jsonify({"jsonrpc": "2.0", "id": rid, "result": run_tool(name, args)})

    return jsonify({"jsonrpc": "2.0", "id": rid,
                    "error": {"code": -32601, "message": "method not found"}})

Lớp thực thi là nơi duy nhất chạm vào API của môi trường và giao thức trang. Chuỗi nhiều bước được hoàn tất ở đây, và lỗi cũng được chuyển thành một định dạng thống nhất tại đây.

# adapter/runner.py
def run_tool(name, args):
    try:
        payload = HANDLERS[name](**args)
        return ok(payload)
    except ToolError as e:
        return fail(e.code, e.retryable, e.message)

Nên xác định sớm định dạng trả về. Đừng ném nguyên văn exception tầng thấp cho Agent; nó sẽ cố phân tích các chuỗi đó rồi có thể đưa ra quyết định kỳ lạ. Hãy quy ước một cấu trúc có mã kết quả, chẳng hạn {"ok": false, "code": "env_unavailable", "retryable": true, "attempts": 3}. Khi đó Agent chỉ cần xét hai việc: có thể thử lại không và có cần chuyển cho con người xử lý không.

Cách thiết kế tham số

Độ hạt của công cụ nên chia theo hành động nghiệp vụ, không phải theo API. Nếu bọc mỗi endpoint tầng thấp thành một công cụ riêng thì gần như chưa trừu tượng hóa gì; Agent vẫn phải tự sắp xếp thứ tự các bước.

Với tham số có một ranh giới rõ ràng: việc nào mô hình nên quyết định và việc nào lớp adapter tự quyết định. Khu vực, mục đích và URL đích thuộc nhóm đầu, để mô hình điền. Cổng debug, tên hàng đợi nội bộ và chọn pool môi trường thuộc nhóm sau; đừng đưa vào schema vì sớm muộn mô hình cũng sẽ điền sai.

Selector là chỗ rất dễ gặp lỗi. Khi cấu trúc trang thay đổi, các lệnh gọi gắn cứng selector có thể hỏng hàng loạt. Có thể để mô hình truyền mục tiêu theo ngữ nghĩa, ví dụ một dấu hiệu tương đối ổn định như “nút đăng nhập”, còn mapping selector được duy trì trong lớp adapter. Khi trang đổi, chỉ cần sửa một chỗ.

Tham số số phải có giới hạn trên. Nếu timeout, số lần retry hoặc số trang không có maximum trong schema, mô hình có thể truyền một giá trị rất lớn và biến một lệnh gọi thành tác vụ kéo dài hơn mười phút. Mọi công cụ có logic lặp đều cần điểm dừng rõ ràng. Dùng trường như max_pages để giới hạn công việc thay vì “tiếp tục lật trang cho đến khi hết dữ liệu”.

Cũng cần tính đến tính idempotent. Yêu cầu bên gọi truyền task_id; request lặp lại có thể trả thẳng kết quả trước đó, tránh việc Agent retry rồi mở môi trường lần thứ hai.

Giữ giá trị trả về nhỏ gọn. Không trả ảnh chụp màn hình dưới dạng base64 mà trả tham chiếu tệp. Với kết quả dạng danh sách, hãy kèm count và cờ cắt ngắn thay vì nhét toàn bộ bảng vào context.

Khi gỡ lỗi, hãy kiểm tra ba điểm này trước

Khi dùng stdio, JSON-RPC chiếm stdout nên tuyệt đối không ghi log vào stdout. Chỉ một dòng print tùy tiện cũng có thể làm parser giao thức hỏng ngay và khiến bạn mất nhiều thời gian dò lỗi. Hãy đưa log thống nhất sang stderr hoặc vào tệp.

tools/list là điểm kiểm tra đầu tiên. Nếu khai báo không được nạp thì toàn bộ các lệnh gọi sau sẽ không xảy ra. Trước hết hãy kiểm tra tên công cụ, cấu trúc schema và xem additionalProperties có đang chặn tham số nào không.

Điểm kiểm tra thứ hai là trace theo từng bước. Với mỗi lệnh gọi, ghi lại task_id, tóm tắt tham số, thời gian và mã kết quả. Khi có sự cố, bạn sẽ biết nó bị kẹt ở khâu tạo môi trường, điều hướng hay kiểm tra. Hãy lưu tham số gốc của các ca thất bại để replay nguyên trạng — bug tái hiện được thì sửa nhanh hơn nhiều.

Điểm kiểm tra thứ ba là một bộ fixtures cố định: một trang thử nghiệm có nội dung ổn định, kèm vài locator phần tử không đổi. Chạy smoke test sau mỗi lần sửa lớp adapter tiết kiệm hơn bất kỳ kiểu xác nhận bằng lời nào.

Ranh giới quyền hạn

Lớp adapter là nơi tập trung quyền hạn cao nhất trong toàn hệ thống: thông tin xác thực, môi trường và thao tác trang đều nằm trong tay nó. Vì vậy ranh giới chỉ có thể được đặt ở đây.

Đặt thông tin xác thực trong cấu hình phía server, không đưa vào tham số công cụ hay context của mô hình. Phân công cụ theo mức rủi ro: công cụ chỉ đọc như chụp màn hình và lấy văn bản được bật mặc định; công cụ ghi như nhấp, gửi và xóa được tắt mặc định, chỉ bật tạm theo tác vụ. Như vậy ngay cả khi mô hình phán đoán sai, thiệt hại vẫn có giới hạn.

Cô lập môi trường theo mục đích. Tài khoản khác nhau và tác vụ khác nhau phải dùng môi trường khác nhau; không trộn lẫn trong một môi trường. Trong kịch bản quản lý nhiều tài khoản, việc tạo môi trường, gắn đường ra mạng và duy trì trạng thái có thể giao cho một lớp môi trường chuyên biệt. Các công cụ như PurpleMark tách riêng lớp đó, còn lớp adapter chỉ phụ trách ghép logic nghiệp vụ và kiểm tra.

Giữ lại nhật ký audit. Cần có thể xuất theo thời gian tác vụ nào đã dùng môi trường nào và đã gọi những công cụ ghi nào. Khi thực sự xảy ra sự cố, bản ghi này là cách duy nhất để khôi phục lại quá trình.

Cuối cùng là giới hạn cứng: lớp adapter không nên cung cấp bất kỳ lối nào để vượt qua quy tắc của nền tảng. Những hành động như đăng ký hàng loạt, vượt qua xác minh hoặc giả mạo danh tính không được đóng gói thành công cụ, kể cả dưới tên công cụ nội bộ. Một khi công cụ đã lộ cho mô hình, lệnh gọi có thể xảy ra tự động; trước đó không có điểm chặn đáng tin cậy, còn sau đó thì không thể hoàn tác hậu quả.

Về phiên bản giao thức và định nghĩa trường, hãy tham khảo tài liệu MCP chính thức.