PythonでMCPサーバーを構築する：ウェブ検索とスクレイピングガイド

Pythonの例を使ってAIアシスタント用のMCPサーバーを構築する

目次

モデルコンテキストプロトコル（MCP）は、AIアシスタントが外部データソースやツールとどのように相互作用するかを革命的に変えてきました。本ガイドでは、ウェブ検索およびスクレイピング機能に焦点を当てた例を用いて、MCPサーバーをPythonで構築する方法について説明します。

MCPロボット

モデルコンテキストプロトコルとは？

モデルコンテキストプロトコル（MCP）は、Anthropicが導入したオープンプロトコルで、AIアシスタントが外部システムに接続する方法を標準化します。各データソースにカスタム統合を構築する代わりに、MCPは統一されたインターフェースを提供し、以下を可能にします：

  • AIアシスタント（Claude、ChatGPT、またはカスタムLLMアプリケーションなど）がツールを発見し使用できるようにする
  • 開発者がデータソース、ツール、プロンプトを標準化されたプロトコルを通じて公開できるようにする
  • 各ユースケースで輪を再発明することなくシームレスな統合を実現する

このプロトコルはクライアント-サーバーのアーキテクチャで動作し、以下のように構成されます：

  • MCPクライアント（AIアシスタント）は、機能を発見し使用する
  • MCPサーバーは、リソース、ツール、プロンプトを公開する
  • 通信はJSON-RPCを介してstdioまたはHTTP/SSEで行われる

他の言語でMCPサーバーを実装したい場合は、GoでMCPサーバーを実装する方法に関するガイドをご覧ください。このガイドでは、プロトコル仕様とメッセージ構造について詳細に説明します。

なぜPythonでMCPサーバーを構築するのか？

PythonはMCPサーバー開発に最適な選択肢です。その理由は以下の通りです：

  1. 豊富なエコシステム：requests、beautifulsoup4、selenium、playwrightなどのライブラリにより、ウェブスクレイピングが簡単になります
  2. MCP SDK：公式のPython SDK（mcp）は、堅牢なサーバー実装をサポートします
  3. 迅速な開発：Pythonのシンプルさにより、プロトタイピングと反復が迅速に行えます
  4. AI/ML統合：langchain、openaiなどのAIライブラリやデータ処理ツールとの統合が容易です
  5. コミュニティサポート：豊富なコミュニティと、詳細なドキュメント、例が用意されています

開発環境の設定

まず、仮想環境を作成し、必要な依存関係をインストールしてください。仮想環境はPythonプロジェクトの分離に不可欠です。もし復習が必要な場合は、venvチートシートをご覧ください。

# 仮想環境の作成とアクティベート
python -m venv mcp-env
source mcp-env/bin/activate  # Windows: mcp-env\Scripts\activate

# MCP SDKとウェブスクレイピングライブラリのインストール
pip install mcp requests beautifulsoup4 playwright lxml
playwright install  # Playwright用のブラウザドライバをインストール

現代的な代替案：もし依存関係の解決とインストールをより高速に行いたい場合は、uv - 現代的なPythonパッケージおよび環境マネージャをご利用ください。これはpipよりも大規模なプロジェクトで大幅に高速です。

基本的なMCPサーバーの構築

最小限のMCPサーバー構造から始めましょう。Pythonに初めて触れるか、構文や一般的なパターンのクイックリファレンスが必要な場合は、Pythonチートシートをご覧ください。Pythonの基本を包括的に概説しています。

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

# サーバーインスタンスの作成
app = Server("websearch-scraper")

@app.list_tools()
async def list_tools() -> list[Tool]:
    """利用可能なツールを定義"""
    return [
        Tool(
            name="search_web",
            description="情報を検索する",
            inputSchema={
                "type": "object",
                "properties": {
                    "query": {
                        "type": "string",
                        "description": "検索クエリ"
                    },
                    "max_results": {
                        "type": "number",
                        "description": "最大結果数",
                        "default": 5
                    }
                },
                "required": ["query"]
            }
        )
    ]

@app.call_tool()
async def call_tool(name: str, arguments: dict) -> list[TextContent]:
    """ツールの実行を処理"""
    if name == "search_web":
        query = arguments["query"]
        max_results = arguments.get("max_results", 5)
        
        # 検索ロジックをここに実装
        results = await perform_web_search(query, max_results)
        
        return [TextContent(
            type="text",
            text=f"'{query}'の検索結果:\n\n{results}"
        )]
    
    raise ValueError(f"不明なツール: {name}")

async def perform_web_search(query: str, max_results: int) -> str:
    """実際の検索実装のプレースホルダー"""
    return f"{max_results}件の結果が見つかりました: {query}"

async def main():
    """サーバーを実行"""
    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())

ウェブ検索機能の実装

今度は、DuckDuckGoを使用して実際のウェブ検索ツールを実装しましょう（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]:
    """DuckDuckGoで検索し、結果を解析"""
    
    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"検索に失敗しました: {str(e)}")

def format_search_results(results: list[dict]) -> str:
    """表示用に検索結果をフォーマット"""
    if not results:
        return "結果が見つかりませんでした。"
    
    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)

ウェブスクレイピング機能の追加

ウェブページからコンテンツをスクレイピングするツールを追加しましょう。LLMと使用する際には、HTMLをMarkdown形式に変換することも検討してください。この目的のために、PythonでHTMLをMarkdownに変換する方法に関する包括的なガイドをご覧ください。6つの異なるライブラリをベンチマークと実用的な推奨事項とともに比較しています。

from playwright.async_api import async_playwright
import asyncio

async def scrape_webpage(url: str, selector: str = None) -> dict:
    """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')
            
            if selector:
                # 特定の要素を抽出
                element = await page.query_selector(selector)
                content = await element.inner_text() if element else "セレクタが見つかりませんでした"
            else:
                # 本文を抽出
                content = await page.inner_text('body')
            
            title = await page.title()
            
            return {
                "title": title,
                "content": content[:5000],  # コンテンツの長さを制限
                "url": url,
                "success": True
            }
            
        except Exception as e:
            return {
                "error": str(e),
                "url": url,
                "success": False
            }
        finally:
            await browser.close()

# MCPサーバーにスクレイパーのツールを追加
@app.list_tools()
async def list_tools() -> list[Tool]:
    return [
        Tool(
            name="search_web",
            description="DuckDuckGoを使用してウェブを検索",
            inputSchema={
                "type": "object",
                "properties": {
                    "query": {"type": "string", "description": "検索クエリ"},
                    "max_results": {"type": "number", "default": 5}
                },
                "required": ["query"]
            }
        ),
        Tool(
            name="scrape_webpage",
            description="ウェブページからコンテンツをスクレイピング",
            inputSchema={
                "type": "object",
                "properties": {
                    "url": {"type": "string", "description": "スクレイピングするURL"},
                    "selector": {
                        "type": "string",
                        "description": "特定のコンテンツを抽出するためのCSSセレクタ（オプション）"
                    }
                },
                "required": ["url"]
            }
        )
    ]

完全なMCPサーバーの実装

検索とスクレイピングの両方の機能を持つ、完全な生産性に適したMCPサーバーの実装例です：

#!/usr/bin/env python3
"""
ウェブ検索およびスクレイピング用のMCPサーバー
ウェブを検索し、ページからコンテンツを抽出するツールを提供
"""

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

# ロギングの設定
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("websearch-scraper")

# サーバーの作成
app = Server("websearch-scraper")

# 検索の実装
async def search_web(query: str, max_results: int = 5) -> str:
    """DuckDuckGoで検索し、フォーマットされた結果を返す"""
    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 ""
                })
        
        # 結果のフォーマット
        if not results:
            return "結果が見つかりませんでした。"
        
        formatted = [f"{len(results)}件の結果が'{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"検索に失敗しました: {e}")
        return f"検索エラー: {str(e)}"

# スクレイパーの実装
async def scrape_page(url: str, selector: str = None) -> str:
    """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 "セレクタが見つかりませんでした"
            else:
                content = await page.inner_text('body')
            
            # コンテンツの長さを制限
            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"スクレイピングに失敗しました: {e}")
            return f"スクレイピングエラー: {str(e)}"
        finally:
            await browser.close()

# MCPツールの定義
@app.list_tools()
async def list_tools() -> list[Tool]:
    """利用可能なMCPツールを一覧表示"""
    return [
        Tool(
            name="search_web",
            description="DuckDuckGoを使用してウェブを検索。タイトル、スニペット、URLを返します。",
            inputSchema={
                "type": "object",
                "properties": {
                    "query": {
                        "type": "string",
                        "description": "検索クエリ"
                    },
                    "max_results": {
                        "type": "number",
                        "description": "最大結果数（デフォルト: 5）",
                        "default": 5
                    }
                },
                "required": ["query"]
            }
        ),
        Tool(
            name="scrape_webpage",
            description="ウェブページからコンテンツを抽出。CSSセレクタを使用して特定の要素を抽出できます。",
            inputSchema={
                "type": "object",
                "properties": {
                    "url": {
                        "type": "string",
                        "description": "スクレイピングするURL"
                    },
                    "selector": {
                        "type": "string",
                        "description": "特定のコンテンツを抽出するためのCSSセレクタ（オプション）"
                    }
                },
                "required": ["url"]
            }
        )
    ]

@app.call_tool()
async def call_tool(name: str, arguments: Any) -> list[TextContent]:
    """ツールの実行を処理"""
    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"不明なツール: {name}")
    
    except Exception as e:
        logger.error(f"ツールの実行に失敗しました: {e}")
        return [TextContent(
            type="text",
            text=f"{name}の実行エラー: {str(e)}"
        )]

async def main():
    """MCPサーバーを実行"""
    logger.info("WebSearch-Scraper MCPサーバーを起動中")
    
    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())

MCPサーバーの設定

Claude Desktopや他のMCPクライアントでMCPサーバーを使用するには、設定ファイルを作成してください：

Claude Desktop用（claude_desktop_config.json）：

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

場所：

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

MCPサーバーのテスト

機能を確認するためのテストスクリプトを作成してください：

import asyncio
import json
import sys
from io import StringIO

async def test_mcp_server():
    """ローカルでMCPサーバーをテスト"""
    
    # 検索テスト
    print("ウェブ検索をテスト中...")
    results = await search_web("Python MCPチュートリアル", 3)
    print(results)
    
    # スクレイパーのテスト
    print("\n\nウェブページスクレイパーをテスト中...")
    content = await scrape_page("https://example.com")
    print(content[:500])

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

高度な機能とベストプラクティス

1. レート制限

ターゲットサーバーを過剰に負荷しないようにレート制限を実装してください：

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("レート制限を超えました")
        
        self.requests[key].append(now)

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

2. キャッシュ

パフォーマンスを向上させるためにキャッシュを追加してください：

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. エラーハンドリング

堅牢なエラーハンドリングを実装してください：

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"エラー ({error_type.value}): {str(error)}"

4. 入力検証

処理前にユーザー入力を検証してください：

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

デプロイに関する考慮事項

ウェブデプロイ用にSSEトランスポートを使用する

ウェブベースのデプロイでは、SSE（Server-Sent Events）トランスポートを使用してください。サーバーレスデプロイを検討している場合は、AWS LambdaのパフォーマンスをJavaScript、Python、Golangで比較に関するガイドを参考に、ランタイムについての決定を下してください：

import mcp.server.sse

async def main_sse():
    """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()

AWS Lambdaデプロイ

MCPサーバーは、SSEトランスポートを使用する場合、AWS Lambda関数としてデプロイすることもできます。Lambdaデプロイに関する包括的なガイド：

Dockerデプロイ

コンテナ化されたデプロイのためにDockerfileを作成してください：

FROM python:3.11-slim

WORKDIR /app

# システム依存関係のインストール
RUN apt-get update && apt-get install -y \
    wget \
    && rm -rf /var/lib/apt/lists/*

# Python依存関係のインストール
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
RUN playwright install chromium
RUN playwright install-deps

# アプリケーションのコピー
COPY mcp_server.py .

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

パフォーマンス最適化

非同期操作

非同期操作を使用して並列処理を行ってください：

async def search_multiple_queries(queries: list[str]) -> list[str]:
    """複数のクエリを同時に検索"""
    tasks = [search_web(query) for query in queries]
    results = await asyncio.gather(*tasks)
    return results

接続プール

パフォーマンス向上のため接続を再利用してください：

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()

セキュリティのベストプラクティス

  1. 入力のサニタイズ: すべてのユーザー入力を常に検証し、サニタイズしてください
  2. URLホワイトリスト: スクレイピングのためにURLホワイトリストの実装を検討してください
  3. タイムアウト制御: リソース枯渇を防ぐために適切なタイムアウトを設定してください
  4. コンテンツ制限: スクレイピングされたコンテンツのサイズを制限してください
  5. 認証: 本番環境での展開のために認証を実装してください
  6. HTTPS: 本番環境でのSSEトランスポートにHTTPSを使用してください

異なるLLMプロバイダーとの連携

MCPはAnthropicがClaudeのために開発したが、このプロトコルはどのLLMとも連携できるように設計されています。複数のAIプロバイダーと連携するMCPサーバーを構築し、構造化された出力を必要とする場合は、人気のあるLLMプロバイダーにおける構造化出力の比較を確認してください。OpenAI、Gemini、Anthropic、Mistral、AWS Bedrockが含まれます。

有用なリンクとリソース

関連リソース

MCPとプロトコルの実装

Python開発

Webスクレイピングとコンテンツ処理

サーバーレス展開リソース

LLM統合

結論

PythonでMCPサーバーを構築することで、カスタムツールやデータソースでAIアシスタントを拡張する強力な可能性が開かれます。ここでは示したウェブ検索とスクレイピングの機能はあくまで始まりに過ぎず、この基盤を拡張してデータベース、API、ファイルシステム、ほぼすべての外部システムを統合できます。

Model Context Protocolはまだ進化中ですが、AIツールの統合に標準的なアプローチを採用しているため、次世代のAIアプリケーションを開発する開発者にとって非常に魅力的な技術です。組織内でのツールの作成であれ、コミュニティ向けのMCPサーバーの構築であれ、Pythonは迅速な開発と展開に最適な基盤を提供します。