---

3.1 引言：动手是最好的学习

在前面两个章节中，我们已经系统地学习了MCP的设计哲学、核心架构以及详细的协议规范。理论知识为我们构建了坚实的认知基础，但真正的掌握源于实践。本章，我们将正式从“理论家”转变为“工程师”，亲手构建我们的第一个MCP Server。

我们将使用Python语言和官方提供的mcp-sdk库来完成这个任务。选择Python是因为其在AI和后端开发领域的广泛应用以及简洁的语法。mcp-sdk则为我们封装了繁琐的JSON-RPC通信细节，让我们能更专注于实现Server的核心业务逻辑——即提供具体的能力（Capabilities）。

本章的目标是构建一个具备以下功能的MCP Server：

1. 实现标准的文件系统能力：包括读取文件（fs/readFile）和列出目录（fs/listDirectory）。
2. 实现一个自定义的项目工具：一个名为project/countLines的工具，用于统计指定文件的代码行数。
3. 通过标准输入/输出（stdio）与Host通信：这是最基础也是最常见的MCP Server运行方式，特别适用于本地开发和集成到IDE等场景。

通过本章的学习，你不仅会掌握构建MCP Server的具体技术步骤，更会深入理解Server在MCP生态中所扮演的“能力提供者”角色，为你后续构建更复杂的“实践项目”打下坚实的基础。

---

3.2 环境准备：搭建你的开发工作台

在编写代码之前，我们需要先配置好Python开发环境并安装必要的库。

3.2.1 安装Python

请确保你的系统中安装了Python 3.8或更高版本。你可以在终端中通过以下命令检查：

python3 --version

如果未安装或版本过低，请从Python官网下载并安装。

3.2.2 创建项目目录与虚拟环境

良好的项目组织习惯是专业开发的开始。让我们为我们的MCP Server创建一个专门的目录，并在其中使用虚拟环境来隔离项目依赖。

# 创建项目主目录
mkdir my-mcp-server
cd my-mcp-server

# 创建Python虚拟环境
python3 -m venv .venv

# 激活虚拟环境
# On macOS/Linux:
source .venv/bin/activate
# On Windows:
# .\.venv\Scripts\activate

激活虚拟环境后，你终端提示符的前面应该会有一个(.venv)的标记，表示当前所有的Python包安装和执行都将局限在这个环境中。

3.2.3 安装mcp-sdk

mcp-sdk是我们将要使用的核心库。在激活的虚拟环境中，使用pip进行安装：

pip install mcp-sdk

这个库提供了启动MCP Server、处理请求、发送响应和通知的所有基础功能。

---

3.3 编写Server主程序 (main.py)

现在，让我们在my-mcp-server目录下创建一个名为main.py的文件，并开始编写我们的Server代码。我们将分步构建，并对每一部分进行详细解析。

3.3.1 完整代码概览

为了让你有一个整体的印象，我们首先展示完整的main.py代码。别担心，我们会在后面逐一拆解和解释。

# main.py

import asyncio
import os
import logging
from typing import List, Dict, Any

from mcp_sdk.mcp import McpServer
from mcp_sdk.protocol import ( # 从SDK导入协议相关的类
    Request, Response, Error, ErrorCode,
    FileContent, DirectoryEntry, FileChange, FileChangeType,
    ToolDefinition, ToolParameter, ToolExecutionResult
)

# 配置日志，便于调试
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')

class MyAwesomeMcpServer(McpServer):
    """我们自定义的MCP Server，继承自mcp-sdk提供的基类"""

    def __init__(self, workspace_root: str):
        super().__init__()
        if not os.path.isdir(workspace_root):
            raise ValueError(f"Workspace root '{workspace_root}' does not exist or is not a directory.")
        self.workspace_root = os.path.abspath(workspace_root)
        logging.info(f"Server initialized with workspace root: {self.workspace_root}")

    def _to_safe_path(self, path: str) -> str:
        """核心安全函数：将相对路径转换为安全的绝对路径，并防止路径穿越"""
        # 规范化路径，消除 '..' 等
        abs_path = os.path.abspath(os.path.join(self.workspace_root, path))
        # 检查最终路径是否仍在工作区内
        if not abs_path.startswith(self.workspace_root):
            raise PermissionError("Attempted to access path outside of workspace")
        return abs_path

    # --- 实现文件系统能力 ---

    async def handle_fs_read_file(self, request: Request) -> Response:
        path = request.params.get('path')
        if not path:
            return self.error(request.id, ErrorCode.InvalidParams, "'path' parameter is required.")

        try:
            safe_path = self._to_safe_path(path)
            logging.info(f"Reading file: {safe_path}")
            with open(safe_path, 'rb') as f:
                content_bytes = f.read()
            
            # MCP要求内容为base64编码
            import base64
            content_base64 = base64.b64encode(content_bytes).decode('utf-8')
            
            result = FileContent(content=content_base64, encoding='base64')
            return self.success(request.id, result.dict())
        except FileNotFoundError:
            return self.error(request.id, ErrorCode.NotFound, f"File not found: {path}")
        except PermissionError as e:
            return self.error(request.id, ErrorCode.PermissionDenied, str(e))
        except Exception as e:
            logging.error(f"Error reading file {path}: {e}", exc_info=True)
            return self.error(request.id, ErrorCode.InternalError, f"An internal error occurred: {e}")

    async def handle_fs_list_directory(self, request: Request) -> Response:
        path = request.params.get('path', '.') # 默认为工作区根目录
        try:
            safe_path = self._to_safe_path(path)
            logging.info(f"Listing directory: {safe_path}")
            entries = []
            for name in os.listdir(safe_path):
                entry_path = os.path.join(safe_path, name)
                if os.path.isdir(entry_path):
                    entry_type = 'directory'
                else:
                    entry_type = 'file'
                entries.append(DirectoryEntry(name=name, type=entry_type).dict())
            return self.success(request.id, entries)
        except FileNotFoundError:
            return self.error(request.id, ErrorCode.NotFound, f"Directory not found: {path}")
        except PermissionError as e:
            return self.error(request.id, ErrorCode.PermissionDenied, str(e))
        except Exception as e:
            logging.error(f"Error listing directory {path}: {e}", exc_info=True)
            return self.error(request.id, ErrorCode.InternalError, f"An internal error occurred: {e}")

    # --- 实现自定义工具能力 ---

    async def handle_project_list_tools(self, request: Request) -> Response:
        tools = [
            ToolDefinition(
                name="project/countLines",
                description="Counts the number of lines in a specified file within the workspace.",
                parameters=[
                    ToolParameter(name="path", description="The relative path to the file.", type="string", required=True)
                ]
            ).dict()
        ]
        return self.success(request.id, tools)

    async def handle_project_execute_tool(self, request: Request) -> Response:
        tool_name = request.params.get('name')
        tool_params = request.params.get('parameters', {})

        if tool_name == "project/countLines":
            return await self._execute_count_lines(request.id, tool_params)
        else:
            return self.error(request.id, ErrorCode.MethodNotFound, f"Tool '{tool_name}' is not supported.")

    async def _execute_count_lines(self, request_id: str, params: Dict[str, Any]) -> Response:
        path = params.get('path')
        if not path:
            return self.error(request_id, ErrorCode.InvalidParams, "'path' parameter is required for countLines tool.")

        try:
            safe_path = self._to_safe_path(path)
            logging.info(f"Executing tool 'countLines' on: {safe_path}")
            line_count = 0
            with open(safe_path, 'r', encoding='utf-8', errors='ignore') as f:
                for _ in f:
                    line_count += 1
            
            result = ToolExecutionResult(result={"lineCount": line_count}, stdout=f"Successfully counted {line_count} lines in {path}.")
            return self.success(request_id, result.dict())
        except FileNotFoundError:
            return self.error(request_id, ErrorCode.NotFound, f"File not found for tool execution: {path}")
        except PermissionError as e:
            return self.error(request_id, ErrorCode.PermissionDenied, str(e))
        except Exception as e:
            logging.error(f"Error executing tool 'countLines' on {path}: {e}", exc_info=True)
            return self.error(request_id, ErrorCode.InternalError, f"An internal error occurred: {e}")

async def main():
    # 定义Server的工作区根目录，这里我们用一个示例目录 'workspace'
    # 在实际使用中，这通常由启动Server的进程（如IDE）动态指定
    workspace_dir = "workspace"
    if not os.path.exists(workspace_dir):
        os.makedirs(workspace_dir)
        # 创建一些示例文件用于测试
        with open(os.path.join(workspace_dir, "welcome.txt"), "w") as f:
            f.write("Hello, MCP World!\nThis is your first file.")
        with open(os.path.join(workspace_dir, "code_example.py"), "w") as f:
            f.write("def hello():\n    print('Hello from a Python file!')\n# End of file")

    server = MyAwesomeMcpServer(workspace_root=workspace_dir)
    logging.info("Starting MCP Server...")
    # run_stdio() 会启动一个循环，监听标准输入并向标准输出写入响应
    await server.run_stdio()

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

3.3.2 代码逐段解析

1. 导入与初始化

import asyncio
import os
import logging
from mcp_sdk.mcp import McpServer
from mcp_sdk.protocol import ...

logging.basicConfig(level=logging.INFO, ...)

class MyAwesomeMcpServer(McpServer):
    def __init__(self, workspace_root: str):
        super().__init__()
        ...
        self.workspace_root = os.path.abspath(workspace_root)

• 我们导入了必要的模块，特别是McpServer基类和协议中定义的各种数据结构类（如Request, FileContent等）。
• 我们创建了一个自己的Server类MyAwesomeMcpServer，它继承自McpServer。这是SDK推荐的做法。
• 构造函数__init__接收一个workspace_root参数。这是我们Server的工作根目录，所有的文件操作都将被限制在这个目录内，这是至关重要的安全措施。

2. 核心安全函数 _to_safe_path

    def _to_safe_path(self, path: str) -> str:
        abs_path = os.path.abspath(os.path.join(self.workspace_root, path))
        if not abs_path.startswith(self.workspace_root):
            raise PermissionError("Attempted to access path outside of workspace")
        return abs_path

• 这个内部方法是整个Server的安全基石。
• 它接收一个由Host提供的路径（可能是相对路径，甚至包含../等），并将其转换为一个绝对路径。
• 最关键的一步是检查这个转换后的绝对路径是否仍然以我们的工作区根目录开头。如果不是，就意味着Host试图进行“路径穿越”攻击（Path Traversal），访问工作区之外的文件。此时我们必须抛出PermissionError并拒绝该请求。

3. 实现fs/readFile

    async def handle_fs_read_file(self, request: Request) -> Response:
        ...
        safe_path = self._to_safe_path(path)
        with open(safe_path, 'rb') as f:
            content_bytes = f.read()
        content_base64 = base64.b64encode(content_bytes).decode('utf-8')
        result = FileContent(content=content_base64, encoding='base64')
        return self.success(request.id, result.dict())

• mcp-sdk通过方法命名约定来分发请求。当收到一个method为fs/readFile的请求时，SDK会自动调用我们定义的handle_fs_read_file方法。
• 方法内部，我们首先获取路径参数，然后立即调用_to_safe_path进行安全检查。
• 我们以二进制模式（'rb'）读取文件，以确保能处理任何文件类型。
• 根据MCP协议规范，文件内容必须以Base64编码后作为字符串返回。我们使用base64库进行编码。
• 最后，我们使用FileContent数据类来构建返回结果，并调用self.success()方法生成一个标准的成功响应。
• 注意完善的错误处理：我们捕获了FileNotFoundError, PermissionError等常见异常，并使用self.error()返回符合JSON-RPC规范的错误响应。

4. 实现project/listTools和project/executeTool

    async def handle_project_list_tools(self, request: Request) -> Response:
        tools = [ ToolDefinition(...) ]
        return self.success(request.id, tools)

    async def handle_project_execute_tool(self, request: Request) -> Response:
        if tool_name == "project/countLines":
            return await self._execute_count_lines(request.id, tool_params)
        ...

• handle_project_list_tools的实现非常直接。我们创建了一个ToolDefinition列表，其中详细描述了我们的project/countLines工具：它的名称、给LLM看的自然语言描述、以及它的参数（需要一个名为path的字符串）。
• handle_project_execute_tool则像一个路由器。它根据请求的工具名称，将任务分发给具体的执行函数（这里是_execute_count_lines）。

5. 实现_execute_count_lines

    async def _execute_count_lines(self, request_id: str, params: Dict[str, Any]) -> Response:
        safe_path = self._to_safe_path(path)
        ...
        with open(safe_path, 'r', ...) as f:
            for _ in f:
                line_count += 1
        result = ToolExecutionResult(result={"lineCount": line_count}, ...)
        return self.success(request_id, result.dict())

• 这个函数是countLines工具的具体实现逻辑。
• 它同样以安全为先，使用_to_safe_path处理路径。
• 它逐行读取文件来计算行数，这是一个简单但有效的实现。
• 最后，它使用ToolExecutionResult来封装结果。注意，result字段可以包含任何JSON可序列化的结构化数据（这里我们返回了一个包含lineCount的对象），而stdout字段则可以提供一些人类可读的日志信息。

6. 主入口 main 函数

async def main():
    workspace_dir = "workspace"
    if not os.path.exists(workspace_dir):
        os.makedirs(workspace_dir)
        # 创建示例文件...

    server = MyAwesomeMcpServer(workspace_root=workspace_dir)
    await server.run_stdio()

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

• main函数是程序的入口。
• 它首先创建了一个名为workspace的目录，并放入了两个示例文件，这样我们启动Server后就有东西可以操作了。
• 然后，它实例化了我们的MyAwesomeMcpServer。
• 最关键的一行是await server.run_stdio()。这个由SDK提供的方法会启动一个无限循环，开始监听进程的标准输入（stdin），并将所有的响应和通知写入到标准输出（stdout）。这就是MCP Server最基本的运行模式。

---

3.4 运行与测试：与你的Server对话

现在我们的代码已经完成，是时候运行它并像一个MCP Host一样与它交互了。

3.4.1 启动Server

在你的终端（确保虚拟环境已激活），运行main.py：

python3 main.py

如果一切顺利，你应该会看到类似以下的日志输出，然后程序会阻塞，等待来自标准输入的指令：

2023-10-27 10:30:00,123 - INFO - Server initialized with workspace root: /path/to/your/my-mcp-server/workspace
2023-10-27 10:30:00,124 - INFO - Starting MCP Server...

3.4.2 手动发送JSON-RPC请求

现在，你可以手动输入JSON-RPC请求来测试Server的功能。注意：你需要将整个JSON请求压缩到一行，然后按回车。

测试1：列出工作区根目录的文件

输入 (粘贴到终端并回车):

{"jsonrpc": "2.0", "id": 1, "method": "fs/listDirectory", "params": {"path": "."}}

Server输出 (会立刻显示在终端):

{"jsonrpc":"2.0","id":1,"result":[{"name":"welcome.txt","type":"file"},{"name":"code_example.py","type":"file"}]}

这验证了fs/listDirectory工作正常。

测试2：读取welcome.txt文件

输入:

{"jsonrpc": "2.0", "id": 2, "method": "fs/readFile", "params": {"path": "welcome.txt"}}

Server输出 (content是base64编码的):

{"jsonrpc":"2.0","id":2,"result":{"content":"SGVsbG8sIE1DUCBXb3JsZCEKVGhpcyBpcyB5b3VyIGZpcnN0IGZpbGUu","encoding":"base64"}}

你可以使用在线的Base64解码器验证content的内容就是"Hello, MCP World!\nThis is your first file."。

测试3：使用自定义工具统计代码行数

输入:

{"jsonrpc": "2.0", "id": 3, "method": "project/executeTool", "params": {"name": "project/countLines", "parameters": {"path": "code_example.py"}}}

Server输出:

{"jsonrpc":"2.0","id":3,"result":{"result":{"lineCount":3},"stdout":"Successfully counted 3 lines in code_example.py.","stderr":null}}

这完美地验证了我们的自定义工具project/countLines按预期工作。

测试4：尝试路径穿越攻击

输入:

{"jsonrpc": "2.0", "id": 4, "method": "fs/readFile", "params": {"path": "../../some_secret_file.txt"}}

Server输出:

{"jsonrpc":"2.0","id":4,"error":{"code":-32003,"message":"Attempted to access path outside of workspace"}}

这证明了我们的_to_safe_path安全机制成功地阻止了非授权的文件访问！

---

3.5 总结

恭喜你！你已经成功地从零开始构建、运行并测试了一个功能完备的MCP Server。通过这个过程，我们不仅学习了mcp-sdk的使用方法，更重要的是，我们将抽象的协议规范转化为了具体、可运行的代码。

我们掌握了：

• 如何通过继承McpServer来创建自定义Server。
• 如何通过handle_前缀的方法名约定来实现MCP的核心能力。
• 如何实现一个至关重要的安全层，防止路径穿越攻击。
• 如何定义和实现自定义工具，极大地扩展Server的能力。
• 如何通过stdio与Server进行交互和测试。

现在，你已经具备了构建任何MCP Server所需要的核心技能。在下一章，我们将转换视角，去构建MCP的另一半——Host，并让它与我们本章创建的Server进行真正的“对话”，从而释放MCP的全部潜力。