Mở đầu: Câu chuyện thực chiến từ dịch vụ thương mại điện tử
Tôi vẫn nhớ rõ cách đây 8 tháng, đội ngũ kỹ thuật của một startup thương mại điện tử quy mô 50 người đã phải đối mặt với cơn ác mộng: 3.000 tin nhắn khách hàng mỗi ngày trên Discord, 8 nhân viên hỗ trợ làm việc 16 tiếng/ca, và tỷ lệ phản hồi chỉ đạt 67%. Sau 3 tuần triển khai hệ thống AI chatbot dựa trên MCP (Model Context Protocol) kết nối đồng thời Slack và Discord, đội ngũ giảm còn 2 người giám sát, thời gian phản hồi trung bình giảm từ 45 phút xuống còn 8 giây, và doanh thu tăng 23% nhờ khách hàng hài lòng hơn. Bài viết này sẽ hướng dẫn bạn xây dựng hệ thống tương tự từ con số 0.
MCP là gì và tại sao nó thay đổi cuộc chơi
Model Context Protocol (MCP) là giao thức chuẩn công nghiệp cho phép các mô hình AI tương tác với công cụ và nguồn dữ liệu bên ngoài một cách an toàn và có cấu trúc. Khác với việc sử dụng API thuần túy, MCP cung cấp:
Kiến trúc MCP cơ bản
┌─────────────┐ MCP Protocol ┌─────────────────┐
│ AI Model │◄────────────────────►│ Tool Server │
│ (Provider) │ │ (Slack/Discord) │
└─────────────┘ └─────────────────┘
│ │
│ ▼
│ ┌─────────────────┐
└─────────────────────────►│ Database/API │
│ (HolySheep AI) │
└─────────────────┘
Ưu điểm vượt trội của MCP so với webhook truyền thống bao gồm: standardized tool discovery (phát hiện công cụ tự động), type-safe schema validation (xác thực kiểu dữ liệu), streaming responses (phản hồi theo luồng), và built-in error recovery (phục hồi lỗi tự động). Điểm mấu chốt là MCP hoạt động như một abstraction layer, giúp bạn không cần quan tâm đến chi tiết implementation của từng nền tảng.
Thiết lập dự án và cài đặt môi trường
Trước khi bắt đầu, hãy đảm bảo bạn có môi trường Python 3.10+ và API key từ HolySheep AI. Với mức giá chỉ từ $0.42/MTok cho DeepSeek V3.2 và độ trễ trung bình dưới 50ms, HolySheep là lựa chọn tối ưu về chi phí cho production workload.
Cài đặt các thư viện cần thiết
pip install mcp slack-sdk discord.py aiohttp python-dotenv
Cấu trúc thư mục dự án
project/
├── mcp_server/
│ ├── __init__.py
│ ├── slack_connector.py
│ ├── discord_connector.py
│ └── tools.py
├── config/
│ └── settings.py
├── .env
└── main.py
File .env - KHÔNG BAO GIỜ commit file này lên git
HOLYSHEEP_API_KEY=YOUR_HOLYSHEEP_API_KEY
SLACK_BOT_TOKEN=xoxb-your-slack-token
SLACK_SIGNING_SECRET=your-signing-secret
DISCORD_BOT_TOKEN=your-discord-token
DISCORD_APPLICATION_ID=your-app-id
MCP_SERVER_PORT=8765
Triển khai MCP Server với HolySheep AI
Điểm khác biệt quan trọng khi sử dụng HolySheep thay vì OpenAI hay Anthropic là base_url phải là
https://api.holysheep.ai/v1. Dưới đây là implementation hoàn chỉnh:
config/settings.py
import os
from dotenv import load_dotenv
load_dotenv()
class Config:
# HolySheep AI Configuration - LUÔN sử dụng endpoint này
HOLYSHEEP_BASE_URL = "https://api.holysheep.ai/v1"
HOLYSHEEP_API_KEY = os.getenv("HOLYSHEEP_API_KEY")
# Pricing reference (2026): GPT-4.1 $8, Claude Sonnet 4.5 $15,
# Gemini 2.5 Flash $2.50, DeepSeek V3.2 $0.42/MTok
# Tỷ giá ¥1=$1 → Tiết kiệm 85%+ so với providers khác
# Slack Configuration
SLACK_BOT_TOKEN = os.getenv("SLACK_BOT_TOKEN")
SLACK_SIGNING_SECRET = os.getenv("SLACK_SIGNING_SECRET")
# Discord Configuration
DISCORD_BOT_TOKEN = os.getenv("DISCORD_BOT_TOKEN")
DISCORD_APPLICATION_ID = os.getenv("DISCORD_APPLICATION_ID")
@classmethod
def validate(cls):
required = ['HOLYSHEEP_API_KEY', 'SLACK_BOT_TOKEN', 'DISCORD_BOT_TOKEN']
missing = [k for k in required if not getattr(cls, k)]
if missing:
raise ValueError(f"Thiếu biến môi trường: {', '.join(missing)}")
return True
mcp_server/tools.py - Định nghĩa các MCP tools
from typing import List, Dict, Any, Optional
from mcp.types import Tool, ToolInputSchema
import aiohttp
import json
class MCPTools:
"""Danh sách tools được đăng ký với MCP server"""
TOOLS: List[Tool] = [
Tool(
name="send_slack_message",
description="Gửi tin nhắn đến một Slack channel hoặc user",
inputSchema=ToolInputSchema(
type="object",
properties={
"channel": {"type": "string", "description": "Channel ID hoặc tên (#general)"},
"text": {"type": "string", "description": "Nội dung tin nhắn"},
"thread_ts": {"type": "string", "description": "Thread timestamp để reply", "optional": True}
},
required=["channel", "text"]
)
),
Tool(
name="search_slack_history",
description="Tìm kiếm lịch sử tin nhắn trên Slack",
inputSchema=ToolInputSchema(
type="object",
properties={
"query": {"type": "string", "description": "Từ khóa tìm kiếm"},
"channel": {"type": "string", "description": "Channel cụ thể (tùy chọn)", "optional": True},
"limit": {"type": "integer", "description": "Số lượng kết quả", "default": 10}
},
required=["query"]
)
),
Tool(
name="send_discord_message",
description="Gửi tin nhắn đến một Discord channel",
inputSchema=ToolInputSchema(
type="object",
properties={
"channel_id": {"type": "string", "description": "Discord channel ID"},
"content": {"type": "string", "description": "Nội dung tin nhắn"},
"embed": {"type": "object", "description": "Discord embed object", "optional": True}
},
required=["channel_id", "content"]
)
),
Tool(
name="get_discord_thread",
description="Lấy thông tin thread trên Discord",
inputSchema=ToolInputSchema(
type="object",
properties={
"thread_id": {"type": "string", "description": "Thread ID"}
},
required=["thread_id"]
)
)
]
async def call_holysheep_api(messages: List[Dict], tools: List[Dict]) -> Dict[str, Any]:
"""
Gọi HolySheep AI API với streaming response
base_url: https://api.holysheep.ai/v1
"""
url = "https://api.holysheep.ai/v1/chat/completions"
headers = {
"Authorization": f"Bearer YOUR_HOLYSHEEP_API_KEY",
"Content-Type": "application/json"
}
payload = {
"model": "gpt-4.1", # Hoặc deepseek-v3.2, claude-sonnet-4.5
"messages": messages,
"tools": tools,
"stream": True
}
async with aiohttp.ClientSession() as session:
async with session.post(url, headers=headers, json=payload) as response:
if response.status != 200:
error_text = await response.text()
raise Exception(f"API Error {response.status}: {error_text}")
# Xử lý streaming response
chunks = []
async for line in response.content:
if line:
chunks.append(line.decode())
return {"status": "success", "chunks": chunks}
Kết nối Slack với MCP
Slack integration đòi hỏi xử lý events API và webhooks. Dưới đây là implementation đầy đủ:
mcp_server/slack_connector.py
import asyncio
import hmac
import hashlib
import time
from typing import Callable, Optional
from slack_sdk import WebClient
from slack_sdk.errors import SlackApiError
from slack_sdk.socket_mode import SocketModeClient
from slack_sdk.socket_mode.request import SocketModeRequest
from config.settings import Config
class SlackConnector:
def __init__(self, token: str, signing_secret: str):
self.client = WebClient(token=token)
self.signing_secret = signing_secret
self.socket_client: Optional[SocketModeClient] = None
self.message_handler: Optional[Callable] = None
def verify_signature(self, body: str, timestamp: str, signature: str) -> bool:
"""Xác thực request từ Slack webhook"""
if abs(time.time() - int(timestamp)) > 60 * 5:
return False
sig_basestring = f"v0:{timestamp}:{body}".encode()
my_signature = "v0=" + hmac.new(
self.signing_secret.encode(),
sig_basestring,
hashlib.sha256
).hexdigest()
return hmac.compare_digest(my_signature, signature)
async def send_message(self, channel: str, text: str, thread_ts: Optional[str] = None) -> dict:
"""Gửi tin nhắn đến Slack channel"""
try:
result = self.client.chat_postMessage(
channel=channel,
text=text,
thread_ts=thread_ts if thread_ts else None
)
return {
"success": True,
"ts": result["ts"],
"channel": result["channel"]
}
except SlackApiError as e:
return {
"success": False,
"error": str(e)
}
async def search_history(self, query: str, channel: Optional[str] = None, limit: int = 10) -> dict:
"""Tìm kiếm lịch sử Slack"""
try:
search_params = {
"query": query,
"count": limit,
"sort": "score",
"sort_dir": "desc"
}
if channel:
search_params["channel"] = channel
result = self.client.search_messages(**search_params)
messages = []
for msg in result.get("messages", {}).get("matches", []):
messages.append({
"text": msg["text"],
"user": msg.get("username", msg.get("user", "unknown")),
"channel": msg["channel"]["name"],
"ts": msg["ts"],
"permalink": msg["permalink"]
})
return {"success": True, "messages": messages, "total": len(messages)}
except SlackApiError as e:
return {"success": False, "error": str(e)}
async def handle_events(self, socket_mode_request: SocketModeRequest):
"""Xử lý Slack events thông qua Socket Mode"""
request_type = socket_mode_request.type
if request_type == "events_api":
# Xác nhận event
self.socket_client.send_socket_mode_response(socket_mode_request)
event = socket_mode_request.payload.get("event", {})
event_type = event.get("type")
if event_type == "message" and not event.get("subtype"):
# Tin nhắn mới từ user
channel_id = event.get("channel")
user_id = event.get("user")
text = event.get("text")
ts = event.get("ts")
# Gọi MCP handler nếu có
if self.message_handler:
await self.message_handler(
platform="slack",
channel=channel_id,
user=user_id,
text=text,
thread_ts=ts
)
async def start_socket_mode(self):
"""Khởi động Socket Mode client cho real-time events"""
self.socket_client = SocketModeClient(
app_token=Config.SLACK_APP_TOKEN,
web_client=self.client,
socket_mode_request_listeners=[self.handle_events]
)
await self.socket_client.connect()
print("✅ Slack Socket Mode đã kết nối")
Kết nối Discord với MCP
Discord sử dụng gateway API và interactions endpoint. Implementation chi tiết:
mcp_server/discord_connector.py
import discord
from discord import app_commands
from typing import Optional, List, Dict, Any
from config.settings import Config
class DiscordConnector(discord.Client):
"""Discord bot với MCP tool integration"""
def __init__(self, token: str):
intents = discord.Intents.default()
intents.message_content = True
intents.guilds = True
intents.members = True
super().__init__(intents=intents)
self.tree = app_commands.CommandTree(self)
self.message_handler: Optional[callable] = None
async def setup_hook(self):
"""Đăng ký slash commands với Discord"""
guild = discord.Object(id=Config.DISCORD_GUILD_ID)
@self.tree.command(name="ask", description="Hỏi AI assistant", guild=guild)
async def ask(interaction: discord.Interaction, question: str):
if self.message_handler:
response = await self.message_handler(
platform="discord",
user_id=str(interaction.user.id),
channel=str(interaction.channel_id),
text=question
)
await interaction.response.send_message(response)
else:
await interaction.response.send_message("⚠️ Bot đang khởi động, thử lại sau")
@self.tree.command(name="status", description="Kiểm tra trạng thái hệ thống", guild=guild)
async def status(interaction: discord.Interaction):
embed = discord.Embed(
title="🤖 System Status",
color=discord.Color.green()
)
embed.add_field(name="MCP Server", value="🟢 Online", inline=True)
embed.add_field(name="AI Model", value="🟢 Active", inline=True)
embed.add_field(name="Latency", value="<50ms", inline=True)
await interaction.response.send_message(embed=embed)
await self.tree.sync(guild=guild)
async def on_message(self, message: discord.Message):
"""Xử lý tin nhắn Discord"""
# Bỏ qua tin nhắn từ bot
if message.author.bot:
return
# Kiểm tra mention hoặc DM
if self.user and (self.user.mentioned_in(message) or isinstance(message.channel, discord.DMChannel)):
if self.message_handler:
response = await self.message_handler(
platform="discord",
user_id=str(message.author.id),
channel=str(message.channel_id),
text=message.content
)
# Discord có giới hạn 2000 ký tự
if len(response) > 2000:
for i in range(0, len(response), 2000):
await message.reply(response[i:i+2000])
else:
await message.reply(response)
async def send_message(self, channel_id: str, content: str, embed: Optional[Dict] = None) -> Dict[str, Any]:
"""Gửi tin nhắn đến Discord channel"""
try:
channel = self.get_channel(int(channel_id))
if not channel:
return {"success": False, "error": "Channel not found"}
if embed:
await channel.send(content, embed=discord.Embed.from_dict(embed))
else:
await channel.send(content)
return {"success": True, "channel_id": channel_id}
except Exception as e:
return {"success": False, "error": str(e)}
async def get_thread_info(self, thread_id: str) -> Dict[str, Any]:
"""Lấy thông tin Discord thread"""
try:
thread = self.get_channel(int(thread_id))
if not thread or not isinstance(thread, discord.Thread):
return {"success": False, "error": "Thread not found"}
return {
"success": True,
"name": thread.name,
"owner_id": thread.owner_id,
"member_count": thread.member_count,
"message_count": thread.message_count,
"created_at": thread.created_at.isoformat()
}
except Exception as e:
return {"success": False, "error": str(e)}
Khởi tạo Discord bot
discord_bot = DiscordConnector(Config.DISCORD_BOT_TOKEN)
Đồng bộ hóa MCP Server và AI Processing
Đây là phần cốt lõi xử lý request từ cả hai nền tảng và gọi HolySheep AI:
main.py - Entry point chính
import asyncio
import json
from typing import Dict, Any
from config.settings import Config
from mcp_server.slack_connector import SlackConnector
from mcp_server.discord_connector import discord_bot
from mcp_server.tools import MCPTools, call_holysheep_api
class MCPBridge:
"""
MCP Bridge - Điều phối request giữa Slack/Discord và AI
Sử dụng HolySheep AI cho inference
"""
def __init__(self):
self.slack = SlackConnector(
token=Config.SLACK_BOT_TOKEN,
signing_secret=Config.SLACK_SIGNING_SECRET
)
self.conversation_history: Dict[str, list] = {}
self.MAX_HISTORY = 20
async def handle_message(self, platform: str, user_id: str, channel: str, text: str) -> str:
"""Xử lý tin nhắn từ Slack hoặc Discord"""
# Tạo conversation ID duy nhất
conv_id = f"{platform}:{channel}"
# Khởi tạo history nếu chưa có
if conv_id not in self.conversation_history:
self.conversation_history[conv_id] = []
# Thêm system prompt
system_prompt = {
"role": "system",
"content": """Bạn là AI assistant cho hệ thống customer service.
Trả lời ngắn gọn, thân thiện, chuyên nghiệp.
Nếu cần thực hiện hành động, sử dụng MCP tools.
Luôn trả lời bằng tiếng Việt."""
}
# Thêm user message vào history
Tài nguyên liên quan
Bài viết liên quan