สวัสดีครับ ผมเป็นนักพัฒนาที่ใช้เวลาหลายเดือนในการศึกษาและทดลองสร้างเครื่องมือดึงข้อมูลจากเว็บไซต์ วันนี้จะมาแบ่งปันประสบการณ์ตรงในการใช้ MCP (Model Context Protocol) ซึ่งเป็นเครื่องมือที่ทำให้การสร้าง Web Scraper ง่ายมากขึ้น แม้คุณไม่เคยมีประสบการณ์เขียนโค้ดมาก่อนก็ตาม
ในบทความนี้ คุณจะได้เรียนรู้วิธีดึงข้อมูลจากเว็บไซต์ต่าง ๆ โดยใช้ MCP ร่วมกับ AI จาก HolySheep AI ซึ่งมีความเร็วในการตอบสนองน้อยกว่า 50 มิลลิวินาที และราคาประหยัดกว่าผู้ให้บริการอื่นถึง 85% ขึ้นไป
MCP คืออะไร และทำไมต้องใช้?
MCP ย่อมาจาก Model Context Protocol เป็นมาตรฐานการเชื่อมต่อระหว่าง AI กับเครื่องมือภายนอก ลองนึกภาพว่า AI เปรียบเสมือนสมอง แต่สมองต้องการตาและมือในการมองเห็นและจับการสิ่งต่าง ๆ MCP ก็เป็นตัวเชื่อมที่ทำให้ AI สามารถเข้าถึงข้อมูลจากเว็บไซต์ได้โดยตรง
ก่อนที่จะเริ่ม คุณต้องมีสิ่งเหล่านี้:
- บัญชี HolySheep AI (สมัครฟรีที่ สมัครที่นี่)
- คอมพิวเตอร์ที่ติดตั้ง Python 3.8 ขึ้นไป
- โปรแกรมแก้ไขโค้ด เช่น VS Code
ขั้นตอนที่ 1: ติดตั้ง MCP Server
ขั้นแรก เราต้องติดตั้ง MCP Server ซึ่งเป็นตัวกลางที่ช่วยให้ AI สามารถเข้าถึงเว็บไซต์ได้ วิธีติดตั้งมีดังนี้:
# เปิด Terminal หรือ Command Prompt แล้วพิมพ์คำสั่งนี้
pip install mcp httpx beautifulsoup4
คำสั่งนี้จะติดตั้ง 3 อย่างพร้อมกัน
mcp คือตัวเชื่อมต่อ
httpx คือเครื่องมือขอข้อมูลจากเว็บ
beautifulsoup4 คือเครื่องมือแยกข้อมูลจากหน้าเว็บ
💡 เคล็ดลับ: หลังจากพิมพ์คำสั่งแล้ว รอสักครู่จนเห็นข้อความว่า "Successfully installed..." แสดงว่าติดตั้งสำเร็จแล้ว
ขั้นตอนที่ 2: สร้าง MCP Server สำหรับ Web Scraper
ตอนนี้เราจะมาสร้างไฟล์ที่ทำหน้าที่เป็น Web Scraper กัน สร้างไฟล์ใหม่ชื่อ web_scraper_server.py แล้วพิมพ์โค้ดด้านล่าง:
import httpx
from bs4 import BeautifulSoup
from mcp.server import Server
from mcp.types import Tool, TextContent
import asyncio
สร้าง MCP Server
server = Server("web-scraper")
@server.list_tools()
async def list_tools():
""" บอกรายการเครื่องมือที่ server มี """
return [
Tool(
name="scrape_website",
description="ดึงข้อมูลจากเว็บไซต์ โดยระบุ URL และ CSS selector",
inputSchema={
"type": "object",
"properties": {
"url": {
"type": "string",
"description": "URL ของเว็บไซต์ที่ต้องการดึงข้อมูล"
},
"selector": {
"type": "string",
"description": "CSS selector สำหรับเลือกองค์ประกอบที่ต้องการ"
}
},
"required": ["url"]
}
)
]
@server.call_tool()
async def call_tool(name: str, arguments: dict):
""" ทำงานเมื่อมีการเรียกใช้เครื่องมือ """
if name == "scrape_website":
url = arguments.get("url")
selector = arguments.get("selector", "body")
# ขอข้อมูลจากเว็บไซต์
async with httpx.AsyncClient() as client:
response = await client.get(url, timeout=30.0)
response.raise_for_status()
# แยกวิเคราะห์ข้อมูล
soup = BeautifulSoup(response.text, "html.parser")
if selector:
elements = soup.select(selector)
results = [elem.get_text(strip=True) for elem in elements]
return TextContent(type="text", text=f"พบ {len(results)} รายการ:\n" + "\n".join(results[:10]))
else:
return TextContent(type="text", text=soup.get_text()[:2000])
รัน server
if __name__ == "__main__":
import mcp.server.stdio
asyncio.run(server.run(mcp.server.stdio.stdio_server()))ขั้นตอนที่ 3: เชื่อมต่อกับ HolySheep AI
หลังจากสร้าง Server แล้ว ต่อไปเราจะสร้างไคลเอนต์ที่ใช้ HolySheep AI เป็นสมองในการควบคุม โค้ดด้านล่างจะทำให้ AI สามารถใช้เครื่องมือ Web Scraper ที่เราสร้างไว้:
import httpx
import json
ตั้งค่า HolySheep API
BASE_URL = "https://api.holysheep.ai/v1"
API_KEY = "YOUR_HOLYSHEEP_API_KEY" # ใส่ API Key ของคุณ
def call_holysheep(user_message: str, tools: list):
""" ส่งข้อความไปถาม HolySheep AI พร้อมเครื่องมือ """
headers = {
"Authorization": f"Bearer {API_KEY}",
"Content-Type": "application/json"
}
# กำหนดเครื่องมือที่ AI สามารถใช้ได้
payload = {
"model": "deepseek-v3.2",
"messages": [
{"role": "user", "content": user_message}
],
"tools": tools,
"temperature": 0.7
}
response = httpx.post(
f"{BASE_URL}/chat/completions",
headers=headers,
json=payload,
timeout=30.0
)
return response.json()
กำหนดเครื่องมือที่ให้ AI ใช้ได้
WEB_SCRAPER_TOOLS = [
{
"type": "function",
"function": {
"name": "scrape_website",
"description": "ดึงข้อมูลจากเว็บไซต์",
"parameters": {
"type": "object",
"properties": {
"url": {
"type": "string",
"description": "URL ของเว็บไซต์"
},
"selector": {
"type": "string",
"description": "CSS selector เช่น h1, .title, #header"
}
},
"required": ["url"]
}
}
}
]
ทดสอบการใช้งาน
if __name__ == "__main__":
result = call_holysheep(
"ไปดึงข้อมูลจาก https://example.com แล้วบอกหัวข้อหลักของเว็บไซต์",
WEB_SCRAPER_TOOLS
)
print(json.dumps(result, indent=2, ensure_ascii=False))📸 ภาพหน้าจอ: เมื่อรันโค้ดด้านบน คุณจะเห็นผลลัพธ์เป็น JSON ที่มีข้อมูลจากเว็บไซต์ example.com ถ้าสำเร็จจะมี field "content" ที่บอกหัวข้อหลักของเว็บไซต์
ขั้นตอนที่ 4: สร้าง Web Scraper ที่ใช้งานจริง
ตอนนี้เรามาสร้างเครื่องมือที่สมบูรณ์ขึ้น ซึ่งสามารถดึงข้อมูลหลายรายการพร้อมกัน และบันทึกลงไฟล์ได้:
import httpx
from bs4 import BeautifulSoup
import json
from datetime import datetime
class WebScraper:
""" คลาสสำหรับดึงข้อมูลจากเว็บไซต์ """
def __init__(self):
self.session = httpx.AsyncClient(
timeout=30.0,
headers={
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36"
}
)
self.results = []
async def scrape(self, url: str, selectors: dict) -> dict:
"""
ดึงข้อมูลจากเว็บไซต์
selectors: dict เช่น {"title": "h1", "links": "a", "paragraphs": "p"}
"""
try:
response = await self.session.get(url)
response.raise_for_status()
soup = BeautifulSoup(response.text, "html.parser")
data = {"url": url, "timestamp": datetime.now().isoformat()}
for key, selector in selectors.items():
elements = soup.select(selector)
data[key] = [elem.get_text(strip=True) for elem in elements]
return {"success": True, "data": data}
except Exception as e:
return {"success": False, "error": str(e)}
async def scrape_multiple(self, urls: list, selectors: dict) -> list:
""" ดึงข้อมูลจากหลายเว็บไซต์พร้อมกัน """
tasks = [self.scrape(url, selectors) for url in urls]
return await asyncio.gather(*tasks)
def save_to_file(self, filename: str = "scraped_data.json"):
""" บันทึกผลลัพธ์ลงไฟล์ JSON """
with open(filename, "w", encoding="utf-8") as f:
json.dump(self.results, f, ensure_ascii=False, indent=2)
print(f"บันทึกข้อมูลลงไฟล์ {filename} เรียบร้อยแล้ว")
async def close(self):
await self.session.aclose()
วิธีใช้งาน
import asyncio
async def main():
scraper = WebScraper()
# ตัวอย่าง: ดึงข่าวจากเว็บไซต์ข่าว
urls = [
"https://news.ycombinator.com/", # Hacker News
]
selectors = {
"titles": "span.titleline > a", # หัวข้อข่าว
"scores": "span.score", # คะแนนโหวต
}
results = await scraper.scrape_multiple(urls, selectors)
for result in results:
if result["success"]:
print(f"✅ ดึงข้อมูลจาก {result['data']['url']} สำเร็จ")
print(f" พบ {len(result['data']['titles'])} ข่าว")
print(f" ตัวอย่าง: {result['data']['titles'][0] if result['data']['titles'] else 'ไม่มี'}")
await scraper.close()
รันโปรแกรม
if __name__ == "__main__":
asyncio.run(main())ข้อผิดพลาดที่พบบ่อยและวิธีแก้ไข
จากประสบการณ์ที่ใช้งานจริง ผมพบข้อผิดพลาดหลายอย่างที่เกิดขึ้นบ่อย ขอแบ่งปันวิธีแก้ไขดังนี้:
กรณีที่ 1: ข้อผิดพลาด 403 Forbidden
ปัญหา: เว็บไซต์ปฏิเสธไม่ให้เข้าถึง แสดงข้อผิดพลาด 403
สาเหตุ: เว็บไซต์ตรวจพบว่าเราไม่ใช่เบราว์เซอร์จริง หรือไอพีถูกบล็อก
วิธีแก้ไข:
# เพิ่ม headers ที่จำลองเบราว์เซอร์จริง
headers = {
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36",
"Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,image/webp,*/*;q=0.8",
"Accept-Language": "th-TH,th;q=0.9,en-US;q=0.8,en;q=0.7",
"Accept-Encoding": "gzip, deflate, br",
"Connection": "keep-alive",
}
หรือใช้ proxy ถ้าถูกบล็อกไอพี
proxies = {
"http://": "http://your-proxy-server:port",
"https://": "http://your-proxy-server:port"
}
async with httpx.AsyncClient(headers=headers, proxies=proxies) as client:
response = await client.get(url)กรณีที่ 2: ข้อผิดพลาด AttributeError: 'NoneType' object has no attribute
ปัญห