Phân tích chuyên sâu về các công cụ MCP: Những hàm mà AI có thể gọi

Công cụ là những thành phần cơ bản bạn sẽ sử dụng nhiều nhất. Chúng là các hàm mà trợ lý AI có thể phát hiện và gọi trong quá trình hội thoại — cơ chế cốt lõi để cung cấp cho AI những khả năng thực tế.

Trong bài học trước, bạn đã xây dựng các công cụ đơn giản. Bây giờ bạn sẽ học cách xây dựng các công cụ chất lượng sản xuất với xác thực đầu vào, xử lý lỗi đúng cách, những thao tác bất đồng bộ và các mẫu hoạt động đáng tin cậy ở quy mô lớn.

🔄 Tóm tắt nhanh: Trong bài học trước, bạn đã xây dựng MCP server đầu tiên của mình với @mcp.tool() và kết nối nó với Claude Desktop. Bây giờ bạn sẽ học bộ công cụ đầy đủ để xây dựng các công cụ mạnh mẽ, sẵn sàng cho sản xuất.

Cách hoạt động của schema công cụ

Khi bạn định nghĩa một công cụ với gợi ý kiểu, SDK sẽ tự động tạo JSON Schema:

@mcp.tool()
def search_users(
    query: str,
    limit: int = 10,
    active_only: bool = True
) -> str:
    """Search for users by name or email."""
    ...

📍 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 trò chuyện 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.

SDK tạo ra schema này (mà AI client nhìn thấy trong quá trình khởi tạo):

✏️ Cách điền thông tin chi tiết của bạn: Thay thế mỗi `[]` 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ẽ dẫn đến kết quả không rõ ràng — hãy cụ thể hơn.

👀 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.

📌 Cách xử lý kết quả: Lưu phản hồi vào file Notes. Chọn gợi ý có hiệu quả cao nhất và thực hiện nó trong tuần này — đừng cố gắng làm tất cả cùng một lúc.

⚠️ Nếu kết quả không ổn: Nếu các gợi ý có vẻ chung chung, hãy dán nội dung sau: "Hãy cụ thể hơn với ngữ cảnh thực tế của tôi. Bỏ qua những lời khuyên chung chung." Nếu nó 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 ngữ 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."

json
{
  "name": "search_users",
  "description": "Search for users by name or email.",
  "inputSchema": {
    "type": "object",
    "properties": {
      "query": { "type": "string" },
      "limit": { "type": "integer", "default": 10 },
      "active_only": { "type": "boolean", "default": true }
    },
    "required": ["query"]
  }
}

Lưu ý: Trường query là bắt buộc (không có giá trị mặc định), trong khi limit và active_only là tùy chọn (có giá trị mặc định). AI sử dụng schema này để xây dựng các lệnh gọi hợp lệ.

Mẹo về schema:

• Sử dụng tên tham số mô tả rõ ràng (user_email chứ không phải e)
• Đặt giá trị mặc định hợp lý cho các tham số tùy chọn
• Viết chuỗi tài liệu rõ ràng — đó là mô tả của công cụ cho AI

✅ Kiểm tra nhanh: Bạn định nghĩa một tham số công cụ là limit: int = 10. Trong schema được tạo, limit là bắt buộc hay tùy chọn?

Câu trả lời: Tùy chọn — các tham số có giá trị mặc định là tùy chọn trong schema. AI sẽ sử dụng giá trị mặc định là 10 trừ khi người dùng chỉ định một số khác.

Xác thực đầu vào

Không bao giờ tin tưởng rằng AI sẽ gửi đầu vào hợp lệ. Luôn xác thực:

@mcp.tool()
def get_user_profile(user_id: int) -> str:
    """Get a user's profile by their ID."""
    if user_id <= 0:
        return "Error: user_id must be a positive integer"

    # Look up user
    user = database.get_user(user_id)
    if user is None:
        return f"No user found with ID {user_id}"

    return f"Name: {user.name}\nEmail: {user.email}\nRole: {user.role}"

Các mẫu xác thực:

• Kiểm tra phạm vi cho số (if amount < 0)
• Kiểm tra định dạng cho chuỗi (if not email.contains("@"))
• Giới hạn độ dài (if len(query) > 500)
• Các giá trị được cho phép (if status not in ["active", "inactive"])

Trả về thông báo lỗi rõ ràng — AI sẽ đọc chúng và giải thích vấn đề cho người dùng.

Xử lý lỗi

Các ngoại lệ không được xử lý sẽ làm sập server của bạn. Luôn bắt lỗi và trả về các thông báo hữu ích:

@mcp.tool()
def fetch_stock_price(symbol: str) -> str:
    """Get the current stock price for a ticker symbol."""
    symbol = symbol.upper().strip()
    if not symbol.isalpha() or len(symbol) > 5:
        return f"Error: '{symbol}' is not a valid ticker symbol"

    try:
        price = stock_api.get_price(symbol)
        return f"{symbol}: ${price:.2f}"
    except ConnectionError:
        return "Error: Could not connect to the stock price service. Try again later."
    except Exception as e:
        return f"Error fetching price for {symbol}: {str(e)}"

Các quy tắc xử lý lỗi:

1. Xác thực đầu vào trước khi thực hiện bất kỳ thao tác nào
2. Nắm bắt các ngoại lệ cụ thể trước, sau đó là các ngoại lệ chung
3. Trả về thông báo lỗi dễ đọc (AI sẽ chuyển tiếp chúng)
4. Không bao giờ hiển thị dấu vết ngăn xếp hoặc chi tiết nội bộ cho người dùng
5. Ghi nhật ký lỗi chi tiết vào stderr để gỡ lỗi riêng của bạn

Các công cụ bất đồng bộ

Đối với các công cụ thực hiện yêu cầu mạng, truy vấn cơ sở dữ liệu hoặc đọc file — hãy sử dụng async:

import httpx

@mcp.tool()
async def get_weather(city: str) -> str:
    """Get current weather for a city."""
    async with httpx.AsyncClient() as client:
        response = await client.get(
            f"https://api.weather.example.com/current",
            params={"city": city}
        )
        data = response.json()
        return f"{city}: {data['temp']}°C, {data['condition']}"

Từ khóa async cho server biết công cụ này có thể nhường quyền điều khiển trong các thao tác I/O, ngăn server bị treo trong khi chờ phản hồi.

Khi nào nên sử dụng bất đồng bộ:

• Gọi API (yêu cầu HTTP)
• Truy vấn cơ sở dữ liệu
• Đọc/ghi file
• Bất kỳ thao tác nào liên quan đến việc chờ đợi

Khi nào nên sử dụng đồng bộ:

• Tính toán thuần túy (toán học, thao tác chuỗi)
• Tra cứu dữ liệu trong bộ nhớ
• Định dạng đơn giản

✅ Kiểm tra nhanh: MCP server của bạn có một công cụ đồng bộ gọi API bên ngoài chậm (mất 5 giây). Điều gì xảy ra nếu AI gọi công cụ này?

Câu trả lời: Server bị chặn trong 5 giây — không thể xử lý bất kỳ yêu cầu nào khác trong thời gian đó. Việc biến công cụ thành bất đồng bộ sẽ cho phép server xử lý các yêu cầu khác trong khi chờ phản hồi từ API.

Trả về rich content

Các công cụ có thể trả về nhiều hơn plain text. MCP hỗ trợ nhiều loại nội dung:

from mcp.types import TextContent, ImageContent

@mcp.tool()
def analyze_data(dataset: str) -> list:
    """Analyze a dataset and return results with a chart."""
    # ... perform analysis ...

    return [
        TextContent(
            type="text",
            text="Analysis complete:\n- Mean: 42.5\n- Median: 38.0"
        ),
        ImageContent(
            type="image",
            data=base64_chart_data,
            mimeType="image/png"
        )
    ]

Đối với hầu hết các công cụ, việc trả về một chuỗi plain text là đủ — SDK sẽ tự động đóng gói nó trong một đối tượng TextContent. Sử dụng các loại nội dung rõ ràng khi bạn cần trả về hình ảnh, dữ liệu có cấu trúc hoặc nhiều phần nội dung.

Các mẫu thiết kế công cụ

Mẫu Wrapper

Bao bọc các API hoặc thư viện hiện có dưới dạng công cụ MCP:

import subprocess

@mcp.tool()
def run_git_log(repo_path: str, count: int = 5) -> str:
    """Show recent git commits for a repository."""
    result = subprocess.run(
        ["git", "log", f"--oneline", f"-{count}"],
        cwd=repo_path,
        capture_output=True,
        text=True
    )
    if result.returncode != 0:
        return f"Error: {result.stderr}"
    return result.stdout

Mẫu Aggregator

Kết hợp nhiều nguồn dữ liệu thành một phản hồi công cụ:

@mcp.tool()
async def project_status(project_id: str) -> str:
    """Get a complete project status overview."""
    tasks = await get_tasks(project_id)
    budget = await get_budget(project_id)
    team = await get_team_members(project_id)

    open_tasks = sum(1 for t in tasks if t.status == "open")
    return (
        f"Project: {project_id}\n"
        f"Tasks: {open_tasks} open / {len(tasks)} total\n"
        f"Budget: ${budget.spent:,.0f} / ${budget.total:,.0f}\n"
        f"Team: {len(team)} members"
    )

Mẫu Guard

Thêm kiểm tra an toàn trước khi thực hiện hành động:

@mcp.tool()
def delete_file(filepath: str) -> str:
    """Delete a file from the allowed directory."""
    allowed_dir = "/data/temp/"

    # Safety: resolve the path and check it's within allowed directory
    resolved = os.path.realpath(filepath)
    if not resolved.startswith(os.path.realpath(allowed_dir)):
        return "Error: Can only delete files in /data/temp/"

    if not os.path.exists(resolved):
        return f"File not found: {filepath}"

    os.remove(resolved)
    return f"Deleted: {filepath}"

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

Xây dựng một MCP server với ba công cụ này:

1. lookup_word — Nhận một từ và trả về định nghĩa của nó (sử dụng API từ điển miễn phí)
2. translate_text — Nhận văn bản và ngôn ngữ đích, trả về văn bản đã dịch (giả lập phản hồi nếu không có API key)
3. summarize_url — Nhận một URL, lấy nội dung trang và trả về bản tóm tắt ngắn gọn

Bao gồm xác thực đầu vào và xử lý lỗi cho cả ba. Kết nối với Claude Desktop và kiểm tra từng công cụ.

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

• Gợi ý kiểu dữ liệu tự động tạo schema công cụ — luôn bao gồm chúng với tên rõ ràng
• Xác thực tất cả đầu vào trước khi xử lý — không bao giờ tin tưởng rằng AI gửi dữ liệu sạch
• Bắt lỗi và trả về thông báo mô tả thay vì gây lỗi
• Sử dụng bất đồng bộ cho bất kỳ hoạt động I/O nào (gọi API, truy vấn cơ sở dữ liệu, truy cập file)
• Mô tả công cụ tốt sẽ xác định xem AI có sử dụng công cụ của bạn một cách chính xác hay không
• Tuân theo các mẫu: Wrapper (bao bọc code hiện có), Aggregator (kết hợp các nguồn), Guard (kiểm tra an toàn)