使用Python设置您的第一个MCP服务器（第3部分）

作者头像东城·2026-01-31 23:21:08·
是时候动手实践了！在这一部分中，我们将设置开发环境并创建我们的第一个MCP服务器。如果您从未编写过代码，也不用担心 - 我们将一步一步来。
我们要构建什么
还记得第1部分中Maria的咖啡馆吗？我们正在创建一个反馈收集系统，该系统：
存储客户反馈
分析情感倾向（满意、中性、不满意）
生成总结报告
识别改进领域
第1步：设置您的环境
安装Python
首先，我们需要Python（3.8版本或更高）：
Windows：
访问python.org
下载安装程序
重要：勾选"Add Python to PATH"
点击安装
Mac：
  1. # 如果您有Homebrew
  2. brew install python3

  4. # 或者从python.org下载
bash
Linux：
  1. sudo apt update
  2. sudo apt install python3 python3-pip
bash
创建您的项目
打开您的终端（Windows上的命令提示符）并运行：
  1. # 创建新目录
  2. mkdir mcp-feedback-system
  3. cd mcp-feedback-system

  5. # 创建虚拟环境
  6. python -m venv venv

  8. # 激活它
  9. # 在Windows上：
  10. venv\Scripts\activate
  11. # 在Mac/Linux上：
  12. source venv/bin/activate

  14. # 安装MCP SDK
  15. pip install mcp
bash
第2步：理解MCP服务器基础
MCP服务器有三个主要部分：
资源（我们有什么数据？）
  1. # 就像菜单上的项目
  2. - "来自客户的最近反馈"
  3. - "本周反馈总结"
  4. - "改进建议列表"
工具（我们能做什么？）
  1. # 就像厨房设备
  2. - "收集新反馈"
  3. - "分析情感"
  4. - "生成报告"
服务器（我们如何提供服务？）
  1. # 就像餐厅本身
  2. - 处理请求
  3. - 管理连接
  4. - 提供接口
第3步：构建我们的第一个MCP服务器
创建一个名为feedback_server.py的文件：
  1. #!/usr/bin/env python3
  2. """
  3. Customer Feedback MCP Server
  4. A simple server for collecting and analyzing customer feedback
  5. """

  7. import json
  8. import asyncio
  9. from datetime import datetime
  10. from typing import Any, Dict, List

  12. # MCP SDK imports
  13. from mcp.server.models import InitializationOptions
  14. from mcp.server import NotificationOptions, Server
  15. from mcp.server.stdio import stdio_server
  16. from mcp.types import Resource, Tool, TextContent

  18. class FeedbackServer:
  19. def __init__(self):
  20. self.server = Server("customer-feedback")
  21. self.feedback_list = []
  22. self.setup_handlers()

  24. def setup_handlers(self):
  25. """Set up all the server handlers"""

  27. @self.server.list_resources()
  28. async def handle_list_resources() -> List[Resource]:
  29. """List available resources"""
  30. return [
  31. Resource(
  32. uri="feedback://recent",
  33. name="Recent Feedback",
  34. description="View recent customer feedback",
  35. mimeType="application/json"
  36. ),
  37. Resource(
  38. uri="feedback://summary",
  39. name="Feedback Summary",
  40. description="Get a summary of all feedback",
  41. mimeType="text/plain"
  42. )
  43. ]

  45. @self.server.read_resource()
  46. async def handle_read_resource(uri: str) -> str:
  47. """Read a specific resource"""
  48. if uri == "feedback://recent":
  49. # Return last 5 feedback entries
  50. recent = self.feedback_list[-5:] if self.feedback_list else []
  51. return json.dumps(recent, indent=2)

  53. elif uri == "feedback://summary":
  54. # Generate summary
  55. if not self.feedback_list:
  56. return "No feedback collected yet."

  58. total = len(self.feedback_list)
  59. sentiments = {"positive": 0, "neutral": 0, "negative": 0}

  61. for feedback in self.feedback_list:
  62. sentiments[feedback["sentiment"]] += 1

  64. return f"""
  65. Feedback Summary
  66. ================
  67. Total Feedback: {total}
  68. Positive: {sentiments['positive']} ({sentiments['positive']/total*100:.1f}%)
  69. Neutral: {sentiments['neutral']} ({sentiments['neutral']/total*100:.1f}%)
  70. Negative: {sentiments['negative']} ({sentiments['negative']/total*100:.1f}%)
  71. """

  73. raise ValueError(f"Unknown resource: {uri}")

  75. @self.server.list_tools()
  76. async def handle_list_tools() -> List[Tool]:
  77. """List available tools"""
  78. return [
  79. Tool(
  80. name="collect_feedback",
  81. description="Collect new customer feedback",
  82. inputSchema={
  83. "type": "object",
  84. "properties": {
  85. "customer_name": {
  86. "type": "string",
  87. "description": "Name of the customer"
  88. },
  89. "feedback": {
  90. "type": "string",
  91. "description": "The feedback text"
  92. },
  93. "rating": {
  94. "type": "integer",
  95. "description": "Rating from 1-5",
  96. "minimum": 1,
  97. "maximum": 5
  98. }
  99. },
  100. "required": ["customer_name", "feedback", "rating"]
  101. }
  102. ),
  103. Tool(
  104. name="analyze_sentiment",
  105. description="Analyze sentiment of feedback text",
  106. inputSchema={
  107. "type": "object",
  108. "properties": {
  109. "text": {
  110. "type": "string",
  111. "description": "Text to analyze"
  112. }
  113. },
  114. "required": ["text"]
  115. }
  116. )
  117. ]

  119. @self.server.call_tool()
  120. async def handle_call_tool(
  121. name: str,
  122. arguments: Dict[str, Any]
  123. ) -> List[TextContent]:
  124. """Handle tool calls"""

  126. if name == "collect_feedback":
  127. # Analyze sentiment based on rating
  128. rating = arguments["rating"]
  129. if rating >= 4:
  130. sentiment = "positive"
  131. elif rating == 3:
  132. sentiment = "neutral"
  133. else:
  134. sentiment = "negative"

  136. # Store feedback
  137. feedback_entry = {
  138. "id": len(self.feedback_list) + 1,
  139. "timestamp": datetime.now().isoformat(),
  140. "customer_name": arguments["customer_name"],
  141. "feedback": arguments["feedback"],
  142. "rating": rating,
  143. "sentiment": sentiment
  144. }

  146. self.feedback_list.append(feedback_entry)

  148. return [TextContent(
  149. type="text",
  150. text=f"Feedback collected successfully! ID: {feedback_entry['id']}"
  151. )]

  153. elif name == "analyze_sentiment":
  154. text = arguments["text"].lower()

  156. # Simple sentiment analysis
  157. positive_words = ["great", "excellent", "love", "amazing", "wonderful"]
  158. negative_words = ["bad", "terrible", "hate", "awful", "horrible"]

  160. positive_count = sum(1 for word in positive_words if word in text)
  161. negative_count = sum(1 for word in negative_words if word in text)

  163. if positive_count > negative_count:
  164. sentiment = "positive"
  165. elif negative_count > positive_count:
  166. sentiment = "negative"
  167. else:
  168. sentiment = "neutral"

  170. return [TextContent(
  171. type="text",
  172. text=f"Sentiment: {sentiment}"
  173. )]

  175. raise ValueError(f"Unknown tool: {name}")

  177. async def run(self):
  178. """Run the server"""
  179. async with stdio_server() as (read_stream, write_stream):
  180. await self.server.run(
  181. read_stream,
  182. write_stream,
  183. InitializationOptions(
  184. server_name="customer-feedback",
  185. server_version="0.1.0",
  186. capabilities=self.server.get_capabilities(
  187. notification_options=NotificationOptions(),
  188. experimental_capabilities={}
  189. )
  190. )
  191. )

  193. # Main entry point
  194. async def main():
  195. server = FeedbackServer()
  196. await server.run()

  198. if __name__ == "__main__":
  199. asyncio.run(main())
python
第4步：测试您的服务器
让我们确保一切正常工作！首先，使脚本可执行：
  1. # 在Mac/Linux上：
  2. chmod +x feedback_server.py

  4. # 在Windows上，您可以跳过这一步
bash
现在使用MCP检查器测试它：
  1. # 安装MCP检查器工具
  2. pip install mcp-inspector

  4. # 使用检查器运行您的服务器
  5. mcp-inspector feedback_server.py
bash
您应该看到：
您的服务器启动
可用资源（最近反馈、反馈总结）
可用工具（collect_feedback、analyze_sentiment）
在检查器中尝试这些命令：
使用一些测试数据调用collect_feedback
读取feedback://recent资源
读取feedback://summary资源
第5步：理解我们构建的内容
让我们分解关键部分：
资源
我们创建了两个Agent可以读取的资源：
feedback://recent：显示最后5个反馈条目
feedback://summary：提供所有反馈的统计信息
工具
我们实现了两个Agent可以使用的工具：
collect_feedback：保存新反馈，根据评分自动分析情感
analyze_sentiment：基于关键词的简单情感分析
服务器
我们的服务器：
在内存中存储反馈（稍后我们将添加持久性）
提供标准MCP接口
可以被任何MCP兼容的AI Agent使用
常见问题和解决方案
"找不到Python"
确保Python在您的PATH中
尝试使用python3而不是python
"找不到模块"
确保您的虚拟环境已激活
重新安装：pip install mcp
"权限被拒绝"
在Mac/Linux上：使用chmod +x feedback_server.py
在Windows上：如果需要，以管理员身份运行
下一步是什么？
恭喜！您已经构建了您的第一个MCP服务器。在第4部分中，我们将：
创建一个使用我们服务器的AI Agent
将其连接到LLM以获得智能响应
构建自动化工作流
添加数据持久性
您的服务器现在已经准备好被AI Agent使用了。您将如何扩展它以处理您的特定用例？在评论中分享您的想法！
准备好更多内容了吗？完整的教程和额外示例可在brandonredmond.com/learn/paths/ai-systems-intro获得。
技术要点总结
MCP服务器的核心概念
资源（Resources）：提供数据访问的接口
工具（Tools）：提供功能操作的接口
服务器（Server）：协调资源和工具的统一入口
开发环境设置
Python环境：确保使用3.8+版本
虚拟环境：隔离项目依赖
MCP SDK：安装必要的开发工具
代码结构分析
异步编程：使用async/await处理并发
装饰器模式：使用@self.server.list_resources()等装饰器
类型提示：使用typing模块提供类型安全
测试和调试
MCP Inspector：官方测试工具
资源测试：验证数据访问功能
工具测试：验证功能操作
扩展建议
数据持久化：添加数据库存储
错误处理：完善异常处理机制
日志记录：添加详细的日志输出
配置管理：支持环境变量和配置文件
通过这个教程，您已经掌握了MCP服务器的基础知识，可以开始构建自己的AI系统了！