MCP Service Discovery: กลไกค้นพบและตั้งค่าแหล่งข้อมูลอัตโนมัติ

ในยุคที่ AI Agent กำลังเปลี่ยนแปลงวิธีการทำงานของระบบอีคอมเมิร์ซและองค์กร การเชื่อมต่อกับแหล่งข้อมูลที่หลากหลายอย่างรวดเร็วเป็นสิ่งจำเป็น บทความนี้จะสอนวิธีสร้างระบบ MCP Service Discovery ที่ช่วยให้ AI ของคุณค้นพบและเชื่อมต่อกับแหล่งข้อมูลต่าง ๆ ได้โดยอัตโนมัติ โดยใช้ HolySheep AI เป็นแพลตฟอร์มหลัก

ปัญหา: ระบบ AI ลูกค้าสัมพันธ์ที่ต้องเชื่อมต่อหลายแหล่งข้อมูล

สมมติว่าคุณพัฒนาระบบ Chatbot อีคอมเมิร์ซที่ต้องดึงข้อมูลจากหลายแหล่งพร้อมกัน ทั้งสินค้าคงคลัง ราคา รีวิว และสถานะการจัดส่ง การตั้งค่าทีละ connection ทำให้โค้ดซับซ้อนและบำรุงรักษายาก

วิธีแก้: MCP Service Discovery อัตโนมัติ

MCP (Model Context Protocol) ช่วยให้ AI ค้นพบแหล่งข้อมูลที่พร้อมใช้งานโดยอัตโนมัติ ลดเวลาพัฒนาลง 70% และเพิ่มความยืดหยุ่นในการขยายระบบ

ส่วนประกอบหลักของระบบ Discovery

import httpx
import asyncio
from typing import List, Dict, Optional
from dataclasses import dataclass
from enum import Enum

class DataSourceType(Enum):
    POSTGRESQL = "postgresql"
    MONGODB = "mongodb"
    REDIS = "redis"
    ELASTICSEARCH = "elasticsearch"
    API = "http_api"

@dataclass
class DataSource:
    name: str
    type: DataSourceType
    endpoint: str
    status: str
    latency_ms: float
    capabilities: List[str]

class MCPServiceDiscovery:
    def __init__(self, api_key: str):
        self.base_url = "https://api.holysheep.ai/v1"
        self.headers = {
            "Authorization": f"Bearer {api_key}",
            "Content-Type": "application/json"
        }
        self.registered_sources: Dict[str, DataSource] = {}
    
    async def discover_sources(self) -> List[DataSource]:
        """ค้นหาแหล่งข้อมูลที่พร้อมใช้งานทั้งหมด"""
        async with httpx.AsyncClient(timeout=30.0) as client:
            response = await client.get(
                f"{self.base_url}/mcp/sources",
                headers=self.headers
            )
            response.raise_for_status()
            data = response.json()
            return [self._parse_source(s) for s in data["sources"]]
    
    async def register_source(self, source: DataSource) -> bool:
        """ลงทะเบียนแหล่งข้อมูลใหม่"""
        async with httpx.AsyncClient(timeout=10.0) as client:
            response = await client.post(
                f"{self.base_url}/mcp/sources/register",
                headers=self.headers,
                json=self._source_to_dict(source)
            )
            return response.status_code == 201
    
    async def health_check(self, source_name: str) -> Optional[DataSource]:
        """ตรวจสอบสถานะแหล่งข้อมูล"""
        async with httpx.AsyncClient(timeout=5.0) as client:
            response = await client.get(
                f"{self.base_url}/mcp/sources/{source_name}/health",
                headers=self.headers
            )
            if response.status_code == 200:
                return self._parse_source(response.json())
            return None

การใช้งานจริง: RAG System สำหรับองค์กร

องค์กรขนาดใหญ่มักมีเอกสารกระจายอยู่หลายที่ ทั้ง SharePoint, Confluence, S3 และ Database ต่าง ๆ ระบบ RAG ที่ใช้ MCP Service Discovery จะสามารถค้นหาและ index เอกสารจากทุกแหล่งโดยอัตโนมัติ

class EnterpriseRAGSystem:
    def __init__(self, api_key: str):
        self.discovery = MCPServiceDiscovery(api_key)
        self.vector_store = []
        self.cache = {}
    
    async def initialize_knowledge_base(self):
        """เริ่มต้น Knowledge Base จากแหล่งข้อมูลที่ค้นพบ"""
        sources = await self.discovery.discover_sources()
        
        for source in sources:
            if source.status == "healthy":
                await self._index_source(source)
        
        # สร้าง Embedding ด้วย HolySheep
        await self._create_embeddings()
    
    async def query_with_context(self, question: str, top_k: int = 5):
        """ค้นหาคำตอบพร้อม Context จากทุกแหล่งข้อมูล"""
        # สร้าง Embedding ของคำถาม
        async with httpx.AsyncClient(timeout=30.0) as client:
            embedding_response = await client.post(
                "https://api.holysheep.ai/v1/embeddings",
                headers={
                    "Authorization": f"Bearer {self.discovery.api_key}",
                    "Content-Type": "application/json"
                },
                json={
                    "model": "text-embedding-3-small",
                    "input": question
                }
            )
            query_embedding = embedding_response.json()["data"][0]["embedding"]
        
        # ค้นหา Documents ที่ใกล้เคียง
        relevant_docs = self._semantic_search(query_embedding, top_k)
        
        # สร้าง Prompt สำหรับ LLM
        context = "\n\n".join([doc["content"] for doc in relevant_docs])
        prompt = f"Context:\n{context}\n\nQuestion: {question}"
        
        # ส่งไปยัง LLM
        async with httpx.AsyncClient(timeout=60.0) as client:
            llm_response = await client.post(
                "https://api.holysheep.ai/v1/chat/completions",
                headers={
                    "Authorization": f"Bearer {self.discovery.api_key}",
                    "Content-Type": "application/json"
                },
                json={
                    "model": "gpt-4.1",
                    "messages": [{"role": "user", "content": prompt}],
                    "temperature": 0.3,
                    "max_tokens": 1000
                }
            )
            return llm_response.json()["choices"][0]["message"]["content"]

การตั้งค่า Configuration อัตโนมัติ

หัวใจสำคัญของ MCP Service Discovery คือการตั้งค่าที่ยืดหยุ่นและอัปเดตอัตโนมัติ ระบบจะตรวจสอบสถานะของแหล่งข้อมูลและปรับ Connection Pool ตามความต้องการ

import json
from typing import Any, Dict

class AutoConfigManager:
    """จัดการ Configuration อัตโนมัติสำหรับ MCP"""
    
    def __init__(self, api_key: str):
        self.api_key = api_key
        self.config: Dict[str, Any] = {}
        self.discovery = MCPServiceDiscovery(api_key)
    
    async def load_config(self, config_path: str = "mcp_config.json"):
        """โหลด Configuration จากไฟล์หรือ Server"""
        try:
            with open(config_path, "r") as f:
                self.config = json.load(f)
        except FileNotFoundError:
            # ดึง Config จาก Server
            self.config = await self._fetch_config_from_server()
        
        return self.config
    
    async def auto_configure_sources(self):
        """ตั้งค่าแหล่งข้อมูลอัตโนมัติตาม Config"""
        sources = await self.discovery.discover_sources()
        
        for source in sources:
            if source.status == "healthy":
                pool_size = self._calculate_pool_size(source)
                timeout = self._calculate_timeout(source)
                
                self.config[f"{source.name}_pool"] = pool_size
                self.config[f"{source.name}_timeout"] = timeout
        
        await self._save_config()
        return self.config
    
    def _calculate_pool_size(self, source: DataSource) -> int:
        """คำนวณขนาด Connection Pool"""
        base_size = 5
        if source.latency_ms < 50:
            return base_size * 2
        elif source.latency_ms < 200:
            return base_size
        else:
            return base_size // 2
    
    def _calculate_timeout(self, source: DataSource) -> int:
        """คำนวณ Timeout ตาม Latency"""
        return max(int(source.latency_ms * 2 / 1000), 5)
    
    async def _fetch_config_from_server(self) -> Dict[str, Any]:
        async with httpx.AsyncClient(timeout=30.0) as client:
            response = await client.get(
                "https://api.holysheep.ai/v1/mcp/config/default",
                headers={"Authorization": f"Bearer {self.api_key}"}
            )
            return response.json()
    
    async def _save_config(self):
        with open("mcp_config.json", "w") as f:
            json.dump(self.config, f, indent=2)

ข้อผิดพลาดที่พบบ่อยและวิธีแก้ไข

ข้อผิดพลาดที่ 1: 401 Unauthorized - API Key ไม่ถูกต้อง

สาเหตุ: API Key หมดอายุ หรือใส่ผิด format

❌ วิธีที่ผิด - Key ไม่ถูกต้อง
headers = {"Authorization": "YOUR_HOLYSHEEP_API_KEY"}  # ขาด Bearer

✅ วิธีที่ถูก - ต้องมี Bearer และ space
headers = {"Authorization": f"Bearer {api_key}"}

ตรวจสอบ Format ของ API Key
def validate_api_key(api_key: str) -> bool:
    if not api_key or len(api_key) < 20:
        return False
    if not api_key.startswith("hs_"):
        return False
    return True

ดึง API Key ใหม่หากหมดอายุ
async def refresh_token_if_needed():
    try:
        response = await client.get(
            "https://api.holysheep.ai/v1/auth/validate",
            headers={"Authorization": f"Bearer {api_key}"}
        )
        if response.status_code == 401:
            # ดึง API Key ใหม่จาก Dashboard
            return await get_new_api_key()
    except httpx.HTTPStatusError as e:
        if e.response.status_code == 401:
            return await get_new_api_key()

ข้อผิดพลาดที่ 2: Connection Timeout - แหล่งข้อมูลตอบสนองช้า

สาเหตุ: แหล่งข้อมูลมี latency สูงกว่า timeout ที่ตั้งไว้ หรือ network มีปัญหา

❌ วิธีที่ผิด - Timeout คงที่ ไม่ยืดหยุ่น
async with httpx.AsyncClient(timeout=10.0) as client:
    # หากแหล่งข้อมูลช้า จะ timeout เสมอ

✅ วิธีที่ถูก - Dynamic Timeout ตามประเภทข้อมูล
class AdaptiveTimeout:
    TIMEOUTS = {
        "health_check": 5.0,
        "small_query": 10.0,
        "bulk_read": 60.0,
        "indexing": 300.0
    }
    
    @classmethod
    def get_timeout(cls, operation: str, source_latency: float) -> float:
        base = cls.TIMEOUTS.get(operation, 30.0)
        # เพิ่ม timeout ตาม latency ของ source
        return max(base, source_latency * 3 / 1000)

async def query_with_adaptive_timeout(source: DataSource, query: str):
    timeout = AdaptiveTimeout.get_timeout("small_query", source.latency_ms)
    async with httpx.AsyncClient(timeout=timeout) as client:
        # retry with exponential backoff
        for attempt in range(3):
            try:
                response = await client.get(
                    f"{source.endpoint}/search",
                    params={"q": query}
                )
                return response.json()
            except httpx.TimeoutException:
                if attempt < 2:
                    await asyncio.sleep(2 ** attempt)
                else:
                    raise

ข้อผิดพลาดที่ 3: Rate Limit Exceeded - เรียก API บ่อยเกินไป

สาเหตุ: เรียก Discovery API บ่อยเกินจำนวนที่กำหนด หรือไม่ได้ใช้ caching

import time
from functools import lru_cache

❌ วิธีที่ผิด - เรียก API ทุกครั้งโดยไม่ cache
async def get_sources_every_time():
    response = await client.get(f"{base_url}/mcp/sources")  # เรียกทุกครั้ง
    return response.json()

✅ วิธีที่ถูก - Caching + Rate Limit
class RateLimitedDiscovery(MCPServiceDiscovery):
    def __init__(self, api_key: str):
        super().__init__(api_key)
        self._cache = {}
        self._cache_ttl = 300  # 5 นาที
        self._last_request = 0
        self._min_interval = 1.0  # วินาทีระหว่าง request
    
    async def discover_sources(self) -> List[DataSource]:
        # ตรวจสอบ cache
        cache_key = "sources_list"
        if cache_key in self._cache:
            cached, timestamp = self._cache[cache_key]
            if time.time() - timestamp < self._cache_ttl:
                return cached
        
        # รอให้ครบ interval
        elapsed = time.time() - self._last_request
        if elapsed < self._min_interval:
            await asyncio.sleep(self._min_interval - elapsed)
        
        # เรียก API
        sources = await super().discover_sources()
        self._last_request = time.time()
        
        # เก็บ cache
        self._cache[cache_key] = (sources, time.time())
        return sources

ใช้ Batch Request แทนหลาย request
async def batch_discover():
    """รวมหลาย operation ใน request เดียว"""
    async with httpx.AsyncClient(timeout=30.0) as client:
        response = await client.post(
            "https://api.holysheep.ai/v1/mcp/batch",
            headers={"Authorization": f"Bearer {api_key}"},
            json={
                "operations": [
                    {"type": "discover_sources"},
                    {"type": "health_check", "source": "db_primary"},
                    {"type": "get_config"}
                ]
            }
        )
        return response.json()

ตารางเปรียบเทียบ Latency และ Cost

โมเดล	ราคา ($/MTok)	Latency เฉลี่ย
GPT-4.1	$8.00	~120ms
Claude Sonnet 4.5	$15.00	~95ms
Gemini 2.5 Flash	$2.50	~50ms
DeepSeek V3.2	$0.42	~80ms

หมายเหตุ: อัตราแลกเปลี่ยนปัจจุบัน ¥1 = $1 ทำให้ประหยัดได้ถึง 85% เมื่อเทียบกับผู้ให้บริการอื่น รองรับการชำระเงินผ่าน WeChat และ Alipay พร้อม latency เฉลี่ยต่ำกว่า 50ms

สรุป

MCP Service Discovery ช่วยให้การเชื่อมต่อ AI กับแหล่งข้อมูลเป็นเรื่องง่าย ลดโค้ดที่ซับซ้อน และเพิ่มความยืดหยุ่นในการขยายระบบ ด้วย HolySheep AI คุณจะได้รับ API ที่เสถียร ราคาประหยัด และ Support ที่รวดเร็ว

👉 สมัคร HolySheep AI — รับเครดิตฟรีเมื่อลงทะเบียน