"""
MCP客户端使用示例
展示大模型如何自主调用工具完成数据可视化任务
"""import asyncio
import json
from typing import List, Dict, Anyclass MCPClient:"""MCP客户端模拟器，展示大模型的自主调用流程"""def __init__(self, server):self.server = serverasync def autonomous_data_analysis(self, user_request: str) -> Dict[str, Any]:"""模拟大模型自主完成数据分析任务的完整流程这就是Tools方式的强大之处 - 大模型可以自主决策和执行"""print(f"🤖 用户请求: {user_request}")print("🚀 大模型开始自主分析...")results = {"user_request": user_request,"workflow_steps": [],"final_output": None}try:print("📊 步骤1: 获取可用数据集列表...")datasets_response = await self.server._list_datasets()datasets_info = json.loads(datasets_response[0].text)results["workflow_steps"].append({"step": 1,"action": "list_available_datasets","result": "获取到数据集列表","data": datasets_info})print("🎯 步骤2: 智能选择数据集...")if "销售" in user_request or "sales" in user_request.lower():selected_dataset = "sales"elif "用户" in user_request or "user" in user_request.lower():selected_dataset = "users"elif "股票" in user_request or "stock" in user_request.lower():selected_dataset = "stocks"else:selected_dataset = list(datasets_info["available_datasets"])[0]print(f"✅ 选择数据集: {selected_dataset}")print("📋 步骤3: 获取数据内容...")data_response = await self.server._get_dataset(selected_dataset, limit=10)data_info = json.loads(data_response[0].text)results["workflow_steps"].append({"step": 3,"action": "get_dataset","result": f"获取 {selected_dataset} 数据集","data": data_info})print("🔍 步骤4: 分析数据结构...")structure_response = await self.server._analyze_structure(selected_dataset)structure_info = json.loads(structure_response[0].text)results["workflow_steps"].append({"step": 4,"action": "analyze_data_structure","result": "分析完成数据结构","data": structure_info})print("🎯 步骤5: 确定分析目标...")if "趋势" in user_request or "trend" in user_request.lower():analysis_goal = "trend"elif "分布" in user_request or "distribution" in user_request.lower():analysis_goal = "distribution"elif "比较" in user_request or "comparison" in user_request.lower():analysis_goal = "comparison"elif "相关" in user_request or "correlation" in user_request.lower():analysis_goal = "correlation"else:if structure_info["column_types"]["datetime"]:analysis_goal = "trend"elif len(structure_info["column_types"]["categorical"]) > 0:analysis_goal = "comparison"else:analysis_goal = "distribution"print(f"✅ 分析目标: {analysis_goal}")print("💡 步骤6: 获取可视化建议...")suggestions_response = await self.server._suggest_visualization(selected_dataset, analysis_goal)suggestions = json.loads(suggestions_response[0].text)results["workflow_steps"].append({"step": 6,"action": "suggest_visualization","result": "获取可视化建议","data": suggestions})print("📈 步骤7: 创建图表...")chart_type = suggestions["recommended_charts"][0] if suggestions["recommended_charts"] else "bar"columns = structure_info["basic_info"]["columns"]datetime_cols = structure_info["column_types"]["datetime"]numeric_cols = structure_info["column_types"]["numeric"]categorical_cols = structure_info["column_types"]["categorical"]x_column = Noney_column = Noneif chart_type == "line" and datetime_cols and numeric_cols:x_column = datetime_cols[0]y_column = numeric_cols[0]elif chart_type == "bar" and categorical_cols and numeric_cols:x_column = categorical_cols[0]y_column = numeric_cols[0]elif chart_type == "scatter" and len(numeric_cols) >= 2:x_column = numeric_cols[0]y_column = numeric_cols[1]elif chart_type == "pie" and categorical_cols:x_column = categorical_cols[0]chart_response = await self.server._create_chart(dataset_name=selected_dataset,chart_type=chart_type,x_column=x_column,y_column=y_column,title=f"{selected_dataset.title()} {analysis_goal.title()} Analysis")results["workflow_steps"].append({"step": 7,"action": "create_chart","result": f"创建 {chart_type} 图表","chart_created": True})print("🔬 步骤8: 生成数据洞察...")insights_response = await self.server._get_insights(selected_dataset)insights = json.loads(insights_response[0].text)results["workflow_steps"].append({"step": 8,"action": "get_data_insights","result": "生成数据洞察","data": insights})final_report = self._generate_final_report(selected_dataset, analysis_goal, chart_type, insights)results["final_output"] = final_reportprint("✅ 自主分析完成!")return resultsexcept Exception as e:print(f"❌ 自主分析过程中出错: {str(e)}")results["error"] = str(e)return resultsdef _generate_final_report(self, dataset: str, goal: str, chart_type: str, insights: Dict) -> Dict[str, Any]:"""生成最终分析报告"""report = {"executive_summary": f"基于 {dataset} 数据集完成了 {goal} 分析，使用 {chart_type} 图表进行可视化。","key_findings": insights.get("key_findings", []),"recommendations": insights.get("recommendations", []),"methodology": ["自动识别最适合的数据集","智能分析数据结构和特征","基于分析目标选择最佳可视化方式","生成数据驱动的洞察和建议"],"next_steps": ["可以尝试不同的可视化角度","深入分析特定的数据子集","结合业务场景进行进一步解读"]}return report
async def demonstrate_tools_vs_resources():"""演示Tools方式 vs Resources方式的区别"""print("=" * 60)print("🚀 MCP Tools方式 vs Resources方式对比演示")print("=" * 60)from __main__ import DataVisualizationServerserver = DataVisualizationServer()client = MCPClient(server)print("\n" + "=" * 40)print("❌ Resources方式 (需要人工干预)")print("=" * 40)print("用户: 帮我用销售数据画个图")print("助手: 请先从资源列表中选择 'sales' 数据资源...")print("用户: [手动选择sales资源]")print("助手: 请选择'分析数据'提示词...")print("用户: [手动选择提示词]")print("助手: 请选择图表类型...")print("用户: [手动选择图表类型]")print("🔄 需要多次人工交互才能完成任务")print("\n" + "=" * 40)print("✅ Tools方式 (大模型自主完成)")print("=" * 40)user_requests = ["帮我用销售数据画个趋势图","分析用户数据的分布情况", "对比股票数据的相关性"]for i, request in enumerate(user_requests, 1):print(f"\n--- 示例 {i} ---")result = await client.autonomous_data_analysis(request)print(f"\n📊 完成的工作流程:")for step in result["workflow_steps"]:print(f" {step['step']}. {step['action']}: {step['result']}")if result.get("final_output"):print(f"\n📋 最终报告:")print(f" 概述: {result['final_output']['executive_summary']}")print(f" 关键发现: {len(result['final_output']['key_findings'])} 项")print(f" 建议: {len(result['final_output']['recommendations'])} 项")print("\n" + "=" * 60)print("🎯 核心优势总结")print("=" * 60)advantages = {"自主性": "大模型可以自己决定使用哪些工具，无需人工干预","智能化": "根据用户需求和数据特征自动选择最佳方案","效率": "一次请求完成整个工作流程，节省时间","一致性": "每次都按照最佳实践执行分析流程","可扩展": "轻松添加新的分析功能，大模型自动学会使用"}for key, value in advantages.items():print(f"✅ {key}: {value}")
class ToolDesignBestPractices:"""工具设计最佳实践指南"""@staticmethoddef demonstrate_good_vs_bad_design():"""演示好的和坏的工具设计"""print("\n" + "=" * 60)print("📚 MCP工具设计最佳实践")print("=" * 60)print("\n❌ 错误设计 - 使用Resources (大模型无法自主调用)")print("-" * 50)bad_design = '''# 错误方式：Resource (需要人工选择)server.setRequestHandler(ReadResourceRequestSchema, async (request) => {if (request.params.uri === "data://sales") {return { contents: [{ uri: request.params.uri, text: "销售数据..." }] };}});# 问题：# 1. 大模型无法自主调用# 2. 需要用户手动选择资源# 3. 工作流程被打断'''print(bad_design)print("\n✅ 正确设计 - 使用Tools (大模型可自主调用)")print("-" * 50)good_design = '''# 正确方式：Tool (模型可自主调用)server.setRequestHandler(CallToolRequestSchema, async (request) => {if (request.params.name === "get_sales_data") {return {content: [{type: "text",text: "销售数据..."}]};}});# 优势：# 1. 大模型可以自主决定何时调用# 2. 支持参数传递和条件调用# 3. 完整的自动化工作流程'''print(good_design)print("\n🎯 设计原则")print("-" * 30)principles = ["原则1: 所有需要大模型自主使用的功能都必须设计为Tools","原则2: Tools应该是原子性的，每个工具完成一个明确的任务","原则3: 提供清晰的参数描述，让大模型知道如何使用","原则4: 返回结构化数据，便于大模型进一步处理","原则5: 包含错误处理，提供有用的错误信息"]for principle in principles:print(f" ✓ {principle}")print("\n🔧 实际应用场景")print("-" * 30)scenarios = {"数据分析": "自动获取数据 → 分析结构 → 选择可视化 → 生成图表","文档处理": "自动读取文档 → 提取关键信息 → 总结内容 → 生成报告", "代码分析": "自动扫描代码 → 识别问题 → 提供建议 → 生成修复方案","业务流程": "自动获取状态 → 执行检查 → 处理异常 → 更新结果"}for scenario, workflow in scenarios.items():print(f" 📋 {scenario}: {workflow}")
async def complete_demo():"""完整的使用演示"""print("🚀 开始完整演示...")await demonstrate_tools_vs_resources()ToolDesignBestPractices.demonstrate_good_vs_bad_design()print("\n" + "=" * 60)print("✅ 演示完成! 总结:")print("=" * 60)summary = ["1. Tools方式让大模型拥有真正的自主性","2. 用户只需要提出需求，大模型自动完成整个工作流程","3. 相比Resources和Prompts，Tools提供了更好的用户体验","4. 正确的工具设计是实现智能自动化的关键","5. 未来的AI应用应该更多地采用Tools模式"]for point in summary:print(f" ✓ {point}")print(f"\n🎯 记住核心原则: 想让大模型自己调用，必须写成Tools!")
if __name__ == "__main__":asyncio.run(complete_demo())
