Posted inAI

Xây dựng MCP Server tùy chỉnh bằng Python: Chuẩn hóa tích hợp công cụ và dữ liệu cho Claude Desktop cùng AI Agent

AI tutorial - IT technology blog
AI tutorial - IT technology blog

Bối cảnh & Lý do

Vào lúc 2:15 sáng thứ Ba, các cảnh báo từ PagerDuty đồng loạt kích hoạt trên hai availability zone. Cụm xác thực (authentication cluster) của chúng tôi bị đình trệ hoàn toàn. Đồng nghiệp của tôi phải xoay xở với bốn tab terminal riêng biệt và dán các đoạn JSON chưa định dạng vào Claude để tìm ra microservice nào đã ném ra lỗi 500 đầu tiên. Mỗi service nội bộ lúc đó đều phụ thuộc vào script bọc (wrapper script) tự chế riêng, định nghĩa schema tùy biến cho OpenAI và các đoạn mã kết nối chắp vá, dễ gãy vỡ. Bất kỳ thay đổi nhỏ nào ở API cũng làm gián đoạn hoàn toàn quy trình xử lý sự cố (triage).

Việc xây dựng các cổng kết nối API riêng lẻ cho từng AI client rõ ràng không thể mở rộng lâu dài. Anthropic đã giới thiệu Model Context Protocol (MCP) nhằm giải quyết bài toán kiến trúc lộn xộn này. Hãy hình dung MCP giống như Language Server Protocol (LSP) dành cho các mô hình ngôn ngữ lớn (LLM). Thay vì phải triển khai lại các tích hợp trên nhiều công cụ khác nhau, bạn chỉ cần xây dựng một giao diện chuẩn hóa duy nhất. Các client có thể tự khám phá tính năng (capabilities), truy vấn tài nguyên ngữ cảnh và thực thi các hành động thông qua standard input/output (stdio) hoặc Server-Sent Events (SSE).

Việc chuẩn hóa theo MCP đã giúp rút ngắn thời gian khắc phục trung bình (MTTR) khi phân tích log sự cố của đội ngũ chúng tôi từ 24 phút xuống dưới 6 phút. Một khi các service đã hỗ trợ MCP, bất kỳ host tương thích nào—cho dù là Claude Desktop, Cursor hay một AI agent tự trị tự phát triển—đều có thể truy vấn các chỉ số vận hành một cách an toàn, kiểm tra trạng thái database và kích hoạt các bước xử lý sự cố có kiểm soát.

Cài đặt

Hãy luôn cô lập các dependency của server để tránh xung đột package với các công cụ hệ thống. Chúng ta sẽ sử dụng Python 3.10 trở lên.

1. Thiết lập môi trường ảo (Virtual Environment)

mkdir mcp-incident-server
cd mcp-incident-server
python3 -m venv .venv
source .venv/bin/activate

2. Cài đặt MCP SDK chính thức

Cài đặt Python SDK chính thức của Anthropic cùng với httpx cho các lệnh gọi HTTP bất đồng bộ và pydantic để xác thực schema chặt chẽ:

pip install --upgrade "mcp[cli]" httpx pydantic

3. Viết Python MCP Server

Tạo một file có tên là server.py. Chúng ta sẽ sử dụng module cấp cao FastMCP của SDK để cung cấp hai tool vận hành và một resource theo dõi chỉ số theo thời gian thực:

import sys
from datetime import datetime, timezone
from typing import Any, Dict, List
from mcp.server.fastmcp import FastMCP

# Khởi tạo instance cho FastMCP server
mcp = FastMCP("production-ops-server")

# Kho dữ liệu vận hành giả lập (mock)
INCIDENT_LOGS = [
    {
        "id": "INC-101",
        "service": "auth-api",
        "level": "CRITICAL",
        "message": "Hết kết nối PostgreSQL connection pool (500/500 kết nối đang hoạt động)",
        "timestamp": "2026-10-03T01:58:12Z",
    },
    {
        "id": "INC-102",
        "service": "payment-gw",
        "level": "WARNING",
        "message": "Độ trễ Stripe webhook gateway tăng cao lên 1420ms (ngưỡng p99: 400ms)",
        "timestamp": "2026-10-03T02:05:44Z",
    },
]

@mcp.tool()
def fetch_incident_logs(service: str = "", severity: str = "") -> List[Dict[str, Any]]:
    """Truy vấn log sự cố đang hoạt động được lọc theo tên service hoặc mức độ nghiêm trọng."""
    results = INCIDENT_LOGS
    if service:
        results = [log for log in results if log["service"].lower() == service.lower()]
    if severity:
        results = [log for log in results if log["level"].lower() == severity.lower()]
    return results

@mcp.tool()
def restart_microservice_pod(service_name: str, force: bool = False) -> str:
    """Kích hoạt rolling restart cho Kubernetes pod gặp sự cố trong thời gian gián đoạn."""
    valid_services = ["auth-api", "payment-gw", "order-processor"]
    if service_name not in valid_services:
        return f"Lỗi: Không tìm thấy service '{service_name}'. Danh sách hợp lệ: {valid_services}"
    
    action = "Bắt buộc dừng ngay lập tức (forced termination)" if force else "Khởi động lại tuần tự (graceful rolling restart)"
    now = datetime.now(timezone.utc).isoformat()
    return f"[{now}] Đã kích hoạt {action} cho target pod '{service_name}' (status: 202 Accepted)."

@mcp.resource("system://metrics/summary")
def get_system_metrics() -> str:
    """Cung cấp dữ liệu đo từ xa (telemetry) tài nguyên cluster dưới dạng context resource có cấu trúc."""
    return "Tình trạng Cluster: DEGRADED | CPU: 78.4% | Memory: 84.1% | Kết nối DB hoạt động: 492/500"

if __name__ == "__main__":
    # Lắng nghe yêu cầu từ client qua standard I/O
    mcp.run(transport="stdio")

Cấu hình

Sau khi đã hoàn thiện server, chúng ta cần đăng ký nó với AI host client. Claude Desktop là môi trường tiêu chuẩn cho việc thử nghiệm cục bộ cũng như vận hành hàng ngày.

1. Cấu hình Claude Desktop

Mở file cấu hình tương ứng trên hệ điều hành của bạn:

  • macOS: ~/Library/Application Support/Claude/claude_desktop_config.json
  • Windows: %APPDATA%\Claude\claude_desktop_config.json
  • Linux: ~/.config/Claude/claude_desktop_config.json

Đăng ký server của bạn trong phần mcpServers. Hãy luôn cung cấp đường dẫn tuyệt đối đến file thực thi Python bên trong môi trường ảo, tránh dùng Python của hệ thống:

{
  "mcpServers": {
    "ops-debugger": {
      "command": "/home/ubuntu/mcp-incident-server/.venv/bin/python",
      "args": [
        "/home/ubuntu/mcp-incident-server/server.py"
      ],
      "env": {
        "LOG_LEVEL": "INFO",
        "DB_TIMEOUT_MS": "5000"
      }
    }
  }
}

2. Bảo mật và cô lập I/O

Tránh nhúng mật khẩu database hoặc API key trực tiếp vào mã nguồn Python. Thay vào đó, hãy truyền chúng qua khối env hoặc biến môi trường của hệ điều hành.

Tuyệt đối không in thông điệp gỡ lỗi trực tiếp bằng print() ở chế độ stdio. FastMCP sử dụng stdout duy nhất cho việc tuần tự hóa các bản tin JSON-RPC. Việc ghi text thô vào stdout sẽ làm hỏng cấu trúc gói tin và ngắt kết nối với client. Thay vào đó, hãy chuyển toàn bộ log chẩn đoán nội bộ sang sys.stderr.

Kiểm thử & Giám sát

Không bao giờ triển khai các công cụ hạ tầng một cách mù quáng. Hãy kiểm tra server cục bộ bằng công cụ test harness chính thức trước khi kết nối với LLM thực tế.

1. Kiểm thử với MCP Inspector

Anthropic cung cấp MCP Inspector, một giao diện web trực quan giúp mô phỏng quá trình bắt tay client-server, kiểm tra schema và bắt các gói tin JSON-RPC thô. Chạy công cụ bằng npx:

npx @modelcontextprotocol/inspector /home/ubuntu/mcp-incident-server/.venv/bin/python /home/ubuntu/mcp-incident-server/server.py

Mở URL cục bộ được cung cấp (thường là http://localhost:5173). Ở thanh bên trái, chọn fetch_incident_logs, nhập {"service": "auth-api"} vào phần tham số, sau đó nhấp vào Run Tool. Xác nhận rằng công cụ trả về bản ghi INC-101 với phản hồi dạng 200 thành công.

2. Kiểm thử trực tiếp trên Client

Thoát hoàn toàn và khởi động lại Claude Desktop. Tìm biểu tượng chiếc búa ở góc dưới bên phải thanh nhập tin nhắn. Bạn sẽ thấy ops-debugger đang hoạt động, hiển thị hai tool (fetch_incident_logs, restart_microservice_pod) và resource chỉ số đo lường của bạn.

Thử nghiệm toàn bộ quy trình với câu lệnh sau: "Kiểm tra log sự cố đang hoạt động của auth-api và cho tôi biết liệu chúng ta có cần restart pod không."

Claude sẽ phân tích schema, yêu cầu cấp quyền thực thi fetch_incident_logs, đánh giá điểm nghẽn kết nối database và đưa ra đề xuất thực hiện rolling restart.

3. Khắc phục các lỗi Runtime phổ biến

  • Lỗi JSON-RPC Parse Error: Do các lệnh print thừa hoặc traceback chưa được xử lý ghi vào stdout. Hãy chuyển hướng toàn bộ logging tùy chỉnh sang stderr thông qua logging.basicConfig(stream=sys.stderr).
  • Subprocess Exit Code 1 / File Not Found: Đường dẫn binary khai báo trong claude_desktop_config.json không chính xác. Chạy lệnh which python trong môi trường ảo .venv đã kích hoạt và dán chính xác đường dẫn đó vào file cấu hình.
  • Client Timeout: Các truy vấn SQL hoặc lệnh gọi API kéo dài làm nghẽn luồng xử lý worker. Hãy bọc các tác vụ I/O bên ngoài trong asyncio.wait_for(..., timeout=5.0) để xử lý lỗi dứt điểm trước khi client bị ngắt kết nối.
Share:
Tags: