MCP 服务器

Step-by-Step Guide to Build Your Own MCP Server and MCP Client with UI Using Python

mkcod

开发者工具

README

使用 Python 构建您自己的带有 UI 的 MCP 服务器和 MCP 客户端的分步指南

本综合指南将引导您完成使用 Python 创建模型上下文协议 (MCP) 服务器和带有图形用户界面的客户端的过程。 MCP 旨在促进语言模型和工具提供商之间的通信，从而实现强大的基于 AI 的应用程序。在本教程结束时，您将拥有一个功能齐全的 MCP 系统，其中包含可以无缝交互的服务器和客户端组件。

先决条件和环境设置

在开始开发之前，请确保您的系统满足以下要求：

安装了最新 Python 版本的 Windows 或 Mac 计算机
熟悉 Python 编程
掌握 Python 中的异步编程基础知识
了解 UI 开发概念

首先，让我们使用必要的工具和包设置我们的开发环境：

# 如果尚未安装，请安装 uv 工具
pip install uv

uv 工具将帮助我们高效地创建和管理 Python 项目。它是一个现代 Python 包管理器和虚拟环境工具，我们将在本指南中全程使用[^1][^2]。

构建 MCP 服务器

步骤 1：创建服务器项目

导航到您所需的项目目录并创建一个新的 MCP 服务器项目：

cd "您所需的目录"
uvx create-mcp-server

在创建过程中，系统会提示您输入项目详细信息，例如名称、描述和配置选项。这将生成一个包含所有必要文件和依赖项的项目结构[^1]。

步骤 2：设置虚拟环境

创建项目后，导航到项目目录并设置虚拟环境：

cd server-project-name
uv sync --dev --all-extras

此命令确保所有依赖项都已正确安装在您的虚拟环境中[^1]。

步骤 3：了解项目结构

生成的服务器项目具有类似于以下的结构：

SERVER-PROJECT-NAME/
├── .venv/
├── Lib/
│ └── site-packages/
├── Scripts/
├── src/
│ └── server_project_name/
│ ├── __pycache__/
│ ├── __init__.py
│ └── server.py
├── .gitignore
├── pyproject.toml
├── README.md
└── uv.lock

您需要关注的关键文件位于 src/server_project_name/ 目录中，特别是包含服务器实现的 server.py[^1]。

步骤 4：实现服务器功能

打开 server.py 文件并实现您的服务器功能。一个基本的 MCP 服务器应该定义客户端可以调用的工具。这是一个实现加法工具的简单示例：

from mcp import Server, ToolCall, ToolResult

class MyMCPServer(Server):
    async def handle_tool_call(self, tool_call: ToolCall) -> ToolResult:
        if tool_call.name == "add":
            # 提取参数
            a = tool_call.arguments.get("a")
            b = tool_call.arguments.get("b")
            
            # 执行加法
            result = a + b
            
            # 返回结果
            return ToolResult(content=str(result))
        
        # 处理未知工具
        return ToolResult(error=f"Unknown tool: {tool_call.name}")
    
    async def list_tools(self):
        # 定义可用工具
        tools = [
            {
                "name": "add",
                "description": "将两个数字相加",
                "input_schema": {
                    "type": "object",
                    "properties": {
                        "a": {"type": "number"},
                        "b": {"type": "number"}
                    },
                    "required": ["a", "b"]
                }
            }
        ]
        return tools

# 运行服务器
if __name__ == "__main__":
    server = MyMCPServer()
    server.run()

此示例创建一个具有简单加法工具的服务器，客户端可以调用该工具[^3]。

步骤 5：运行服务器

要运行您的 MCP 服务器：

python -m server_project_name.server

默认情况下，服务器将在 localhost 的 8000 端口上运行。如果需要，您可以在服务器配置中自定义此设置[^3]。

构建 MCP 客户端

步骤 1：创建客户端项目

为您的 MCP 客户端创建一个新目录：

uv init mcp-client
cd mcp-client

设置一个虚拟环境并安装所需的包：

uv venv
# 激活虚拟环境
# 在 Windows 上：
.venv\Scripts\activate
# 在 macOS/Linux 上：
source .venv/bin/activate

# 安装所需的包
uv add mcp anthropic python-dotenv

为您的客户端实现创建一个新文件：

touch client.py

步骤 2：实现客户端

创建一个可以连接到服务器并调用其工具的基本 MCP 客户端。这是一个示例实现：

import asyncio
from typing import Optional
from contextlib import AsyncExitStack
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client
from dotenv import load_dotenv

load_dotenv()  # 加载环境变量

class MCPClient:
    def __init__(self):
        # 初始化会话和客户端对象
        self.session: Optional[ClientSession] = None
        self.exit_stack = AsyncExitStack()

    async def connect_to_server(self, server_script_path: str):
        """连接到 MCP 服务器"""
        is_python = server_script_path.endswith('.py')
        is_js = server_script_path.endswith('.js')
        
        if not (is_python or is_js):
            raise ValueError("Server script must be a .py or .js file")
            
        command = "python" if is_python else "node"
        
        server_params = StdioServerParameters(
            command=command,
            args=[server_script_path],
            env=None
        )
        
        stdio_transport = await self.exit_stack.enter_async_context(stdio_client(server_params))
        self.stdio, self.write = stdio_transport
        
        self.session = await self.exit_stack.enter_async_context(ClientSession(self.stdio, self.write))
        await self.session.initialize()
        
        # 列出可用工具
        response = await self.session.list_tools()
        tools = response.tools
        print("\nConnected to server with tools:", [tool.name for tool in tools])
        return tools

    async def call_add_tool(self, a: int, b: int):
        """调用服务器上的 add 工具"""
        if not self.session:
            raise RuntimeError("Not connected to server")
            
        result = await self.session.call_tool(
            name="add",
            arguments={"a": a, "b": b}
        )
        return result.content[^0].text

    async def cleanup(self):
        """清理资源"""
        await self.exit_stack.aclose()

async def main():
    client = MCPClient()
    try:
        await client.connect_to_server("path/to/your/server.py")
        result = await client.call_add_tool(4, 5)
        print(f"Result: {result}")
    finally:
        await client.cleanup()

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

此客户端连接到 MCP 服务器，使用参数 4 和 5 调用 "add" 工具，并打印结果[^2][^3]。

使用 Tkinter 向 MCP 客户端添加 UI

步骤 1：设置基本 UI

现在，让我们通过使用 Tkinter 添加图形用户界面来增强我们的客户端。首先，如果您的 Python 安装中尚未包含 Tkinter，请安装它：

uv add tk

现在，让我们创建一个名为 client_ui.py 的新文件：

import asyncio
import tkinter as tk
from tkinter import ttk, messagebox
from client import MCPClient  # 导入我们之前创建的客户端类

class MCPClientUI:
    def __init__(self, root):
        self.root = root
        self.root.title("MCP Client")
        self.root.geometry("500x400")
        
        self.client = MCPClient()
        self.connected = False
        
        self.setup_ui()
        
    def setup_ui(self):
        # 创建主框架
        main_frame = ttk.Frame(self.root, padding="10")
        main_frame.pack(fill=tk.BOTH, expand=True)
        
        # 服务器连接部分
        server_frame = ttk.LabelFrame(main_frame, text="Server Connection", padding="10")
        server_frame.pack(fill=tk.X, pady=5)
        
        ttk.Label(server_frame, text="Server Script Path:").grid(row=0, column=0, sticky=tk.W, pady=5)
        self.server_path = tk.StringVar(value="path/to/your/server.py")
        ttk.Entry(server_frame, textvariable=self.server_path, width=40).grid(row=0, column=1, padx=5, pady=5)
        
        self.connect_btn = ttk.Button(server_frame, text="Connect", command=self.connect_to_server)
        self.connect_btn.grid(row=0, column=2, padx=5, pady=5)
        
        # 工具部分
        tool_frame = ttk.LabelFrame(main_frame, text="Add Tool", padding="10")
        tool_frame.pack(fill=tk.X, pady=5)
        
        ttk.Label(tool_frame, text="Number A:").grid(row=0, column=0, sticky=tk.W, pady=5)
        self.num_a = tk.IntVar(value=0)
        ttk.Entry(tool_frame, textvariable=self.num_a, width=10).grid(row=0, column=1, padx=5, pady=5)
        
        ttk.Label(tool_frame, text="Number B:").grid(row=0, column=2, sticky=tk.W, pady=5)
        self.num_b = tk.IntVar(value=0)
        ttk.Entry(tool_frame, textvariable=self.num_b, width=10).grid(row=0, column=3, padx=5, pady=5)
        
        self.calculate_btn = ttk.Button(tool_frame, text="Calculate", command=self.call_add_tool, state=tk.DISABLED)
        self.calculate_btn.grid(row=0, column=4, padx=5, pady=5)
        
        # 结果部分
        result_frame = ttk.LabelFrame(main_frame, text="Results", padding="10")
        result_frame.pack(fill=tk.BOTH, expand=True, pady=5)
        
        self.result_text = tk.Text(result_frame, height=10, width=50)
        self.result_text.pack(fill=tk.BOTH, expand=True)
        
    def connect_to_server(self):
        server_path = self.server_path.get()
        self.result_text.insert(tk.END, f"Connecting to server at {server_path}...\n")
        
        # 创建并运行一个协程来连接到服务器
        async def connect():
            try:
                tools = await self.client.connect_to_server(server_path)
                self.connected = True
                self.root.after(0, self.update_ui_after_connect, tools)
            except Exception as e:
                self.root.after(0, lambda: self.show_error(f"Connection error: {str(e)}"))
        
        asyncio.run(connect())
    
    def update_ui_after_connect(self, tools):
        self.result_text.insert(tk.END, f"Connected to server. Available tools: {[tool.name for tool in tools]}\n")
        self.connect_btn.config(text="Connected", state=tk.DISABLED)
        self.calculate_btn.config(state=tk.NORMAL)
    
    def call_add_tool(self):
        if not self.connected:
            self.show_error("Not connected to server")
            return
            
        try:
            a = self.num_a.get()
            b = self.num_b.get()
            
            self.result_text.insert(tk.END, f"Calculating {a} + {b}...\n")
            
            # 创建并运行一个协程来调用 add 工具
            async def add():
                try:
                    result = await self.client.call_add_tool(a, b)
                    self.root.after(0, lambda: self.show_result(result))
                except Exception as e:
                    self.root.after(0, lambda: self.show_error(f"Tool call error: {str(e)}"))
            
            asyncio.run(add())
        except Exception as e:
            self.show_error(f"Input error: {str(e)}")
    
    def show_result(self, result):
        self.result_text.insert(tk.END, f"Result: {result}\n")
    
    def show_error(self, message):
        messagebox.showerror("Error", message)
        self.result_text.insert(tk.END, f"Error: {message}\n")
    
    def cleanup(self):
        asyncio.run(self.client.cleanup())
        self.root.destroy()

if __name__ == "__main__":
    root = tk.Tk()
    app = MCPClientUI(root)
    root.protocol("WM_DELETE_WINDOW", app.cleanup)
    root.mainloop()

此 UI 提供用于输入服务器路径和加法运算值的字段、用于连接到服务器和计算结果的按钮以及用于显示输出的文本区域[^4]。

步骤 2：在 Tkinter 中处理异步操作

Tkinter 本身并非设计用于与 asyncio 一起使用，而我们的 MCP 客户端依赖于 asyncio。上面的解决方案使用了一种简单的阻塞模式运行 asyncio 操作的方法，该方法适用于简单的操作，但对于更复杂的应用程序来说并不理想。

为了获得更强大的解决方案，让我们创建一个增强版本，以更好地处理异步操作：

import asyncio
import tkinter as tk
from tkinter import ttk, messagebox
from client import MCPClient
import threading

class AsyncTkApp:
    def __init__(self, root):
        self.root = root
        self.root.title("Advanced MCP Client")
        self.root.geometry("550x450")
        
        self.client = MCPClient()
        self.connected = False
        self.loop = asyncio.new_event_loop()
        
        # 启动一个新线程来运行异步事件循环
        self.thread = threading.Thread(target=self._start_loop, daemon=True)
        self.thread.start()
        
        self.setup_ui()
    
    def _start_loop(self):
        asyncio.set_event_loop(self.loop)
        self.loop.run_forever()
    
    def setup_ui(self):
        # 创建带有填充的主框架
        main_frame = ttk.Frame(self.root, padding="15")
        main_frame.pack(fill=tk.BOTH, expand=True)
        
        # 服务器连接部分
        server_frame = ttk.LabelFrame(main_frame, text="Server Connection", padding="10")
        server_frame.pack(fill=tk.X, pady=8)
        
        ttk.Label(server_frame, text="Server Script Path:").grid(row=0, column=0, sticky=tk.W, pady=5)
        self.server_path = tk.StringVar(value="path/to/your/server.py")
        ttk.Entry(server_frame, textvariable=self.server_path, width=40).grid(row=0, column=1, padx=5, pady=5)
        
        self.connect_btn = ttk.Button(server_frame, text="Connect", command=self.connect_to_server)
        self.connect_btn.grid(row=0, column=2, padx=5, pady=5)
        
        # 带有改进布局的工具部分
        tool_frame = ttk.LabelFrame(main_frame, text="Add Tool", padding="10")
        tool_frame.pack(fill=tk.X, pady=8)
        
        ttk.Label(tool_frame, text="Number A:").grid(row=0, column=0, sticky=tk.W, pady=5)
        self.num_a = tk.IntVar(value=0)
        ttk.Entry(tool_frame, textvariable=self.num_a, width=10).grid(row=0, column=1, padx=5, pady=5)
        
        ttk.Label(tool_frame, text="Number B:").grid(row=0, column=2, sticky=tk.W, pady=5)
        self.num_b = tk.IntVar(value=0)
        ttk.Entry(tool_frame, textvariable=self.num_b, width=10).grid(row=0, column=3, padx=5, pady=5)
        
        self.calculate_btn = ttk.Button(tool_frame, text="Calculate", command=self.call_add_tool, state=tk.DISABLED)
        self.calculate_btn.grid(row=0, column=4, padx=5, pady=5)
        
        # 带有滚动条的结果部分
        result_frame = ttk.LabelFrame(main_frame, text="Results", padding="10")
        result_frame.pack(fill=tk.BOTH, expand=True, pady=8)
        
        # 向文本小部件添加滚动条
        scrollbar = ttk.Scrollbar(result_frame)
        scrollbar.pack(side=tk.RIGHT, fill=tk.Y)
        
        self.result_text = tk.Text(result_frame, height=12, width=50, yscrollcommand=scrollbar.set)
        self.result_text.pack(fill=tk.BOTH, expand=True)
        scrollbar.config(command=self.result_text.yview)
        
        # 状态栏
        status_frame = ttk.Frame(main_frame)
        status_frame.pack(fill=tk.X, pady=5)
        self.status_var = tk.StringVar(value="Ready")
        status_label = ttk.Label(status_frame, textvariable=self.status_var, anchor=tk.W)
        status_label.pack(side=tk.LEFT)
    
    def connect_to_server(self):
        server_path = self.server_path.get()
        self.log_message(f"Connecting to server at {server_path}...")
        self.status_var.set("Connecting...")
        
        async def connect_async():
            try:
                tools = await self.client.connect_to_server(server_path)
                self.root.after(0, lambda: self.update_ui_after_connect(tools))
            except Exception as e:
                self.root.after(0, lambda: self.show_error(f"Connection error: {str(e)}"))
        
        # 安排协程在 asyncio 事件循环中运行
        asyncio.run_coroutine_threadsafe(connect_async(), self.loop)
    
    def update_ui_after_connect(self, tools):
        self.connected = True
        self.log_message(f"Connected to server. Available tools: {[tool.name for tool in tools]}")
        self.connect_btn.config(text="Connected", state=tk.DISABLED)
        self.calculate_btn.config(state=tk.NORMAL)
        self.status_var.set("Connected")
    
    def call_add_tool(self):
        if not self.connected:
            self.show_error("Not connected to server")
            return
            
        try:
            a = self.num_a.get()
            b = self.num_b.get()
            
            self.log_message(f"Calculating {a} + {b}...")
            self.status_var.set("Calculating...")
            
            async def add_async():
                try:
                    result = await self.client.call_add_tool(a, b)
                    self.root.after(0, lambda: self.show_result(result))
                except Exception as e:
                    self.root.after(0, lambda: self.show_error(f"Tool call error: {str(e)}"))
            
            # 安排协程在 asyncio 事件循环中运行
            asyncio.run_coroutine_threadsafe(add_async(), self.loop)
        except Exception as e:
            self.show_error(f"Input error: {str(e)}")
    
    def log_message(self, message):
        self.result_text.insert(tk.END, f"{message}\n")
        self.result_text.see(tk.END)  # 滚动到末尾
    
    def show_result(self, result):
        self.log_message(f"Result: {result}")
        self.status_var.set("Ready")
    
    def show_error(self, message):
        messagebox.showerror("Error", message)
        self.log_message(f"Error: {message}")
        self.status_var.set("Error occurred")
    
    def cleanup(self):
        async def cleanup_async():
            await self.client.cleanup()
            self.loop.stop()
        
        # 在事件循环中安排清理
        asyncio.run_coroutine_threadsafe(cleanup_async(), self.loop)
        self.root.after(100, self.root.destroy)  # 给清理完成留出时间

if __name__ == "__main__":
    root = tk.Tk()
    app = AsyncTkApp(root)
    root.protocol("WM_DELETE_WINDOW", app.cleanup)
    root.mainloop()

这个增强版本使用一个单独的线程来运行 asyncio 事件循环，这可以更好地处理异步操作，而不会阻塞 UI[^4]。

测试完整系统

要测试您的 MCP 服务器和客户端：

在一个终端中启动 MCP 服务器：

cd server-project-directory
python -m server_project_name.server

在另一个终端中运行 MCP 客户端 UI：

cd mcp-client
python client_ui.py

在客户端 UI 中：
- 输入您的服务器脚本的路径
- 单击“连接”以建立连接
- 在输入字段中输入两个数字
- 单击“计算”以将请求发送到服务器
- 在结果文本区域中查看结果

扩展系统

一旦您拥有了基本服务器和客户端，您可以通过以下方式扩展系统：

向服务器添加更多工具，例如文本处理或数据分析功能
使用更多功能和更好的错误处理来增强客户端 UI
实施身份验证和安全措施
添加对多个并发连接的支持
创建具有图形、图表或其他可视化的更高级的 UI

结论

在本分步指南中，我们构建了一个完整的带有图形用户界面的 MCP 服务器和客户端系统。模型上下文协议允许客户端和服务器之间进行强大的通信，从而实现利用 AI 和其他工具的各种应用程序。

我们构建的系统演示了 MCP 的基本概念：定义工具、从客户端调用它们以及在用户友好的界面中呈现结果。这些知识为开发用于各种用例的更复杂的基于 MCP 的应用程序奠定了坚实的基础。

随着 MCP 的不断发展，及时了解协议的最新发展将有助于您构建更强大、更高效的应用程序。即使技术进步，这里学到的核心概念仍然适用。