Mở Đầu: Câu Chuyện Thực Tế Từ Hệ Thống Chăm Sóc Khách Hàng AI Thương Mại Điện Tử
Tôi vẫn nhớ rõ đêm mùng 3 Tết năm nay — hệ thống chăm sóc khách hàng AI của một trong những sàn thương mại điện tử lớn nhất Việt Nam bất ngờ chậm như rùa bò. Hàng nghìn khách hàng đang truy vấn về đơn hàng, tình trạng giao hàng, và chính sách đổi trả — nhưng mọi thứ đều timeout. Trong 45 phút đầu tiên, đội kỹ thuật của tôi gần như mò mẫm trong bóng tối vì không có log chi tiết, không biết request bị chặn ở đâu, và latency tăng từ 80ms lên hơn 8 giây. Đó là khoảnh khắc tôi quyết định đầu tư toàn bộ thời gian để xây dựng một hệ thống MCP Tool debugging chuẩn chỉnh. Kết quả? 6 tháng sau, đội của tôi xử lý sự cố trung bình trong 3 phút thay vì 45 phút như trước.
Bài viết này sẽ chia sẻ toàn bộ kinh nghiệm thực chiến về cách debug MCP Tool một cách hiệu quả, từ những lỗi đơn giản nhất cho đến những vấn đề phức tạp chỉ xuất hiện dưới tải cao. Tôi sẽ hướng dẫn bạn từng bước với code mẫu có thể sao chép và chạy ngay, đồng thời tích hợp với HolySheep AI — nền tảng mà tôi sử dụng cho hầu hết các dự án vì chi phí chỉ bằng 15% so với các provider phương Tây.
MCP Tool Là Gì và Tại Sao Cần Debugging?
MCP (Model Context Protocol) Tool là cách để các mô hình AI tương tác với external tools và services. Khi bạn build một hệ thống RAG cho doanh nghiệp hoặc một ứng dụng AI agent phức tạp, MCP Tool đóng vai trò như cầu nối giữa LLM và thế giới thực — truy vấn database, gọi API bên thứ ba, thao tác file system, hoặc kết nối với các dịch vụ như Slack, Notion, GitHub.
Vấn đề là khi mọi thứ hoạt động tốt ở local nhưng fail khi deploy lên production, hoặc khi latency tăng đột ngột vào giờ cao điểm, bạn cần một phương pháp systematic để trace và debug. Đây là lúc MCP Tool debugging trở nên quan trọng.
Cài Đặt Môi Trường Debug MCP Tool
Trước khi đi vào các kỹ thuật debugging, chúng ta cần setup môi trường phù hợp. Tôi khuyên dùng một project riêng biệt để debug thay vì debug trực tiếp trong production code.
# Tạo virtual environment cho MCP debugging
python -m venv mcp-debug-env
source mcp-debug-env/bin/activate # Linux/Mac
hoặc: mcp-debug-env\Scripts\activate # Windows
Cài đặt dependencies cần thiết
pip install anthropic openai httpx structlog python-dotenv
pip install mcp # MCP SDK nếu bạn tự build tools
Cài đặt logging để trace requests
pip install loguru rich
Verify installation
python -c "import mcp; print(f'MCP version: {mcp.__version__}')"Sau khi cài đặt xong, tạo file cấu hình môi trường với API key từ HolySheep AI — nơi tôi nhận được latency trung bình chỉ 45ms cho các request GPT-4o và chi phí chỉ $0.42/1M tokens cho DeepSeek V3.2.
# config.py - Cấu hình MCP Tool với HolySheep AI
import os
from dotenv import load_dotenv
load_dotenv() # Load .env file
=== MCP Tool Configuration ===
MCP_CONFIG = {
"base_url": "https://api.holysheep.ai/v1", # KHÔNG dùng api.openai.com
"api_key": os.getenv("YOUR_HOLYSHEEP_API_KEY"),
"model": "gpt-4o", # Hoặc deepseek-v3.2 cho chi phí thấp
"timeout": 30, # Timeout 30 giây cho mỗi request
"max_retries": 3,
}
=== Logging Configuration ===
LOGGING_CONFIG = {
"level": "DEBUG", # Bật DEBUG để xem chi tiết
"format": "{time:YYYY-MM-DD HH:mm:ss.SSS} | {level: <8} | {name}:{function}:{line} - {message}",
"rotation": "100 MB",
"retention": "7 days",
"output": "./logs/mcp_debug.log",
}
=== Tool Registry ===
AVAILABLE_TOOLS = {
"search_database": {
"endpoint": "/tools/search",
"timeout": 10,
"retry_on_failure": True,
},
"get_customer_info": {
"endpoint": "/tools/customer",
"timeout": 5,
"retry_on_failure": True,
},
"update_order_status": {
"endpoint": "/tools/order",
"timeout": 15,
"retry_on_failure": False, # Không retry cho operations ghi dữ liệu
}
}
print("✅ MCP Configuration loaded successfully")
print(f" Base URL: {MCP_CONFIG['base_url']}")
print(f" Model: {MCP_CONFIG['model']}")Kỹ Thuật Log Tracing Cho MCP Tool
Khi debug MCP Tool, log là vũ khí quan trọng nhất của bạn. Tôi đã xây dựng một logging wrapper tùy chỉnh giúp track mọi request/response một cách chi tiết, bao gồm cả token usage và latency thực tế.
# mcp_debug_logger.py - Advanced logging cho MCP Tool
import time
import structlog
import httpx
from datetime import datetime
from typing import Any, Dict, Optional
from config import MCP_CONFIG, LOGGING_CONFIG
Configure structlog cho structured logging
structlog.configure(
processors=[
structlog.stdlib.filter_by_level,
structlog.stdlib.add_logger_name,
structlog.stdlib.add_log_level,
structlog.processors.TimeStamper(fmt="iso"),
structlog.processors.StackInfoRenderer(),
structlog.processors.format_exc_info,
structlog.processors.JSONRenderer()
],
wrapper_class=structlog.stdlib.BoundLogger,
context_class=dict,
logger_factory=structlog.stdlib.LoggerFactory(),
cache_logger_on_first_use=True,
)
logger = structlog.get_logger("mcp_debug")
class MCPDebugClient:
"""Enhanced MCP client với built-in debugging và tracing"""
def __init__(self, config: Dict[str, Any]):
self.base_url = config["base_url"]
self.api_key = config["api_key"]
self.model = config["model"]
self.timeout = config["timeout"]
self.max_retries = config["max_retries"]
# Metrics tracking
self.metrics = {
"total_requests": 0,
"failed_requests": 0,
"total_tokens": 0,
"latencies": [],
"errors_by_type": {}
}
# HTTP Client với connection pooling
self.client = httpx.Client(
base_url=self.base_url,
headers={
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
},
timeout=self.timeout,
limits=httpx.Limits(max_connections=100, max_keepalive_connections=20)
)
logger.info("mcp_debug_client_initialized",
base_url=self.base_url,
model=self.model)
def call_tool(self, tool_name: str, parameters: Dict[str, Any]) -> Dict[str, Any]:
"""
Gọi MCP tool với full tracing và error tracking
"""
start_time = time.perf_counter()
request_id = f"req_{datetime.now().strftime('%Y%m%d_%H%M%S_%f')}"
logger.info("mcp_tool_call_started",
request_id=request_id,
tool_name=tool_name,
parameters=parameters)
try:
# Gọi API
response = self._execute_with_retry(tool_name, parameters)
# Calculate metrics
latency_ms = (time.perf_counter() - start_time) * 1000
self._update_metrics("success", latency_ms, response)
logger.info("mcp_tool_call_completed",
request_id=request_id,
tool_name=tool_name,
latency_ms=round(latency_ms, 2),
status="success")
return {
"success": True,
"data": response,
"latency_ms": latency_ms,
"request_id": request_id
}
except httpx.TimeoutException as e:
latency_ms = (time.perf_counter() - start_time) * 1000
self._update_metrics("timeout", latency_ms, None)
logger.error("mcp_tool_timeout",
request_id=request_id,
tool_name=tool_name,
latency_ms=round(latency_ms, 2),
error=str(e))
return {
"success": False,
"error": "timeout",
"message": f"Request timeout sau {self.timeout}s",
"latency_ms": latency_ms,
"request_id": request_id
}
except httpx.HTTPStatusError as e:
latency_ms = (time.perf_counter() - start_time) * 1000
self._update_metrics("http_error", latency_ms, None)
logger.error("mcp_tool_http_error",
request_id=request_id,
tool_name=tool_name,
status_code=e.response.status_code,
response_body=e.response.text[:500],
latency_ms=round(latency_ms, 2))
return {
"success": False,
"error": "http_error",
"status_code": e.response.status_code,
"message": e.response.text,
"latency_ms": latency_ms,
"request_id": request_id
}
except Exception as e:
latency_ms = (time.perf_counter() - start_time) * 1000
self._update_metrics("unknown_error", latency_ms, None)
logger.exception("mcp_tool_unknown_error",
request_id=request_id,
tool_name=tool_name,
error_type=type(e).__name__,
error_message=str(e))
return {
"success": False,
"error": type(e).__name__,
"message": str(e),
"latency_ms": latency_ms,
"request_id": request_id
}
def _execute_with_retry(self, tool_name: str, parameters: Dict) -> Dict:
"""Execute request với retry logic"""
for attempt in range(self.max_retries):
try:
response = self.client.post(
"/chat/completions",
json={
"model": self.model,
"messages": [
{"role": "user", "content": f"Execute tool: {tool_name} with params: {parameters}"}
],
"tools": [{"type": "function", "function": {"name": tool_name}}]
}
)
response.raise_for_status()
return response.json()
except httpx.HTTPStatusError as e:
if e.response.status_code >= 500 and attempt < self.max_retries - 1:
logger.warning("retry_attempt",
attempt=attempt + 1,
max_retries=self.max_retries,
status_code=e.response.status_code)
time.sleep(2 ** attempt) # Exponential backoff
continue
raise
raise httpx.HTTPStatusError("Max retries exceeded", request=None, response=None)
def _update_metrics(self, status: str, latency_ms: float, response: Optional[Dict]):
"""Update internal metrics"""
self.metrics["total_requests"] += 1
if status != "success":
self.metrics["failed_requests"] += 1
self.metrics["latencies"].append(latency_ms)
if status not in self.metrics["errors_by_type"]:
self.metrics["errors_by_type"][status] = 0
self.metrics["errors_by_type"][status] += 1
# Track token usage nếu có response
if response and "usage" in response:
usage = response["usage"]
self.metrics["total_tokens"] += usage.get("total_tokens", 0)
def get_metrics_summary(self) -> Dict[str, Any]:
"""Trả về summary của các metrics"""
latencies = self.metrics["latencies"]
return {
"total_requests": self.metrics["total_requests"],
"failed_requests": self.metrics["failed_requests"],
"success_rate": round(
(self.metrics["total_requests"] - self.metrics["failed_requests"])
/ self.metrics["total_requests"] * 100, 2
) if self.metrics["total_requests"] > 0 else 0,
"avg_latency_ms": round(sum(latencies) / len(latencies), 2) if latencies else 0,
"p50_latency_ms": round(sorted(latencies)[len(latencies)//2], 2) if latencies else 0,
"p95_latency_ms": round(sorted(latencies)[int(len(latencies)*0.95)], 2) if latencies else 0,
"p99_latency_ms": round(sorted(latencies)[int(len(latencies)*0.99)], 2) if latencies else 0,
"total_tokens": self.metrics["total_tokens"],
"errors_by_type": self.metrics["errors_by_type"]
}
def close(self):
self.client.close()
logger.info("mcp_client_closed", metrics=self.get_metrics_summary())
=== Demo Usage ===
if __name__ == "__main__":
# Initialize client
client = MCPDebugClient(MCP_CONFIG)
# Test với sample tool call
result = client.call_tool(
tool_name="search_database",
parameters={"query": "customer orders", "limit": 10}
)
print("\n" + "="*60)
print("METRICS SUMMARY")
print("="*60)
for key, value in client.get_metrics_summary().items():
print(f" {key}: {value}")
client.close()Các Chiến Lược Debugging MCP Tool Hiệu Quả
1. Request/Response Interception
Một trong những kỹ thuật quan trọng nhất là intercept tất cả requests và responses để xem chính xác dữ liệu đang được truyền qua lại. Điều này giúp bạn phát hiện các vấn đề như malformed JSON, thiếu fields, hoặc data type mismatch.
# mcp_interceptor.py - Request/Response Interception
import json
import base64
from functools import wraps
from typing import Callable, Any
import httpx
class MCPInterceptor:
"""
Interceptor để debug tất cả HTTP traffic
Rất hữu ích để phát hiện vấn đề về data format hoặc authentication
"""
def __init__(self):
self.request_log = []
self.response_log = []
self.error_log = []
def intercept_request(self, method: str, url: str, headers: dict, body: Any) -> dict:
"""Log request trước khi gửi"""
request_entry = {
"type": "request",
"method": method,
"url": url,
"timestamp": datetime.now().isoformat(),
"headers": self._sanitize_headers(headers),
"body": body
}
self.request_log.append(request_entry)
# Pretty print for debugging
print(f"\n🔵 REQUEST [{method}] {url}")
print(f" Headers: {json.dumps(request_entry['headers'], indent=2)}")
if body:
print(f" Body: {json.dumps(body, indent=2, ensure_ascii=False)}")
return request_entry
def intercept_response(self, response: httpx.Response, request_entry: dict) -> dict:
"""Log response sau khi nhận"""
response_entry = {
"type": "response",
"status_code": response.status_code,
"headers": dict(response.headers),
"body": response.text[:5000] if response.text else None, # Limit size
"timestamp": datetime.now().isoformat(),
"request_url": request_entry["url"]
}
self.response_log.append(response_entry)
# Pretty print
status_emoji = "✅" if response.status_code < 400 else "❌"
print(f"\n{status_emoji} RESPONSE [{response.status_code}] {request_entry['url']}")
print(f" Time: {response_entry['timestamp']}")
if response_entry['body']:
try:
parsed = json.loads(response_entry['body'])
print(f" Body: {json.dumps(parsed, indent=2, ensure_ascii=False)[:1000]}")
except:
print(f" Body: {response_entry['body'][:500]}")
return response_entry
def intercept_error(self, error: Exception, request_entry: dict):
"""Log errors"""
error_entry = {
"type": "error",
"error_type": type(error).__name__,
"error_message": str(error),
"timestamp": datetime.now().isoformat(),
"request_url": request_entry["url"]
}
self.error_log.append(error_entry)
print(f"\n💥 ERROR: {error_entry['error_type']}")
print(f" Message: {error_entry['error_message']}")
print(f" Request: {error_entry['request_url