MCP: Giao thức giúp "dọn dẹp" và chuẩn hóa kiến trúc Agent

Cách đây vài tuần, một thành viên trong đội dữ liệu đã hỏi liệu chúng tôi có thể cập nhật lược đồ cơ sở dữ liệu đang được một trong các công cụ của hệ thống agent phức tạp điền vào hay không. Bản cập nhật rất đơn giản: chỉ cần thêm hai cột mới vào bảng.

Vấn đề là định nghĩa của công cụ này nằm ở bộ điều phối (orchestrator) của agent. Một phiên bản tương tự thứ hai nằm ở agent xác thực. Một phiên bản thứ ba hơi khác và đã lỗi thời lại nằm trong một mô-đun tiện ích mà ai đó đã viết ba sprint trước. Logic phê duyệt có sự can thiệp của con người (human-in-the-loop) được kết nối trực tiếp vào các cạnh của đồ thị (graph edges), với một triển khai tùy chỉnh cho mỗi công cụ. Việc thay đổi lược đồ đồng nghĩa với việc phải chạm vào bốn tệp, kiểm tra lại từng agent riêng lẻ và hy vọng không có gì bị hỏng âm thầm ở phía sau.

Chúng tôi đã khắc phục nó, nhưng điều này đặt ra một câu hỏi nghiêm túc: tại sao chúng ta lại xây dựng theo cách này?

Câu trả lời trung thực là chúng ta không có lựa chọn nào khác. Việc gọi công cụ (tool calling) trong LangGraph về bản chất là một mối quan tâm cục bộ. Bạn định nghĩa công cụ ở nơi bạn cần, gọi chúng ở nơi bạn gọi và sở hữu toàn bộ phần kết nối (plumbing). Điều này có thể quản lý được khi bạn chỉ có hai agent, nhưng nó trở thành vấn đề khi bảy agent chia sẻ các công cụ chồng chéo với một cổng kiểm soát của con người.

Sau khi nghiên cứu, chúng tôi quyết định rằng thay vì định nghĩa công cụ cục bộ cho mọi agent, chúng ta nên sử dụng một tài nguyên được chia sẻ có thể lưu trữ tất cả các công cụ và bất kỳ agent nào cũng có thể sử dụng chúng.

Trong bài viết này, chúng ta sẽ tìm hiểu sâu hơn về giải pháp này.

MCP là gì?

Model Context Protocol là một tiêu chuẩn mở được Anthropic công bố vào cuối năm 2024. Nó chuẩn hóa cách một AI agent khám phá và gọi các công cụ. Thay vì định nghĩa công cụ bên trong bộ điều phối, bạn chạy chúng trên một máy chủ riêng biệt. Agent kết nối với máy chủ đó tại thời điểm chạy, hỏi xem có công cụ nào available và nhận lại danh sách.

Một kỹ sư cấp cao đọc bài viết này sẽ ngay lập tức hỏi: "Tôi không thể chỉ xây dựng một sổ đăng ký công cụ tập trung và tiêm nó vào mỗi agent khi khởi động sao?". Tôi đã tự hỏi điều đó và đã sử dụng sổ đăng ký công cụ thay vì MCP trong một hệ thống khác.

Đúng là bạn có thể làm vậy, và nếu bạn đã có thứ gì đó hoạt động tương tự, MCP không phải là một vấn đề khẩn cấp. Những gì một sổ đăng ký tùy chỉnh không mang lại cho bạn là ranh giới khả năng tương tác (interoperability boundary). MCP là một giao thức, không phải một thư viện. Bất kỳ máy khách tương thích MCP nào cũng có thể kết nối với máy chủ của bạn, là LangGraph hôm nay, có thể là một khung công tác khác vào năm sau. Một máy khách TypeScript có thể gọi máy chủ Python của bạn mà không cần bất kỳ công việc tích hợp nào thêm. Sổ đăng ký công cụ không cung cấp chức năng này.

Còn có một vấn đề về quyền sở hữu của nhóm. Trong trường hợp của chúng tôi, đội ML sở hữu các công cụ, còn đội ứng dụng sở hữu đồ thị. MCP đã mang lại cho họ một hợp đồng rõ ràng mà không cần chia sẻ cơ sở mã.

Xây dựng MCP Server

Một máy chủ MCP có thể hiển thị ba thứ: Công cụ (Tools - các hành động có thể gọi), Tài nguyên (Resources - dữ liệu chỉ đọc), và Mẫu (Prompts - các mẫu có thể tái sử dụng). Đối với một hệ thống agent cần thực hiện một số hành động, công cụ là mối quan tâm chính.

Python SDK đi kèm với FastMCP, xử lý việc tạo lược đồ từ các gợi ý kiểu (type hints) và quản lý vòng đời giao thức. Bạn chỉ cần viết một hàm và trang trí nó bằng bộ trang trí công cụ, máy chủ sẽ lo phần còn lại.

Một điều khiến nhiều người mắc bẫy với phương thức vận chuyển stdio: không bao giờ ghi vào stdout. Giao thức MCP sử dụng stdout làm kênh giao tiếp của nó. Bất kỳ lệnh gọi print() lạc nào cũng sẽ làm hỏng luồng thông báo theo cách rất khó gỡ lỗi.

import sys
import logging
from mcp.server.fastmcp import FastMCP

logging.basicConfig(level=logging.INFO, stream=sys.stderr)
logger = logging.getLogger("analyst-tools")

mcp = FastMCP("analyst-tools")

@mcp.tool()
async def run_analysis(code: str, dataset: str) -> dict:
    """
    Executes a Python snippet against live data and returns the result.
    Use when the user wants to compute aggregates, filter records,
    or derive insights. The code must assign its final output to a
    variable named 'output'.
    
    Args:
        code: Python code to execute.
        dataset: One of 'sales', 'inventory', 'pipeline'.
    """
    logger.info(f"run_analysis | dataset={dataset}")
    return await execute_in_sandbox(code, dataset)

@mcp.tool()
async def write_to_db(table: str, payload: dict) -> dict:
    """
    Persists a result record to the analyst results table.
    Only call this after run_analysis has returned a verified output.
    
    Args:
        table: Target table name.
        payload: Key-value pairs to write as a new record.
    """
    logger.info(f"write_to_db | table={table}")
    return await persist_result(table, payload)

if __name__ == "__main__":
    mcp.run(transport="stdio")

Các chuỗi tài liệu (docstrings) được LLM sử dụng để giúp agent quyết định gọi công cụ nào. Vì vậy, việc viết một docstring tốt là rất quan trọng.

Stdio so với HTTP

Quyết định này xuất hiện trong mọi triển khai sản xuất và hầu hết các bài viết đều bỏ qua nó.

Stdio chạy máy chủ như một quy trình con của máy khách. Giao tiếp diễn ra qua đầu vào và đầu ra chuẩn. Độ trễ chỉ ở mức vài mili-giây, không có mạng liên quan và việc thiết lập là tối thiểu. Đây là lựa chọn đúng cho phát triển cục bộ, triển khai trên một máy, hoặc bất cứ đâu máy chủ và máy khách sống trong cùng một cây quy trình.

Streamable HTTP chạy máy chủ như một dịch vụ độc lập. Sử dụng điều này khi máy chủ cần được chia sẻ trên nhiều máy khách hoặc máy, khi bạn muốn triển khai nó dưới dạng container, hoặc khi bạn cần mở rộng quy mô ngang (horizontal scaling). Các triển khai Serverless như Cloud Run hoạt động tốt ở đây. Stdio hoàn toàn không phù hợp với mô hình serverless vì nó giả định một quy trình cha tồn tại lâu dài.

Việc chuyển đổi giữa hai chế độ này trong FastMCP chỉ cần một dòng:

mcp.run(transport="streamable-http", host="0.0.0.0", port=8080)

Chúng ta chỉ cần thay đổi phương thức vận chuyển trong mcp.run() và mọi thứ khác vẫn giữ nguyên.

Đối với các yêu cầu về nơi lưu trữ dữ liệu, một máy chủ MCP chạy tại chỗ (on-premise) với các công cụ không bao giờ chạm vào API bên ngoài mang lại cho bạn một câu chuyện rõ ràng cho đội tuân thủ của bạn. Giao thức không quan tâm máy chủ chạy ở đâu.

Kết nối với LangGraph

Thư viện langchain-mcp-adapters quản lý vòng đời quy trình con, thực hiện bắt tay khám phá công cụ và chuyển đổi lược đồ công cụ MCP thành các đối tượng công cụ tương thích với LangChain.

from langchain_mcp_adapters.client import MultiServerMCPClient
from langgraph.graph import StateGraph, MessagesState, START
from langgraph.prebuilt import ToolNode, tools_condition
from langchain_google_vertexai import ChatVertexAI

llm = ChatVertexAI(
    model="gemini-2.5-flash",
    temperature=0,
    max_tokens=None
)

async def run(query: str):
    async with MultiServerMCPClient({
        "analyst-tools": {
            "command": "python",
            "args": ["./mcp_server.py"],
            "transport": "stdio",
        }
    }) as client:
        tools = await client.get_tools()
        llm_with_tools = llm.bind_tools(tools)

        def agent_node(state: MessagesState):
            return {"messages": [llm_with_tools.invoke(state["messages"])]}

        graph = StateGraph(MessagesState)
        graph.add_node("agent", agent_node)
        graph.add_node("tools", ToolNode(tools))
        graph.add_edge(START, "agent")
        graph.add_conditional_edges("agent", tools_condition)
        graph.add_edge("tools", "agent")

        app = graph.compile()

        result = await app.ainvoke({
            "messages": [{"role": "user", "content": query}]
        })
        print(result["messages"][-1].content)

tools_condition là một mô-đun tích hợp sẵn của LangGraph kiểm tra xem tin nhắn cuối cùng có chứa lệnh gọi công cụ hay không. Nếu có, định tuyến tới bộ thực thi công cụ và nếu không, chúng ta đã xong. Việc sử dụng nó thay vì viết hàm định tuyến của riêng bạn rất quan trọng vì nó xử lý các trường hợp cạnh và thiếu sót triển khai.

Một hành vi đáng để biết: MultiServerMCPClient tạo một phiên MCP mới cho mỗi lệnh gọi công cụ theo mặc định. Đối với một yêu cầu duy nhất thực hiện năm lệnh gọi công cụ tuần tự, đó là năm lần bắt tay. Điều này ổn cho stdio trên cùng một máy, nhưng đáng chú ý trên phương thức vận chuyển HTTP với máy chủ từ xa. Đối với khối lượng công việc sản xuất có các lệnh gọi công cụ chuỗi, hãy sử dụng async with client.session("analyst-tools") để ghim nhiều lệnh gọi vào một phiên.

Human-in-the-loop tại ranh giới giao thức

Trước MCP, cổng phê duyệt của chúng tôi sống trong đồ thị. Chúng tôi sử dụng interrupt_before trên các nút cụ thể, kết nối logic xác nhận tùy chỉnh vào các cạnh đồ thị và cập nhật UI mỗi khi thêm công cụ nhạy cảm mới. Nó hoạt động nhưng cũng có nghĩa là thêm một công cụ cần phê duyệt là một bài tập phối hợp của ba đội.

Sau MCP, cổng chuyển sang một lớp duy nhất giữa bộ thực thi LangGraph và máy khách MCP. Bất kỳ công cụ nào khớp với chính sách nhạy cảm sẽ chạm vào cổng trước khi đến máy chủ. Đồ thị không biết gì về nó.

SENSITIVE_TOOLS = frozenset({"write_to_db", "send_notification", "trigger_webhook"})

async def gated_call(tool_name: str, arguments: dict, execute) -> dict:
    if tool_name in SENSITIVE_TOOLS:
        # In production: push to Slack / internal UI / audit queue
        print(f"\nAPPROVAL REQUIRED {tool_name}")
        print(f"Arguments: {arguments}")
        decision = input("Approve? (y/n): ").strip().lower()
        if decision != "y":
            return {
                "status": "rejected",
                "reason": f"Operator declined '{tool_name}'."
            }
    return await execute(tool_name, arguments)

SENSITIVE_TOOLS là một tập hợp duy nhất, được tham khảo cho mọi lệnh gọi công cụ bất kể agent nào kích hoạt nó. Thêm công cụ nhạy cảm mới vào máy chủ? Chỉ cần thêm tên vào tập hợp này. Đồ thị không thay đổi. UI phê duyệt không thay đổi. Trong hệ thống nội bộ của chúng tôi, chúng tôi tải cái này từ tệp cấu hình khi khởi động. Đội sản phẩm và tuân thủ có thể cập nhật nó mà không cần triển khai mã.

Điều gì có thể bị hỏng trong Sản xuất và tại sao?

Máy chủ bị sập giữa chừng. Máy khách sẽ nhận được lỗi trên lệnh gọi công cụ tiếp theo. ToolNode của LangGraph sẽ trả lại điều này cho LLM dưới dạng thông báo lỗi công cụ. Mô hình có phục hồi hay bị lặp trong sự nhầm lẫn phụ thuộc vào lời nhắc hệ thống của bạn. Tối thiểu, hãy ghi nhật ký stderr của quy trình con riêng biệt để bạn có thể thấy điều gì đã giết chết máy chủ, nếu không thì việc gỡ lỗi chỉ là phỏng đoán.

LLM gọi sai công cụ. MCP không bảo vệ bạn khỏi điều này. Nếu mô tả công cụ của bạn mơ hồ hoặc trùng lặp về nghĩa, mô hình sẽ đưa ra quyết định định tuyến sai. Chúng tôi đã dành nhiều thời gian để tinh chỉnh các chuỗi tài liệu trong máy chủ của mình cụ thể vì một mô tả được viết kém đã gây ra write_to_db được gọi trước khi run_analysis kết thúc. Hãy coi mô tả công cụ là một vấn đề kỹ thuật prompt engineering.

Cổng phê duyệt trên quy trình công việc chạy dài. Nếu một con người cần phê duyệt một lệnh gọi công cụ và mất năm phút, đồ thị agent bị treo chờ. LangGraph hỗ trợ lưu trữ trạng thái đồ thị thông qua kiểm tra (checkpointing), vì vậy bạn có thể để quy trình thoát và tiếp tục khi quyết định đến. Điều đó phức tạp hơn những gì được hiển thị ở đây nhưng đó là kiến trúc đúng cho các quy trình công việc không thể chặn một luồng vô thời hạn.

Tác động của MCP đối với Hệ thống Agent của chúng tôi

Chúng tôi đã di chuyển bảy công cụ lên máy chủ, ba trong số đó được kiểm soát bởi cổng phê duyệt. Bộ điều phối gọi chúng không biết bất kỳ điều gì về chúng.

Chúng tôi đã loại bỏ hoàn toàn sự trùng lặp công cụ. Bây giờ, run_analysis được định nghĩa chính xác ở một nơi phục vụ bảy quy trình công việc đồng thời. Để cập nhật lược đồ đầu ra, chúng tôi chỉ cần thực hiện thay đổi trong máy chủ và mọi người tiêu dùng sẽ nhận được thay đổi.

Việc thêm các khả năng mới trở nên nhanh chóng. Ví dụ, chúng tôi đã thêm một công cụ generate_visualisation vào tuần sau và agent đã sử dụng nó vào ngay ngày hôm sau. Không có thay đổi nào đối với bộ điều phối.

Chúng tôi kết thúc với một đội sở hữu các công cụ, đội khác sở hữu đồ thị, và một hợp đồng rõ ràng giữa họ. Khi đội phân tích muốn một khả năng mới, họ nói chuyện với đội ML về máy chủ, không phải đội ứng dụng hay đội đồ thị.

Tôi muốn chia sẻ một điều mà MCP không khắc phục được: Nó sẽ không làm cho các công cụ không đáng tin cậy trở nên đáng tin cậy. Nó sẽ không giúp LLM đưa ra quyết định định tuyến tốt hơn nếu mô tả của bạn tồi. Và nó không thay thế khả năng quan sát (observability), bạn vẫn cần ghi nhật ký các lệnh gọi công cụ và theo dõi các đường dẫn thực thi. Cấu trúc giúp việc này dễ dàng đo lường hơn, nhưng công việc vẫn là của bạn.

Kết luận

Bằng cách chuyển đổi sang MCP và chuyển các công cụ khỏi bộ điều phối agent cục bộ sang một máy chủ chuyên dụng, chúng tôi đã dọn dẹp cơ sở mã, tách biệt các ràng buộc kỹ thuật và làm cho toàn bộ hệ thống agent dễ triển khai hơn.

Nhờ sự chuyển đổi này, đội ML của chúng tôi hiện có thể triển khai và phiên bản các công cụ độc lập mà không cần chạm vào đồ thị ứng dụng.