Membangun Server MCP dalam Python: Panduan Pencarian Web & Scraping

Bangun server MCP untuk asisten AI dengan contoh Python

Konten Halaman

Model Context Protocol (MCP) sedang merevolusi cara asisten AI berinteraksi dengan sumber data eksternal dan alat. Dalam panduan ini, kita akan menjelaskan bagaimana membangun server MCP dalam Python, dengan contoh yang berfokus pada kemampuan pencarian web dan pengambilan data.

MCP robots

Apa itu Model Context Protocol?

Model Context Protocol (MCP) adalah protokol terbuka yang diperkenalkan oleh Anthropic untuk menstandarkan cara asisten AI terhubung dengan sistem eksternal. Daripada membangun integrasi khusus untuk setiap sumber data, MCP menyediakan antarmuka yang terpadu yang memungkinkan:

  • Asisten AI (seperti Claude, ChatGPT, atau aplikasi LLM kustom) untuk menemukan dan menggunakan alat
  • Pengembang untuk menampilkan sumber data, alat, dan prompt melalui protokol standar
  • Integrasi yang mulus tanpa perlu merekayasa ulang untuk setiap kasus penggunaan

Protokol ini beroperasi pada arsitektur klien-server di mana:

  • Klien MCP (asisten AI) menemukan dan menggunakan kemampuan
  • Server MCP menampilkan sumber daya, alat, dan prompt
  • Komunikasi terjadi melalui JSON-RPC melalui stdio atau HTTP/SSE

Jika Anda tertarik menerapkan server MCP dalam bahasa lain, lihat panduan kami tentang implementasi server MCP dalam Go, yang membahas spesifikasi protokol dan struktur pesan secara rinci.

Mengapa Membangun Server MCP dalam Python?

Python adalah pilihan yang sangat baik untuk pengembangan server MCP karena:

  1. Ekosistem yang Kaya: Perpustakaan seperti requests, beautifulsoup4, selenium, dan playwright membuat pengambilan data web menjadi sederhana
  2. SDK MCP: SDK Python resmi (mcp) menyediakan dukungan implementasi server yang kuat
  3. Pengembangan Cepat: Sederhananya Python memungkinkan prototipe dan iterasi yang cepat
  4. Integrasi AI/ML: Mudah terintegrasi dengan perpustakaan AI seperti langchain, openai, dan alat pemrosesan data
  5. Dukungan Komunitas: Komunitas besar dengan dokumentasi dan contoh yang luas

Menyiapkan Lingkungan Pengembangan Anda

Pertama, buat lingkungan virtual dan instal dependensi yang diperlukan. Menggunakan lingkungan virtual sangat penting untuk isolasi proyek Python - jika Anda membutuhkan refresher, lihat panduan kami tentang cheatsheet venv untuk perintah dan praktik terbaik yang rinci.

# Membuat dan mengaktifkan lingkungan virtual
python -m venv mcp-env
source mcp-env/bin/activate  # Di Windows: mcp-env\Scripts\activate

# Menginstal SDK MCP dan perpustakaan pengambilan data web
pip install mcp requests beautifulsoup4 playwright lxml
playwright install  # Menginstal driver browser untuk Playwright

Alternatif Modern: Jika Anda lebih suka resolusi dan instalasi dependensi yang lebih cepat, pertimbangkan untuk menggunakan uv - manajer paket dan lingkungan Python modern yang dapat jauh lebih cepat daripada pip untuk proyek besar.

Membangun Server MCP Dasar

Mari mulai dengan struktur server MCP minimal. Jika Anda baru dengan Python atau membutuhkan referensi cepat untuk sintaks dan pola umum, cheatsheet Python kami menyediakan gambaran menyeluruh tentang dasar-dasar Python.

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

# Membuat instance server
app = Server("websearch-scraper")

@app.list_tools()
async def list_tools() -> list[Tool]:
    """Mendefinisikan alat yang tersedia"""
    return [
        Tool(
            name="search_web",
            description="Mencari informasi di web",
            inputSchema={
                "type": "object",
                "properties": {
                    "query": {
                        "type": "string",
                        "description": "Query pencarian"
                    },
                    "max_results": {
                        "type": "number",
                        "description": "Jumlah maksimum hasil",
                        "default": 5
                    }
                },
                "required": ["query"]
            }
        )
    ]

@app.call_tool()
async def call_tool(name: str, arguments: dict) -> list[TextContent]:
    """Menangani eksekusi alat"""
    if name == "search_web":
        query = arguments["query"]
        max_results = arguments.get("max_results", 5)
        
        # Implementasikan logika pencarian di sini
        results = await perform_web_search(query, max_results)
        
        return [TextContent(
            type="text",
            text=f"Hasil pencarian untuk '{query}':\n\n{results}"
        )]
    
    raise ValueError(f"Alat tidak dikenal: {name}")

async def perform_web_search(query: str, max_results: int) -> str:
    """Penempat untuk implementasi pencarian sebenarnya"""
    return f"Menemukan {max_results} hasil untuk: {query}"

async def main():
    """Menjalankan server"""
    async with mcp.server.stdio.stdio_server() as (read_stream, write_stream):
        await app.run(
            read_stream,
            write_stream,
            app.create_initialization_options()
        )

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

Menerapkan Fungsi Pencarian Web

Sekarang mari kita implementasikan alat pencarian web nyata menggunakan DuckDuckGo (yang tidak memerlukan kunci API):

import asyncio
import requests
from bs4 import BeautifulSoup
from urllib.parse import quote_plus

async def search_duckduckgo(query: str, max_results: int = 5) -> list[dict]:
    """Mencari DuckDuckGo dan memasak hasil"""
    
    url = f"https://html.duckduckgo.com/html/?q={quote_plus(query)}"
    headers = {
        "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36"
    }
    
    try:
        response = requests.get(url, headers=headers, timeout=10)
        response.raise_for_status()
        
        soup = BeautifulSoup(response.text, 'html.parser')
        results = []
        
        for result in soup.select('.result')[:max_results]:
            title_elem = result.select_one('.result__title')
            snippet_elem = result.select_one('.result__snippet')
            url_elem = result.select_one('.result__url')
            
            if title_elem and snippet_elem:
                results.append({
                    "title": title_elem.get_text(strip=True),
                    "snippet": snippet_elem.get_text(strip=True),
                    "url": url_elem.get_text(strip=True) if url_elem else "N/A"
                })
        
        return results
        
    except Exception as e:
        raise Exception(f"Pencarian gagal: {str(e)}")

def format_search_results(results: list[dict]) -> str:
    """Memformat hasil pencarian untuk tampilan"""
    if not results:
        return "Tidak ada hasil yang ditemukan."
    
    formatted = []
    for i, result in enumerate(results, 1):
        formatted.append(
            f"{i}. {result['title']}\n"
            f"   {result['snippet']}\n"
            f"   URL: {result['url']}\n"
        )
    
    return "\n".join(formatted)

Menambahkan Kemampuan Pengambilan Data Web

Mari tambahkan alat untuk mengambil dan mengekstrak konten dari halaman web. Saat mengambil konten HTML untuk digunakan dengan LLM, Anda mungkin juga ingin mengubahnya menjadi format Markdown untuk pemrosesan yang lebih baik. Untuk tujuan ini, lihat panduan menyeluruh kami tentang mengubah HTML ke Markdown dengan Python, yang membandingkan 6 perpustakaan berbeda dengan benchmark dan rekomendasi praktis.

from playwright.async_api import async_playwright
import asyncio

async def scrape_webpage(url: str, selector: str = None) -> dict:
    """Mengambil konten dari halaman web menggunakan Playwright"""
    
    async with async_playwright() as p:
        browser = await p.chromium.launch(headless=True)
        page = await browser.new_page()
        
        try:
            await page.goto(url, timeout=30000)
            
            # Tunggu konten untuk dimuat
            await page.wait_for_load_state('networkidle')
            
            if selector:
                # Ekstrak elemen tertentu
                element = await page.query_selector(selector)
                content = await element.inner_text() if element else "Selector tidak ditemukan"
            else:
                # Ekstrak konten utama
                content = await page.inner_text('body')
            
            title = await page.title()
            
            return {
                "title": title,
                "content": content[:5000],  # Batasi panjang konten
                "url": url,
                "success": True
            }
            
        except Exception as e:
            return {
                "error": str(e),
                "url": url,
                "success": False
            }
        finally:
            await browser.close()

# Tambahkan alat scraper ke server MCP
@app.list_tools()
async def list_tools() -> list[Tool]:
    return [
        Tool(
            name="search_web",
            description="Mencari web menggunakan DuckDuckGo",
            inputSchema={
                "type": "object",
                "properties": {
                    "query": {"type": "string", "description": "Query pencarian"},
                    "max_results": {"type": "number", "default": 5}
                },
                "required": ["query"]
            }
        ),
        Tool(
            name="scrape_webpage",
            description="Mengambil konten dari halaman web",
            inputSchema={
                "type": "object",
                "properties": {
                    "url": {"type": "string", "description": "URL untuk diambil"},
                    "selector": {
                        "type": "string",
                        "description": "Selector CSS opsional untuk konten tertentu"
                    }
                },
                "required": ["url"]
            }
        )
    ]

Implementasi Server MCP Lengkap

Berikut adalah implementasi server MCP lengkap yang siap diproduksi dengan kemampuan pencarian dan pengambilan data:

#!/usr/bin/env python3
"""
Server MCP untuk Pencarian Web dan Pengambilan Data
Menyediakan alat untuk mencari web dan mengekstrak konten dari halaman
"""

import asyncio
import logging
from typing import Any
import requests
from bs4 import BeautifulSoup
from urllib.parse import quote_plus
from playwright.async_api import async_playwright

from mcp.server import Server
from mcp.types import Tool, TextContent, ImageContent, EmbeddedResource
import mcp.server.stdio

# Konfigurasi logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("websearch-scraper")

# Membuat server
app = Server("websearch-scraper")

# Implementasi pencarian
async def search_web(query: str, max_results: int = 5) -> str:
    """Mencari DuckDuckGo dan mengembalikan hasil yang diformat"""
    url = f"https://html.duckduckgo.com/html/?q={quote_plus(query)}"
    headers = {"User-Agent": "Mozilla/5.0"}
    
    try:
        response = requests.get(url, headers=headers, timeout=10)
        soup = BeautifulSoup(response.text, 'html.parser')
        results = []
        
        for result in soup.select('.result')[:max_results]:
            title = result.select_one('.result__title')
            snippet = result.select_one('.result__snippet')
            link = result.select_one('.result__url')
            
            if title and snippet:
                results.append({
                    "title": title.get_text(strip=True),
                    "snippet": snippet.get_text(strip=True),
                    "url": link.get_text(strip=True) if link else ""
                })
        
        # Format hasil
        if not results:
            return "Tidak ada hasil yang ditemukan."
        
        formatted = [f"Menemukan {len(results)} hasil untuk '{query}':\n"]
        for i, r in enumerate(results, 1):
            formatted.append(f"\n{i}. **{r['title']}**")
            formatted.append(f"   {r['snippet']}")
            formatted.append(f"   {r['url']}")
        
        return "\n".join(formatted)
        
    except Exception as e:
        logger.error(f"Pencarian gagal: {e}")
        return f"Kesalahan pencarian: {str(e)}"

# Implementasi scraper
async def scrape_page(url: str, selector: str = None) -> str:
    """Mengambil konten halaman web menggunakan Playwright"""
    async with async_playwright() as p:
        browser = await p.chromium.launch(headless=True)
        page = await browser.new_page()
        
        try:
            await page.goto(url, timeout=30000)
            await page.wait_for_load_state('networkidle')
            
            title = await page.title()
            
            if selector:
                element = await page.query_selector(selector)
                content = await element.inner_text() if element else "Selector tidak ditemukan"
            else:
                content = await page.inner_text('body')
            
            # Batasi panjang konten
            content = content[:8000] + "..." if len(content) > 8000 else content
            
            result = f"**{title}**\n\nURL: {url}\n\n{content}"
            return result
            
        except Exception as e:
            logger.error(f"Pengambilan data gagal: {e}")
            return f"Kesalahan pengambilan data: {str(e)}"
        finally:
            await browser.close()

# Definisi alat MCP
@app.list_tools()
async def list_tools() -> list[Tool]:
    """Daftar alat MCP yang tersedia"""
    return [
        Tool(
            name="search_web",
            description="Mencari web menggunakan DuckDuckGo. Mengembalikan judul, snippet, dan URL.",
            inputSchema={
                "type": "object",
                "properties": {
                    "query": {
                        "type": "string",
                        "description": "Query pencarian"
                    },
                    "max_results": {
                        "type": "number",
                        "description": "Jumlah maksimum hasil (default: 5)",
                        "default": 5
                    }
                },
                "required": ["query"]
            }
        ),
        Tool(
            name="scrape_webpage",
            description="Mengekstrak konten dari halaman web. Dapat menargetkan elemen tertentu dengan selector CSS.",
            inputSchema={
                "type": "object",
                "properties": {
                    "url": {
                        "type": "string",
                        "description": "URL untuk diambil"
                    },
                    "selector": {
                        "type": "string",
                        "description": "Selector CSS opsional untuk mengekstrak konten tertentu"
                    }
                },
                "required": ["url"]
            }
        )
    ]

@app.call_tool()
async def call_tool(name: str, arguments: Any) -> list[TextContent]:
    """Menangani eksekusi alat"""
    try:
        if name == "search_web":
            query = arguments["query"]
            max_results = arguments.get("max_results", 5)
            result = await search_web(query, max_results)
            return [TextContent(type="text", text=result)]
        
        elif name == "scrape_webpage":
            url = arguments["url"]
            selector = arguments.get("selector")
            result = await scrape_page(url, selector)
            return [TextContent(type="text", text=result)]
        
        else:
            raise ValueError(f"Alat tidak dikenal: {name}")
    
    except Exception as e:
        logger.error(f"Eksekusi alat gagal: {e}")
        return [TextContent(
            type="text",
            text=f"Kesalahan eksekusi {name}: {str(e)}"
        )]

async def main():
    """Menjalankan server MCP"""
    logger.info("Memulai Server MCP WebSearch-Scraper")
    
    async with mcp.server.stdio.stdio_server() as (read_stream, write_stream):
        await app.run(
            read_stream,
            write_stream,
            app.create_initialization_options()
        )

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

Mengonfigurasi Server MCP Anda

Untuk menggunakan server MCP Anda dengan Claude Desktop atau klien MCP lainnya, buat file konfigurasi:

Untuk Claude Desktop (claude_desktop_config.json):

{
  "mcpServers": {
    "websearch-scraper": {
      "command": "python",
      "args": [
        "/path/to/your/mcp_server.py"
      ],
      "env": {}
    }
  }
}

Lokasi:

  • macOS: ~/Library/Application Support/Claude/claude_desktop_config.json
  • Windows: %APPDATA%\Claude\claude_desktop_config.json
  • Linux: ~/.config/Claude/claude_desktop_config.json

Menguji Server MCP Anda

Buat skrip pengujian untuk memverifikasi fungsi:

import asyncio
import json
import sys
from io import StringIO

async def test_mcp_server():
    """Menguji server MCP secara lokal"""
    
    # Uji pencarian
    print("Menguji pencarian web...")
    results = await search_web("Tutorial MCP Python", 3)
    print(results)
    
    # Uji scraper
    print("\n\nMenguji scraper halaman web...")
    content = await scrape_page("https://example.com")
    print(content[:500])

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

Fitur Lanjutan dan Praktik Terbaik

1. Pembatasan Tingkat

Implementasikan pembatasan tingkat untuk menghindari membanjiri server target:

import time
from collections import defaultdict

class RateLimiter:
    def __init__(self, max_requests: int = 10, time_window: int = 60):
        self.max_requests = max_requests
        self.time_window = time_window
        self.requests = defaultdict(list)
    
    async def acquire(self, key: str):
        now = time.time()
        self.requests[key] = [
            t for t in self.requests[key] 
            if now - t < self.time_window
        ]
        
        if len(self.requests[key]) >= self.max_requests:
            raise Exception("Melebihi batas tingkat")
        
        self.requests[key].append(now)

limiter = RateLimiter(max_requests=10, time_window=60)

2. Penyimpanan Cache

Tambahkan penyimpanan cache untuk meningkatkan kinerja:

from functools import lru_cache
import hashlib

@lru_cache(maxsize=100)
async def cached_search(query: str, max_results: int):
    return await search_web(query, max_results)

3. Penanganan Kesalahan

Implementasikan penanganan kesalahan yang kuat:

from enum import Enum

class ErrorType(Enum):
    NETWORK_ERROR = "network_error"
    PARSE_ERROR = "parse_error"
    RATE_LIMIT = "rate_limit_exceeded"
    INVALID_INPUT = "invalid_input"

def handle_error(error: Exception, error_type: ErrorType) -> str:
    logger.error(f"{error_type.value}: {str(error)}")
    return f"Kesalahan ({error_type.value}): {str(error)}"

4. Validasi Input

Validasi input pengguna sebelum diproses:

from urllib.parse import urlparse

def validate_url(url: str) -> bool:
    try:
        result = urlparse(url)
        return all([result.scheme, result.netloc])
    except:
        return False

def validate_query(query: str) -> bool:
    return len(query.strip()) > 0 and len(query) < 500

Pertimbangan Pengujian

Menggunakan Transport SSE untuk Pengujian Web

Untuk pengujian berbasis web, gunakan transport SSE (Server-Sent Events). Jika Anda mempertimbangkan pengujian serverless, Anda mungkin tertarik membandingkan kinerja AWS Lambda di seluruh JavaScript, Python, dan Golang untuk membuat keputusan yang terinformasi tentang runtime Anda:

import mcp.server.sse

async def main_sse():
    """Menjalankan server dengan transport SSE"""
    from starlette.applications import Starlette
    from starlette.routing import Mount
    
    sse = mcp.server.sse.SseServerTransport("/messages")
    
    starlette_app = Starlette(
        routes=[
            Mount("/mcp", app=sse.get_server())
        ]
    )
    
    import uvicorn
    await uvicorn.Server(
        config=uvicorn.Config(starlette_app, host="0.0.0.0", port=8000)
    ).serve()

Pengujian AWS Lambda

Server MCP juga dapat dideploy sebagai fungsi AWS Lambda, terutama ketika menggunakan transport SSE. Untuk panduan menyeluruh tentang pengujian Lambda:

Pengujian Docker

Buat Dockerfile untuk pengujian berbasis kontainer:

FROM python:3.11-slim

WORKDIR /app

# Instal dependensi sistem
RUN apt-get update && apt-get install -y \
    wget \
    && rm -rf /var/lib/apt/lists/*

# Instal dependensi Python
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
RUN playwright install chromium
RUN playwright install-deps

# Salin aplikasi
COPY mcp_server.py .

CMD ["python", "mcp_server.py"]

Optimisasi Kinerja

Operasi Async

Gunakan asyncio untuk operasi konkuren:

async def search_multiple_queries(queries: list[str]) -> list[str]:
    """Mencari beberapa query secara konkuren"""
    tasks = [search_web(query) for query in queries]
    results = await asyncio.gather(*tasks)
    return results

Pem pooling Koneksi

Ulangi koneksi untuk kinerja yang lebih baik:

import aiohttp

session = None

async def get_session():
    global session
    if session is None:
        session = aiohttp.ClientSession()
    return session

async def fetch_url(url: str) -> str:
    session = await get_session()
    async with session.get(url) as response:
        return await response.text()

Praktik Keamanan Terbaik

  1. Sanitasi Input: Selalu validasi dan bersihkan input pengguna
  2. Daftar Putih URL: Pertimbangkan menerapkan daftar putih URL untuk pengambilan data
  3. Kontrol Timeout: Tetapkan timeout yang tepat untuk mencegah kehabisan sumber daya
  4. Batas Konten: Batasi ukuran konten yang diambil
  5. Autentikasi: Implementasikan autentikasi untuk deployment produksi
  6. HTTPS: Gunakan HTTPS untuk transport SSE dalam produksi

Bekerja dengan Berbagai Penyedia LLM

Meskipun MCP dikembangkan oleh Anthropic untuk Claude, protokol dirancang untuk bekerja dengan LLM apa pun. Jika Anda sedang membangun server MCP yang berinteraksi dengan berbagai penyedia AI dan membutuhkan output terstruktur, Anda perlu melihat perbandingan output terstruktur di berbagai penyedia LLM populer termasuk OpenAI, Gemini, Anthropic, Mistral, dan AWS Bedrock.

Tautan dan Sumber Daya yang Berguna

Sumber Daya Terkait

MCP dan Implementasi Protokol

Pengembangan Python

Pengambilan Data Web dan Pemrosesan Konten

Sumber Daya Deployment Serverless

Integrasi LLM

Kesimpulan

Membangun server MCP dalam Python membuka kemungkinan kuat untuk memperluas asisten AI dengan alat kustom dan sumber daya data. Kemampuan pencarian web dan pengambilan data yang ditunjukkan di sini hanyalah awal—Anda dapat memperluas fondasi ini untuk mengintegrasikan database, API, sistem file, dan hampir setiap sistem eksternal.

Model Context Protocol masih berkembang, tetapi pendekatannya yang standar untuk integrasi alat AI membuatnya menjadi teknologi yang menarik bagi pengembang yang membangun generasi berikutnya aplikasi berbasis AI. Baik Anda sedang membuat alat internal untuk organisasi Anda atau membangun server MCP publik untuk komunitas, Python memberikan fondasi yang sangat baik untuk pengembangan dan deployment yang cepat.