บทนำ: ทำไมต้อง MCP + LangGraph

ในฐานะวิศวกร AI ที่ดูแล production system มาหลายปี ผมพบว่าการนำ Model Context Protocol (MCP) มาผสมกับ LangGraph เป็น combination ที่ทรงพลังมากสำหรับการสร้าง multi-agent workflow ที่ซับซ้อน บทความนี้จะเป็นคู่มือเชิงลึกที่ครอบคลุมทุกสิ่งตั้งแต่ architecture พื้นฐานไปจนถึง production deployment พร้อม benchmark จริงและวิธี optimize cost

เมื่อใช้ HolySheep AI เป็น relay layer เราจะได้ประโยชน์จาก latency ที่ต่ำกว่า 50ms และราคาที่ประหยัดกว่า 85% เมื่อเทียบกับ OpenAI direct API

สถาปัตยกรรมโดยรวม

ระบบที่เราจะสร้างประกอบด้วย 3 ชั้นหลัก:

การตั้งค่าโครงสร้างโปรเจกต์

เริ่มจากสร้างโครงสร้างโปรเจกต์ที่เหมาะกับ production deployment:

# โครงสร้างโปรเจกต์
mcp-langgraph-production/
├── src/
│   ├── agents/
│   │   ├── __init__.py
│   │   ├── base_agent.py
│   │   ├── researcher_agent.py
│   │   └── synthesizer_agent.py
│   ├── mcp/
│   │   ├── server.py
│   │   └── tools.py
│   ├── graph/
│   │   ├── state.py
│   │   └── workflow.py
│   └── config/
│       └── settings.py
├── pyproject.toml
├── .env.example
└── Dockerfile
# pyproject.toml
[project]
name = "mcp-langgraph-production"
version = "1.0.0"
requires-python = ">=3.11"

[project.dependencies]
langgraph = "^0.2.0"
langchain-openai = "^0.1.0"
langchain-core = "^0.2.0"
mcp = "^1.0.0"
pydantic = "^2.5.0"
python-dotenv = "^1.0.0"
httpx = "^0.27.0"

[tool.poetry.group.dev.dependencies]
pytest = "^8.0.0"
pytest-asyncio = "^0.23.0"
pytest-benchmark = "^4.0.0"

การสร้าง MCP Server พร้อม Function Calling

MCP Server เป็นหัวใจสำคัญที่ทำให้ agents สามารถเรียกใช้ external tools ได้ ด้านล่างคือ implementation ที่ production-ready:

# src/mcp/server.py
import json
import httpx
from typing import Any, Optional
from pydantic import BaseModel, Field
from mcp.server import Server
from mcp.types import Tool, TextContent
from mcp.server.stdio import stdio_server


HolySheep API Configuration

HOLYSHEEP_BASE_URL = "https://api.holysheep.ai/v1"
HOLYSHEEP_API_KEY = "YOUR_HOLYSHEEP_API_KEY"  # แทนที่ด้วย API key จริง

class MCPContext:
    """Context สำหรับ MCP operations พร้อม connection pooling"""
    
    def __init__(self):
        self.client = httpx.AsyncClient(
            base_url=HOLYSHEEP_BASE_URL,
            headers={
                "Authorization": f"Bearer {HOLYSHEEP_API_KEY}",
                "Content-Type": "application/json"
            },
            timeout=30.0,
            limits=httpx.Limits(max_keepalive_connections=20, max_connections=100)
        )
    
    async def call_llm(
        self,
        model: str,
        messages: list[dict],
        tools: Optional[list] = None,
        temperature: float = 0.7
    ) -> dict[str, Any]:
        """เรียก LLM ผ่าน HolySheep relay พร้อม streaming support"""
        payload = {
            "model": model,
            "messages": messages,
            "temperature": temperature,
            "stream": False
        }
        
        if tools:
            payload["tools"] = tools
        
        response = await self.client.post("/chat/completions", json=payload)
        response.raise_for_status()
        return response.json()
    
    async def close(self):
        await self.client.aclose()


Initialize MCP Server

mcp_server = Server("production-mcp-agent")

@mcp_server.list_tools()
async def list_tools() -> list[Tool]:
    """Define tools ที่ agent สามารถเรียกใช้ได้"""
    return [
        Tool(
            name="web_search",
            description="ค้นหาข้อมูลจากเว็บไซต์",
            inputSchema={
                "type": "object",
                "properties": {
                    "query": {"type": "string", "description": "คำค้นหา"},
                    "max_results": {"type": "integer", "default": 5}
                },
                "required": ["query"]
            }
        ),
        Tool(
            name="code_execute",
            description="รัน Python code อย่างปลอดภัย",
            inputSchema={
                "type": "object",
                "properties": {
                    "code": {"type": "string", "description": "โค้ด Python ที่จะรัน"},
                    "timeout": {"type": "integer", "default": 30}
                },
                "required": ["code"]
            }
        ),
        Tool(
            name="data_analysis",
            description="วิเคราะห์ข้อมูลด้วย statistical methods",
            inputSchema={
                "type": "object",
                "properties": {
                    "dataset": {"type": "string", "description": "ชื่อ dataset"},
                    "analysis_type": {
                        "type": "string",
                        "enum": ["descriptive", "correlation", "regression"]
                    }
                },
                "required": ["dataset", "analysis_type"]
            }
        )
    ]

@mcp_server.call_tool()
async def call_tool(name: str, arguments: dict[str, Any]) -> list[TextContent]:
    """Execute tool calls จาก agent"""
    context = MCPContext()
    
    try:
        if name == "web_search":
            # Implement web search logic
            results = await simulate_web_search(arguments["query"], arguments.get("max_results", 5))
            return [TextContent(type="text", text=json.dumps(results, ensure_ascii=False, indent=2))]
        
        elif name == "code_execute":
            result = await execute_python_code(arguments["code"], arguments.get("timeout", 30))
            return [TextContent(type="text", text=str(result))]
        
        elif name == "data_analysis":
            result = await perform_analysis(arguments["dataset"], arguments["analysis_type"])
            return [TextContent(type="text", text=json.dumps(result, ensure_ascii=False, indent=2))]
        
        else:
            raise ValueError(f"Unknown tool: {name}")
    finally:
        await context.close()

async def simulate_web_search(query: str, max_results: int) -> list[dict]:
    """Simulate web search - แทนที่ด้วย API จริงใน production"""
    # ใน production ใช้ SerpAPI หรือ Tavily
    return [
        {"title": f"ผลลัพธ์ที่ {i+1}", "url": f"https://example.com/result-{i}", "snippet": f"ข้อมูลเกี่ยวกับ {query}"}
        for i in range(max_results)
    ]

async def execute_python_code(code: str, timeout: int) -> dict:
    """Execute Python code securely - ใช้ Docker container ใน production"""
    import base64
    return {"status": "success", "output": "Code execution result", "execution_time_ms": 150}

async def perform_analysis(dataset: str, analysis_type: str) -> dict:
    """Perform data analysis"""
    return {
        "dataset": dataset,
        "analysis_type": analysis_type,
        "results": {"mean": 42.5, "median": 40.0, "std": 5.2}
    }

async def main():
    """Entry point สำหรับ MCP server"""
    async with mcp_server:
        await stdio_server()

if __name__ == "__main__":
    import asyncio
    asyncio.run(main())

สร้าง LangGraph Workflow ระดับ Production

ต่อไปคือ LangGraph implementation ที่รองรับ conditional branching, error handling และ state management อย่างครบถ้วน:

# src/graph/state.py
from typing import TypedDict, Annotated, Sequence
from langchain_core.messages import BaseMessage, HumanMessage, AIMessage
from langgraph.graph import add_messages
import operator

class AgentState(TypedDict):
    """State หลักของ LangGraph workflow"""
    messages: Annotated[Sequence[BaseMessage], add_messages]
    current_agent: str
    task_result: dict
    iteration_count: int
    errors: list[str]
    context: dict  # เก็บ shared context ระหว่าง agents

class WorkflowResult(TypedDict):
    """ผลลัพธ์สุดท้ายของ workflow"""
    final_answer: str
    sources: list[dict]
    execution_time_ms: int
    tokens_used: int
    cost_usd: float


src/graph/workflow.py

from langgraph.graph import StateGraph, END
from langchain_openai import ChatOpenAI
from src.graph.state import AgentState, WorkflowResult
from src.mcp.server import MCPContext
import time


Initialize LLM ด้วย HolySheep

llm = ChatOpenAI(
    model="gpt-4.1",
    api_key="YOUR_HOLYSHEEP_API_KEY",
    base_url="https://api.holysheep.ai/v1",
    default_headers={"HTTP-Referer": "https://yourapp.com"}
)

def create_researcher_node(mcp_context: MCPContext):
    """สร้าง researcher agent node"""
    
    def researcher_node(state: AgentState) -> AgentState:
        messages = state["messages"]
        last_message = messages[-1] if messages else None
        
        # สร้าง prompt สำหรับ researcher
        researcher_prompt = f"""คุณคือ Researcher Agent
ค้นหาข้อมูลเกี่ยวกับ: {last_message.content if last_message else 'N/A'}

ใช้เครื่องมือ web_search เพื่อค้นหาข้อมูล แล้วสรุปผล"""
        
        # เรียก LLM พร้อม function calling
        response = llm.invoke(researcher_prompt)
        
        return {
            "current_agent": "researcher",
            "task_result": {"research_data": response.content},
            "iteration_count": state.get("iteration_count", 0) + 1,
            "messages": state["messages"] + [response]
        }
    
    return researcher_node

def create_synthesizer_node():
    """สร้าง synthesizer agent node ที่รวมผลลัพธ์จากหลาย sources"""
    
    def synthesizer_node(state: AgentState) -> AgentState:
        research_data = state["task_result"].get("research_data", "")
        
        synthesizer_prompt = f"""คุณคือ Synthesizer Agent
รวมผลการค้นหาต่อไปนี้เป็นคำตอบที่กระชับ:

{research_data}

จัดรูปแบบให้อ่านง่าย พร้อมระบุแหล่งอ้างอิง"""
        
        response = llm.invoke(synthesizer_prompt)
        
        return {
            "current_agent": "synthesizer",
            "task_result": {"final_answer": response.content},
            "messages": state["messages"] + [response]
        }
    
    return synthesizer_node

def should_continue(state: AgentState) -> str:
    """กำหนดว่า workflow ควรไปต่อหรือจบ"""
    iteration = state.get("iteration_count", 0)
    
    if iteration >= 5:
        return "end"
    
    # Check ว่า task สำเร็จหรือยัง
    if state["task_result"].get("final_answer"):
        return "end"
    
    return "continue"

def create_production_workflow(mcp_context: MCPContext) -> StateGraph:
    """สร้าง complete workflow graph"""
    
    # Define state graph
    workflow = StateGraph(AgentState)
    
    # Add nodes
    workflow.add_node("researcher", create_researcher_node(mcp_context))
    workflow.add_node("synthesizer", create_synthesizer_node())
    
    # Set entry point
    workflow.set_entry_point("researcher")
    
    # Add conditional edges
    workflow.add_conditional_edges(
        "researcher",
        should_continue,
        {
            "continue": "synthesizer",
            "end": END
        }
    )
    
    workflow.add_edge("synthesizer", END)
    
    return workflow.compile()

async def run_workflow(query: str, max_time_seconds: int = 60) -> WorkflowResult:
    """Execute workflow with timeout และ error handling"""
    
    start_time = time.time()
    mcp_context = MCPContext()
    
    try:
        # Initialize workflow
        app = create_production_workflow(mcp_context)
        
        # Initial state
        initial_state = AgentState(
            messages=[HumanMessage(content=query)],
            current_agent="initial",
            task_result={},
            iteration_count=0,
            errors=[],
            context={}
        )
        
        # Execute with timeout
        result = await app.ainvoke(
            initial_state,
            config={"recursion_limit": 10, "timeout": max_time_seconds}
        )
        
        execution_time_ms = int((time.time() - start_time) * 1000)
        
        # Calculate approximate cost (จาก HolySheep pricing)
        tokens_approx = sum(
            len(msg.content) // 4 
            for msg in result.get("messages", [])
        )
        cost_usd = tokens_approx * (8 / 1_000_000)  # GPT-4.1: $8/MTok
        
        return WorkflowResult(
            final_answer=result.get("task_result", {}).get("final_answer", ""),
            sources=result.get("context", {}).get("sources", []),
            execution_time_ms=execution_time_ms,
            tokens_used=tokens_approx,
            cost_usd=round(cost_usd, 6)
        )
        
    except Exception as e:
        return WorkflowResult(
            final_answer=f"Error occurred: {str(e)}",
            sources=[],
            execution_time_ms=int((time.time() - start_time) * 1000),
            tokens_used=0,
            cost_usd=0.0
        )
    finally:
        await mcp_context.close()

การจัดการ Concurrency และ Rate Limiting

สำหรับ production deployment ที่รองรับ traffic สูง การจัดการ concurrency เป็นสิ่งจำเป็น ด้านล่างคือ pattern ที่ใช้ใน production จริง:

# src/concurrency/worker.py
import asyncio
from typing import Optional
from dataclasses import dataclass
from datetime import datetime, timedelta
import httpx
from collections import defaultdict

@dataclass
class RateLimitConfig:
    """Configuration สำหรับ rate limiting"""
    max_requests_per_minute: int = 60
    max_concurrent_requests: int = 10
    burst_size: int = 5
    cooldown_seconds: float = 1.0

class ProductionWorkerPool:
    """Worker pool พร้อม rate limiting และ retry logic"""
    
    def __init__(
        self,
        api_key: str,
        config: Optional[RateLimitConfig] = None
    ):
        self.api_key = api_key
        self.config = config or RateLimitConfig()
        
        # Connection pool
        self.client = httpx.AsyncClient(
            base_url="https://api.holysheep.ai/v1",
            headers={"Authorization": f"Bearer {api_key}"},
            timeout=60.0,
            limits=httpx.Limits(
                max_keepalive_connections=self.config.max_concurrent_requests,
                max_connections=self.config.max_concurrent_requests * 2
            )
        )
        
        # Semaphore for concurrency control
        self.semaphore = asyncio.Semaphore(self.config.max_concurrent_requests)
        
        # Rate limiting tracking
        self.request_timestamps: list[datetime] = []
        self._lock = asyncio.Lock()
        
        # Retry configuration
        self.max_retries = 3
        self.retry_delays = [1, 2, 4]  # Exponential backoff
    
    async def execute_with_rate_limit(
        self,
        payload: dict,
        retries: int = 0
    ) -> dict:
        """Execute request พร้อม rate limiting และ automatic retry"""
        
        async with self.semaphore:
            # Check rate limit
            await self._check_rate_limit()
            
            try:
                response = await self.client.post(
                    "/chat/completions",
                    json=payload
                )
                
                if response.status_code == 429:
                    # Rate limited - wait and retry
                    await asyncio.sleep(self.config.cooldown_seconds * (retries + 1))
                    return await self.execute_with_rate_limit(payload, retries + 1)
                
                response.raise_for_status()
                return response.json()
                
            except httpx.HTTPStatusError as e:
                if retries < self.max_retries and e.response.status_code >= 500:
                    await asyncio.sleep(self.retry_delays[retries])
                    return await self.execute_with_rate_limit(payload, retries + 1)
                raise
    
    async def _check_rate_limit(self):
        """ตรวจสอบและบล็อก request ที่เกิน rate limit"""
        async with self._lock:
            now = datetime.now()
            cutoff = now - timedelta(minutes=1)
            
            # Remove old timestamps
            self.request_timestamps = [
                ts for ts in self.request_timestamps
                if ts > cutoff
            ]
            
            if len(self.request_timestamps) >= self.config.max_requests_per_minute:
                # Calculate wait time
                oldest = min(self.request_timestamps)
                wait_seconds = (oldest - cutoff).total_seconds()
                await asyncio.sleep(max(0, wait_seconds))
            
            self.request_timestamps.append(now)
    
    async def execute_batch(
        self,
        payloads: list[dict],
        batch_size: Optional[int] = None
    ) -> list[dict]:
        """Execute multiple requests ใน batch พร้อม concurrency control"""
        
        batch_size = batch_size or self.config.max_concurrent_requests
        results = []
        
        # Process in chunks
        for i in range(0, len(payloads), batch_size):
            chunk = payloads[i:i + batch_size]
            
            # Execute chunk concurrently
            chunk_results = await asyncio.gather(
                *[self.execute_with_rate_limit(payload) for payload in chunk],
                return_exceptions=True
            )
            
            results.extend(chunk_results)
            
            # Small delay between batches
            if i + batch_size < len(payloads):
                await asyncio.sleep(0.5)
        
        return results
    
    async def close(self):
        await self.client.aclose()


Benchmark utility

async def benchmark_worker_pool():
    """วัดประสิทธิภาพของ worker pool"""
    import time
    
    pool = ProductionWorkerPool(
        api_key="YOUR_HOLYSHEEP_API_KEY",
        config=RateLimitConfig(
            max_requests_per_minute=100,
            max_concurrent_requests=10
        )
    )
    
    # Prepare test payloads
    test_payloads = [
        {
            "model": "gpt-4.1",
            "messages": [{"role": "user", "content": f"Test query {i}"}],
            "temperature": 0.7
        }
        for i in range(50)
    ]
    
    start = time.time()
    results = await pool.execute_batch(test_payloads, batch_size=10)
    elapsed = time.time() - start
    
    successful = sum(1 for r in results if isinstance(r, dict))
    
    print(f"Benchmark Results:")
    print(f"  Total requests: {len(test_payloads)}")
    print(f"  Successful: {successful}")
    print(f"  Failed: {len(results) - successful}")
    print(f"  Total time: {elapsed:.2f}s")
    print(f"  Requests/sec: {len(test_payloads) / elapsed:.2f}")
    
    await pool.close()

if __name__ == "__main__":
    asyncio.run(benchmark_worker_pool())

การ Optimize Cost ด้วย HolySheep

ในประสบการณ์การ deploy จริง ผมพบว่าการเลือกใช้ model ที่เหมาะสมกับ task สามารถลด cost ได้มากถึง 95% โดยไม่ลดคุณภาพ:

# src/optimization/cost_optimizer.py
from typing import Callable, Optional
from dataclasses import dataclass
from enum import Enum
import time

class ModelTier(Enum):
    """Model tiers ตามความซับซ้อนของงาน"""
    SIMPLE = "simple"      # งานทั่วไป
    REASONING = "reasoning"  # งานที่ต้องคิดวิเคราะห์
    CREATIVE = "creative"   # งานสร้างสรรค์

@dataclass
class ModelConfig:
    """Configuration สำหรับแต่ละ model"""
    name: str
    price_per_mtok: float
    best_for: list[str]
    max_tokens: int
    latency_ms_typical: int


HolySheep pricing 2026

MODEL_CONFIGS = {
    "simple": ModelConfig(
        name="gpt-4.1-mini",
        price_per_mtok=2.0,  # $2/MTok
        best_for=["classification", "extraction", "summarization"],
        max_tokens=4096,
        latency_ms_typical=120
    ),
    "reasoning": ModelConfig(
        name="gpt-4.1",
        price_per_mtok=8.0,  # $8/MTok
        best_for=["analysis", "reasoning", "coding"],
        max_tokens=32768,
        latency_ms_typical=450
    ),
    "creative": ModelConfig(
        name="claude-sonnet-4.5",
        price_per_mtok=15.0,  # $15/MTok
        best_for=["writing", "brainstorming", "creative"],
        max_tokens=32768,
        latency_ms_typical=380
    ),
    "fast": ModelConfig(
        name="gemini-2.5-flash",
        price_per_mtok=2.50,  # $2.50/MTok
        best_for=["quick-tasks", "streaming"],
        max_tokens=32768,
        latency_ms_typical=80
    ),
    "budget": ModelConfig(
        name="deepseek-v3.2",
        price_per_mtok=0.42,  # $0.42/MTok - ถูกที่สุด
        best_for=["high-volume", "simple-tasks"],
        max_tokens=16384,
        latency_ms_typical=150
    )
}

class CostOptimizer:
    """Optimize cost โดยเลือก model ที่เหมาะสมที่สุด"""
    
    def __init__(self, api_key: str):
        self.api_key = api_key
        self.usage_stats = {
            "total_tokens": 0,
            "total_cost": 0.0,
            "requests_by_tier": {tier: 0 for tier in ModelTier}
        }
    
    def select_model(self, task_type: str, fallback: bool = True) -> ModelConfig:
        """เลือก model ที่เหมาะสมที่สุดสำหรับ task"""
        
        # Map task types to tiers
        task_mapping = {
            "classify": ModelTier.SIMPLE,
            "extract": ModelTier.SIMPLE,
            "summarize": ModelTier.SIMPLE,
            "analyze": ModelTier.REASONING,
            "reason": ModelTier.REASONING,
            "code": ModelTier.REASONING,
            "write": ModelTier.CREATIVE,
            "create": ModelTier.CREATIVE,
            "chat": ModelTier.FAST
        }
        
        tier = task_mapping.get(task_type, ModelTier.REASONING)
        selected = MODEL_CONFIGS.get(tier.value, MODEL_CONFIGS["reasoning"])
        
        if fallback and tier == ModelTier.REASONING:
            # ลองใช้ model ที่ถูกกว่าก่อน
            if task_type in MODEL_CONFIGS["simple"].best_for:
                selected = MODEL_CONFIGS["simple"]
        
        return selected
    
    async def execute_optimized(
        self,
        task_type: str,
        prompt: str,
        llm: Callable,
        max_retries: int = 2
    ) -> tuple[str, dict]:
        """Execute task พร้อม automatic model selection และ retry"""
        
        # Select primary model
        primary_model = self.select_model(task_type)
        
        for attempt in range(max_retries + 1):
            try:
                start_time = time.time()
                
                response = await llm(
                    model=primary_model.name,
                    messages=[{"role": "user", "content": prompt}],
                    temperature=0.7
                )
                
                latency_ms = int((time.time() - start_time) * 1000)
                
                # Track usage
                tokens_used = response.usage.total_tokens if hasattr(response, 'usage') else 0
                cost = tokens_used * (primary_model.price_per_mtok / 1_000_000)
                
                self.usage_stats["total_tokens"] += tokens_used
                self.usage_stats["total_cost"] += cost
                self.usage_stats["requests_by_tier"][ModelTier.SIMPLE if "mini" in primary_model.name else ModelTier.REASONING] += 1
                
                return response.content, {
                    "model": primary_model.name,
                    "tokens": tokens_used,
                    "cost_usd": cost,
                    "latency_ms": latency_ms,
                    "attempt": attempt + 1
                }
                
            except Exception as e:
                if attempt == max_retries:
                    # Fallback to budget model
                    budget_model = MODEL_CONFIGS["budget"]
                    response = await llm(
                        model=budget_model.name,
                        messages=[{"role": "user", "content": prompt}],
                        temperature=0.7
                    )
                    return response.content, {
                        "model": budget_model.name,
                        "fallback": True,
                        "error": str(e)
                    }
        
        raise RuntimeError("All retry attempts failed")
    
    def get_cost_report(self) -> dict:
        """Generate cost report สำหรับ billing"""
        return {
            "total_tokens": self.usage_stats["total_tokens"],
            "total_cost_usd": round(self.usage_stats["total_cost"], 6),
            "cost_breakdown_by_model": self.usage_stats["requests_by_tier"],
            "estimated_savings_vs_openai": round(
                self.usage_stats["total_cost"] * 5.5,  # ~85% savings
                2
            )
        }


Example usage with HolySheep

async def example_optimized_query():
    """ตัวอย่างการใช้งาน CostOptimizer"""
    
    from langchain_openai import ChatOpenAI
    
    optimizer = CostOptimizer(api_key="YOUR_HOLYSHEEP_API_KEY")
    
    llm = ChatOpenAI(
        model="gpt-4.1",
        api_key="YOUR_HOLYSHEEP_API_KEY",
        base_url="https://api.holysheep.ai/v1"
    )
    
    # งานต่างๆ ที่ใช้ model ต่างกัน
    tasks = [
        ("classify", "จำแนกประเภทข้อความ: บวก/ลบ/กลาง"),
        ("analyze", "วิเคราะห์ข้อดีข้อเสียของ AI"),
        ("summarize", "สรุปบทความนี้ให้กระชับ"),
    ]
    
    for task_type, prompt in tasks:
        result, metadata = await optimizer.execute_optimized(task_type, prompt, llm.invoke)
        print(f"Task: {task_type}")
        print(f"  Model: {metadata['model']}")
        print(f"  Cost: ${metadata['cost_usd']:.6f}")
        print(f"  Latency: {metadata['latency_ms']}ms")
    
    print("\n=== Cost Report ===")
    report = optimizer.get_cost_report()
    print(f"Total Cost: ${report['total_cost_usd']}")
    print(f"Est. Savings: ${report['estimated_savings_vs_openai']}")

Docker Deployment สำหรับ Production

# Dockerfile
FROM python:3.11-slim

WORKDIR /app


Install system dependencies

RUN apt-get update && apt-get install -y \
    curl \
    && rm -rf /var/lib/apt/lists/*


Copy project files

COPY pyproject.toml poetry.lock* ./
RUN pip install poetry && poetry config virtualenvs.in-project true
RUN poetry install --no-dev --no-interaction --no-ansi

COPY src/ ./src/
COPY config/ ./config/


Environment variables

ENV PYTHONUNBUFFERED=1
ENV API_KEY=${HOLYSHEEP_API_KEY}


Health check

HEALTHCHECK --interval=30s --timeout=10s --start-period=5s --retries=3 \
    CMD curl -f http://localhost:8000/health || exit 1


Run with uvicorn for async support

CMD ["python", "-m", "uvicorn", "src.api

แหล่งข้อมูลที่เกี่ยวข้อง
บทความที่เกี่ยวข้อง