Python MCP Server Client

🚀 從零開始搭建 MCP Server 和 MCP Client 的完整指南（中文版）

本指南將詳細介紹如何使用 Python 從零開始搭建 Model Context Protocol (MCP) 服務端（Server）和客戶端（Client），助你輕鬆上手 MCP 的開發。

🚀 快速開始

安裝依賴

在開始之前，確保你已經安裝了以下 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 的核心代碼，負責與 Server 通信並處理用戶查詢。

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)}")

啟動命令

在終端中運行以下命令啟動 Server 和 Client：

啟動 Server

python server.py serve

啟動 Client

python client.py main

📚 詳細文檔

參考資料

• Model Context Protocol (MCP)
• Serper.dev API
• FastAPI 官方文檔

以上就是如何使用 Python 搭建 MCP Server 和 Client 的完整指南。如果有任何問題，請隨時查閱相關文檔或在社區中尋求幫助！