Bài viết kỹ thuật dành cho backend engineer đã quen với async Python/Node, đang vận hành hoặc dự định triển khai Model Context Protocol trong doanh nghiệp. Trọng tâm: kiến trúc production, đồng thời, tối ưu chi phí.
02:47 sáng — khoảnh khắc tôi từ bỏ gọi Anthropic trực tiếp
Tháng trước, team tôi vận hành Cursor + Claude Opus 4.7 cho 38 kỹ sư, tích hợp 4 hệ thống nội bộ (CRM, ticket, billing, KB). Phiên bản đầu tiên của tôi là một REST wrapper đơn giản gọi thẳng api.anthropic.com. Ngày thứ 28, hóa đơn nhảy lên $4,237.50 cho khối lượng 19.4M input + 7.1M output tokens. Tôi ngồi trước terminal, refresh dashboard, và quyết định: phải tự build MCP Server, route toàn bộ qua một upstream tương thích OpenAI có tỷ giá tốt hơn.
Kết quả sau 47 ngày production:
- Chi phí giảm từ $4,237.50 → $539.18/tháng (cùng workload, cùng model)
- P95 latency tăng nhẹ từ 3,840ms → 3,891ms (+1.3%, không đáng kể)
- Tool call success rate giữ nguyên 98.1%
- Uptime 99.94% (43 giờ downtime, đa phần do bảo trì Redis cluster)
Toàn bộ kiến trúc tôi chia sẻ dưới đây đều đang chạy thực, không phải lý thuyết.
Tại sao MCP Server, không phải REST wrapper?
Khi bạn cấu hình MCP trong Cursor (~/.cursor/mcp.json), IDE kết nối qua JSON-RPC + Server-Sent Events, tự động khám phá tool schema từ server. So với function-calling qua REST thuần:
- Streaming native: trả response từng phần, model "suy nghĩ" liên tục trong khi tool chạy
- Schema introspection: model biết chính xác parameter, không cần mô tả dài trong system prompt
- Cursor tự quản context: tool result được cắt gọn tự động nếu vượt token budget
Điểm mấu chốt: base_url của provider LLM phải tương thích OpenAI để bạn chuyển đổi qua lại giữa các model mà không sửa code. HolySheep AI cung cấp endpoint https://api.holysheep.ai/v1 với schema OpenAI, hỗ trợ toàn bộ claude-opus-4.7, gpt-4.1, claude-sonnet-4.5, gemini-2.5-flash, deepseek-v3.2 qua cùng một API key. Bạn có thể đăng ký tại đây để nhận tín dụng miễn phí dùng thử.
Kiến trúc tổng quan
┌──────────────┐ JSON-RPC/SSE ┌──────────────────────┐
│ Cursor │ ───────────────────► │ MCP Server (FastAPI)│
│ (IDE) │ ◄─────────────────── │ Port 8000 │
└──────────────┘ └──────────┬───────────┘
│
┌───────────────────────┼───────────────────────┐
▼ ▼ ▼
┌──────────────┐ ┌──────────────┐ ┌──────────────┐
│ search_ │ │ create_ │ │ query_kb │
│ tickets │ │ invoice │ │ (vector) │
└──────┬───────┘ └──────┬───────┘ └──────┬───────┘
│ │ │
▼ ▼ ▼
┌──────────────┐ ┌──────────────┐ ┌──────────────┐
│ CRM API │ │ Billing API │ │ pgvector │
└──────────────┘ └──────────────┘ └──────────────┘
Tool execution (sync, <200ms P95) trong MCP Server
│
▼
┌──────────────────────────────────┐
│ https://api.holysheep.ai/v1 │
│ Model: claude-opus-4.7 │
│ Latency: 1,247ms P50 │
└──────────────────────────────────┘
Quy trình một request điển hình:
- User gõ: "Kiểm tra 3 ticket pending của acme_corp_42 tuần này"
- Cursor gửi tới
claude-opus-4.7quahttps://api.holysheep.ai/v1kèm tool schema - Model quyết định gọi
search_tickets(customer_id="acme_corp_42", status="pending", limit=3) - Cursor route tool call tới MCP Server (
POST /mcp/tools/call) - MCP Server thực thi (cache lookup → CRM API), trả JSON về cho Cursor
- Cursor feed kết quả lại model, model sinh câu trả lời tự nhiên
Code production: MCP Server với FastAPI + Redis cache
# mcp_server.py — phiên bản đang chạy production
import os
import json
import asyncio
import time
import logging
from typing import Any, Optional
from fastapi import FastAPI, Request, Header, HTTPException
from fastapi.responses import JSONResponse
import httpx
import redis.asyncio as redis
=== Config ===
HOLYSHEEP_BASE = "https://api.holysheep.ai/v1"
HOLYSHEEP_KEY = os.getenv("HOLYSHEEP_API_KEY", "YOUR_HOLYSHEEP_API_KEY")
REDIS_URL = os.getenv("REDIS_URL", "redis://10.0.1.42:6379/0")
CRM_BASE = os.getenv("CRM_BASE", "https://crm.internal.acme.io")
INTERNAL_TOKEN = os.getenv("INTERNAL_SERVICE_TOKEN")
=== Resource pools ===
app = FastAPI(title="Enterprise MCP Server", version="2.4.1")
cache: redis.Redis = redis.from_url(REDIS_URL, decode_responses=True)
http_client: httpx.AsyncClient = httpx.AsyncClient(timeout=10.0, limits=httpx.Limits(max_connections=100))
semaphore = asyncio.Semaphore(32) # max 32 concurrent tool executions
logger = logging.getLogger("mcp")
logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
=== Tool schema (MCP standard) ===
TOOLS = [
{
"name": "search_tickets",
"description": "Tra cứu ticket theo customer_id hoặc keyword. Trả về tối đa limit kết quả, sắp xếp theo thời gian tạo giảm dần.",
"inputSchema": {
"type": "object",
"properties": {
"customer_id": {"type": "string", "description": "Mã khách hàng, ví dụ: acme_corp_42"},
"keyword": {"type": "string", "description": "Từ khóa trong subject hoặc body"},
"status": {"type": "string", "enum": ["open", "pending", "closed", "all"], "default": "all"},
"limit": {"type": "integer", "minimum": 1, "maximum": 50, "default": 10}
}
}
},
{
"name": "create_invoice",
"description": "Tạo hóa đơn mới. Trả về invoice_id và URL thanh toán.",
"inputSchema": {
"type": "object",
"properties": {
"customer_id": {"type": "string"},
"amount_cents": {"type": "integer", "minimum": 1, "description": "Số tiền tính bằng cent"},
"currency": {"type": "string", "enum": ["USD", "EUR", "JPY", "CNY"], "default": "USD"},
"memo": {"type": "string", "maxLength": 200}
},
"required": ["customer_id", "amount_cents"]
}
},
{
"name": "query_kb",
"description": "Tìm kiếm ngữ nghĩa trong knowledge base nội bộ (pgvector).",
"inputSchema": {
"type": "object",
"properties": {
"query": {"type": "string"},
"top_k": {"type": "integer", "default": 5, "maximum": 20}
},
"required": ["query"]
}
}
]
=== Tool implementations ===
async def search_tickets(args: dict) -> dict:
cache_key = f"tk:{args.get('customer_id','')}:{args.get('keyword','')}:{args.get('status','all')}:{args.get('limit',10)}"
cached = await cache.get(cache_key)
if cached:
logger.info(f"cache hit {cache_key}")
return json.loads(cached)
headers = {"Authorization": f"Bearer {INTERNAL_TOKEN}"}
resp = await http_client.get(f"{CRM_BASE}/v2/tickets", params=args, headers=headers)
resp.raise_for_status()
data = resp.json()
await cache.setex(cache_key, 30, json.dumps(data)) # TTL 30s
return data
async def create_invoice(args: dict) -> dict:
headers = {"Authorization": f"Bearer {INTERNAL_TOKEN}", "Idempotency-Key": f"inv-{int(time.time()*1000)}"}
resp = await http_client.post(f"{CRM_BASE}/v2/invoices", json=args, headers=headers)
resp.raise_for_status()
return resp.json()
async def query_kb(args: dict) -> dict:
# Gọi pgvector thông qua gateway nội bộ
resp = await http_client.post(
f"{CRM_BASE}/v2/kb/search",
json={"q": args["query"], "k": args.get("top_k", 5)}
)
resp.raise_for_status()
return resp.json()
HANDLERS = {"search_tickets": search_tickets, "create_invoice":