从零开始搭建 MCP Server 和 MCP Client 的完整指南（中文版）

以下是如何使用 Python 从零开始搭建 Model Context Protocol (MCP) 服务端（Server）和客户端（Client）的完整教程。

目录

1. 安装依赖
2. 初始化项目
3. 编写代码
4. 启动命令

1. 安装依赖

在开始之前，确保你已经安装了以下 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 解析网页内容。

2. 初始化项目

创建一个新目录，并在其中添加以下文件：

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

3. 启动命令

在终端中运行以下命令启动 Server 和 Client：

启动 Server

python server.py serve

启动 Client

python client.py main

4. 参考资料

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

以上就是如何使用 Python 搭建 MCP Server 和 Client 的完整指南。如果有任何问题，请随时查阅相关文档或在社区中寻求帮助！