用Python构建一个简单的MCP服务器

在本文中，您将了解Model Context Protocol (MCP)是什么，并且如何使用FastMCP在Python中构建一个简单、实用的任务跟踪MCP服务器。

我们将涵盖的主题包括：

• MCP的工作原理，包括主机、客户端、服务器和三个核心原语。
• 如何使用 FastMCP 实施 MCP 工具、资源和提示。
• 如何使用FastMCP客户端运行和测试您的MCP服务器。

你是否曾经尝试过将语言模型连接到你自己的数据或工具？如果是的话，你就会知道这通常意味着编写自定义集成、管理API模式以及处理身份验证。每一个新的AI应用程序都像是从头开始重建相同的连接逻辑。

Model Context Protocol (MCP) 通过标准化大语言模型（LLMs）和其他AI模型与外部系统互动的方式解决了这个问题。FastMCP 是一个使构建MCP服务器变得简单的框架。

在本文中，您将了解MCP是什么，它是如何工作的，以及如何使用FastMCP构建一个实际的任务跟踪服务器。您将创建管理任务的工具、查看任务列表的资源以及指导AI互动的提示。

理解模型上下文协议

正如提到的，Model Context Protocol (MCP) 是一个开放协议，定义了 AI 应用程序如何与外部系统进行通信。

MCP 如何工作

MCP 有三个组成部分：

Hosts 是用户实际与之互动的由AI驱动的应用程序。主机可以是Claude Desktop，一个具有AI功能的IDE，或者是你自己构建的定制应用程序。主机包含（或与）语言模型，并发起与MCP服务器的连接。

客户端 连接到服务器。当主机需要与MCP服务器通信时，它会创建一个客户端实例来管理该特定连接。一个主机可以同时运行多个客户端，每个客户端连接到不同的服务器。客户端处理所有协议级通信。

服务器 是你构建的。它们暴露特定的功能——数据库访问、文件操作、API 集成——并通过提供工具、资源和提示来响应客户端请求。

因此，用户与主机互动，主机使用客户端与你的服务器通信，服务器将结构化结果返回给链中的各个环节。

要了解更多关于MCP的信息，请阅读《模型上下文协议完整指南》。

三个核心原语

MCP服务器暴露三种类型的功能：

工具是执行操作的函数。它们就像是LLM可以调用的可执行命令。 add_task 、send_an_email 和 query_a_database 是一些工具的例子。

资源 提供只读访问数据。它们允许查看信息而不更改它。示例包括任务列表、配置文件和用户配置文件。

提示是指导AI互动的模板。它们结构化了模型如何处理特定任务。例如，“分析这些任务并建议优先级”和“审查此代码以查找安全问题。”

在实际操作中，你会将这些基本元素结合起来。一个AI模型可能会使用一个资源来查看任务，然后使用一个工具来根据定义工作流程的提示更新一个任务。

设置您的环境

你需要 Python 3.10 或更高版本。使用 pip 安装 FastMCP（或 uv 如果你更喜欢）：

pip install fastmcp

让我们开始吧！

构建任务跟踪器服务器

我们将构建一个管理简单任务列表的服务器。创建一个名为  task_server.py  的文件，并添加导入语句：

from fastmcp import FastMCP

from datetime import datetime

这些为我们提供了FastMCP框架和用于跟踪任务创建时间的日期时间处理功能。

初始化服务器

现在设置服务器和一个简单的内存存储：

mcp = FastMCP("TaskTracker")# Simple in-memory task storagetasks = []task_id_counter = 1

这会做如下事情：

• FastMCP("TaskTracker") 创建您的MCP服务器，并命名为描述性名称。
• tasks 是一个存储所有任务的列表。
• task_id_counter 为每个任务生成唯一的ID。

在实际应用中，你会使用数据库。为了这个教程，我们会保持简单。

创建工具

工具是用 @mcp.tool() 装饰的函数。让我们创建三个有用的工具。

工具 1：添加新任务

首先，让我们创建一个工具，将任务添加到我们的列表中：

@mcp.tool()def add_task(title: str, description: str = "") -> dict:    """Add a new task to the task list."""    global task_id_counter    task = {        "id": task_id_counter,        "title": title,        "description": description,        "status": "pending",        "created_at": datetime.now().isoformat()    }    tasks.append(task)    task_id_counter += 1    return task

该工具执行以下操作：

• 输入一个任务标题（必填）和一个可选的描述。
• 创建一个任务字典，包含唯一ID、状态和时间戳。
• 将其添加到我们的 tasks 列表。
• 返回创建的任务。

该模型现在可以调用  add_task("Write documentation", "Update API docs")  并获得一个结构化的任务对象。

工具2：完成任务

接下来，让我们添加一个工具来标记任务已完成：

@mcp.tool()def complete_task(task_id: int) -> dict:    """Mark a task as completed."""    for task in tasks:        if task["id"] == task_id:            task["status"] = "completed"            task["completed_at"] = datetime.now().isoformat()            return task    return {"error": f"Task {task_id} not found"}

该工具在任务列表中搜索匹配的ID，将其状态更新为“已完成”，并盖上完成时间戳。然后，如果找到匹配项，返回更新后的任务，否则返回错误消息。

工具 3：删除任务

最后，添加一个工具来删除任务：

@mcp.tool()def delete_task(task_id: int) -> dict:    """Delete a task from the list."""    for i, task in enumerate(tasks):        if task["id"] == task_id:            deleted_task = tasks.pop(i)            return {"success": True, "deleted": deleted_task}    return {"success": False, "error": f"Task {task_id} not found"}

此工具搜索任务，将其从列表中删除，并返回带有已删除任务数据的确认信息。

这三个工具为任务管理提供了模型的创建、读取、更新和删除（CRUD）操作。

添加资源

资源让AI应用程序可以查看数据而无需修改它。让我们创建两个资源。

资源 1：查看所有任务

此资源返回完整的任务列表：

@mcp.resource("tasks://all")def get_all_tasks() -> str:    """Get all tasks as formatted text."""    if not tasks:        return "No tasks found"    result = "Current Tasks:\n\n"    for task in tasks:        status_emoji = "✅" if task["status"] == "completed" else "⏳"        result += f"{status_emoji} [{task['id']}] {task['title']}\n"        if task["description"]:            result += f"   Description: {task['description']}\n"        result += f"   Status: {task['status']}\n"        result += f"   Created: {task['created_at']}\n\n"    return result

以下是运作方式：

• 装饰器 @mcp.resource("tasks://all") 创建一个具有URI样标识符的资源。
• 该功能将所有任务格式化为带有表情符号的可读文本，以提高视觉清晰度。
• 如果不存在任务，它会返回一个简单的消息。

AI应用程序可以读取此资源以了解所有任务的当前状态。

资源 2：仅查看待办任务

此资源过滤未完成的任务：

@mcp.resource("tasks://pending")def get_pending_tasks() -> str:    """Get only pending tasks."""    pending = [t for t in tasks if t["status"] == "pending"]    if not pending:        return "No pending tasks!"    result = "Pending Tasks:\n\n"    for task in pending:        result += f"⏳ [{task['id']}] {task['title']}\n"        if task["description"]:            result += f"   {task['description']}\n"        result += "\n"    return result

该资源将任务列表过滤为仅待处理项目，格式化以便于阅读，并在没有剩余任务时返回一条消息。

资源非常适合于模型需要频繁读取但不进行更改的数据。

定义提示

提示引导AI应用程序如何与您的服务器互动。让我们创建一个有用的提示：

@mcp.prompt()def task_summary_prompt() -> str:    """Generate a prompt for summarizing tasks."""    return """Please analyze the current task list and provide:1. Total number of tasks (completed vs pending)2. Any overdue or high-priority items3. Suggested next actions4. Overall progress assessmentUse the tasks://all resource to access the complete task list."""

此提示定义了一个用于任务分析的结构化模板，告诉AI应包括哪些信息，并引用用于数据的资源。

提示使人工智能互动更加一致和有用。当人工智能模型使用这个提示时，它知道要获取任务数据并以这种特定格式进行分析。

运行和测试服务器

将此代码添加到运行您的服务器：

if __name__ == "__main__":    mcp.run()

从你的终端启动服务器：

fastmcp run task_server.py

你会看到输出确认服务器正在运行。现在服务器已经准备好接受来自MCP客户端的连接。

使用FastMCP客户端进行测试

您可以使用FastMCP内置的客户端测试您的服务器。创建一个名为的测试文件 test_client.py 并运行它：

from fastmcp import Clientimport asyncioasync def test_server():    async with Client("task_server.py") as client:        # List available tools        tools = await client.list_tools()        print("Available tools:", [t.name for t in tools.tools])        # Add a task        result = await client.call_tool("add_task", {            "title": "Learn MCP",            "description": "Build a task tracker with FastMCP"        })        print("\nAdded task:", result.content[0].text)        # View all tasks        resources = await client.list_resources()        print("\nAvailable resources:", [r.uri for r in resources.resources])        task_list = await client.read_resource("tasks://all")        print("\nAll tasks:\n", task_list.contents[0].text)asyncio.run(test_server())

你会看到你的工具执行并资源返回数据。这确认了一切都正常工作。