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

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

本综合指南将带你了解如何使用 Python 创建 Model Context Protocol（MCP）服务和带图形用户界面的客户端。MCP 旨在促进大模型与工具提供方之间的通信，从而支持构建功能强大的 AI 应用。完成本教程后，你将拥有一个由服务端和客户端组件构成、可无缝交互的可用 MCP 系统。

前置条件和环境设置

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

• 安装了最新版本 Python 的 Windows 或 Mac 电脑
• 熟悉 Python 编程
• 了解 Python 异步编程的基本概念
• 了解 UI 开发概念

首先，使用必要的工具和软件包设置开发环境：

# Install uv tool if not already installed
pip install uv

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

构建 MCP 服务

步骤 1：创建服务项目

转到所需的项目目录，并创建一个新的 MCP 服务项目：

cd "your_desired_directory"
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":
            # Extract arguments
            a = tool_call.arguments.get("a")
            b = tool_call.arguments.get("b")
            
            # Perform addition
            result = a + b
            
            # Return the result
            return ToolResult(content=str(result))
        
        # Handle unknown tool
        return ToolResult(error=f"Unknown tool: {tool_call.name}")
    
    async def list_tools(self):
        # Define available tools
        tools = [
            {
                "name": "add",
                "description": "Adds two numbers together",
                "input_schema": {
                    "type": "object",
                    "properties": {
                        "a": {"type": "number"},
                        "b": {"type": "number"}
                    },
                    "required": ["a", "b"]
                }
            }
        ]
        return tools

# Run the server
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
# Activate the virtual environment
# On Windows:
.venv\Scripts\activate
# On macOS/Linux:
source .venv/bin/activate

# Install required packages
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()  # Load environment variables

class MCPClient:
    def __init__(self):
        # Initialize session and client objects
        self.session: Optional[ClientSession] = None
        self.exit_stack = AsyncExitStack()

    async def connect_to_server(self, server_script_path: str):
        """Connect to an MCP server"""
        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()
        
        # List available tools
        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):
        """Call the add tool on the server"""
        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):
        """Clean up resources"""
        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  # Import the client class we created earlier

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):
        # Create main frame
        main_frame = ttk.Frame(self.root, padding="10")
        main_frame.pack(fill=tk.BOTH, expand=True)
        
        # Server connection section
        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 section
        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)
        
        # Results section
        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")
        
        # Create and run a coroutine to connect to the server
        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")
            
            # Create and run a coroutine to call the add tool
            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()
        
        # Start a new thread for running the async 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):
        # Create main frame with padding
        main_frame = ttk.Frame(self.root, padding="15")
        main_frame.pack(fill=tk.BOTH, expand=True)
        
        # Server connection section
        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 section with improved layout
        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)
        
        # Results section with scrollbar
        result_frame = ttk.LabelFrame(main_frame, text="Results", padding="10")
        result_frame.pack(fill=tk.BOTH, expand=True, pady=8)
        
        # Add a scrollbar to the text widget
        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 bar
        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)}"))
        
        # Schedule the coroutine to run in the asyncio event loop
        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)}"))
            
            # Schedule the coroutine to run in the asyncio event loop
            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)  # Scroll to the 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()
        
        # Schedule cleanup in the event loop
        asyncio.run_coroutine_threadsafe(cleanup_async(), self.loop)
        self.root.after(100, self.root.destroy)  # Give time for cleanup to complete

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

这个增强版本为 asyncio 事件循环使用了一个独立线程，这样可以更好地处理异步操作，同时不会阻塞 UI[^4]。

测试完整系统

要测试 MCP 服务和客户端：

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

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

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

cd mcp-client
python client_ui.py

3. 在客户端 UI 中：
   • 输入服务脚本的路径
   • 点击“Connect”以建立连接
   • 在输入字段中输入两个数字
   • 点击“Calculate”向服务发送请求
   • 在结果文本区域中查看结果

扩展系统

基本的服务和客户端运行起来后，你可以通过以下方式扩展系统：

1. 向服务添加更多工具，例如文本处理或数据分析函数
2. 为客户端 UI 添加更多功能并改进错误处理
3. 实施身份验证和安全措施
4. 添加对多个并发连接的支持
5. 创建更高级的 UI，包含图形、图表或其他可视化内容

结论

在本分步指南中，我们构建了一个完整的 MCP 服务和客户端系统，并配有图形用户界面。Model Context Protocol 支持客户端与服务之间进行强大的通信，可用于构建各种利用 AI 和其他工具的应用程序。

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

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