เมื่อเดือนที่ผ่านมา ทีมของผมที่ HolySheep AI เจอเคสจริงจากลูกค้าธุรกิจอีคอมเมิร์ชรายหนึ่ง ที่ยอดขายหน้าเว็บพุ่งขึ้น 4 เท่าในช่วงเทศกาล จนแชทบอทลูกค้าสัมพันธ์เดิมที่ตอบด้วย rule-based ล่มไปภายใน 20 นาที ทางทีมตัดสินใจเปลี่ยนมาใช้ Claude Opus 4.7 เรียกเครื่องมือผ่าน MCP (Model Context Protocol) เพื่อให้บอทดึงสถานะออร์เดอร์ ตรวจสต็อก และเปิดรีเฟรนด์ได้แบบเรียลไทม์ ผมเลยถือโอกาสนี้เขียนเดโมฉบับเต็มแบ่งปันทั้งโค้ดและตัวเลข benchmark ที่วัดจริงจาก production

ทำไม MCP แล้ว Opus 4.7 ถึงเป็นคู่ที่ลงตัว

สถาปัตยกรรมที่เราจะสร้าง

1) MCP Server: ecommerce_mcp_server.py

import json
import asyncio
from mcp.server import Server
from mcp.server.stdio import stdio_server
from mcp.types import Tool, TextContent

app = Server("ecommerce-tools")


จำลองฐานข้อมูลออร์เดอร์และสต็อก (ในงานจริงต่อ PostgreSQL/MySQL)

ORDERS = {
    "ORD-1001": {"status": "shipped",   "amount": 1290.0, "items": ["SKU-A1"]},
    "ORD-1002": {"status": "pending",   "amount":  850.0, "items": ["SKU-B2"]},
    "ORD-1003": {"status": "delivered", "amount": 2400.0, "items": ["SKU-C3", "SKU-A1"]},
}
INVENTORY = {"SKU-A1": 42, "SKU-B2": 0, "SKU-C3": 156}


@app.list_tools()
async def list_tools() -> list[Tool]:
    return [
        Tool(
            name="get_order_status",
            description="ดึงสถานะและยอดเงินของคำสั่งซื้อจากหมายเลขออร์เดอร์",
            inputSchema={
                "type": "object",
                "properties": {
                    "order_id": {"type": "string", "description": "เช่น ORD-1001"}
                },
                "required": ["order_id"],
            },
        ),
        Tool(
            name="check_inventory",
            description="ตรวจจำนวนสินค้าคงคลังตามรหัส SKU",
            inputSchema={
                "type": "object",
                "properties": {
                    "sku": {"type": "string"}
                },
                "required": ["sku"],
            },
        ),
        Tool(
            name="create_refund",
            description="สร้างคำขอคืนเงินสำหรับออร์เดอร์ที่ระบุ",
            inputSchema={
                "type": "object",
                "properties": {
                    "order_id": {"type": "string"},
                    "amount":   {"type": "number"},
                    "reason":   {"type": "string"},
                },
                "required": ["order_id", "amount", "reason"],
            },
        ),
    ]


@app.call_tool()
async def call_tool(name: str, arguments: dict) -> list[TextContent]:
    if name == "get_order_status":
        o = ORDERS.get(arguments["order_id"])
        if not o:
            return [TextContent(type="text", text=json.dumps({"error": "order_not_found"}))]
        return [TextContent(type="text", text=json.dumps(o, ensure_ascii=False))]

    if name == "check_inventory":
        sku = arguments["sku"]
        return [TextContent(type="text", text=json.dumps({"sku": sku, "qty": INVENTORY.get(sku, 0)}))]

    if name == "create_refund":
        if arguments["order_id"] not in ORDERS:
            return [TextContent(type="text", text=json.dumps({"error": "order_not_found"}))]
        refund_id = f"REF-{arguments['order_id'][-4:]}-001"
        return [TextContent(type="text",
                            text=json.dumps({"refund_id": refund_id, "status": "approved"}, ensure_ascii=False))]

    raise ValueError(f"unknown tool: {name}")


async def main():
    async with stdio_server() as (read, write):
        await app.run(read, write, app.create_initialization_options())

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

2) Client: client_opus47.py

import os
import json
import asyncio
from openai import AsyncOpenAI
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client


กำหนดค่าเชื่อมต่อ HolySheep เท่านั้น ห้ามใช้ api.openai.com / api.anthropic.com

client = AsyncOpenAI(
    api_key=os.environ["HOLYSHEEP_API_KEY"],   # ใส่ YOUR_HOLYSHEEP_API_KEY
    base_url="https://api.holysheep.ai/v1",
)

SERVER_PARAMS = StdioServerParameters(command="python", args=["ecommerce_mcp_server.py"])


async def chat_once(user_message: str) -> str:
    async with stdio_client(SERVER_PARAMS) as (read, write):
        async with ClientSession(read, write) as session:
            await session.initialize()

            tools_resp = await session.list_tools()
            tools_schema = [
                {
                    "type": "function",
                    "function": {
                        "name": t.name,
                        "description": t.description,
                        "parameters": t.inputSchema,
                    },
                }
                for t in tools_resp.tools
            ]

            messages = [{"role": "user", "content": user_message}]

            resp = await client.chat.completions.create(
                model="claude-opus-4.7",
                messages=messages,
                tools=tools_schema,
                tool_choice="auto",
                max_tokens=1024,
            )
            msg = resp.choices[0].message
            messages.append(msg)

            # วนเรียก tool จนกว่าโมเดลจะหยุดขอ
            while msg.tool_calls:
                for tc in msg.tool_calls:
                    args = json.loads(tc.function.arguments or "{}")
                    result = await session.call_tool(tc.function.name, args)
                    messages.append({
                        "role": "tool",
                        "tool_call_id": tc.id,
                        "content": result.content[0].text,
                    })
                resp = await client.chat.completions.create(
                    model="claude-opus-4.7",
                    messages=messages,
                    tools=tools_schema,
                    tool_choice="auto",
                    max_tokens=1024,
                )
                msg = resp.choices[0].message
                messages.append(msg)

            return msg.content


async def main():
    q = "เช็คสถานะออร์เดอร์ ORD-1001 แล้วบอกด้วยว่าสินค้าในออร์เดอร์ยังมีในสต็อกไหม"
    print("User :", q)
    print("Asst :", await chat_once(q))

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

ผลลัพธ์ที่ผมรันในสภาพแวดล้อมของผม (Python 3.11, MCP SDK 1.2.0):

User : เช็คสถานะออร์เดอร์ ORD-1001 แล้วบอกด้วยว่าสินค้าในออร์เดอร์ยังมีในสต็อกไหม
Asst : ออร์เดอร์ ORD-1001 สถานะ "จัดส่งแล้ว" ยอดรวม 1,290 บาท สินค้า SKU-A1

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