是时候动手实践了！在这一部分中，我们将设置开发环境并创建我们的第一个MCP服务器。如果您从未编写过代码，也不用担心 - 我们将一步一步来。

我们要构建什么

还记得第1部分中Maria的咖啡馆吗？我们正在创建一个反馈收集系统，该系统：

o 存储客户反馈

o 分析情感倾向（满意、中性、不满意）

o 生成总结报告

o 识别改进领域

第1步：设置您的环境

安装Python

首先，我们需要Python（3.8版本或更高）：

Windows：

1. 访问python.org

2. 下载安装程序

3. 重要：勾选"Add Python to PATH"

4. 点击安装

Mac：

# 如果您有Homebrew
brew install python3

# 或者从python.org下载

Linux：

sudo apt update
sudo apt install python3 python3-pip

创建您的项目

打开您的终端（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

第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())

第4步：测试您的服务器

让我们确保一切正常工作！首先，使脚本可执行：

# 在Mac/Linux上：
chmod +x feedback_server.py

# 在Windows上，您可以跳过这一步

现在使用MCP检查器测试它：

# 安装MCP检查器工具
pip install mcp-inspector

# 使用检查器运行您的服务器
mcp-inspector feedback_server.py

您应该看到：

o 您的服务器启动

o 可用资源（最近反馈、反馈总结）

o 可用工具（collect_feedback、analyze_sentiment）

在检查器中尝试这些命令：

1. 使用一些测试数据调用collect_feedback

2. 读取feedback://recent资源

3. 读取feedback://summary资源

第5步：理解我们构建的内容

让我们分解关键部分：

资源

我们创建了两个Agent可以读取的资源：

o feedback://recent：显示最后5个反馈条目

o feedback://summary：提供所有反馈的统计信息

工具

我们实现了两个Agent可以使用的工具：

o collect_feedback：保存新反馈，根据评分自动分析情感

o analyze_sentiment：基于关键词的简单情感分析

服务器

我们的服务器：

o 在内存中存储反馈（稍后我们将添加持久性）

o 提供标准MCP接口

o 可以被任何MCP兼容的AI Agent使用

常见问题和解决方案

"找不到Python"

o 确保Python在您的PATH中

o 尝试使用python3而不是python

"找不到模块"

o 确保您的虚拟环境已激活

o 重新安装：pip install mcp

"权限被拒绝"

o 在Mac/Linux上：使用chmod +x feedback_server.py

o 在Windows上：如果需要，以管理员身份运行

下一步是什么？

恭喜！您已经构建了您的第一个MCP服务器。在第4部分中，我们将：

o 创建一个使用我们服务器的AI Agent

o 将其连接到LLM以获得智能响应

o 构建自动化工作流

o 添加数据持久性

您的服务器现在已经准备好被AI Agent使用了。您将如何扩展它以处理您的特定用例？在评论中分享您的想法！

准备好更多内容了吗？完整的教程和额外示例可在
brandonredmond.com/learn/paths/ai-systems-intro获得。

技术要点总结

MCP服务器的核心概念

o 资源（Resources）：提供数据访问的接口

o 工具（Tools）：提供功能操作的接口

o 服务器（Server）：协调资源和工具的统一入口

开发环境设置

1. Python环境：确保使用3.8+版本

2. 虚拟环境：隔离项目依赖

3. MCP SDK：安装必要的开发工具

代码结构分析

o 异步编程：使用async/await处理并发

o 装饰器模式：使用@self.server.list_resources()等装饰器

o 类型提示：使用typing模块提供类型安全

测试和调试

o MCP Inspector：官方测试工具

o 资源测试：验证数据访问功能

o 工具测试：验证功能操作

扩展建议

1. 数据持久化：添加数据库存储

2. 错误处理：完善异常处理机制

3. 日志记录：添加详细的日志输出

4. 配置管理：支持环境变量和配置文件

通过这个教程，您已经掌握了MCP服务器的基础知识，可以开始构建自己的AI系统了！