跳至主要内容

使用 FastMCP 2.0 构建 MCP 服务器与客户端

构建、测试与集成 ArXiv Explorer MCP 服务器。
· 8分钟 阅读

使用 AI 探索

在 AI 领域,Model Context Protocol(MCP,模型上下文协议)迅速成为热门话题。MCP 是一项开放标准,为像 Claude 4 这样的 AI 模型提供了一种一致的方式来连接外部工具、服务和实时数据源。这种连接能力改变了游戏规则,因为它通过弥合 AI 与各类系统之间的鸿沟,使大语言模型(LLM)能够给出更相关、最新且可执行的响应。

在本教程中,我们将深入介绍 FastMCP 2.0——一个强大的框架,只需几行代码就能轻松构建自己的 MCP 服务器。我们将了解 FastMCP 的核心组件,学习如何同时构建 MCP 服务器和客户端,并将其无缝集成到您的工作流程中。如果您是 MCP 新手,请查看我们的模型上下文协议指南与演示项目以了解更多信息。 

什么是 FastMCP 2.0?

FastMCP 2.0 是一个开源 Python 框架,旨在让构建 Model Context Protocol(MCP)服务器和客户端变得简单高效。开发者只需几行代码即可创建自定义 MCP 服务器,并将其连接到自己的 AI 应用,如代码编辑器或聊天助手。

FastMCP 2.0 是 FastMCP 1.0 的继任者,后者现称为MCP Python SDK,它为在 MCP 生态中工作提供了一整套工具。

FastMCP 2.0 的独特之处在于其全面的功能集,远超核心 MCP 规范。它简化了部署、身份验证、客户端与服务器代理,甚至可以基于 REST API 生成服务器。  

FastMCP 2.0 的核心组件

FastMCP 2.0 围绕三个关键组件构建:工具(Tools)、资源(Resources)和提示(Prompts)。

1. 工具(Tools)

工具是通过 MCP 协议向 LLM 暴露的 Python 函数。通过为函数添加 @mcp.tool 装饰器,您就允许 AI 模型在对话中调用它。这使得 LLM 能执行现实世界中的操作,如查询数据库、调用 API 或运行计算。

from fastmcp import FastMCPmcp = FastMCP(name="UtilityServer")@mcp.tooldef convert_usd_to_eur(amount: float, rate: float = 0.91) -> float:    """Converts a given amount in USD to EUR using the provided rate."""    return round(amount * rate, 2)

2. 资源(Resources)

资源为 LLM 或客户端应用提供对数据的只读访问,如文件、数据库记录、配置或动态生成的内容。例如:

from fastmcp import FastMCPmcp = FastMCP(name="WeatherResourceServer")@mcp.resource("config://weather-settings")def get_weather_settings() -> dict:    """Provides weather configuration."""    return {        "default_units": "metric",        "default_location": "Berlin",        "features_enabled": ["forecast", "alerts"]    }@mcp.resource("resource://welcome-message")def welcome_message() -> str:    """Returns a dynamic greeting for users based on time of day."""    from datetime import datetime    hour = datetime.now().hour    if hour < 12:        return "Good morning! Here's your weather update."    elif hour < 18:        return "Good afternoon! Need a forecast?"    else:        return "Good evening! Let's check the weather."

3. 提示(Prompts)

提示是可复用、可参数化的消息模板,用于引导 LLM 的响应。通过将提示定义为函数,您可以确保一致、上下文丰富的指令,供客户端调用。 

这些对于按顺序执行多个工具并引导 LLM 使用可用的工具与资源完成特定任务是必不可少的。

from fastmcp import FastMCPfrom fastmcp.prompts.prompt import PromptMessage, TextContentmcp = FastMCP(name="TutorPromptServer")@mcp.promptdef explain_math_concept(concept: str) -> PromptMessage:    """Creates a message asking an AI tutor to explain a math concept simply."""    text = f"Can you explain the math concept of '{concept}' in a way that's easy for a 12-year-old to understand?"    return PromptMessage(role="user", content=TextContent(type="text", text=text))

使用 FastMCP 2.0 与 Tavily 构建 AI ArXiv Explorer

本节我们将构建一个“ArXiv Explorer” MCP 服务器。该服务器包含两个工具:一个用于提取研究论文的链接和标题,另一个用于对论文进行总结。除了这些工具,服务器还将包含资源和提示,以帮助围绕所选主题创建一份完整的报告。

1. 环境准备

开始之前,您需要获取 Tavily 的 API 密钥。您可以在tavily.com 免费申请,每月可发起 1,000 次请求。然后,安装必要的 Python 库:

pip install fastmcp tavily-python

2. 构建 “ArxivExplorer” 服务器

我们先创建服务器的 Python 文件。该文件包含以下部分:

  1. 初始化: 通过提供服务器名称 "ArxivExplorer" 创建一个 FastMCP 实例。
  2. 资源:使用 @mcp.resource 装饰器将 arxiv_topics 函数暴露为只读数据源。如果用户不确定要研究什么主题,AI 客户端可以获取这些主题来获得灵感。
  3. 工具: @mcp.tool 装饰器可将常规的 Python 函数变成 LLM 可执行的操作。
    1. search_arxiv:该工具使用 Tavily API 将搜索限定在特定域(arxiv.org),确保我们只从 arxiv.org 获取相关的科学论文。
    2. summarize_paper:该工具使用 Tavily 的 qna_search,其针对提供直接、简洁答案进行了优化,非常适合生成摘要。
  4. 提示: @mcp.prompt 装饰器创建一个可复用模板。这不仅仅是个简单的提示;它是一份策略计划,指导 LLM 如何按顺序使用服务器的工具来执行复杂任务。
  5. HTTP 传输: mcp.run(transport="http") 是关键一行。它不是使用本地进程的 “stdio”,而是启动一个 Web 服务器。这样网络上的任何客户端(或同一台机器上的客户端)都可以通过 URL 连接到我们的服务器,使其真正成为一个可远程共享的工具。

server.py:

import osfrom typing import Dict, Listfrom fastmcp import FastMCPfrom tavily import TavilyClient# --- Configuration ---TAVILY_API_KEY = os.environ.get("TAVILY_API_KEY")if not TAVILY_API_KEY:    raise ValueError("Please set the TAVILY_API_KEY environment variable.")tavily = TavilyClient(api_key=TAVILY_API_KEY)mcp = FastMCP(name="ArxivExplorer")print("✅ ArxivExplorer server initialized.")# --- Dynamic Resource: Suggested AI research topics [email protected]("resource://ai/arxiv_topics")def arxiv_topics() -> List[str]:    return [        "Transformer interpretability",        "Efficient large-scale model training",        "Federated learning privacy",        "Neural network pruning",    ]print("✅ Resource 'resource://ai/arxiv_topics' registered.")# --- Tool: Search ArXiv for recent papers [email protected](annotations={"title": "Search Arxiv"})def search_arxiv(query: str, max_results: int = 5) -> List[Dict]:    """    Queries ArXiv via Tavily, returning title + link for each paper,    and *only* ArXiv results.    """    resp = tavily.search(        query=f"site:arxiv.org {query}",        max_results=max_results    )    return [        {"title": r["title"].strip(), "url": r["url"]} for r in resp.get("results", [])    ]# --- Tool: Summarize an ArXiv paper [email protected](annotations={"title": "Summarize Paper"})def summarize_paper(paper_url: str) -> str:    """    Returns a one-paragraph summary of the paper at the given URL.    """    prompt = f"Summarize the key contributions of this ArXiv paper: {paper_url}"    return tavily.qna_search(query=prompt)print("✅ Tools 'Search Arxiv' and 'Summarize Paper' registered.")# --- Prompt Template: Explore a topic thoroughly [email protected] explore_topic_prompt(topic: str) -> str:    return (        f"I want to explore recent work on '{topic}'.\n"        f"1. Call the 'Search Arxiv' tool to find the 5 most recent papers.\n"        f"2. For each paper URL, call 'Summarize Paper' to extract its key contributions.\n"        f"3. Combine all summaries into an overview report."    )print("✅ Prompt 'explore_topic_prompt' registered.")if __name__ == "__main__":    print("\n🚀 Starting ArxivExplorer Server...")    mcp.run(transport="http")

3. 构建交互式客户端

在将其集成到 AI 应用之前,我们需要测试 MCP 服务器的功能与端点。为此,FastMCP 提供了简洁的 API 来连接 MCP 服务器并测试其所有功能。

客户端文件包括:

  1. HTTP 传输: StreamableHttpTransport 专为连接以 HTTP 模式运行的 FastMCP 服务器而设计。
  2. unwrap_tool_result 辅助函数:这是一个关键的实用函数。FastMCP 客户端不会返回原始数据;而是返回结构化对象。该函数可安全检查响应对象,并提取您真正需要的 Python 数据。
  3. 工作流:主函数执行一个完整的测试流程。
    1. Ping 服务器以确保其在线。
    2. 列出所有可用的工具、资源和提示。
    3. 读取 arxiv_topics 资源并正确解析结果。
    4. 调用 search_arxiv 工具,并使用辅助函数解包论文列表。
    5. 将第一个搜索结果中的 URL 传递给 summarize_paper 工具,从而实现工具链式调用。
    6. 最后,获取 explore_topic_prompt,展示将要发送给 LLM 的最终可用指令。

client.py:

import astimport asyncioimport pprintfrom fastmcp import Clientfrom fastmcp.client.transports import StreamableHttpTransport# --- Configuration ---SERVER_URL = "http://localhost:8000/mcp"  # adjust if hosted elsewherepp = pprint.PrettyPrinter(indent=2, width=100)def unwrap_tool_result(resp):    """    Safely unwraps the content from a FastMCP tool call result object.    """    if hasattr(resp, "content") and resp.content:        # The content is a list containing a single content object        content_object = resp.content[0]        # It could be JSON or plain text        if hasattr(content_object, "json"):            return content_object.json        if hasattr(content_object, "text"):            try:                # Use ast.literal_eval for safely evaluating a string containing a Python literal                return ast.literal_eval(content_object.text)            except (ValueError, SyntaxError):                # If it's not a literal, return the raw text                return content_object.text    return respasync def main():    transport = StreamableHttpTransport(url=SERVER_URL)    client = Client(transport)    print("\n🚀 Connecting to FastMCP server at:", SERVER_URL)    async with client:        # 1. Ping to test connectivity        print("\n🔗 Testing server connectivity...")        await client.ping()        print("✅ Server is reachable!\n")        # 2. Discover server capabilities        print("🛠️  Available tools:")        pp.pprint(await client.list_tools())        print("\n📚 Available resources:")        pp.pprint(await client.list_resources())        print("\n💬 Available prompts:")        pp.pprint(await client.list_prompts())        # 3. Fetch the topics resource        print("\n\n📖 Fetching resource: resource://ai/arxiv_topics")        res = await client.read_resource("resource://ai/arxiv_topics")        topics = ast.literal_eval(res[0].text)        print("Today's AI topics:")        for i, t in enumerate(topics, 1):            print(f"  {i}. {t}")        # 4. Test the search tool        print("\n\n🔍 Testing tool: search_arxiv")        raw_search = await client.call_tool(            "search_arxiv",            {"query": "Transformer interpretability", "max_results": 3},        )        search_results = unwrap_tool_result(raw_search)        for i, paper in enumerate(search_results, 1):            print(f"  {i}. {paper['title']}\n     {paper['url']}")                # 5. Test the summarize tool on the first result        if search_results:            first_url = search_results[0]["url"]            print("\n\n📝 Testing tool: summarize_paper")            raw_summary = await client.call_tool(                "summarize_paper", {"paper_url": first_url}            )            summary = unwrap_tool_result(raw_summary)            print("\nSummary of first paper:\n", summary)        # 6. Test the prompt generator        print("\n\n🚀 Testing prompt: explore_topic_prompt")        prompt_resp = await client.get_prompt(            "explore_topic_prompt", {"topic": "Transformer interpretability"}        )        print("\nGenerated prompt for an LLM:")        for msg in prompt_resp.messages:            print(f"{msg.role.upper()}: {msg.content.text}\n")if __name__ == "__main__":    asyncio.run(main())

所有文件与配置均已发布至kingabzpro/FastMCP-Project GitHub 仓库。如遇问题,您可以参考该仓库。我还附上了本地运行的安装指南。

4. 运行应用程序

现在我们同时运行 MCP 服务器和客户端。 

首先,在本地系统中将 Tavily API 密钥设置为环境变量:  

export TAVILY_API_KEY='your_tavily_api_key'

接着,运行服务器:  

python server.py

几秒内,您会看到 FastMCP 服务器已在 http://localhost:8000/mcp 运行。

FastMCP server is running at http://localhost:8000/mcp

打开一个新的终端,运行客户端脚本,测试一切是否正常:  

python client.py

可以看到,所有函数、资源、工具与提示均工作正常。我们的 MCP 服务器在进行多次 Tavily API 请求时也表现良好。

🚀  Connecting to FastMCP server at: http://localhost:8000/mcp🔗 Testing server connectivity...✅ Server is reachable!🛠️  Available tools:[ Tool(name='search_arxiv', title='Search Arxiv', description='Queries ArXiv via Tavily, returning title + link for each paper,\nand *only* ArXiv results.', inputSchema={'properties': {'query': {'title': 'Query', 'type': 'string'}, 'max_results': {'default': 5, 'title': 'Max Results', 'type': 'integer'}}, 'required': ['query'], 'type': 'object'}, outputSchema={'properties': {'result': {'items': {'additionalProperties': True, 'type': 'object'}, 'title': 'Result', 'type': 'array'}}, 'required': ['result'], 'title': '_WrappedResult', 'type': 'object', 'x-fastmcp-wrap-result': True}, annotations=ToolAnnotations(title='Search Arxiv', readOnlyHint=None, destructiveHint=None, idempotentHint=None, openWorldHint=None), meta=None),  Tool(name='summarize_paper', title='Summarize Paper', description='Returns a one-paragraph summary of the paper at the given URL.', inputSchema={'properties': {'paper_url': {'title': 'Paper Url', 'type': 'string'}}, 'required': ['paper_url'], 'type': 'object'}, outputSchema={'properties': {'result': {'title': 'Result', 'type': 'string'}}, 'required': ['result'], 'title': '_WrappedResult', 'type': 'object', 'x-fastmcp-wrap-result': True}, annotations=ToolAnnotations(title='Summarize Paper', readOnlyHint=None, destructiveHint=None, idempotentHint=None, openWorldHint=None), meta=None)]📚 Available resources:[ Resource(name='arxiv_topics', title=None, uri=AnyUrl('resource://ai/arxiv_topics'), description=None, mimeType='text/plain', size=None, annotations=None, meta=None)]💬 Available prompts:[ Prompt(name='explore_topic_prompt', title=None, description=None, arguments=[PromptArgument(name='topic', description=None, required=True)], meta=None)]📖 Fetching resource: resource://ai/arxiv_topicsToday's AI topics:  1. Transformer interpretability  2. Efficient large-scale model training  3. Federated learning privacy  4. Neural network pruning🔍 Testing tool: search_arxiv  1. Transformer Interpretability Beyond Attention Visualization     https://arxiv.org/abs/2012.09838  2. A Practical Review of Mechanistic Interpretability for ...     https://arxiv.org/abs/2407.02646  3. Mechanistic Interpretability of Fine-Tuned Vision ...     https://arxiv.org/abs/2503.18762📝 Testing tool: summarize_paperSummary of first paper: The paper proposes a novel method for interpreting Transformer networks beyond attention visualization, introducing a new way to compute relevancy. It aims to enhance interpretability in vision tasks. The method is shown to provide more accurate insights into model decisions.🚀 Testing prompt: explore_topic_promptGenerated prompt:USER: type='text' text="I want to explore recent work on 'Transformer interpretability'.\n1. Call the 'Search Arxiv' tool to find the 5 most recent papers.\n2. For each paper URL, call 'Summarize Paper' to extract its key contributions.\n3. Combine all summaries into an overview report." annotations=None meta=None

在 Cursor AI 中集成 MCP 服务器

保持 MCP 服务器运行,接下来我们将其集成到 Cursor AI 代码编辑器中。 

  1. 进入“Settings(设置)”。
  2. 点击“Tools & Integrations(工具与集成)”选项卡。
  3. 点击“New MCP Server(新建 MCP 服务器)”按钮。
  4. 添加以下配置:
{  "mcpServers": {    "ArxivExplorer": {      "url": "http://127.0.0.1:8000/mcp/"    }  }}

将上述配置保存到 mcp.json 文件后,您会在 “Tools & Integrations(工具与集成)” 区域看到新添加的 MCP 服务器,界面会显示如下所示的两个工具。

ArxivExplorer configured in Cursor AI

在 Cursor AI 中测试 MCP 服务器

接下来进入有趣的部分。让我们看看在Cursor AI 代码编辑器中,我们的 MCP 服务器表现如何。我们将学习如何在编辑器内通过提示调用工具、访问资源并执行自定义提示。

1. 测试资源

首先检查您的 MCP 服务器是否能通过其资源提供有用的数据。

提示: “what are the some popular arxiv topics”(有哪些热门的 arxiv 主题)

Testing the Resources

如果一切设置妥当,AI 模型会理解您的请求,并自动从 MCP 服务器的资源端点检索话题列表。

2. 测试单个工具

接下来,试用您通过 MCP 暴露的工具。我们先测试 search_arxiv 工具,让 AI 模型列出关于大型语言模型的前十篇研究论文。  

提示: “Please search for the top 10 research papers on large language models.”(请搜索大型语言模型领域的前 10 篇研究论文。)

AI 将调用 search_arxiv 工具,您应能看到一份快速且准确的相关论文列表,直接来自 arXiv。

接下来测试摘要工具,它接收研究论文的 URL 并返回该报告的摘要。  

提示: “Please summarize the following paper for me: https://arxiv.org/abs/2402.06196”(请为我总结以下论文:https://arxiv.org/abs/2402.06196)

摘要工具会抓取该论文并返回简明扼要的总结。

3. 测试提示

我们现在从 MCP 调用一个自定义提示,帮助我们生成报告。

提示:“Create a report on Quantum computing using the latest work.”(基于最新研究创建一份关于量子计算的报告。)

Testing the Prompt

查看服务器代码,您会发现 AI 模型理解了任务,并使用了 explore_topic_prompt 函数。

@mcp.promptdef explore_topic_prompt(topic: str) -> str:    return (        f"I want to explore recent work on '{topic}'.\n"        f"1. Call the 'Search Arxiv' tool to find the 5 most recent papers.\n"        f"2. For each paper URL, call 'Summarize Paper' to extract its key contributions.\n"        f"3. Combine all summaries into an overview report."    )

首先,它会调用 “Search Arxiv” 工具查找论文链接。 

Tool result output

然后,它会将这些链接逐一传递给 “Summarize Paper” 工具生成摘要。最后,我将创建报告。

Tool result output

结语

FastMCP 2.0 不仅仅提供了一个简单的方式来创建 MCP 服务器。它更是一个用于构建安全、模块化、可扩展 AI 集成的强大框架。借助身份验证、高级代理(允许一个 FastMCP 服务器作为另一个的前端)以及通过挂载与导入将多个 FastMCP 服务器组合为单一应用等功能,您可以轻松构建复杂且易维护的系统。

中间件支持可让您为所有 MCP 请求与响应添加横切功能,例如日志记录或请求修改。您还可以通过 MCP 上下文将日志消息与进度更新回传给客户端,从而轻松监控和管理长时间运行的操作。

在本教程中,我们学习了如何构建一个功能完整的 MCP 服务器并测试其特性。这些工具与 MCP 协议正逐渐成为现代 AI 应用的关键组件,使 LLM 能更好地执行高级任务并更高效地自动化工作流。 

若想继续学习 MCP 服务器,请务必查看我们的其他资源: 

Abid Ali Awan

作者

Abid Ali Awan

作为一名持证的数据科学家,我热衷于利用前沿技术打造创新的机器学习应用。凭借在语音识别、数据分析与报告、MLOps、对话式人工智能以及自然语言处理方面的扎实背景,我不断打磨构建智能系统的能力,力求带来切实影响。除技术专长外,我也擅长沟通,能够将复杂概念提炼为清晰、简明的表述。因此,我成为数据科学领域备受关注的博主,与不断壮大的数据专业人士社区分享洞见与实践经验。目前,我专注于内容创作与编辑,借助大语言模型打造有力且吸引人的内容,帮助企业与个人更好地发挥数据价值。

主题