Tôi đã thử sử dụng Langchain với máy chủ MCP, đây là các bước:

Giao thức Ngữ cảnh Mô hình (MCP) là một giao thức mã nguồn mở được phát triển bởi Anthropic nhằm giải quyết một thách thức cơ bản trong các ứng dụng Mô hình Ngôn ngữ Lớn (LLM): sự cô lập của chúng khỏi các nguồn dữ liệu và công cụ bên ngoài. Hướng dẫn toàn diện này sẽ dẫn dắt bạn trong việc triển khai MCP với LangChain, cung cấp cho bạn kiến thức để tạo ra các ứng dụng tinh vi kết hợp hiệu quả cả hai công nghệ này.

💡

Khi triển khai thử nghiệm cho các ứng dụng dựa trên API, các nhà phát triển và người thử nghiệm ngày càng tìm đến các công cụ chuyên biệt như Apidog, một sự thay thế toàn diện cho Postman giúp đơn giản hóa vòng đời phát triển API.

button

Hiểu Về MCP và Mục Đích Của Nó

Giao thức Ngữ cảnh Mô hình nhằm tiêu chuẩn hóa cách các ứng dụng dựa trên LLM kết nối với các hệ thống bên ngoài đa dạng. Hãy xem MCP như "USB-C cho AI" - một giao diện phổ quát cho phép trao đổi dữ liệu liền mạch, an toàn và có khả năng mở rộng giữa các LLM/đại lý AI và các nguồn tài nguyên bên ngoài.

MCP sử dụng kiến trúc máy chủ - máy khách:

Các máy chủ MCP: Các ứng dụng AI cần truy cập dữ liệu bên ngoài
Các máy chủ MCP: Các nhà cung cấp dữ liệu hoặc công cụ cung cấp thông tin cho các máy chủ

Giao thức tạo điều kiện phân tách rõ ràng các mối quan tâm, cho phép các nhà phát triển tạo ra các bộ kết nối mô-đun, tái sử dụng trong khi duy trì bảo mật vững chắc thông qua kiểm soát quyền hạn chi tiết.

Kiến Trúc Kỹ Thuật

Kiến trúc của MCP bao gồm ba thành phần chính:

Máy chủ: Máy chủ MCP cung cấp công cụ và nguồn dữ liệu thông qua một API tiêu chuẩn hóa
Khách hàng: Ứng dụng khách hàng giao tiếp với máy chủ để truy cập công cụ và dữ liệu
Bộ chuyển đổi: LangChain cung cấp các bộ chuyển đổi giúp đơn giản hóa việc tích hợp giữa các máy chủ MCP và các ứng dụng LLM

Luồng giao tiếp theo mẫu này:

Ứng dụng LangChain yêu cầu thực thi dữ liệu/công cụ
Bộ chuyển đổi MCP chuyển đổi yêu cầu thành định dạng giao thức MCP
Máy chủ xử lý yêu cầu và trả về kết quả
Bộ chuyển đổi chuyển đổi phản hồi trở lại định dạng có thể sử dụng bởi LangChain

Yêu Cầu Trước Khi Bắt Đầu

Trước khi bắt đầu, hãy đảm bảo bạn đã có những điều sau:

Cài đặt Python 3.8+
Khóa API OpenAI (để sử dụng các mô hình GPT với LangChain)
Kiến thức cơ bản về các khái niệm của LangChain
Quyền truy cập terminal (các ví dụ được hiển thị trên macOS)

Thiết Lập Môi Trường

Đầu tiên, hãy tạo và cấu hình môi trường phát triển của chúng ta:

# Tạo một môi trường ảo
python3 -m venv MCP_Demo

# Kích hoạt môi trường ảo
source MCP_Demo/bin/activate

# Cài đặt các gói cần thiết
pip install langchain-mcp-adapters
pip install langchain-openai

# Đặt khóa API OpenAI của bạn
export OPENAI_API_KEY=your_api_key

Tạo Một Máy Chủ MCP Đơn Giản

Chúng ta sẽ bắt đầu bằng cách xây dựng một máy chủ MCP cơ bản cung cấp các phép toán toán học. Tạo một tệp có tên math_server.py:

# math_server.py
from mcp.server.fastmcp import FastMCP

mcp = FastMCP("Math")

@mcp.tool()
def add(a: int, b: int) -> int:
    """Cộng hai số"""
    return a + b

@mcp.tool()
def multiply(a: int, b: int) -> int:
    """Nhân hai số"""
    return a * b

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

Máy chủ này cung cấp hai công cụ toán học: add và multiply. Lớp FastMCP giúp đơn giản hóa việc tạo máy chủ, tự động xử lý các chi tiết của giao thức. Mỗi hàm được trang trí với @mcp.tool() trở nên khả dụng cho các khách hàng, với tài liệu được lấy từ các chuỗi tài liệu.

Triển Khai Khách Hàng LangChain

Bước tiếp theo, tạo một khách hàng LangChain để tương tác với máy chủ MCP. Lưu lại với tên client.py:

# Tạo tham số máy chủ cho kết nối stdio
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client
from langchain_mcp_adapters.tools import load_mcp_tools
from langgraph.prebuilt import create_react_agent
from langchain_openai import ChatOpenAI
import asyncio

model = ChatOpenAI(model="gpt-4o")

# Cấu hình tham số máy chủ
server_params = StdioServerParameters(
    command="python",
    # Chỉ định đường dẫn đến tệp máy chủ của bạn
    args=["math_server.py"],
)

async def run_agent():
    async with stdio_client(server_params) as (read, write):
        async with ClientSession(read, write) as session:
            # Khởi tạo kết nối
            await session.initialize()
            
            # Tải công cụ MCP vào định dạng LangChain
            tools = await load_mcp_tools(session)
            
            # Tạo và chạy tác nhân
            agent = create_react_agent(model, tools)
            agent_response = await agent.ainvoke({"messages": "kết quả của (3 + 5) x 12 là gì?"})
            return agent_response

# Chạy hàm bất đồng bộ
if __name__ == "__main__":
    result = asyncio.run(run_agent())
    print(result)

Khách hàng này thiết lập một kết nối đến máy chủ MCP, tải các công cụ có sẵn, và tạo ra một tác nhân LangChain có thể sử dụng các công cụ này để giải quyết vấn đề.

Chạy Ví Dụ

Để chạy ví dụ này:

Bắt đầu máy chủ MCP trong một tab terminal:

python3 math_server.py

Trong một tab terminal khác, chạy khách hàng:

python3 client.py

Khách hàng sẽ gọi tác nhân LangChain, tác nhân sẽ:

Phân tích câu hỏi "(3 + 5) x 12"
Gọi công cụ add với các tham số 3 và 5
Nhận được kết quả 8
Gọi công cụ multiply với các tham số 8 và 12
Trả về câu trả lời cuối cùng: 96

Triển Khai Máy Chủ MCP Nâng Cao

Hãy mở rộng triển khai của chúng ta để tạo ra một máy chủ MCP tinh vi hơn cung cấp truy cập cơ sở dữ liệu. Ví dụ này mô tả cách xây dựng các bộ kết nối với các hệ thống bên ngoài:

# db_server.py
from mcp.server.fastmcp import FastMCP
import sqlite3
from typing import List, Dict, Any

class DatabaseConnector:
    def __init__(self, db_path):
        self.conn = sqlite3.connect(db_path)
        self.cursor = self.conn.cursor()
    
    def execute_query(self, query: str) -> List[Dict[str, Any]]:
        self.cursor.execute(query)
        columns = [desc[0] for desc in self.cursor.description]
        results = []
        for row in self.cursor.fetchall():
            results.append({columns[i]: row[i] for i in range(len(columns))})
        return results

mcp = FastMCP("DatabaseTools")
db_connector = DatabaseConnector("example.db")

@mcp.tool()
def run_sql_query(query: str) -> List[Dict[str, Any]]:
    """Thực thi một truy vấn SQL trên cơ sở dữ liệu và trả về kết quả"""
    try:
        return db_connector.execute_query(query)
    except Exception as e:
        return {"error": str(e)}

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

Tích Hợp Nhiều Máy Chủ MCP Với LangChain

Đối với các ứng dụng phức tạp hơn, bạn có thể cần tích hợp nhiều máy chủ MCP. Đây là cách tạo ra một khách hàng kết nối đến nhiều máy chủ:

# multi_server_client.py
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client
from langchain_mcp_adapters.tools import load_mcp_tools
from langgraph.prebuilt import create_react_agent
from langchain_openai import ChatOpenAI
import asyncio
from typing import List, Dict

# Định nghĩa cấu hình máy chủ của chúng ta
servers = [
    {
        "name": "math",
        "params": StdioServerParameters(
            command="python", 
            args=["math_server.py"]
        )
    },
    {
        "name": "database",
        "params": StdioServerParameters(
            command="python", 
            args=["db_server.py"]
        )
    }
]

async def connect_to_server(server_config):
    """Kết nối đến một máy chủ MCP và tải công cụ của nó"""
    name = server_config["name"]
    params = server_config["params"]
    
    read, write = await stdio_client(params).__aenter__()
    session = ClientSession(read, write)
    await session.__aenter__()
    await session.initialize()
    
    tools = await load_mcp_tools(session)
    return {
        "name": name,
        "session": session,
        "tools": tools,
        "cleanup": lambda: asyncio.gather(
            session.__aexit__(None, None, None),
            stdio_client(params).__aexit__(None, None, None)
        )
    }

async def run_multi_server_agent():
    # Kết nối đến tất cả các máy chủ
    connections = await asyncio.gather(
        *[connect_to_server(server) for server in servers]
    )
    
    try:
        # Tập hợp tất cả công cụ từ tất cả các máy chủ
        all_tools = []
        for connection in connections:
            all_tools.extend(connection["tools"])
        
        # Tạo tác nhân với tất cả công cụ
        model = ChatOpenAI(model="gpt-4o")
        agent = create_react_agent(model, all_tools)
        
        # Chạy tác nhân với một truy vấn phức tạp mà có thể sử dụng nhiều máy chủ
        response = await agent.ainvoke({
            "messages": "Tìm các khách hàng đã chi tiêu nhiều hơn giá trị đơn hàng trung bình và tính tổng chi tiêu của họ."
        })
        
        return response
    
    finally:
        # Dọn dẹp tất cả các kết nối
        for connection in connections:
            await connection["cleanup"]()

# Chạy tác nhân đa máy chủ
if __name__ == "__main__":
    result = asyncio.run(run_multi_server_agent())
    print(result)

Xử Lý Lỗi và Chiến Lược Dự Phòng

Các triển khai MCP vững mạnh nên bao gồm việc xử lý lỗi. Đây là phiên bản nâng cao của khách hàng cho thấy cách xử lý lỗi và chiến lược dự phòng:

# robust_client.py
async def run_agent_with_fallbacks():
    try:
        # Thử kết nối chính
        async with stdio_client(server_params) as (read, write):
            async with ClientSession(read, write) as session:
                try:
                    await session.initialize()
                    tools = await load_mcp_tools(session)
                    agent = create_react_agent(model, tools)
                    return await agent.ainvoke({"messages": "kết quả của (3 + 5) x 12 là gì?"})
                except Exception as e:
                    print(f"Lỗi khi sử dụng các công cụ MCP: {e}")
                    # Dự phòng để gọi mô hình trực tiếp không dùng công cụ
                    return await model.ainvoke([
                        HumanMessage(content="kết quả của (3 + 5) x 12 là gì?")
                    ])
    except Exception as connection_error:
        print(f"Lỗi kết nối: {connection_error}")
        # Dự phòng cuối cùng
        return {"error": "Không thể thiết lập kết nối đến máy chủ MCP"}

Các Xem Xét Về Bảo Mật

Khi triển khai MCP với LangChain, hãy xem xét các thực hành bảo mật tốt nhất sau:

Xác Thực Đầu Vào: Luôn xác thực đầu vào cho các công cụ MCP để ngăn chặn các cuộc tấn công tiêm chích
Quyền Truy Cập Công Cụ: Thực hiện quyền truy cập chi tiết cho mỗi công cụ
Giới Hạn Tốc Độ: Áp dụng giới hạn tốc độ để ngăn chặn lạm dụng các công cụ
Xác Thực: Thực hiện xác thực đúng cách giữa khách hàng và máy chủ

Dưới đây là một ví dụ về việc triển khai quyền truy cập công cụ:

from mcp.server.fastmcp import FastMCP, Permission

mcp = FastMCP("SecureTools")

# Định nghĩa các mức quyền truy cập
READ_PERMISSION = Permission("read", "Có thể đọc dữ liệu")
WRITE_PERMISSION = Permission("write", "Có thể chỉnh sửa dữ liệu")

@mcp.tool(permissions=[READ_PERMISSION])
def get_data(key: str) -> str:
    """Lấy dữ liệu theo khóa (cần quyền đọc)"""
    # Triển khai...
    return f"Dữ liệu cho {key}"

@mcp.tool(permissions=[WRITE_PERMISSION])
def update_data(key: str, value: str) -> bool:
    """Cập nhật dữ liệu (cần quyền viết)"""
    # Triển khai...
    return True

Tối Ưu Hiệu Suất

Đối với các triển khai sản xuất, hãy xem xét các tối ưu hóa hiệu suất này:

Định Kỳ Kết Nối: Tái sử dụng các kết nối MCP thay vì tạo mới cho mỗi yêu cầu
Xử Lý Theo Lô: Nhóm nhiều lệnh gọi công cụ khi có thể
Xử Lý Bất Đồng Bộ: Sử dụng asyncio để xử lý nhiều yêu cầu đồng thời

Ví dụ về định kỳ kết nối:

class MCPConnectionPool:
    def __init__(self, max_connections=10):
        self.available_connections = asyncio.Queue(max_connections)
        self.max_connections = max_connections
        self.current_connections = 0
    
    async def initialize(self):
        # Tạo trước một số kết nối
        for _ in range(3):  # Bắt đầu với 3 kết nối
            await self._create_connection()
    
    async def _create_connection(self):
        if self.current_connections >= self.max_connections:
            raise Exception("Số kết nối tối đa đã đạt")
        
        read, write = await stdio_client(server_params).__aenter__()
        session = await ClientSession(read, write).__aenter__()
        await session.initialize()
        
        self.current_connections += 1
        await self.available_connections.put(session)
    
    async def get_connection(self):
        if self.available_connections.empty() and self.current_connections < self.max_connections:
            await self._create_connection()
        
        return await self.available_connections.get()
    
    async def release_connection(self, connection):
        await self.available_connections.put(connection)

Kiểm Thử Các Triển Khai MCP

Việc kiểm thử kỹ lưỡng là rất quan trọng để đảm bảo các triển khai MCP đáng tin cậy. Đây là một phương pháp kiểm thử sử dụng pytest:

# test_mcp.py
import pytest
import asyncio
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client
from langchain_mcp_adapters.tools import load_mcp_tools

@pytest.fixture
async def mcp_session():
    server_params = StdioServerParameters(
        command="python",
        args=["math_server.py"],
    )
    
    read, write = await stdio_client(server_params).__aenter__()
    session = ClientSession(read, write)
    await session.__aenter__()
    await session.initialize()
    
    yield session
    
    await session.__aexit__(None, None, None)
    await stdio_client(server_params).__aexit__(None, None, None)

@pytest.mark.asyncio
async def test_add_tool(mcp_session):
    tools = await load_mcp_tools(mcp_session)
    add_tool = next(tool for tool in tools if tool.name == "add")
    
    result = await add_tool.invoke({"a": 5, "b": 7})
    assert result == 12

@pytest.mark.asyncio
async def test_multiply_tool(mcp_session):
    tools = await load_mcp_tools(mcp_session)
    multiply_tool = next(tool for tool in tools if tool.name == "multiply")
    
    result = await multiply_tool.invoke({"a": 6, "b": 8})
    assert result == 48

Kết Luận

Giao thức Ngữ cảnh Mô hình cung cấp một khung mạnh mẽ để kết nối các ứng dụng LangChain với các công cụ và nguồn dữ liệu bên ngoài. Bằng cách tiêu chuẩn hóa những kết nối này, MCP cho phép các nhà phát triển tạo ra các tác nhân AI tinh vi có thể tương tác liền mạch với môi trường của chúng.

Sự kết hợp giữa khả năng của tác nhân LangChain với khả năng kết nối của MCP tạo ra một nền tảng để xây dựng các ứng dụng thực sự mạnh mẽ, ý thức về ngữ cảnh. Khi hệ sinh thái MCP tiếp tục phát triển, chúng ta có thể mong đợi sẽ có thêm nhiều máy chủ và công cụ được xây dựng sẵn, đơn giản hóa hơn nữa quy trình phát triển.

Hướng dẫn này đã đề cập đến các khái niệm cơ bản và chi tiết triển khai việc sử dụng MCP với LangChain, từ cài đặt cơ bản đến các mô hình nâng cao như định kỳ kết nối và xử lý lỗi. Bằng cách làm theo những thực hành này, bạn có thể tạo ra các ứng dụng chắc chắn, sẵn sàng cho sản xuất, tận dụng tốt nhất cả hai công nghệ.

Để khám phá thêm, hãy xem xét việc nghiên cứu hệ sinh thái ngày càng lớn của các máy chủ MCP có sẵn trên GitHub, hoặc đóng góp các máy chủ của riêng bạn cho cộng đồng. Tương lai của các tác nhân AI nằm ở khả năng tận dụng hiệu quả các công cụ và dữ liệu bên ngoài, và MCP là một bước quan trọng tiến tới hiện thực hóa tầm nhìn đó.