当前位置： 首页 > news >正文

MCP系列之实践篇：搭建你的第一个MCP应用

news 来源：原创 2025/4/26 19:00:08

前言

在前两篇文章中，我们已经介绍了MCP（模型上下文协议）的基本概念和技术架构。本篇文章将从理论走向实践，通过一个简单但完整的案例，手把手教你如何搭建和调试一个基于MCP的应用。我们将一起构建一个天气查询和活动推荐的MCP服务器，并将其与AI助手集成。

无论你是AI开发者、工具创建者，还是对MCP感兴趣的技术爱好者，通过本文的实践，你将能够掌握MCP的基本开发流程，为后续开发更复杂的MCP应用打下基础。

准备工作

在开始实践之前，我们需要准备好开发环境和相关工具。

环境要求

  • Python 3.8+（我们将使用Python实现MCP服务器）
  • 基本的命令行和代码编辑器使用经验
  • 初步的Python编程知识
  • 对JSON和API的基本了解

安装依赖

我们将使用Anthropic提供的MCP Python SDK来简化开发过程。首先，创建一个新项目文件夹，然后安装必要的依赖：

# 创建项目文件夹
mkdir mcp-weather-demo
cd mcp-weather-demo# 创建虚拟环境
python -m venv venv
source venv/bin/activate  # Windows上使用: venv\Scripts\activate# 安装MCP SDK和其他依赖
pip install mcp-python requests

项目概述

在本教程中，我们将构建一个包含以下功能的MCP服务器：

  1. 天气查询工具：根据城市名称返回当前天气信息
  2. 活动推荐工具：根据天气情况推荐适合的活动
  3. 城市信息资源：提供城市基本信息作为上下文资源

这个服务器将通过Server-Sent Events (SSE)协议与MCP客户端通信，允许任何符合MCP标准的AI应用调用我们的服务。

步骤一：设计MCP服务器的基本结构

首先，让我们设计MCP服务器的基本结构。创建一个名为weather_server.py的文件：

# weather_server.py
from mcp.server.fastmcp import FastMCP
import asyncio
import logging# 配置日志
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)# 创建MCP服务器实例
mcp_server = FastMCP(name="weather-assistant",description="天气查询和活动推荐服务",host="0.0.0.0",port=1234
)# 主函数
async def main():logger.info("启动MCP天气服务器，监听地址: http://localhost:1234")await mcp_server.run_sse_async()if __name__ == "__main__":asyncio.run(main())

这段代码创建了一个基本的MCP服务器，它将监听1234端口，等待MCP客户端的连接。我们使用FastMCP类，这是MCP Python SDK提供的便捷服务器实现。

步骤二：实现天气查询工具

接下来，让我们实现天气查询工具。在真实应用中，你可能会调用外部天气API，但为了简化教程，我们将使用硬编码的天气数据：

# 在weather_server.py中添加# 模拟的天气数据库
WEATHER_DATA = {"北京": {"condition": "晴", "temperature": 25, "humidity": 40},"上海": {"condition": "多云", "temperature": 27, "humidity": 60},"广州": {"condition": "雨", "temperature": 30, "humidity": 80},"深圳": {"condition": "多云", "temperature": 29, "humidity": 70},"杭州": {"condition": "晴", "temperature": 26, "humidity": 50}
}@mcp_server.tool(name="get_weather", description="获取指定城市的天气信息")
async def get_weather(city: str) -> str:"""获取指定城市的天气信息参数:city: 城市名称，如"北京"、"上海"返回:天气信息描述"""logger.info(f"获取天气信息: {city}")if city in WEATHER_DATA:weather = WEATHER_DATA[city]return f"{city}：{weather['condition']}，{weather['temperature']}°C，湿度{weather['humidity']}%"else:return f"抱歉，未找到{city}的天气信息"

这段代码定义了一个名为get_weather的工具，它接受一个城市名称参数，返回该城市的天气信息。我们使用@mcp_server.tool装饰器将函数注册为MCP工具，这样MCP客户端就能发现并调用它。

步骤三：实现活动推荐工具

接下来，让我们实现活动推荐工具：

# 在weather_server.py中添加# 基于天气的活动推荐
ACTIVITY_RECOMMENDATIONS = {"晴": ["户外散步", "骑自行车", "野餐", "参观公园", "户外运动"],"多云": ["参观博物馆", "购物", "咖啡馆休息", "室外摄影", "城市探索"],"雨": ["室内电影", "博物馆参观", "购物中心", "室内游泳", "图书馆阅读"],"雪": ["室内活动", "温泉", "热饮", "电影院", "室内游戏"]
}@mcp_server.tool(name="suggest_activity", description="根据天气状况推荐适合的活动")
async def suggest_activity(weather_condition: str) -> str:"""根据天气状况推荐适合的活动参数:weather_condition: 天气状况描述，如"晴"、"多云"、"雨"返回:推荐活动列表"""logger.info(f"推荐活动: 基于天气 {weather_condition}")# 提取天气关键词（简单处理）for key in ACTIVITY_RECOMMENDATIONS:if key in weather_condition:activities = ACTIVITY_RECOMMENDATIONS[key]return f"基于当前{key}天气，推荐的活动：{', '.join(activities[:3])}"return "无法根据提供的天气状况推荐活动，建议选择适合当地气候的活动"

这个工具接受天气状况作为输入，根据天气关键词推荐适合的活动。这展示了MCP工具如何实现基于一个工具输出的结果来调用另一个工具的模式。

步骤四：添加城市信息资源

除了工具外，MCP还支持资源（Resources），这些是只读的信息，可以帮助模型更好地理解上下文。让我们添加一些城市信息作为资源：

# 在weather_server.py中添加# 城市信息资源
CITY_INFO = {"北京": "中国首都，政治、文化中心，著名景点包括故宫、长城等","上海": "中国最大城市，经济金融中心，著名景点有外滩、东方明珠等","广州": "广东省省会，中国南方重要商业城市，著名景点有白云山、陈家祠等","深圳": "创新科技中心，毗邻香港，著名景点有世界之窗、欢乐谷等","杭州": "浙江省省会，风景秀丽，著名景点有西湖、灵隐寺等"
}# 注册城市信息资源
for city, info in CITY_INFO.items():@mcp_server.resource(uri=f"cities/{city}", description=f"{city}的基本信息")async def city_resource(query=None):city_name = query["path"].split("/")[-1]return CITY_INFO.get(city_name, f"未找到{city_name}的信息")

这段代码为每个城市创建了一个资源，可以通过cities/{城市名}的URI访问。资源与工具的区别在于，资源是只读的信息，不执行复杂操作，主要用于提供背景知识。

步骤五：集成并测试完整服务器

现在，让我们整合所有代码并运行我们的MCP服务器：

# 完整的weather_server.py
from mcp.server.fastmcp import FastMCP
import asyncio
import logging# 配置日志
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)# 创建MCP服务器实例
mcp_server = FastMCP(name="weather-assistant",description="天气查询和活动推荐服务",host="0.0.0.0",port=1234
)# 模拟的天气数据库
WEATHER_DATA = {"北京": {"condition": "晴", "temperature": 25, "humidity": 40},"上海": {"condition": "多云", "temperature": 27, "humidity": 60},"广州": {"condition": "雨", "temperature": 30, "humidity": 80},"深圳": {"condition": "多云", "temperature": 29, "humidity": 70},"杭州": {"condition": "晴", "temperature": 26, "humidity": 50}
}@mcp_server.tool(name="get_weather", description="获取指定城市的天气信息")
async def get_weather(city: str) -> str:"""获取指定城市的天气信息参数:city: 城市名称，如"北京"、"上海"返回:天气信息描述"""logger.info(f"获取天气信息: {city}")if city in WEATHER_DATA:weather = WEATHER_DATA[city]return f"{city}：{weather['condition']}，{weather['temperature']}°C，湿度{weather['humidity']}%"else:return f"抱歉，未找到{city}的天气信息"# 基于天气的活动推荐
ACTIVITY_RECOMMENDATIONS = {"晴": ["户外散步", "骑自行车", "野餐", "参观公园", "户外运动"],"多云": ["参观博物馆", "购物", "咖啡馆休息", "室外摄影", "城市探索"],"雨": ["室内电影", "博物馆参观", "购物中心", "室内游泳", "图书馆阅读"],"雪": ["室内活动", "温泉", "热饮", "电影院", "室内游戏"]
}@mcp_server.tool(name="suggest_activity", description="根据天气状况推荐适合的活动")
async def suggest_activity(weather_condition: str) -> str:"""根据天气状况推荐适合的活动参数:weather_condition: 天气状况描述，如"晴"、"多云"、"雨"返回:推荐活动列表"""logger.info(f"推荐活动: 基于天气 {weather_condition}")# 提取天气关键词（简单处理）for key in ACTIVITY_RECOMMENDATIONS:if key in weather_condition:activities = ACTIVITY_RECOMMENDATIONS[key]return f"基于当前{key}天气，推荐的活动：{', '.join(activities[:3])}"return "无法根据提供的天气状况推荐活动，建议选择适合当地气候的活动"# 城市信息资源
CITY_INFO = {"北京": "中国首都，政治、文化中心，著名景点包括故宫、长城等","上海": "中国最大城市，经济金融中心，著名景点有外滩、东方明珠等","广州": "广东省省会，中国南方重要商业城市，著名景点有白云山、陈家祠等","深圳": "创新科技中心，毗邻香港，著名景点有世界之窗、欢乐谷等","杭州": "浙江省省会，风景秀丽，著名景点有西湖、灵隐寺等"
}# 注册城市信息资源
for city, info in CITY_INFO.items():@mcp_server.resource(uri=f"cities/{city}", description=f"{city}的基本信息")async def city_resource_gen(city=city):return CITY_INFO.get(city, f"未找到{city}的信息")# 主函数
async def main():logger.info("启动MCP天气服务器，监听地址: http://localhost:1234")await mcp_server.run_sse_async()if __name__ == "__main__":asyncio.run(main())

将上述代码保存到weather_server.py文件中，然后通过以下命令运行服务器：

python weather_server.py

如果一切正常，你应该会看到类似以下的输出：

INFO - 启动MCP天气服务器，监听地址: http://localhost:1234
INFO - MCP服务器启动完成，提供2个工具和5个资源

步骤六：创建MCP客户端测试工具

现在，我们的MCP服务器已经运行起来，但我们需要一个MCP客户端来测试它。创建一个名为weather_client.py的文件：

# weather_client.py
import asyncio
from mcp.client.session import ClientSession
from mcp.client.sse import sse_clientclass WeatherMCPClient:def __init__(self, server_url="http://localhost:1234/sse"):self.server_url = server_urlself._sse_context = Noneself._session = Noneasync def __aenter__(self):# 创建SSE连接self._sse_context = sse_client(self.server_url)self.read, self.write = await self._sse_context.__aenter__()# 创建MCP会话self._session = ClientSession(self.read, self.write)await self._session.__aenter__()await self._session.initialize()return selfasync def __aexit__(self, exc_type, exc_val, exc_tb):if self._session:await self._session.__aexit__(exc_type, exc_val, exc_tb)if self._sse_context:await self._sse_context.__aexit__(exc_type, exc_val, exc_tb)async def list_tools(self):"""列出可用工具"""return await self._session.list_tools()async def list_resources(self):"""列出可用资源"""return await self._session.list_resources()async def call_tool(self, name, arguments):"""调用工具"""return await self._session.call_tool(name, arguments)async def get_resource(self, uri):"""获取资源"""return await self._session.get_resource(uri)async def main():"""测试MCP客户端与服务器的交互"""async with WeatherMCPClient() as client:print("✅ 连接MCP服务器成功")# 列出工具tools = await client.list_tools()print(f"\n🔧 可用工具列表:")for tool in tools.tools:print(f"  - {tool.name}: {tool.description}")# 列出资源resources = await client.list_resources()print(f"\n📚 可用资源列表:")for resource in resources.resources:print(f"  - {resource.uri}: {resource.description}")# 测试天气工具city = "上海"print(f"\n🌤 获取{city}天气:")weather_result = await client.call_tool("get_weather", {"city": city})weather_info = weather_result.content[0].textprint(f"  结果: {weather_info}")# 测试活动推荐工具print(f"\n🎯 根据天气推荐活动:")activity_result = await client.call_tool("suggest_activity", {"weather_condition": weather_info})print(f"  推荐: {activity_result.content[0].text}")# 测试资源访问print(f"\n🏙 获取{city}信息:")city_info = await client.get_resource(f"cities/{city}")print(f"  信息: {city_info.content[0].text}")if __name__ == "__main__":asyncio.run(main())

这个客户端程序连接到我们的MCP服务器，列出可用的工具和资源，然后依次调用它们，展示一个完整的交互流程。

步骤七：运行并测试MCP客户端

确保MCP服务器仍在运行，然后在另一个终端窗口中执行以下命令：

python weather_client.py

如果一切正常，你应该会看到类似以下的输出：

✅ 连接MCP服务器成功🔧 可用工具列表:- get_weather: 获取指定城市的天气信息- suggest_activity: 根据天气状况推荐适合的活动📚 可用资源列表:- cities/北京: 北京的基本信息- cities/上海: 上海的基本信息- cities/广州: 广州的基本信息- cities/深圳: 深圳的基本信息- cities/杭州: 杭州的基本信息🌤 获取上海天气:结果: 上海：多云，27°C，湿度60%🎯 根据天气推荐活动:推荐: 基于当前多云天气，推荐的活动：参观博物馆, 购物, 咖啡馆休息🏙 获取上海信息:信息: 中国最大城市，经济金融中心，著名景点有外滩、东方明珠等

恭喜！你已经成功创建并测试了一个完整的MCP服务器和客户端应用！

步骤八：与AI助手集成

现在，我们已经有了一个功能完善的MCP服务器，接下来我们将展示如何将其与AI大语言模型集成。这里我们将使用一个简化的AI助手模型来模拟集成过程。

创建一个名为ai_assistant.py的文件：

# ai_assistant.py
import asyncio
import json
from weather_client import WeatherMCPClient# 模拟LLM的简单函数
def simulate_llm_response(user_query, context=None):"""模拟大语言模型的响应生成"""# 这只是一个极简的模拟，真实LLM会复杂得多query = user_query.lower()if "天气" in query:# 从查询中提取城市（简化处理）cities = ["北京", "上海", "广州", "深圳", "杭州"]city = next((city for city in cities if city in query), "北京")# 返回要调用的工具return {"thought": f"用户询问{city}的天气，我需要调用天气工具","tool_call": {"name": "get_weather","arguments": {"city": city}}}elif "活动" in query or "做什么" in query:# 如果有天气上下文，推荐活动if context and "天气结果" in context:return {"thought": "用户询问活动建议，基于已知天气推荐","tool_call": {"name": "suggest_activity","arguments": {"weather_condition": context["天气结果"]}}}else:return {"thought": "用户询问活动但没有天气信息，先获取默认城市天气","tool_call": {"name": "get_weather","arguments": {"city": "北京"}}}else:# 简单对话回复return {"thought": "一般问候或闲聊","response": "你好！我是天气助手，可以帮你查询天气和推荐活动。请问你想了解哪个城市的天气？"}async def chat_session():"""模拟与用户的对话会话"""context = {}print("🤖 天气助手已启动！输入'退出'结束对话。")print("🤖 请问有什么可以帮助你的？")async with WeatherMCPClient() as client:while True:# 获取用户输入user_input = input("\n👤 你: ")if user_input.lower() in ["退出", "exit", "quit"]:print("🤖 助手: 再见！祝你有美好的一天！")break# 模拟LLM处理llm_output = simulate_llm_response(user_input, context)print(f"🧠 思考: {llm_output['thought']}")# 处理LLM输出if "tool_call" in llm_output:tool = llm_output["tool_call"]print(f"🔧 调用工具: {tool['name']}({json.dumps(tool['arguments'], ensure_ascii=False)})")# 执行工具调用result = await client.call_tool(tool["name"], tool["arguments"])tool_output = result.content[0].textprint(f"🔧 工具结果: {tool_output}")# 更新上下文if tool["name"] == "get_weather":context["天气结果"] = tool_output# 生成最终回复if tool["name"] == "get_weather":# 如果是天气查询，提示用户是否需要活动推荐response = f"{tool_output}\n\n需要我给你推荐适合这种天气的活动吗？"else:response = tool_outputelse:# 直接使用LLM生成的回复response = llm_output.get("response", "抱歉，我无法理解你的问题")# 输出助手回复print(f"\n🤖 助手: {response}")if __name__ == "__main__":asyncio.run(chat_session())

这个程序模拟了一个简单的AI助手，能够理解用户关于天气和活动的问题，并通过MCP客户端调用相应的工具。虽然这里使用的是一个非常简化的"AI模型"，但基本流程与实际的大语言模型集成是一致的。

步骤九：运行AI助手模拟程序

确保MCP服务器仍在运行，然后在另一个终端窗口中执行：

python ai_assistant.py

现在你可以与模拟的AI助手进行对话，例如：

🤖 天气助手已启动！输入'退出'结束对话。
🤖 请问有什么可以帮助你的？👤 你: 上海今天天气怎么样
🧠 思考: 用户询问上海的天气，我需要调用天气工具
🔧 调用工具: get_weather({"city": "上海"})
🔧 工具结果: 上海：多云，27°C，湿度60%🤖 助手: 上海：多云，27°C，湿度60%需要我给你推荐适合这种天气的活动吗？👤 你: 是的，请推荐
🧠 思考: 用户询问活动建议，基于已知天气推荐
🔧 调用工具: suggest_activity({"weather_condition": "上海：多云，27°C，湿度60%"})
🔧 工具结果: 基于当前多云天气，推荐的活动：参观博物馆, 购物, 咖啡馆休息🤖 助手: 基于当前多云天气，推荐的活动：参观博物馆, 购物, 咖啡馆休息👤 你: 退出
🤖 助手: 再见！祝你有美好的一天！

实际应用扩展方向

我们的示例虽然简单，但展示了MCP的核心工作流程。在实际应用中，你可以沿着以下方向扩展：

1. 集成真实API

将示例中的硬编码数据替换为实际的API调用，例如：

@mcp_server.tool(name="get_weather", description="获取指定城市的天气信息")
async def get_weather(city: str) -> str:"""实际应用中调用真实天气API"""api_key = "你的API密钥"url = f"https://api.weather.com/v1/forecast?city={city}&key={api_key}"async with aiohttp.ClientSession() as session:async with session.get(url) as response:if response.status == 200:data = await response.json()# 解析API返回的JSON数据condition = data["current"]["condition"]temp = data["current"]["temp"]humidity = data["current"]["humidity"]return f"{city}：{condition}，{temp}°C，湿度{humidity}%"else:return f"抱歉，无法获取{city}的天气信息"

2. 添加认证机制

为你的MCP服务器添加安全认证：

# 添加基本认证
from mcp.server.auth import BasicAuthauth_provider = BasicAuth({"user1": "password1", "user2": "password2"})
mcp_server = FastMCP(name="weather-assistant",description="天气查询和活动推荐服务",host="0.0.0.0",port=1234,auth_provider=auth_provider
)

3. 增加更复杂的工具

添加更复杂的工具和资源，如天气预报、旅游推荐等：

@mcp_server.tool(name="get_forecast", description="获取未来5天天气预报")
async def get_forecast(city: str) -> str:# 实现5天天气预报逻辑pass@mcp_server.tool(name="recommend_tourist_spots", description="根据天气推荐旅游景点")
async def recommend_tourist_spots(city: str, weather_condition: str) -> str:# 实现景点推荐逻辑pass

4. 与实际LLM集成

将模拟的LLM替换为实际的API调用：

import openaiasync def get_llm_response(user_query, context=None):messages = [{"role": "system", "content": "你是一个天气助手，可以帮助用户查询天气和推荐活动"}]# 添加上下文if context:messages.append({"role": "system", "content": f"当前已知信息: {json.dumps(context, ensure_ascii=False)}"})# 添加用户查询messages.append({"role": "user", "content": user_query})# 调用OpenAI APIresponse = await openai.ChatCompletion.acreate(model="gpt-4",messages=messages,function_call="auto",functions=[{"name": "get_weather","description": "获取指定城市的天气信息","parameters": {"type": "object","properties": {"city": {"type": "string","description": "城市名称，如北京、上海"}},"required": ["city"]}},# 定义其他工具...])return response

部署考虑

当你准备将MCP服务部署到生产环境时，考虑以下因素：

  1. 安全性：

    • 实施TLS加密（HTTPS）
    • 添加认证机制
    • 限制IP访问
    • 实施请求速率限制

  2. 可靠性：

    • 添加错误处理和重试逻辑
    • 实现健康检查端点
    • 考虑容器化（Docker）部署
    • 监控和日志收集

  3. 性能：

    • 针对高并发进行优化
    • 考虑使用异步I/O和连接池
    • 缓存频繁请求的结果

  4. 可维护性：

    • 添加全面的文档
    • 实现版本控制和变更日志
    • 设计可扩展的架构
    • 编写单元测试和集成测试

结语

在本文中，我们从零开始构建了一个功能完整的MCP服务器和客户端，并展示了如何将其与AI助手集成。虽然我们的示例相对简单，但它包含了MCP开发的核心流程和概念，为你开发更复杂的MCP应用奠定了基础。

MCP的强大之处在于它的标准化和模块化设计，使得AI模型能够无缝地与各种外部工具和资源交互。通过实现MCP服务器，你可以将自己的功能和数据暴露给支持MCP的AI应用，扩展它们的能力范围。

在接下来的系列文章中，我们将深入探讨MCP的更多高级主题，包括工具调用的细节、安全性考虑以及在真实企业环境中的应用案例。敬请期待！

参考资源

  1. MCP Python SDK 文档
  2. MCP 官方协议规范
  3. Anthropic Claude 官方文档
  4. MCP 服务器最佳实践

相关文章：

  • DemoGen：用于数据高效视觉运动策略学习的合成演示生成
  • Python 文本和字节序列(支持字符串和字节序列的双模式API)
  • Webview+Python：用HTML打造跨平台桌面应用的创新方案
  • DHTMLX宣布推出支持 Redux、TypeScript 和 MUI 的 React Gantt甘特图控件
  • xml+html 概述
  • 【前端HTML生成条形码——MQ】
  • 极狐GitLab 项目导入导出设置介绍？
  • #Linux动态大小裁剪以及包大小变大排查思路
  • ApiHug 前端解决方案 - M1 内侧
  • Clickhouse 配置参考
  • 类型补充，scan 和数据库管理命令
  • 一本通 2063：【例1.4】牛吃牧草 1005：地球人口承载力估计
  • 下载electron 22.3.27 源码错误集锦
  • 记录一次问题排查，前台传的日期参数到后台取到的时候少了一天。
  • 考研系列-计算机网络-第二章、物理层
  • IntelliJ IDEA clean git password
  • 广搜bfs-P1443 马的遍历
  • 8.Rust+Axum 数据库集成实战：从 ORM 选型到用户管理系统开发
  • Python爬虫实战: 有道翻译
  • Qt 创建QWidget的界面库(DLL)
  • 因商标近似李小龙形象被裁定无效，真功夫起诉国家知产局，法院判了
  • 2025年度人大立法工作计划将公布：研究启动法律清理工作
  • AI翻译技术已走向大规模商用，应用场景覆盖多个关键领域
  • 牛市早报｜商务部：目前中美之间未进行任何经贸谈判
  • 朝中社发表评论文章，谴责美军部署B1-B轰炸机至日本
  • 临汾攻坚PM2.5：一座曾经“爆表”城市的空气治理探索