金大哥 - python-mcp-server-client MCP 详情

article

README

🚀 从零开始搭建 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

📚 详细文档

参考资料

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

python-mcp-server-client