MCP协议实战：如何用Python快速搭建一个支持文件读取的AI上下文服务器（附完整代码）

最近在折腾本地大模型应用时，我遇到了一个挺典型的问题：想让AI助手帮我分析项目文档，但每次都得手动复制粘贴，或者写一堆临时脚本去读取文件。这种割裂的体验让我开始寻找更优雅的解决方案。直到接触到MCP（Model Context Protocol），我才发现原来AI与本地数据的连接可以如此顺畅。MCP就像是给大模型装上了标准化的“数据线”，让模型能够按需、安全地读取外部资源，而无需我们反复进行繁琐的中间操作。

对于Python开发者来说，尤其是那些正在构建需要接入本地知识库、文档分析或私有数据查询的AI应用的朋友，理解并实践MCP的资源（Resources）机制，能立刻将你的项目从“玩具级”提升到“可用级”。本文不会重复那些枯燥的理论，而是直接带你动手，从零开始，用大约一百行代码，构建一个能够响应AI查询、实时读取指定文件的上下文服务器。你会发现，让AI“看懂”你的本地文件，其实比想象中简单得多。

1. 环境准备与核心概念速览

在开始敲代码之前，我们花几分钟理清几个关键点，这能帮你更好地理解后续每一步在做什么。MCP，即模型上下文协议，它定义了一套标准，让大模型（客户端）能够以一种可控、可发现的方式，调用外部服务器提供的功能。你可以把它想象成AI世界的“插件系统”标准。

在这个体系中，资源（Resource） 是一个核心概念。它特指那些只读的、可供模型获取的数据实体，比如一个文本文件、一张图片的元数据，或者一段数据库查询结果。资源通过唯一的URI（如 file://knowledge.txt）来标识。服务器声明自己拥有哪些资源，当AI模型需要相关信息时，它会通过客户端向服务器请求读取对应的资源。整个过程是无副作用的——仅仅是读取数据，不会修改或删除任何东西，这为数据安全提供了基础保障。

提示：MCP协议目前由Anthropic主导推进，但因其开放标准的设计，可以兼容各种遵循协议的大模型客户端和服务端实现，生态正在快速成长。

我们的实战目标很明确：搭建一个MCP服务器，它对外暴露一个文件资源；同时编写一个客户端程序，这个客户端会连接一个本地的大模型服务（例如基于vLLM部署的Qwen），并引导模型在需要时调用我们的服务器来获取文件内容，最终生成回答。

接下来，看看我们需要准备哪些“食材”：

  • Python 3.8+：这是我们的开发语言基础。
  • MCP SDK：我们将使用官方推荐的 mcp Python 库，它提供了构建服务器和客户端所需的所有工具。
  • 异步文件库：由于MCP服务器基于异步IO，我们选择 aiofiles 来优雅地处理文件读取，避免阻塞。
  • 一个本地大模型服务：你需要一个能够提供OpenAI兼容API的模型服务端点。这可以是本地用vLLM、Ollama等工具启动的模型，也可以是某些云服务提供的兼容端点。本文假设你已经在 http://localhost:9000/v1 运行了这样的服务。
  • 一个待读取的文本文件：例如，一份关于广州历史的文档 guangzhou_history.txt。

安装核心依赖非常简单，一条命令搞定：

pip install mcp aiofiles openai

这里安装了三个包：mcp 是协议实现的核心；aiofiles 用于异步文件操作；openai 库则用于我们的客户端与本地模型API进行通信。

2. 构建MCP服务器：让文件成为可被发现的资源

服务器是整个系统的数据提供方。它的职责很简单：启动一个服务，声明自己拥有哪些资源，并在收到合法的资源读取请求时，返回对应的内容。我们使用 mcp 库中的 FastMCP 来快速构建，它抽象了底层通信细节，让我们能专注于业务逻辑。

首先，创建一个名为 mcp_server.py 的文件。我们来逐步拆解其中的代码：

# mcp_server.py
from mcp.server.fastmcp import FastMCP
import aiofiles
import os

# 初始化一个FastMCP服务器实例，命名为"FileReader"，并指定服务端口为9999
mcp = FastMCP("FileReader", port=9999)

# 使用装饰器声明一个资源
@mcp.resource(
    uri="file://guangzhou_history.txt",  # 资源的唯一标识符
    name="get_guangzhou_history",        # 资源在工具列表中的显示名称
    description="获取关于广州城市名称由来和历史演变的详细文档", # 资源描述，用于引导AI理解何时调用
    mime_type="text/plain"               # 资源的媒体类型，这里是纯文本
)
async def read_guangzhou_history():
    """
    当客户端请求读取 'file://guangzhou_history.txt' 资源时，此函数将被调用。
    它负责异步读取本地文件并返回其内容。
    """
    # 构建文件的绝对路径，这里假设文件放在与脚本同目录下的 `data` 文件夹中
    file_path = os.path.join(os.path.dirname(__file__), 'data', 'guangzhou_history.txt')
    
    # 使用aiofiles异步打开文件，避免在IO等待时阻塞整个服务器
    async with aiofiles.open(file_path, mode='r', encoding='utf-8') as f:
        content = await f.read()
        # 可选：在服务器日志中打印读取信息，便于调试
        print(f"[Server] 已读取资源: file://guangzhou_history.txt, 长度: {len(content)} 字符")
        return content

if __name__ == "__main__":
    # 启动服务器，使用SSE (Server-Sent Events)作为传输协议，这是一种简单的HTTP流协议
    print("MCP文件资源服务器启动，监听端口 9999 ...")
    mcp.run(transport="sse")

这段代码的精髓在于 @mcp.resource 装饰器。它做了以下几件事：

  1. 注册：告诉MCP框架，本服务器提供了一个资源。
  2. 定义元数据：uri、name、description 这些信息会被服务器在初始化时告知客户端。客户端（以及其背后的大模型）正是通过这些元数据来了解“有什么资源可用”以及“每个资源是干什么的”。
  3. 绑定处理函数：当有读取该URI的请求到来时，框架会自动调用被装饰的异步函数 read_guangzhou_history()。

确保在脚本同级目录下创建 data 文件夹，并将你的 guangzhou_history.txt 文件放入其中。文件内容可以是我们之前提到的广州历史文本。

现在，在终端运行 python mcp_server.py。如果看到启动日志，说明你的MCP资源服务器已经在 http://localhost:9999 上就绪，正等待客户端的连接和查询。

3. 编写智能客户端：连接模型与资源的桥梁

客户端扮演着“中间人”或“翻译官”的角色。它既要与MCP服务器对话，获取资源列表；又要与本地的大模型服务对话，将用户的自然语言查询转化为对特定资源的工具调用请求，并整合资源内容，最终让模型生成完整的回答。

创建一个新文件 mcp_client.py。这里的逻辑稍复杂，我们分模块实现一个 MCPClient 类。

# mcp_client.py
import asyncio
from openai import OpenAI
from mcp.client.sse import sse_client
from mcp import ClientSession
from contextlib import AsyncExitStack

class MCPClient:
    def __init__(self, api_key: str, llm_server_url: str, mcp_server_url: str):
        """
        初始化客户端。
        :param api_key: 本地模型服务的API密钥（若无需验证可设为"EMPTY"）。
        :param llm_server_url: 本地大模型服务的OpenAI兼容API地址。
        :param mcp_server_url: MCP服务器的SSE端点地址。
        """
        # 初始化OpenAI客户端，用于与本地模型对话
        self.llm_client = OpenAI(api_key=api_key, base_url=llm_server_url)
        # 获取可用模型，通常取第一个
        models = self.llm_client.models.list()
        self.model = models.data[0].id
        self.mcp_server_url = mcp_server_url
        # AsyncExitStack用于优雅地管理多个异步上下文管理器
        self.exit_stack = AsyncExitStack()
        # 缓存从服务器获取的资源元信息
        self.resources = {}

    async def connect_to_mcp_server(self):
        """连接到MCP服务器，并获取其提供的所有资源列表。"""
        # 建立SSE连接
        read_stream, write_stream = await self.exit_stack.enter_async_context(
            sse_client(self.mcp_server_url)
        )
        # 创建MCP会话
        self.session: ClientSession = await self.exit_stack.enter_async_context(
            ClientSession(read_stream, write_stream)
        )
        # 执行初始化握手
        await self.session.initialize()
        print("[Client] 已成功连接至MCP服务器。")

        # 列出服务器所有可用资源
        list_response = await self.session.list_resources()
        print(f"[Client] 发现 {len(list_response.resources)} 个资源。")
        
        # 将资源信息转换为OpenAI工具调用格式，并缓存起来
        tools_for_llm = []
        for resource in list_response.resources:
            tool_schema = {
                "type": "function",
                "function": {
                    "name": resource.name,
                    "description": resource.description,
                    "parameters": {"type": "object", "properties": {}}  # 此资源无需参数
                }
            }
            tools_for_llm.append(tool_schema)
            
            # 缓存资源信息，键为资源名，值为包含URI等信息的字典
            self.resources[resource.name] = {
                "uri": resource.uri,
                "name": resource.name,
                "description": resource.description,
            }
        return tools_for_llm

    async def query_with_llm(self, user_query: str, tools: list):
        """
        执行核心查询逻辑：将用户问题、可用工具交给大模型，并处理其工具调用请求。
        """
        # 第一轮对话：将用户问题和可用工具信息发送给模型
        messages = [{"role": "user", "content": user_query}]
        
        print(f"[Client] 向模型发送查询: '{user_query}'")
        first_response = self.llm_client.chat.completions.create(
            model=self.model,
            messages=messages,
            tools=tools,
            tool_choice="auto",  # 让模型自行决定是否调用工具
            stream=False
        )
        
        assistant_message = first_response.choices[0].message
        messages.append(assistant_message)  # 将模型的回复加入对话历史

        # 检查模型是否决定调用工具
        if assistant_message.tool_calls:
            tool_call = assistant_message.tool_calls[0]
            function_name = tool_call.function.name
            print(f"[Client] 模型决定调用工具: {function_name}")
            
            # 根据工具名，找到对应的资源URI
            resource_uri = self.resources[function_name]["uri"]
            
            # 向MCP服务器发起资源读取请求
            print(f"[Client] 正在从MCP服务器读取资源: {resource_uri}")
            read_response = await self.session.read_resource(resource_uri)
            resource_content = read_response.contents[0].text  # 获取资源文本内容
            print(f"[Client] 资源内容获取成功，长度: {len(resource_content)} 字符")
            
            # 将获取到的资源内容作为工具调用结果，追加到对话历史中
            messages.append({
                "role": "tool",
                "tool_call_id": tool_call.id,
                "name": function_name,
                "content": resource_content
            })
            
            # 第二轮对话：将包含资源内容的上下文再次发送给模型，让它生成最终答案
            print("[Client] 模型正在结合资源内容生成最终回答...")
            final_response = self.llm_client.chat.completions.create(
                model=self.model,
                messages=messages,
                stream=True  # 使用流式输出，体验更好
            )
            
            print("\n--- AI 助手回答 ---")
            full_answer = ""
            for chunk in final_response:
                if chunk.choices and chunk.choices[0].delta.content:
                    content_piece = chunk.choices[0].delta.content
                    print(content_piece, end='', flush=True)
                    full_answer += content_piece
            print("\n--- 回答结束 ---")
            return full_answer
        else:
            # 如果模型没有调用工具，直接输出其回复
            print("\n--- AI 助手回答 (未使用工具) ---")
            print(assistant_message.content)
            return assistant_message.content

    async def run(self, user_query: str):
        """主执行流程"""
        try:
            # 步骤1: 连接MCP服务器并获取工具列表
            available_tools = await self.connect_to_mcp_server()
            # 步骤2: 执行查询
            await self.query_with_llm(user_query, available_tools)
        finally:
            # 步骤3: 清理资源，关闭连接
            await self.exit_stack.aclose()
            print("[Client] 连接已关闭。")

async def main():
    # 配置参数：请根据你的实际环境修改
    LOCAL_LLM_API_KEY = "EMPTY"  # 本地部署的vLLM等服务通常无需密钥
    LOCAL_LLM_BASE_URL = "http://localhost:9000/v1"  # 你的本地模型服务地址
    MCP_SERVER_SSE_URL = "http://localhost:9999/sse"  # MCP服务器SSE端点
    
    client = MCPClient(LOCAL_LLM_API_KEY, LOCAL_LLM_BASE_URL, MCP_SERVER_SSE_URL)
    # 执行一个示例查询
    await client.run("请告诉我广州为什么被称为‘羊城’？")

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

这个客户端的运作流程是一个清晰的**“发现-决策-获取-合成”**循环：

  1. 发现：连接MCP服务器，获取其暴露的 get_guangzhou_history 资源，并将其格式化成大模型能理解的“工具”描述。
  2. 决策：将用户问题“广州为什么被称为‘羊城’？”和工具描述一同发送给大模型。模型根据工具描述，判断需要调用 get_guangzhou_history 来获取信息。
  3. 获取：客户端截获模型的工具调用请求，解析出要读取的资源URI（file://guangzhou_history.txt），然后向MCP服务器发起读取请求，拿到文件的具体内容。
  4. 合成：客户端将获取到的文件内容作为“工具调用结果”反馈给模型。模型此时拥有了问题相关的具体资料，从而能够生成一个准确、详实的最终回答。

4. 运行与调试：观察整个系统的协作

现在，让我们启动整个系统，看看它们是如何协同工作的。你需要打开两个终端窗口。

终端一（服务器端）：

python mcp_server.py

输出应类似于：

MCP文件资源服务器启动，监听端口 9999 ...

终端二（客户端）： 在确保你的本地大模型服务（例如在 localhost:9000 运行的vLLM+Qwen）已经启动后，运行：

python mcp_client.py

接下来，你将看到一系列日志输出，生动展示了整个交互过程：

  1. 客户端连接：[Client] 已成功连接至MCP服务器。 和 [Client] 发现 1 个资源。
  2. 模型决策：[Client] 向模型发送查询: '请告诉我广州为什么被称为‘羊城’？'，紧接着 [Client] 模型决定调用工具: get_guangzhou_history。这说明模型聪明地识别出，要回答这个问题，需要去调用那个能获取广州历史文档的工具。
  3. 资源获取：[Client] 正在从MCP服务器读取资源: file://guangzhou_history.txt，同时服务器终端会打印 [Server] 已读取资源: file://guangzhou_history.txt, 长度: xxxx 字符，表明文件读取成功。
  4. 最终合成：客户端打印 [Client] 资源内容获取成功，长度: xxxx 字符 和 [Client] 模型正在结合资源内容生成最终回答...，然后流式输出模型生成的、基于文件内容的答案。

如果一切顺利，你将在客户端终端看到AI助手给出的、引用了文件中关于“五仙乘羊赠穗”传说的详细解释。这证明你的MCP上下文服务器成功运行，并赋能大模型完成了一次基于外部知识的精准回答。

5. 进阶探索与实用技巧

一个基本的文件读取服务器已经跑通了，但这只是MCP能力的冰山一角。你可以基于这个框架，轻松扩展出更强大、更实用的功能。

扩展一：支持多个文件和动态URI 现实中，我们往往需要让AI能访问整个目录下的文件。只需修改服务器，使其能根据请求动态读取不同文件。

# 在mcp_server.py中增加一个动态资源
@mcp.resource(
    uri="file://documents/{filename}",
    name="read_document",
    description="根据文件名读取指定文档的内容",
    mime_type="text/plain"
)
async def read_document(filename: str):  # 文件名作为参数
    """动态读取`data`目录下的文件"""
    # 安全警告：务必对filename进行严格的路径遍历检查！
    safe_filename = os.path.basename(filename)  # 防止目录遍历攻击
    file_path = os.path.join(os.path.dirname(__file__), 'data', safe_filename)
    if os.path.exists(file_path):
        async with aiofiles.open(file_path, 'r', encoding='utf-8') as f:
            return await f.read()
    else:
        return f"错误：文件 '{safe_filename}' 未找到。"

对应的，客户端在调用时，模型需要学会在工具调用中传入 {"filename": "xxx.txt"} 这样的参数。这要求模型对工具的描述（包括参数模式）有更好的理解。

扩展二：支持更多资源类型 MCP支持多种MIME类型。除了文本，你还可以暴露图片、JSON数据等。

@mcp.resource(
    uri="data://system/status",
    name="get_system_status",
    description="获取当前系统的状态信息，如CPU、内存使用率",
    mime_type="application/json"
)
async def get_system_status():
    """返回系统状态的JSON数据"""
    import psutil
    status = {
        "cpu_percent": psutil.cpu_percent(interval=1),
        "memory_percent": psutil.virtual_memory().percent,
        "disk_usage": psutil.disk_usage('/').percent
    }
    return json.dumps(status)  # 返回JSON字符串

扩展三：错误处理与日志完善 在生产环境中，健壮性至关重要。我们需要为服务器和客户端添加更完善的错误处理。

  • 服务器端：在资源处理函数中使用 try...except 包裹文件读取操作，返回友好的错误信息而非抛出异常导致服务崩溃。
  • 客户端端：
    • 检查与MCP服务器的连接是否成功。
    • 处理模型不调用工具的情况，提供备选回答逻辑。
    • 对模型可能调用的不存在的工具名进行防御性检查。
# 客户端query_with_llm方法中，工具调用部分可加强
if assistant_message.tool_calls:
    tool_call = assistant_message.tool_calls[0]
    function_name = tool_call.function.name
    
    if function_name not in self.resources:
        error_msg = f"请求的工具 '{function_name}' 未在资源列表中找到。"
        print(f"[Client] 错误: {error_msg}")
        # 可以将错误信息反馈给模型，让其重新思考
        messages.append({
            "role": "tool",
            "tool_call_id": tool_call.id,
            "name": function_name,
            "content": error_msg
        })
        # ... 再次调用模型
        return
    # ... 正常的资源读取流程

性能与安全考量

  • 异步优势：我们全程使用 async/await，这意味着服务器可以高效处理多个并发请求，不会因为一个文件的IO操作而阻塞其他请求。
  • 权限控制：MCP协议本身包含了权限申请机制。对于更敏感的操作（如写入文件、执行命令），服务器可以声明需要用户授权，客户端则会向用户请求许可。在我们的只读文件示例中，这一步被简化了。
  • 输入验证：对于动态路径（如 {filename}），必须进行严格的验证和净化，防止路径遍历攻击（如 ../../../etc/passwd）。使用 os.path.basename() 是简单的第一步，在生产环境中可能需要更严格的白名单机制。

通过以上步骤，你已经掌握了使用Python和MCP协议构建AI上下文服务器的核心技能。从静态文件读取到动态资源发现，从文本到结构化数据，这个框架的扩展性极强。下次当你的AI应用需要接入公司Wiki、项目日志或实时数据库时，不妨考虑用MCP来搭建一个专属的、安全的数据桥梁。

Logo
AI Agent技术社区

Agent 垂直技术社区，欢迎活跃、内容共建。

更多推荐

cover

Windows部署OpenClaw：PowerShell安装、自定义模型与远程访问实测

avatar AI Agent技术社区
cover

深入理解 AI Agent · MEMORY #02：记忆的工程机制

avatar AI Agent技术社区
cover

DeepSeek V4 Pro正式版上线：AI Agent能力跃升背后的行业趋势

avatar AI Agent技术社区