Python MCP Server Client

🚀 从零からのMCP ServerとMCP Clientの構築完全ガイド

このドキュメントは、Pythonを使用してModel Context Protocol (MCP) のサーバー（Server）とクライアント（Client）をゼロから構築する方法を詳しく説明します。

🚀 クイックスタート

目次

1. 依存関係のインストール
2. プロジェクトの初期化
3. コードの記述
4. 起動コマンド

📦 インストール

始める前に、以下のPythonパッケージがインストールされていることを確認してください。

pip install python-multipart requests asyncio websockets uvicorn serper-dev-api

これらのパッケージには以下のものが含まれます：

• python-multipart: multipartリクエストを処理します。
• requests: HTTPリクエストを送信します。
• asyncio と websockets: 非同期通信とWebSocketサポートを実現します。
• uvicorn: FastAPIアプリケーションを実行するために使用します。
• serper-dev-api: Serper.devのAPIを使用してウェブページの内容を解析します。

💻 使用例

プロジェクトの初期化

新しいディレクトリを作成し、以下のファイルを追加します。

server.py

これはMCP Serverの核心コードで、WebSocket接続とツール呼び出しを処理します。

import asyncio
import websockets
import json
from typing import List, Dict
from fastapi import FastAPI, HTTPException
from fastapi.responses import JSONResponse
from serper-dev-api import Serper

app = FastAPI()
serper = Serper(your_api_key)  # あなたのSerper API Keyを入力してください

# ツールインターフェースを定義します
class Tool:
    def __init__(self, name: str, description: str, input_schema: Dict):
        self.name = name
        self.description = description
        self.input_schema = input_schema

async def serve():
    async with websockets.connect("ws://localhost:8000") as websocket:
        await websocket.accept()
        print("接続が確立されました！")
        
        while True:
            try:
                # ツール呼び出しリクエストを受信します
                tool_call = await websocket.recv()
                print(f"ツール呼び出しを受け取りました：{tool_call}")
                
                # ツールパラメータを解析します
                params = json.loads(tool_call)
                tool_name = params["name"]
                arguments = params["arguments"]
                
                if tool_name == "search":
                    result = serper.search(arguments, num_results=3)
                    await websocket.send(json.dumps(result))
                elif tool_name == "parse_website":
                    # ウェブページの内容を解析する関数があると仮定します
                    content = parse_website(arguments)
                    await websocket.send(content)
                
            except Exception as e:
                print(f"エラー：{str(e)}")

client.py

これはMCP Clientの核心コードで、サーバーと通信し、ユーザーのクエリを処理します。

import asyncio
from typing import Dict, List
from fastapi import FastAPI, HTTPException
from fastapi.responses import JSONResponse
from serper-dev-api import Serper

app = FastAPI()
serper = Serper(your_api_key)  # あなたのSerper API Keyを入力してください

async def main():
    async with websockets.connect("ws://localhost:8001") as websocket:
        await websocket.accept()
        print("MCP Serverに接続しました！")
        
        while True:
            try:
                # ツール呼び出しリクエストを送信します
                tool_name = "search"
                arguments = {"query": "Python チュートリアル"}
                tool_call = json.dumps({"name": tool_name, "arguments": arguments})
                
                await websocket.send(tool_call)
                response = await websocket.recv()
                print(f"応答を受け取りました：{response}")
            except Exception as e:
                print(f"エラー：{str(e)}")

起動コマンド

ターミナルで以下のコマンドを実行して、サーバーとクライアントを起動します。

サーバーの起動

python server.py serve

クライアントの起動

python client.py main

📚 ドキュメント

参考資料

• Model Context Protocol (MCP)
• Serper.dev API
• FastAPI 公式ドキュメント

以上がPythonを使用してMCP ServerとClientを構築する完全ガイドです。何か問題があれば、関連ドキュメントを参照するか、コミュニティで支援を求めてください！