[附代码]MCP革命：AI时代万物互联的技术基石 - 2025年3月最新研究报告

# 1. 安装uv包管理工具(推荐替代pip)# pip install uv
# 2. 初始化MCP服务器项目# uv init server
# 3. 添加MCP依赖# uv add 'mcp[cli]'

# server.py - 基础MCP服务器实现from mcp.server.fastmcp import FastMCPfrom typing import List, Dict, Any, Optionalimport datetime
# 创建MCP服务器实例mcp = FastMCP("AdvancedAssistantTools")
# 定义日历工具@mcp.tool("calendar", description="日历管理与查询工具")class CalendarTool:    @mcp.function(        name="query_events",        description="查询指定日期的日历事件"    )    def query_events(        self,        date: str = mcp.parameter(description="查询日期，格式:YYYY-MM-DD")    ) -> List[Dict[str, Any]]:        """查询日历事件演示函数"""        # 在实际应用中，这里会连接真实的日历API        # 本示例仅作演示        try:            query_date = datetime.datetime.strptime(date, "%Y-%m-%d").date()            today = datetime.date.today()
            # 模拟数据            if (query_date - today).days > 30:                return []            elif query_date.weekday() >= 5:  # 周末                return [                    {"title": "家庭聚餐", "start": "12:00", "end": "14:00", "location": "家"}                ]            else:  # 工作日                return [                    {"title": "团队例会", "start": "09:30", "end": "10:30", "location": "会议室A"},                    {"title": "项目评审", "start": "14:00", "end": "16:00", "location": "线上"}                ]        except ValueError:            return {"error": "日期格式错误，请使用YYYY-MM-DD格式"}
# 定义文档处理工具@mcp.tool("document", description="文档处理工具")class DocumentTool:    @mcp.function(        name="summarize",        description="自动总结文档内容"    )    def summarize(        self,        text: str = mcp.parameter(description="需要总结的文本内容"),        max_length: int = mcp.parameter(description="总结的最大长度", default=200)    ) -> Dict[str, Any]:        """文档总结演示函数"""        # 实际实现中会调用NLP模型进行总结        # 本示例仅作演示        if not text:            return {"error": "文本内容为空"}
        # 简单实现：返回前N个字符        summary = text[:min(max_length, len(text))]        return {            "summary": summary,            "original_length": len(text),            "summary_length": len(summary)        }
# 定义数据分析工具@mcp.tool("data_analytics", description="数据分析与可视化工具")class DataAnalyticsTool:    @mcp.function(        name="analyze_time_series",        description="分析时间序列数据并返回统计结果"    )    def analyze_time_series(        self,        data: List[float] = mcp.parameter(description="时间序列数据点列表"),        metric: str = mcp.parameter(description="分析指标：mean, median, trend", default="mean")    ) -> Dict[str, Any]:        """时间序列数据分析演示函数"""        if not data:            return {"error": "数据为空"}
        result = {"data_points": len(data)}
        if metric == "mean":            result["mean"] = sum(data) / len(data)        elif metric == "median":            sorted_data = sorted(data)            mid = len(sorted_data) // 2            result["median"] = sorted_data[mid] if len(data) % 2 == 1 else (sorted_data[mid-1] + sorted_data[mid]) / 2        elif metric == "trend":            # 简单线性趋势计算            if len(data) < 2:                result["trend"] = "insufficient_data"            else:                first_half = sum(data[:len(data)//2]) / (len(data)//2)                second_half = sum(data[len(data)//2:]) / (len(data) - len(data)//2)                result["trend"] = "increasing" if second_half > first_half else "decreasing" if second_half < first_half else "stable"
        return result
# 启动服务器if __name__ == "__main__":    mcp.run(host="0.0.0.0", port=8000)

# advanced_server.py - 包含动态工具发现与权限控制from mcp.server.fastmcp import FastMCPfrom mcp.security import Permission, Role, SecurityContextfrom typing import Dict, Any, Listimport os
mcp = FastMCP("EnterpriseTools")
# 定义权限模型admin_role = Role("admin", "管理员角色，拥有全部权限")user_role = Role("user", "普通用户角色，仅有查询权限")
read_permission = Permission("read", "读取数据权限")write_permission = Permission("write", "写入数据权限")
admin_role.add_permissions([read_permission, write_permission])user_role.add_permissions([read_permission])
# 注册安全上下文security_context = SecurityContext()security_context.add_roles([admin_role, user_role])mcp.set_security_context(security_context)
# 企业数据库工具@mcp.tool("database", description="企业数据库访问工具")class DatabaseTool:    @mcp.function(        name="query_data",        description="查询企业数据",        required_permissions=["read"]  # 需要read权限    )    def query_data(        self,        table: str = mcp.parameter(description="表名"),        filters: Dict[str, Any] = mcp.parameter(description="查询条件")    ) -> Dict[str, Any]:        # 实际实现中会连接数据库        # 示例仅作演示        return {            "status": "success",            "data": [{"id": 1, "name": "示例数据"}],            "metadata": {"table": table, "filters": filters}        }
    @mcp.function(        name="update_data",        description="更新企业数据",        required_permissions=["write"]  # 需要write权限    )    def update_data(        self,        table: str = mcp.parameter(description="表名"),        record_id: int = mcp.parameter(description="记录ID"),        updates: Dict[str, Any] = mcp.parameter(description="更新内容")    ) -> Dict[str, Any]:        # 权限检查在框架层自动完成        return {            "status": "success",            "updated_id": record_id,            "metadata": {"table": table, "updates": updates}        }
# 动态工具注册class ToolRegistry:    def __init__(self, mcp_server):        self.mcp = mcp_server        self.tool_directory = "./plugins"
    def scan_and_register(self):        """扫描插件目录并注册工具"""        if not os.path.exists(self.tool_directory):            os.makedirs(self.tool_directory)
        for filename in os.listdir(self.tool_directory):            if filename.endswith(".py") and not filename.startswith("_"):                module_path = os.path.join(self.tool_directory, filename)                self.register_tool_from_file(module_path)
    def register_tool_from_file(self, file_path):        """从文件动态注册工具"""        # 实际实现会动态加载Python模块        # 此处仅作示例说明        tool_name = os.path.basename(file_path).replace(".py", "")        print(f"发现工具：{tool_name} 从 {file_path}")        # 实际实现中的动态加载代码
# 启动服务器前扫描并注册工具if __name__ == "__main__":    registry = ToolRegistry(mcp)    registry.scan_and_register()    mcp.run(host="0.0.0.0", port=8000)

# knowledge_management.pyfrom mcp.server.fastmcp import FastMCPfrom typing import Dict, Any, List, Optionalimport datetime
mcp = FastMCP("EnterpriseKnowledge")
@mcp.tool("knowledge_base", description="企业知识库管理工具")class KnowledgeBaseTool:    def __init__(self):        # 实际实现中会连接到真实的知识库系统        # 以下仅为演示        self.knowledge_base = {            "product_specs": {                "last_updated": "2025-03-15",                "categories": ["硬件", "软件", "服务"],                "total_documents": 1250            },            "technical_docs": {                "last_updated": "2025-03-10",                "categories": ["API文档", "架构设计", "操作手册"],                "total_documents": 3780            },            "company_policies": {                "last_updated": "2025-02-28",                "categories": ["人力资源", "财务", "IT安全"],                "total_documents": 420            }        }
    @mcp.function(        name="search_documents",        description="搜索企业知识库文档"    )    def search_documents(        self,        query: str = mcp.parameter(description="搜索关键词"),        category: Optional[str] = mcp.parameter(description="文档类别", default=None),        max_results: int = mcp.parameter(description="最大结果数", default=10)    ) -> Dict[str, Any]:        """搜索企业知识库"""        # 实际实现中会连接到搜索引擎        # 示例返回模拟数据        results = []
        if "product" in query.lower():            results.append({                "title": "产品规格说明书v2.5",                "category": "产品文档",                "last_updated": "2025-03-01",                "relevance": 0.95,                "summary": "详细介绍了公司所有产品的技术规格、兼容性信息和使用建议。"            })
        if "security" in query.lower() or "安全" in query:            results.append({                "title": "企业IT安全规范2025版",                "category": "公司政策",                "last_updated": "2025-01-15",                "relevance": 0.88,                "summary": "规定了企业IT系统的安全使用规范、访问控制政策和数据保护要求。"            })
        # 根据category过滤        if category:            results = [doc for doc in results if category.lower() in doc["category"].lower()]
        # 限制结果数量        results = results[:min(len(results), max_results)]
        return {            "query": query,            "total_matches": len(results),            "results": results,            "metadata": {                "search_time": datetime.datetime.now().isoformat(),                "index_coverage": "完整",                "applied_filters": {"category": category} if category else {}            }        }
    @mcp.function(        name="get_knowledge_stats",        description="获取知识库统计信息"    )    def get_knowledge_stats(self) -> Dict[str, Any]:        """获取知识库统计信息"""        total_docs = sum(category["total_documents"] for category in self.knowledge_base.values())        all_categories = []        for kb_section in self.knowledge_base.values():            all_categories.extend(kb_section["categories"])
        return {            "total_documents": total_docs,            "categories": list(set(all_categories)),            "last_system_update": "2025-03-15T08:30:00",            "sections": list(self.knowledge_base.keys())        }
# 启动服务器if __name__ == "__main__":    mcp.run(host="0.0.0.0", port=8001)

# customer_service.pyfrom mcp.server.fastmcp import FastMCPfrom typing import Dict, Any, List, Optionalimport uuidimport datetime
mcp = FastMCP("CustomerServiceSystem")
@mcp.tool("customer_support", description="客户服务与工单管理工具")class CustomerSupportTool:    def __init__(self):        # 模拟工单数据库        self.tickets = {}        self.knowledge_articles = [            {"id": "KA-001", "title": "如何重置密码", "category": "账户管理"},            {"id": "KA-002", "title": "订单退款流程", "category": "支付与退款"},            {"id": "KA-003", "title": "产品激活指南", "category": "产品使用"}        ]
    @mcp.function(        name="create_ticket",        description="创建客户工单"    )    def create_ticket(        self,        customer_email: str = mcp.parameter(description="客户邮箱"),        subject: str = mcp.parameter(description="工单主题"),        description: str = mcp.parameter(description="问题描述"),        priority: str = mcp.parameter(description="优先级：low, medium, high", default="medium"),        category: str = mcp.parameter(description="问题类别", default="general")    ) -> Dict[str, Any]:        """创建新的客户工单"""        # 生成工单ID        ticket_id = f"TK-{uuid.uuid4().hex[:8]}"
        # 创建工单记录        ticket = {            "id": ticket_id,            "customer_email": customer_email,            "subject": subject,            "description": description,            "priority": priority,            "category": category,            "status": "open",            "created_at": datetime.datetime.now().isoformat(),            "updated_at": datetime.datetime.now().isoformat(),            "assigned_to": None        }
        # 存储工单        self.tickets[ticket_id] = ticket
        return {            "status": "success",            "ticket_id": ticket_id,            "message": "工单创建成功",            "ticket": ticket        }
    @mcp.function(        name="update_ticket",        description="更新工单状态"    )    def update_ticket(        self,        ticket_id: str = mcp.parameter(description="工单ID"),        status: Optional[str] = mcp.parameter(description="工单状态", default=None),        notes: Optional[str] = mcp.parameter(description="更新说明", default=None),        assigned_to: Optional[str] = mcp.parameter(description="指派给", default=None)    ) -> Dict[str, Any]:        """更新工单状态"""        if ticket_id not in self.tickets:            return {                "status": "error",                "message": f"找不到工单 {ticket_id}"            }
        ticket = self.tickets[ticket_id]
        # 更新工单信息        if status:            ticket["status"] = status        if notes:            if "notes" not in ticket:                ticket["notes"] = []            ticket["notes"].append({                "content": notes,                "timestamp": datetime.datetime.now().isoformat()            })        if assigned_to:            ticket["assigned_to"] = assigned_to
        ticket["updated_at"] = datetime.datetime.now().isoformat()
        return {            "status": "success",            "message": "工单已更新",            "ticket": ticket        }
    @mcp.function(        name="search_knowledge_base",        description="搜索客服知识库"    )    def search_knowledge_base(        self,        query: str = mcp.parameter(description="搜索关键词"),        category: Optional[str] = mcp.parameter(description="文章类别", default=None)    ) -> Dict[str, Any]:        """搜索客服知识库文章"""        results = []
        # 简单关键词匹配        for article in self.knowledge_articles:            if query.lower() in article["title"].lower():                if category and category != article["category"]:                    continue                results.append(article)
        return {            "query": query,            "results_count": len(results),            "results": results        }
# 启动服务器if __name__ == "__main__":    mcp.run(host="0.0.0.0", port=8002)

# multi_agent_system.pyfrom mcp.server.fastmcp import FastMCPfrom mcp.client import McpClientfrom typing import Dict, Any, Listimport asyncioimport json
# 主协调智能体服务器coordinator_mcp = FastMCP("CoordinatorAgent")
@coordinator_mcp.tool("task_management", description="任务分解与协调工具")class TaskManagementTool:    @coordinator_mcp.function(        name="decompose_task",        description="将复杂任务分解为子任务"    )    def decompose_task(        self,        task_description: str = coordinator_mcp.parameter(description="任务描述"),        complexity: str = coordinator_mcp.parameter(description="任务复杂度: low, medium, high", default="medium")    ) -> Dict[str, Any]:        """将复杂任务分解为子任务"""        # 实际实现中会使用更复杂的任务分解算法        subtasks = []
        if "数据分析" in task_description:            subtasks.extend([                {"id": "subtask-1", "type": "data_collection", "description": "收集相关数据"},                {"id": "subtask-2", "type": "data_processing", "description": "数据清洗与预处理"},                {"id": "subtask-3", "type": "data_analysis", "description": "数据分析与洞察提取"},                {"id": "subtask-4", "type": "report_generation", "description": "生成分析报告"}            ])        elif "内容创作" in task_description:            subtasks.extend([                {"id": "subtask-1", "type": "research", "description": "主题研究"},                {"id": "subtask-2", "type": "outline", "description": "创建内容大纲"},                {"id": "subtask-3", "type": "draft", "description": "撰写初稿"},                {"id": "subtask-4", "type": "edit", "description": "编辑与润色"}            ])        else:            # 默认分解方案            subtasks.extend([                {"id": "subtask-1", "type": "research", "description": "背景调研"},                {"id": "subtask-2", "type": "execution", "description": "执行主要任务"},                {"id": "subtask-3", "type": "review", "description": "结果审核"}            ])
        return {            "original_task": task_description,            "subtasks": subtasks,            "total_subtasks": len(subtasks),            "estimated_complexity": complexity        }
    @coordinator_mcp.function(        name="assign_subtasks",        description="将子任务分配给专业智能体"    )    def assign_subtasks(        self,        subtasks: List[Dict[str, Any]] = coordinator_mcp.parameter(description="子任务列表"),        available_agents: List[str] = coordinator_mcp.parameter(description="可用智能体列表")    ) -> Dict[str, Any]:        """将子任务分配给专业智能体"""        assignments = {}
        # 简单的任务分配逻辑        for subtask in subtasks:            subtask_type = subtask["type"]
            # 根据任务类型选择合适的智能体            if subtask_type == "data_collection" or subtask_type == "data_processing":                if "data_agent" in available_agents:                    agent = "data_agent"                else:                    agent = available_agents[0]  # 默认使用第一个可用智能体            elif subtask_type == "research":                if "research_agent" in available_agents:                    agent = "research_agent"                else:                    agent = available_agents[0]            else:                # 轮询分配其他类型任务                agent = available_agents[len(assignments) % len(available_agents)]
            # 记录分配结果            if agent not in assignments:                assignments[agent] = []            assignments[agent].append(subtask["id"])
        return {            "assignments": assignments,            "total_assigned": sum(len(tasks) for tasks in assignments.values())        }
# 专业智能体服务器示例 - 数据分析智能体data_agent_mcp = FastMCP("DataAnalysisAgent")
@data_agent_mcp.tool("data_analysis", description="数据分析工具")class DataAnalysisTool:    @data_agent_mcp.function(        name="process_data",        description="处理和分析数据"    )    def process_data(        self,        data_source: str = data_agent_mcp.parameter(description="数据源"),        analysis_type: str = data_agent_mcp.parameter(description="分析类型：descriptive, predictive, prescriptive")    ) -> Dict[str, Any]:        """处理和分析数据"""        # 实际实现中会执行真实的数据分析        result = {            "status": "completed",            "data_source": data_source,            "analysis_type": analysis_type,            "results": {                "summary_statistics": {                    "count": 1000,                    "mean": 45.7,                    "median": 42.3,                    "std_dev": 15.2                },                "key_findings": [                    "发现异常值聚集在特定时间段",                    "数据呈现明显的季节性趋势",                    "两个关键变量存在显著相关性"                ]            }        }
        return result
# 智能体协作客户端示例class AgentCollaborationSystem:    def __init__(self):        self.coordinator_client = McpClient("http://localhost:8010")        self.data_agent_client = McpClient("http://localhost:8011")        self.research_agent_client = McpClient("http://localhost:8012")
    async def execute_complex_task(self, task_description):        """执行复杂任务的完整流程"""        print(f"接收任务: {task_description}")
        # 1. 任务分解        decomposition_result = await self.coordinator_client.invoke(            "task_management", "decompose_task",            {"task_description": task_description, "complexity": "high"}        )
        subtasks = decomposition_result["subtasks"]        print(f"任务已分解为 {len(subtasks)} 个子任务")
        # 2. 任务分配        available_agents = ["data_agent", "research_agent"]        assignment_result = await self.coordinator_client.invoke(            "task_management", "assign_subtasks",            {"subtasks": subtasks, "available_agents": available_agents}        )
        print(f"任务分配结果: {json.dumps(assignment_result['assignments'], indent=2)}")
        # 3. 并行执行子任务        execution_tasks = []
        for agent, task_ids in assignment_result["assignments"].items():            for task_id in task_ids:                # 查找对应的子任务详情                subtask = next((t for t in subtasks if t["id"] == task_id), None)                if subtask:                    # 根据不同智能体执行不同任务                    if agent == "data_agent" and subtask["type"] in ["data_collection", "data_processing"]:                        execution_tasks.append(self.data_agent_client.invoke(                            "data_analysis", "process_data",                            {"data_source": "database", "analysis_type": "descriptive"}                        ))
        # 等待所有子任务完成        results = await asyncio.gather(*execution_tasks)
        # 4. 整合结果        final_result = {            "original_task": task_description,            "subtask_results": results,            "completion_status": "success",            "timestamp": datetime.datetime.now().isoformat()        }
        return final_result
# 启动多智能体系统示例if __name__ == "__main__":    # 在实际应用中，这些服务器会在不同的进程或机器上运行    import threading
    def run_coordinator():        coordinator_mcp.run(host="0.0.0.0", port=8010)
    def run_data_agent():        data_agent_mcp.run(host="0.0.0.0", port=8011)
    # 启动服务器    threading.Thread(target=run_coordinator).start()    threading.Thread(target=run_data_agent).start()
    # 等待服务器启动    time.sleep(2)
    # 创建协作系统并执行任务    collaboration_system = AgentCollaborationSystem()    asyncio.run(collaboration_system.execute_complex_task("分析过去三个月的销售数据并生成报告"))