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

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

# 创建虚拟环境
python -m venv venv

# 激活它
# 在Windows上：
venv\Scripts\activate
# 在Mac/Linux上：
source venv/bin/activate

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

import json
import asyncio
from datetime import datetime
from typing import Any, Dict, List

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

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

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

@self.server.list_resources()
async def handle_list_resources() -> List[Resource]:
"""List available resources"""
return [
Resource(
uri="feedback://recent",
name="Recent Feedback",
description="View recent customer feedback",
mimeType="application/json"
),
Resource(
uri="feedback://summary",
name="Feedback Summary",
description="Get a summary of all feedback",
mimeType="text/plain"
)
]

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

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

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

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

return f"""
Feedback Summary
================
Total Feedback: {total}
Positive: {sentiments['positive']} ({sentiments['positive']/total*100:.1f}%)
Neutral: {sentiments['neutral']} ({sentiments['neutral']/total*100:.1f}%)
Negative: {sentiments['negative']} ({sentiments['negative']/total*100:.1f}%)
"""

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

@self.server.list_tools()
async def handle_list_tools() -> List[Tool]:
"""List available tools"""
return [
Tool(
name="collect_feedback",
description="Collect new customer feedback",
inputSchema={
"type": "object",
"properties": {
"customer_name": {
"type": "string",
"description": "Name of the customer"
},
"feedback": {
"type": "string",
"description": "The feedback text"
},
"rating": {
"type": "integer",
"description": "Rating from 1-5",
"minimum": 1,
"maximum": 5
}
},
"required": ["customer_name", "feedback", "rating"]
}
),
Tool(
name="analyze_sentiment",
description="Analyze sentiment of feedback text",
inputSchema={
"type": "object",
"properties": {
"text": {
"type": "string",
"description": "Text to analyze"
}
},
"required": ["text"]
}
)
]

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

if name == "collect_feedback":
# Analyze sentiment based on rating
rating = arguments["rating"]
if rating >= 4:
sentiment = "positive"
elif rating == 3:
sentiment = "neutral"
else:
sentiment = "negative"

# Store feedback
feedback_entry = {
"id": len(self.feedback_list) + 1,
"timestamp": datetime.now().isoformat(),
"customer_name": arguments["customer_name"],
"feedback": arguments["feedback"],
"rating": rating,
"sentiment": sentiment
}

self.feedback_list.append(feedback_entry)

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

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

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

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

if positive_count > negative_count:
sentiment = "positive"
elif negative_count > positive_count:
sentiment = "negative"
else:
sentiment = "neutral"

return [TextContent(
type="text",
text=f"Sentiment: {sentiment}"
)]

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

async def run(self):
"""Run the server"""
async with stdio_server() as (read_stream, write_stream):
await self.server.run(
read_stream,
write_stream,
InitializationOptions(
server_name="customer-feedback",
server_version="0.1.0",
capabilities=self.server.get_capabilities(
notification_options=NotificationOptions(),
experimental_capabilities={}
)
)
)

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

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

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

# 使用检查器运行您的服务器
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系统了！
Aa