ในฐานะวิศวกรอาวุโสที่ดูแลระบบ AI ภายในองค์กร ผมเคยเจอปัญหาคลาสสิกที่หลายท่านน่าจะคุ้นเคย: ทีม Data Science ต้องการให้ Claude ตอบคำถามจากข้อมูล PostgreSQL และ S3 ของบริษัทโดยตรง แต่ไม่อยากอัปโหลดข้อมูลไปยัง Claude API ภายนอกด้วยเหตุผลด้านความปลอดภัยและ compliance บทความนี้จะแชร์ประสบการณ์ตรงในการสร้าง MCP Server แบบ production-ready พร้อมเทคนิคเพิ่มประสิทธิภาพ การควบคุม concurrency และลดต้นทุน Claude API ลง 85%+
ทำไมต้อง MCP Server?
Model Context Protocol (MCP) เป็นโปรโตคอลมาตรฐานที่ Anthropic เปิดตัวในปี 2024 ทำหน้าที่เป็น "USB-C สำหรับ AI" ช่วยให้ LLM เรียกใช้ tools ภายนอกได้อย่างปลอดภัยและเป็นระบบ ข้อดีเมื่อเทียบกับวิธี RAG แบบเดิม:
- Schema-based tool definitions — Claude รู้จัก parameters และ types ล่วงหน้า ไม่ต้อง parse prompt
- stdio/SSE transport — รัน local หรือ remote ก็ได้ เลือกตาม use case
- Permission control ระดับ resource — บังคับ read-only, scoped query, row limit
- Audit log ครบถ้วนทุก tool call — สำคัญมากสำหรับ SOC2/ISO27001
- ไม่ต้อง ETL ข้อมูลเข้า vector store — query ตรงจาก source of truth
สถาปัตยกรรมระบบ
ผมเลือก Python 3.11 + FastMCP เพราะ ecosystem ของ async I/O แข็งแรงที่สุดและ community contribution เยอะที่สุด (8.2k stars บน GitHub) สถาปัตยกรรมแบ่งเป็น 3 ชั้นชัดเจน:
- Transport layer: stdio (รัน local บนเครื่อง developer) หรือ SSE (รันบน server ส่วนกลาง)
- Tool layer: PostgreSQL tools, S3 tools — ทำงานแบบ async ทั้งหมด
- Connection layer: asyncpg (PostgreSQL), aioboto3 (S3) พร้อม connection pool และ auto-reconnect
ขั้นตอนที่ 1: ติดตั้งและเตรียมโปรเจกต์
mkdir mcp-enterprise && cd mcp-enterprise
python3.11 -m venv .venv
source .venv/bin/activate
pip install fastmcp==0.4.0 asyncpg==0.29.0 aioboto3==12.0.0 anthropic pydantic==2.5ขั้นตอนที่ 2: PostgreSQL MCP Server (Production-grade)
from fastmcp import FastMCP
import asyncpg
from typing import Optional
import os
import asyncio
mcp = FastMCP("postgres-enterprise")
class PostgresPool:
def __init__(self):
self.pool: Optional[asyncpg.Pool] = None
async def connect(self):
self.pool = await asyncpg.create_pool(
dsn=os.environ["POSTGRES_DSN"],
min_size=2,
max_size=10,
max_inactive_connection_lifetime=300,
command_timeout=30,
statement_cache_size=1024,
)
async def close(self):
if self.pool:
await self.pool.close()
db = PostgresPool()
@mcp.tool()
async def query_database(sql: str, limit: int = 100) -> dict:
"""รัน read-only SQL query กับ PostgreSQL พร้อม row limit และ read-only enforcement"""
sql_clean = sql.strip().lower()
if not sql_clean.startswith("select") and not sql_clean.startswith("with"):
return {"error": "อนุญาตเฉพาะ SELECT/WITH เท่านั้น", "rows": []}
safe_limit = min(int(limit), 1000)
async with db.pool.acquire() as conn:
async with conn.transaction(readonly=True):
rows = await conn.fetch(f"SELECT * FROM ({sql}) AS sub LIMIT {safe_limit}")
return {
"columns": list(rows[0].keys()) if rows else [],
"rows": [dict(r) for r in rows],
"row_count": len(rows),
}
@mcp.tool()
async def list_tables(schema: str = "public") -> list:
"""แสดงรายชื่อ tables ใน schema พร้อมขนาด"""
async with db.pool.acquire() as conn:
records = await conn.fetch("""
SELECT table_name,
pg_size_pretty(pg_total_relation_size(quote_ident(table_name))) AS size
FROM information_schema.tables
WHERE table_schema = $1
ORDER BY table_name
""", schema)
return [dict(r) for r in records]
if __name__ == "__main__":
asyncio.run(db.connect())
try:
mcp.run(transport="stdio")
finally:
asyncio.run(db.close())ขั้นตอนที่ 3: S3 MCP Server
from fastmcp import FastMCP
import aioboto3
from typing import Optional
import os
s3_mcp = FastMCP("s3-enterprise")
@s3_mcp.tool()
async def list_objects(bucket: str, prefix: str = "", max_keys: int = 1000) -> list:
"""แสดงรายการ objects ใน S3 bucket พร้อม prefix filter"""
session = aioboto3.Session()
safe_max = min(int(max_keys), 1000)
async with session.client("s3", region_name=os.environ.get("AWS_REGION", "ap-southeast-1")) as s3:
response = await s3.list_objects_v2(
Bucket=bucket, Prefix=prefix, MaxKeys=safe_max
)
return [
{
"key": obj["Key"],
"size": obj["Size"],
"modified": obj["LastModified"].isoformat(),
}
for obj in response.get("Contents", [])
]
@s3_mcp.tool()
async def read_object(bucket: str, key: str, max_bytes: int = 10240) -> str:
"""อ่านเนื้อหาไฟล์จาก S3 (จำกัด 10KB เพื่อความปลอดภัย)"""
session = aioboto3.Session()
async with session.client("s3") as s3:
response = await s3.get_object(Bucket=bucket, Key=key)
async with response["Body"] as stream:
data = await stream.read(min(int(max_bytes), 10240))
return data.decode("utf-8", errors="replace")
if __name__ == "__main__":
s3_mcp.run(transport="stdio")