MCP server thực tế: Database, API và file

Bạn đã biết cách xây dựng các tool, resource và prompt. Giờ hãy kết nối chúng với các hệ thống thực tế — cơ sở dữ liệu, API REST và hệ thống file. Ba mô hình này bao phủ 80% các trường hợp sử dụng MCP thực tế.

🔄 Tóm tắt nhanh: Trong bài học trước, bạn đã học cả ba thành phần cơ bản của MCP: Tools cho các hành động, Resources cho dữ liệu và Prompts cho các template. Giờ bạn sẽ áp dụng chúng để kết nối các trợ lý AI với những hệ thống mà tổ chức của bạn thực sự sử dụng.

Mẫu 1: Database server

MCP server phổ biến nhất kết nối AI với cơ sở dữ liệu. AI có thể khám phá schema, viết truy vấn và phân tích kết quả — tất cả thông qua cuộc hội thoại tự nhiên.

Kiến trúc

📍 Nơi dán: Mở ChatGPT (chat.openai.com), Claude (claude.ai) hoặc Gemini (gemini.google.com) và bắt đầu một cuộc hội thoại mới.

📋 Cách sao chép prompt này: Nhấp vào bất kỳ đâu bên trong khối màu xám, nhấn Cmd+A rồi Cmd+C (Mac) hoặc Ctrl+A rồi Ctrl+C (Windows). Hoặc sử dụng biểu tượng sao chép xuất hiện.

Người dùng: "Tháng trước có bao nhiêu đơn hàng được đặt?"
  ↓
Claude → gọi công cụ query_database
  ↓
MCP Server → thực thi SQL đối với PostgreSQL
  ↓
Kết quả trả về: "1.247 đơn hàng với tổng trị giá 89.340 USD"

✏️ ​​Cách điền thông tin chi tiết của bạn: Thay thế mỗi dấu ngoặc vuông [] và trình giữ chỗ trong ngoặc bằng thông tin cụ thể từ tình huống thực tế của bạn. Thông tin đầu vào không rõ ràng sẽ tạo ra kết quả không rõ ràng — hãy cụ thể.

👀 Những gì bạn sẽ thấy: Trong vòng vài giây, AI sẽ trả về một phản hồi có cấu trúc dựa trên prompt ở trên. Hãy đọc kỹ và coi đó là bản nháp, không phải câu trả lời cuối cùng.

📌 Nên làm gì với kết quả: Lưu phản hồi vào file Notes. Hãy chọn đề xuất có tác động mạnh nhất và thực hiện nó trong tuần này — đừng cố gắng làm mọi thứ cùng một lúc.

⚠️ Nếu thấy không ổn: Nếu các đề xuất có vẻ chung chung, hãy dán nội dung này: "Hãy cụ thể hơn với bối cảnh thực tế của tôi. Bỏ qua những lời khuyên chung chung đi." Nếu bỏ qua các chi tiết quan trọng bạn đã cung cấp, hãy hỏi: "Bạn đã bỏ sót [X] trong bối cảnh của tôi — hãy thực hiện lại với điều đó làm ràng buộc chính."

Thực hiện

import asyncpg
from mcp.server.fastmcp import FastMCP

mcp = FastMCP("Database Assistant")

# Connection pool (created on startup)
pool = None

@mcp.tool()
async def query_database(sql: str) -> str:
    """Execute a read-only SQL query and return results."""
    # Safety: block write operations
    forbidden = ["INSERT", "UPDATE", "DELETE", "DROP", "ALTER", "TRUNCATE"]
    sql_upper = sql.upper().strip()
    for keyword in forbidden:
        if keyword in sql_upper:
            return f"Error: {keyword} operations are not allowed. This tool is read-only."

    try:
        async with pool.acquire() as conn:
            rows = await conn.fetch(sql)
            if not rows:
                return "Query returned no results."
            # Format as table
            headers = list(rows[0].keys())
            lines = [" | ".join(headers)]
            lines.append("-" * len(lines[0]))
            for row in rows[:50]:  # Limit to 50 rows
                lines.append(" | ".join(str(row[h]) for h in headers))
            return "\n".join(lines)
    except Exception as e:
        return f"Query error: {str(e)}"

@mcp.resource("db://schema")
def get_schema() -> str:
    """Database table schemas for reference."""
    return open("schema.sql").read()

Các quy tắc an toàn chính

1. Mặc định là chỉ đọc. Chặn các thao tác INSERT, UPDATE, DELETE, DROP, ALTER, TRUNCATE
2. Giới hạn kích thước kết quả. Giới hạn ở 50-100 hàng để tránh tràn ngữ cảnh
3. Sử dụng nhóm kết nối. Một kết nối cho mỗi truy vấn sẽ chậm và dễ bị lỗi
4. Hiển thị schema dưới dạng Resource. AI cần biết cấu trúc bảng để viết SQL chính xác

✅ Kiểm tra nhanh: Tại sao chúng ta hiển thị schema cơ sở dữ liệu dưới dạng Resource thay vì Tool?

Câu trả lời: Schema là dữ liệu tham chiếu tĩnh mà AI cần làm ngữ cảnh — nó không thay đổi trong suốt cuộc hội thoại. Resources được lấy một lần để làm ngữ cảnh. Một Tools sẽ cần được gọi mỗi khi AI muốn kiểm tra cấu trúc bảng, gây lãng phí các lệnh gọi.

Mẫu 2: API Server

Đóng gói các API REST bên ngoài để AI có thể gọi chúng thông qua ngôn ngữ tự nhiên:

Triển khai

import httpx
import os
from mcp.server.fastmcp import FastMCP

mcp = FastMCP("GitHub Assistant")

GITHUB_TOKEN = os.environ.get("GITHUB_TOKEN")
BASE_URL = "https://api.github.com"

@mcp.tool()
async def list_issues(repo: str, state: str = "open", limit: int = 10) -> str:
    """List GitHub issues for a repository (format: owner/repo)."""
    if "/" not in repo:
        return "Error: repo must be in 'owner/repo' format"

    async with httpx.AsyncClient() as client:
        response = await client.get(
            f"{BASE_URL}/repos/{repo}/issues",
            headers={"Authorization": f"Bearer {GITHUB_TOKEN}"},
            params={"state": state, "per_page": min(limit, 30)}
        )
        if response.status_code == 404:
            return f"Repository '{repo}' not found"
        if response.status_code != 200:
            return f"GitHub API error: {response.status_code}"

        issues = response.json()
        if not issues:
            return f"No {state} issues in {repo}"

        lines = [f"Issues in {repo} ({state}):"]
        for issue in issues:
            lines.append(f"  #{issue['number']}: {issue['title']}")
        return "\n".join(lines)

@mcp.tool()
async def create_issue(repo: str, title: str, body: str = "") -> str:
    """Create a new GitHub issue."""
    if "/" not in repo:
        return "Error: repo must be in 'owner/repo' format"

    async with httpx.AsyncClient() as client:
        response = await client.post(
            f"{BASE_URL}/repos/{repo}/issues",
            headers={"Authorization": f"Bearer {GITHUB_TOKEN}"},
            json={"title": title, "body": body}
        )
        if response.status_code == 201:
            issue = response.json()
            return f"Created issue #{issue['number']}: {issue['html_url']}"
        return f"Failed to create issue: {response.status_code}"

Cấu hình API key

Truyền thông tin bí mật thông qua các biến môi trường trong file claude_desktop_config.json:

{
  "mcpServers": {
    "github": {
      "command": "python",
      "args": ["/path/to/github_server.py"],
      "env": {
        "GITHUB_TOKEN": "ghp_your_token_here"
      }
    }
  }
}

Không bao giờ hardcode các API key trong code server của bạn. Biến môi trường giúp giữ các bí mật khỏi hệ thống kiểm soát phiên bản và giúp dễ dàng sử dụng các key khác nhau cho mỗi môi trường.

Các thực tiễn tốt nhất cho API server

• Giới hạn tốc độ: Theo dõi các cuộc gọi API và trả về sớm nếu sắp đạt đến giới hạn
• Xử lý thời gian chờ: Đặt thời gian chờ HTTP (timeout=10.0) để tránh bị treo
• Thông báo lỗi: Dịch mã trạng thái HTTP thành các giải thích dễ hiểu
• Phân trang: Đối với các endpoint dạng danh sách, hỗ trợ tham số giới hạn và đặt giới hạn cho nó

✅ Kiểm tra nhanh: Tại sao chúng ta sử dụng biến môi trường cho API token thay vì hardcode chúng?

Câu trả lời: Biến môi trường giúp giữ các bí mật khỏi mã nguồn và hệ thống kiểm soát phiên bản. Các môi trường khác nhau (phát triển, dàn dựng, sản xuất) có thể sử dụng những token khác nhau. Cấu hình MCP client chuyển chúng cho tiến trình server khi khởi động.

Mẫu 3: Server hệ thống file

Cung cấp cho AI quyền truy cập vào các file trên máy cục bộ — với những ranh giới an toàn nghiêm ngặt:

Triển khai

import os
from mcp.server.fastmcp import FastMCP

mcp = FastMCP("File Assistant")

# SAFETY: Only allow access within this directory
ALLOWED_DIR = os.path.expanduser("~/projects")

def safe_path(filepath: str) -> str:
    """Resolve and validate a file path is within the allowed directory."""
    resolved = os.path.realpath(os.path.join(ALLOWED_DIR, filepath))
    if not resolved.startswith(os.path.realpath(ALLOWED_DIR)):
        raise ValueError(f"Access denied: path outside {ALLOWED_DIR}")
    return resolved

@mcp.tool()
def read_file(filepath: str) -> str:
    """Read a file from the projects directory."""
    try:
        full_path = safe_path(filepath)
        with open(full_path, 'r') as f:
            content = f.read()
        # Truncate very large files
        if len(content) > 10000:
            return content[:10000] + f"\n\n[Truncated — file is {len(content)} characters]"
        return content
    except ValueError as e:
        return f"Error: {str(e)}"
    except FileNotFoundError:
        return f"File not found: {filepath}"

@mcp.tool()
def list_directory(dirpath: str = ".") -> str:
    """List files and directories in a given path."""
    try:
        full_path = safe_path(dirpath)
        entries = os.listdir(full_path)
        dirs = sorted(e + "/" for e in entries if os.path.isdir(os.path.join(full_path, e)))
        files = sorted(e for e in entries if os.path.isfile(os.path.join(full_path, e)))
        return "Directories:\n" + "\n".join(f"  {d}" for d in dirs) + \
               "\n\nFiles:\n" + "\n".join(f"  {f}" for f in files)
    except ValueError as e:
        return f"Error: {str(e)}"

Các quy tắc bảo mật hệ thống file

1. Luôn xác định thư mục được phép truy cập. Không bao giờ cấp quyền truy cập hệ thống file không hạn chế.
2. Giải quyết đường dẫn bằng os.path.realpath(). Điều này giúp ngăn chặn các cuộc tấn công duyệt thư mục ../../.
3. Cắt bớt các file lớn. Một file nhật ký 10MB sẽ làm quá tải cửa sổ ngữ cảnh của AI.
4. Cẩn thận với các thao tác ghi. Cân nhắc việc tạo một công cụ ghi riêng biệt, có nhiều hạn chế hơn.

Kết hợp các mẫu: Server đa hệ thống

Các dự án thực tế thường cần truy cập vào nhiều hệ thống:

mcp = FastMCP("DevOps Assistant")

# Database tools
@mcp.tool()
async def query_production_db(sql: str) -> str: ...

# GitHub tools
@mcp.tool()
async def list_open_prs(repo: str) -> str: ...

# File system tools
@mcp.tool()
def read_config(filename: str) -> str: ...

# Context resources
@mcp.resource("project://architecture")
def get_architecture_docs() -> str: ...

Ngoài ra, bạn cũng có thể chạy các MCP server riêng biệt cho mỗi hệ thống và kết nối tất cả chúng với AI client của bạn. Cả hai cách tiếp cận đều hiệu quả — server riêng lẻ đơn giản hơn, nhiều server có tính mô-đun cao hơn.

Bài tập thực hành

Xây dựng một trong những MCP server kiểu sản xuất sau:

1. SQLite explorer — Truy vấn chỉ đọc đối với cơ sở dữ liệu SQLite cục bộ, với schema là một réource
2. REST API wrapper — Kết nối với bất kỳ API công khai nào (thời tiết, tin tức, API truyện cười) với khả năng xử lý lỗi phù hợp
3. Trình đọc file dự án — Điều hướng và đọc các file trong một thư mục dự án cụ thể với khả năng bảo vệ chống lại việc duyệt thư mục

Những điểm chính cần ghi nhớ

• Máy chủ cơ sở dữ liệu: mặc định là chỉ đọc, hiển thị lược đồ dưới dạng Resource, giới hạn số hàng kết quả
• API Server: sử dụng biến môi trường cho các bí mật, xử lý thời gian chờ và giới hạn tốc độ
• File server: luôn giới hạn trong một thư mục được cho phép, giải quyết đường dẫn để ngăn chặn việc duyệt thư mục
• Đối với tất cả các mẫu: xác thực đầu vào, trả về lỗi rõ ràng, sử dụng bất đồng bộ cho I/O
• Server đa hệ thống kết hợp các mẫu — trong một server hoặc nhiều server được kết nối