Model Context Protocol (MCP) กำลังเปลี่ยนแปลงวิธีที่ AI Agent ทำงานร่วมกับระบบภายนอกอย่างสิ้นเชิง บทความนี้จะพาคุณสร้าง MCP Server ที่รวม Filesystem, Database และ API Tools เข้าด้วยกันอย่างครบวงจร พร้อมสำหรับการ deploy ระดับ production
MCP คืออะไรและทำไมต้องสนใจ
MCP เป็น protocol มาตรฐานที่พัฒนาโดย Anthropic ช่วยให้ AI model สามารถเรียกใช้ tools ภายนอกได้อย่างเป็นมาตรฐาน ไม่ว่าจะเป็นการอ่านไฟล์, query database, หรือเรียก API ต่างๆ โดย MCP Server ที่ดีต้องมีคุณสมบัติดังนี้
- Type-safe: ทุก tool ต้องมี schema ชัดเจน
- Error handling: จัดการ exception อย่างครบถ้วน
- Resource management: ควบคุม concurrent connections
- Performance: รองรับ high-throughput scenarios
- Security: ป้องกัน injection attacks และ unauthorized access
สถาปัตยกรรม MCP Server แบบบูรณาการ
ก่อนเข้าสู่โค้ด มาดูภาพรวมสถาปัตยกรรมของระบบของเรากัน
┌─────────────────────────────────────────────────────────────┐
│ MCP Client (AI Agent) │
└─────────────────────┬───────────────────────────────────────┘
│ JSON-RPC 2.0 over stdio/http
▼
┌─────────────────────────────────────────────────────────────┐
│ MCP Server Gateway │
│ ┌─────────────┐ ┌─────────────┐ ┌─────────────┐ │
│ │ FileSystem │ │ Database │ │ API │ │
│ │ Tool │ │ Tool │ │ Tool │ │
│ └─────────────┘ └─────────────┘ └─────────────┘ │
│ ┌─────────────┐ ┌─────────────┐ ┌─────────────┐ │
│ │Connection │ │Query │ │Rate │ │
│ │Pool Manager │ │Optimizer │ │Limiter │ │
│ └─────────────┘ └─────────────┘ └─────────────┘ │
└─────────────────────────────────────────────────────────────┘
│ │ │
▼ ▼ ▼
Local Files PostgreSQL External APIs
(any path) / SQLite (REST/GraphQL)
สถาปัตยกรรมนี้แบ่งเป็น 3 ชั้นหลัก ชั้น Gateway รับ request จาก AI client, ชั้น Tools จัดการ business logic แต่ละประเภท และชั้น Infrastructure จัดการทรัพยากรระบบ เช่น connection pool และ rate limiting
การตั้งค่า Project และ Dependencies
เริ่มต้นด้วยการสร้าง project structure และติดตั้ง dependencies ที่จำเป็น
Project Structure
mcp-integrated-server/
├── src/
│ ├── __init__.py
│ ├── server.py # Main MCP Server
│ ├── tools/
│ │ ├── __init__.py
│ │ ├── filesystem.py # File operations
│ │ ├── database.py # Database queries
│ │ └── api.py # External API calls
│ ├── core/
│ │ ├── __init__.py
│ │ ├── connection_pool.py # Resource management
│ │ ├── rate_limiter.py # Rate limiting
│ │ └── security.py # Security utilities
│ └── config.py # Configuration
├── tests/
├── pyproject.toml
└── README.md
pyproject.toml
[project]
name = "mcp-integrated-server"
version = "1.0.0"
requires-python = ">=3.11"
dependencies = [
"mcp[dev]>=1.0.0",
"fastapi>=0.115.0",
"uvicorn>=0.32.0",
"asyncpg>=0.30.0",
"aiosqlite>=0.20.0",
"httpx>=0.28.0",
"pydantic>=2.10.0",
"tenacity>=9.0.0",
"structlog>=24.0.0",
]
[project.optional-dependencies]
dev = [
"pytest>=8.3.0",
"pytest-asyncio>=0.25.0",
"pytest-benchmark>=4.0.0",
"ruff>=0.8.0",
"mypy>=1.14.0",
]
[tool.mypy]
python_version = "3.11"
strict = true
plugins = ["pydantic.mypy"]
[tool.ruff]
line-length = 100
target-version = "py311"
Core Components: Connection Pool Manager
Connection pool เป็นหัวใจสำคัญของ MCP Server ที่ต้องรองรับ concurrent requests จำนวนมาก โค้ดด้านล่างใช้ asyncio semaphore เพื่อควบคุมจำนวน connections พร้อมกัน
src/core/connection_pool.py
import asyncio
from contextlib import asynccontextmanager
from dataclasses import dataclass, field
from typing import Any, AsyncGenerator, Optional
import structlog
logger = structlog.get_logger()
@dataclass
class PoolConfig:
max_size: int = 10
min_size: int = 2
timeout: float = 30.0
max_lifetime: float = 3600.0
class ConnectionPool[T]:
"""Generic async connection pool with lifecycle management."""
def __init__(
self,
create_fn: callable,
config: PoolConfig = None
):
self._create_fn = create_fn
self._config = config or PoolConfig()
self._semaphore = asyncio.Semaphore(self._config.max_size)
self._pools: dict[str, T] = {}
self._locks: dict[str, asyncio.Lock] = {}
self._stats = {"acquired": 0, "released": 0, "errors": 0}
self._created_at = asyncio.get_event_loop().time()
@asynccontextmanager
async def acquire(self, key: str = "default") -> AsyncGenerator[T, None]:
"""Acquire a connection from the pool."""
async with self._semaphore:
await self._ensure_connection(key)
try:
conn = self._pools[key]
self._stats["acquired"] += 1
yield conn
except Exception as e:
self._stats["errors"] += 1
logger.error("connection_acquire_error", key=key, error=str(e))
# Recreate connection on error
await self._recreate_connection(key)
raise
finally:
self._stats["released"] += 1
async def _ensure_connection(self, key: str) -> None:
if key not in self._locks:
self._locks[key] = asyncio.Lock()
async with self._locks[key]:
if key not in self._pools or not await self._is_healthy(key):
await self._recreate_connection(key)
async def _is_healthy(self, key: str) -> bool:
conn = self._pools.get(key)
if conn is None:
return False
try:
if hasattr(conn, 'execute'):
await conn.execute("SELECT 1")
return True
except Exception:
return False
async def _recreate_connection(self, key: str) -> None:
if key in self._pools:
old_conn = self._pools.pop(key)
await self._close_connection(old_conn)
self._pools[key] = await self._create_fn(key)
logger.info("connection_recreated", key=key)
async def _close_connection(self, conn: Any) -> None:
try:
if hasattr(conn, 'close'):
await conn.close()
except Exception as e:
logger.warning("connection_close_error", error=str(e))
def get_stats(self) -> dict[str, Any]:
return {
**self._stats,
"pool_size": len(self._pools),
"uptime": asyncio.get_event_loop().time() - self._created_at,
"available": self._semaphore._value,
}
async def close(self) -> None:
for conn in self._pools.values():
await self._close_connection(conn)
self._pools.clear()
logger.info("pool_closed", stats=self._stats)
Database-specific pool implementations
class DatabasePool(ConnectionPool):
"""PostgreSQL connection pool with query optimization."""
def __init__(self, dsn: str, config: PoolConfig = None):
self._dsn = dsn
super().__init__(self._create_postgres, config)
async def _create_postgres(self, key: str):
import asyncpg
return await asyncpg.create_pool(
self._dsn,
min_size=self._config.min_size,
max_size=self._config.max_size,
command_timeout=self._config.timeout,
)
@asynccontextmanager
async def acquire(self, key: str = "default") -> AsyncGenerator:
async with super().acquire(key) as pool:
async with pool.acquire() as conn:
yield conn
class SQLitePool(ConnectionPool):
"""SQLite connection pool for lightweight operations."""
def __init__(self, db_path: str, config: PoolConfig = None):
self._db_path = db_path
super().__init__(self._create_sqlite, config or PoolConfig(max_size=5))
async def _create_sqlite(self, key: str):
import aiosqlite
conn = await aiosqlite.connect(self._db_path, check_same_thread=False)
conn.row_factory = aiosqlite.Row
await conn.execute("PRAGMA journal_mode=WAL")
await conn.execute("PRAGMA synchronous=NORMAL")
await conn.execute("PRAGMA cache_size=-64000") # 64MB cache
return conn
Filesystem Tool: Safe File Operations
Filesystem Tool ต้องรักษา security เป็นหลัก ป้องกัน path traversal attacks และจำกัดสิทธิ์การเข้าถึงได้อย่างเคร่งครัด
src/tools/filesystem.py
import asyncio
import os
from pathlib import Path
from typing import Any, Literal
from dataclasses import dataclass
import structlog
from mcp.types import Tool, TextContent
from mcp.server import Server
from mcp.server.stdio import stdio_server
from ..core.security import SecurityValidator, AllowedPath
logger = structlog.get_logger()
@dataclass
class FileOperation:
operation: Literal["read", "write", "list", "stat", "search"]
path: str
content: str | None = None
pattern: str | None = None
max_size: int = 10 * 1024 * 1024 # 10MB default
class FileSystemTool:
"""Secure filesystem operations with sandboxing."""
def __init__(
self,
allowed_paths: list[str],
max_file_size: int = 10 * 1024 * 1024,
read_only: bool = True,
):
self._allowed = [Path(p).resolve() for p in allowed_paths]
self._max_size = max_file_size
self._read_only = read_only
self._semaphore = asyncio.Semaphore(5) # Max concurrent ops
self._validator = SecurityValidator()
self._cache: dict[str, tuple[str, float]] = {}
self._cache_ttl = 5.0 # seconds
def _validate_path(self, path: str) -> Path:
"""Validate and resolve path within allowed boundaries."""
try:
requested = Path(path).resolve()
# Check if path is within any allowed directory
for allowed in self._allowed:
try:
requested.relative_to(allowed)
return requested
except ValueError:
continue
raise PermissionError(f"Path '{path}' is outside allowed directories")
except Exception as e:
logger.error("path_validation_failed", path=path, error=str(e))
raise
def _get_cached(self, key: str) -> str | None:
if key in self._cache:
content, timestamp = self._cache[key]
if asyncio.get_event_loop().time() - timestamp < self._cache_ttl:
return content
del self._cache[key]
return None
def _set_cached(self, key: str, content: str) -> None:
if len(self._cache) > 100: # Simple LRU-like eviction
oldest = min(self._cache.keys(),
key=lambda k: self._cache[k][1])
del self._cache[oldest]
self._cache[key] = (content, asyncio.get_event_loop().time())
async def read_file(self, path: str, encoding: str = "utf-8") -> str:
"""Read file contents with size and path validation."""
async with self._semaphore:
validated = self._validate_path(path)
# Check cache first
cached =