创建使用MCP服务器的AI Agent（第4部分）

作者头像东城·2026-01-31 23:29:24·
现在到了神奇的部分！在第3部分中，我们构建了一个可以收集和分析反馈的MCP服务器。今天，我们将创建一个使用这个服务器的AI Agent来智能处理客户交互。到最后，您将拥有一个可工作的AI系统！
我们要构建什么
我们将创建一个AI Agent，它：
连接到我们的MCP服务器
使用LLM理解客户需求
自动收集和分析反馈
生成智能报告
建议业务改进
第1步：设置LLM集成
首先，让我们安装所需的内容：
  1. # 确保您的虚拟环境已激活
  2. pip install openai python-dotenv aiohttp
bash
创建一个.env文件来存储您的API密钥：
  1. # .env文件
  2. OPENAI_API_KEY=your-api-key-here
  3. # 从以下地址获取您的密钥：https://platform.openai.com/api-keys
第2步：构建我们的AI Agent
创建feedback_agent.py：
  1. #!/usr/bin/env python3
  2. """
  3. AI Agent for Customer Feedback System
  4. Intelligently handles feedback collection and analysis
  5. """

  7. import os
  8. import json
  9. import asyncio
  10. from typing import Dict, Any, List
  11. from datetime import datetime
  12. from dotenv import load_dotenv

  14. # Load environment variables
  15. load_dotenv()

  17. # For LLM integration
  18. from openai import AsyncOpenAI

  20. # For MCP client
  21. from mcp import ClientSession, StdioServerParameters
  22. from mcp.client.stdio import stdio_client

  24. class FeedbackAgent:
  25. def __init__(self):
  26. self.name = "Feedback Assistant"
  27. self.llm = AsyncOpenAI(api_key=os.getenv("OPENAI_API_KEY"))
  28. self.mcp_session = None
  29. self.server_params = StdioServerParameters(
  30. command="python",
  31. args=["feedback_server.py"]
  32. )

  34. async def connect_to_server(self):
  35. """Connect to our MCP server"""
  36. print(f"🔌 Connecting to feedback server...")

  38. async with stdio_client(self.server_params) as (read, write):
  39. async with ClientSession(read, write) as session:
  40. self.mcp_session = session

  42. # Initialize connection
  43. await session.initialize()

  45. # Get available tools and resources
  46. tools = await session.list_tools()
  47. resources = await session.list_resources()

  49. print(f"✅ Connected! Found {len(tools)} tools and {len(resources)} resources")

  51. # Keep the session active
  52. await self.run_agent_loop()

  54. async def think(self, task: str) -> str:
  55. """Use LLM to understand and plan actions"""
  56. response = await self.llm.chat.completions.create(
  57. model="gpt-3.5-turbo",
  58. messages=[
  59. {
  60. "role": "system",
  61. "content": f"""You are {self.name}, an AI assistant for a café.
  62. You help collect and analyze customer feedback.
  63. Be friendly, professional, and insightful."""
  64. },
  65. {
  66. "role": "user",
  67. "content": task
  68. }
  69. ],
  70. temperature=0.7
  71. )

  73. return response.choices[0].message.content

  75. async def collect_customer_feedback(self, conversation: str) -> Dict[str, Any]:
  76. """Intelligently extract feedback from conversation"""

  78. # Use LLM to extract information
  79. extraction_prompt = f"""
  80. Extract the following from this customer conversation:
  81. 1. Customer name (if mentioned)
  82. 2. The main feedback points
  83. 3. Overall rating (1-5)
  84. 4. Key topics mentioned

  86. Conversation:
  87. {conversation}

  89. Return as JSON format.
  90. """

  92. response = await self.llm.chat.completions.create(
  93. model="gpt-3.5-turbo",
  94. messages=[
  95. {"role": "system", "content": "Extract information and return valid JSON only."},
  96. {"role": "user", "content": extraction_prompt}
  97. ],
  98. temperature=0.3,
  99. response_format={"type": "json_object"}
  100. )

  102. extracted = json.loads(response.choices[0].message.content)

  104. # Use MCP tool to collect feedback
  105. result = await self.mcp_session.call_tool(
  106. "collect_feedback",
  107. arguments={
  108. "customer_name": extracted.get("customer_name", "Anonymous"),
  109. "feedback": extracted.get("feedback", conversation),
  110. "rating": extracted.get("rating", 3)
  111. }
  112. )

  114. return {
  115. "status": "collected",
  116. "details": extracted,
  117. "server_response": result
  118. }

  120. async def analyze_feedback_trends(self) -> str:
  121. """Analyze all feedback and generate insights"""

  123. # Get recent feedback from MCP server
  124. recent_feedback = await self.mcp_session.read_resource("feedback://recent")
  125. summary_data = await self.mcp_session.read_resource("feedback://summary")

  127. # Use LLM to generate insights
  128. analysis_prompt = f"""
  129. Analyze this customer feedback data and provide:
  130. 1. Key themes and patterns
  131. 2. Areas needing immediate attention
  132. 3. Positive aspects to maintain
  133. 4. Specific recommendations for improvement

  135. Recent Feedback:
  136. {recent_feedback}

  138. Summary Statistics:
  139. {summary_data}
  140. """

  142. response = await self.llm.chat.completions.create(
  143. model="gpt-3.5-turbo",
  144. messages=[
  145. {
  146. "role": "system",
  147. "content": "You are a business analyst. Provide actionable insights."
  148. },
  149. {"role": "user", "content": analysis_prompt}
  150. ],
  151. temperature=0.5
  152. )

  154. return response.choices[0].message.content

  156. async def handle_customer_interaction(self, message: str) -> str:
  157. """Main interaction handler"""

  159. # Determine intent
  160. intent_prompt = f"""
  161. Classify this message intent:
  162. - 'give_feedback': Customer wants to share feedback
  163. - 'check_status': Customer asking about their previous feedback
  164. - 'general_question': Other questions

  166. Message: {message}

  168. Return only the intent classification.
  169. """

  171. intent_response = await self.llm.chat.completions.create(
  172. model="gpt-3.5-turbo",
  173. messages=[
  174. {"role": "system", "content": "Classify intent. Return only the classification."},
  175. {"role": "user", "content": intent_prompt}
  176. ],
  177. temperature=0.1
  178. )

  180. intent = intent_response.choices[0].message.content.strip().lower()

  182. if "give_feedback" in intent:
  183. # Collect feedback
  184. feedback_result = await self.collect_customer_feedback(message)

  186. # Generate friendly response
  187. response = await self.think(
  188. f"Customer gave feedback. Details: {feedback_result}. "
  189. "Thank them and mention any immediate actions we'll take."
  190. )

  192. return response

  194. elif "check_status" in intent:
  195. # Get feedback summary
  196. summary = await self.mcp_session.read_resource("feedback://summary")

  198. response = await self.think(
  199. f"Customer asking about feedback status. Our summary: {summary}. "
  200. "Provide a helpful update."
  201. )

  203. return response

  205. else:
  206. # General response
  207. return await self.think(f"Respond helpfully to: {message}")

  209. async def run_agent_loop(self):
  210. """Main agent interaction loop"""
  211. print(f"\n🤖 {self.name} is ready! Type 'quit' to exit.")
  212. print("Try: 'Hi, I'm Sarah. I loved the new latte recipe! 5 stars!'")

  214. while True:
  215. try:
  216. # Get user input
  217. user_input = input("\n👤 You: ").strip()

  219. if user_input.lower() in ['quit', 'exit', 'bye']:
  220. print(f"👋 {self.name}: Goodbye! Have a great day!")
  221. break

  223. if user_input.lower() == 'analyze':
  224. # Run analysis
  225. print(f"\n📊 {self.name}: Analyzing feedback trends...")
  226. analysis = await self.analyze_feedback_trends()
  227. print(f"📈 Analysis:\n{analysis}")
  228. continue

  230. # Handle regular interaction
  231. print(f"\n🤖 {self.name}: Processing...")
  232. response = await self.handle_customer_interaction(user_input)
  233. print(f"🤖 {self.name}: {response}")

  235. except KeyboardInterrupt:
  236. print(f"\n👋 {self.name}: Goodbye!")
  237. break
  238. except Exception as e:
  239. print(f"❌ Error: {e}")

  241. # Main entry point
  242. async def main():
  243. agent = FeedbackAgent()
  244. await agent.connect_to_server()

  246. if __name__ == "__main__":
  247. asyncio.run(main())
python
第3步：添加工作流引擎
创建workflow_engine.py来管理自动化任务：
  1. #!/usr/bin/env python3
  2. """
  3. Workflow Engine for Feedback System
  4. Manages automated tasks and scheduled operations
  5. """

  7. import asyncio
  8. from typing import Dict, Any
  9. from datetime import datetime, timedelta

  11. class FeedbackWorkflow:
  12. def __init__(self, agent):
  13. self.agent = agent
  14. self.workflows = {
  15. "daily_analysis": self.daily_analysis_workflow,
  16. "negative_feedback_alert": self.negative_feedback_alert,
  17. "weekly_report": self.weekly_report_workflow
  18. }

  20. async def daily_analysis_workflow(self):
  21. """Run daily feedback analysis"""
  22. print(f"\n📊 Running daily analysis...")

  24. # Get current feedback summary
  25. summary = await self.agent.mcp_session.read_resource("feedback://summary")
  27. # Analyze trends
  28. analysis = await self.agent.analyze_feedback_trends()
  30. print(f"📈 Daily Analysis Complete:\n{analysis}")
  32. # Check for urgent issues
  33. if "negative" in summary.lower() and "50" in summary:
  34. print("⚠️ High negative feedback detected!")
  35. await self.negative_feedback_alert()

  37. async def negative_feedback_alert(self, feedback_data: Dict[str, Any] = None):
  38. """Handle negative feedback alerts"""
  39. print(f"\n🚨 Negative feedback alert!")

  41. if feedback_data:
  42. # Create action plan
  43. response_plan = await self.agent.think(
  44. f"Create action plan for this negative feedback: {feedback_data['feedback']}. "
  45. "Include: immediate response, follow-up actions, and prevention measures."
  46. )

  48. print(f"📋 Action Plan:\n{response_plan}")

  50. # In production:
  51. # - Send alert to manager
  52. # - Create follow-up task
  53. # - Track resolution

  55. return response_plan

  57. async def weekly_report_workflow(self):
  58. """Generate comprehensive weekly report"""
  59. print(f"\n📊 Generating weekly report...")

  61. # Get all data
  62. summary = await self.agent.mcp_session.read_resource("feedback://summary")
  63. recent = await self.agent.mcp_session.read_resource("feedback://recent")

  65. # Generate executive summary
  66. report = await self.agent.think(
  67. f"Create executive summary for this week's feedback. "
  68. f"Data: {summary}\nRecent examples: {recent}\n"
  69. "Include: key metrics, trends, recommendations, and success stories."
  70. )

  72. print(f"📄 Weekly Report:\n{report}")
  73. return report

  75. async def run_workflow(self, workflow_name: str, *args, **kwargs):
  76. """Execute a specific workflow"""
  77. if workflow_name in self.workflows:
  78. return await self.workflows[workflow_name](*args, **kwargs)
  79. else:
  80. raise ValueError(f"Unknown workflow: {workflow_name}")

  82. class WorkflowScheduler:
  83. def __init__(self, workflow_engine):
  84. self.engine = workflow_engine
  85. self.scheduled_tasks = []

  87. def schedule_daily(self, hour: int, minute: int, workflow: str):
  88. """Schedule daily workflow"""
  89. self.scheduled_tasks.append({
  90. "type": "daily",
  91. "time": {"hour": hour, "minute": minute},
  92. "workflow": workflow
  93. })

  95. def schedule_on_event(self, event_type: str, workflow: str):
  96. """Schedule event-triggered workflow"""
  97. self.scheduled_tasks.append({
  98. "type": "event",
  99. "trigger": event_type,
  100. "workflow": workflow
  101. })

  103. async def run(self):
  104. """Run the scheduler"""
  105. print("⏰ Workflow scheduler started")

  107. # For demo: simulate some workflows
  108. await asyncio.sleep(2)
  109. await self.engine.run_workflow("daily_analysis")

  111. # Simulate negative feedback
  112. await asyncio.sleep(3)
  113. await self.engine.run_workflow("negative_feedback_alert", {
  114. "customer_name": "Demo User",
  115. "feedback": "The coffee was cold and service was slow",
  116. "rating": 2
  117. })
python
第4步：整合所有组件
创建run_system.py来运行完整系统：
  1. #!/usr/bin/env python3
  2. """
  3. Run the Complete AI Feedback System
  4. """

  6. import asyncio
  7. import subprocess
  8. import time
  9. from feedback_agent import FeedbackAgent
  10. from workflow_engine import FeedbackWorkflow, WorkflowScheduler

  12. async def run_complete_system():
  13. """Run server and agent together"""

  15. # Start the MCP server in background
  16. print("🚀 Starting MCP Feedback Server...")
  17. server_process = subprocess.Popen(
  18. ["python", "feedback_server.py"],
  19. stdout=subprocess.PIPE,
  20. stderr=subprocess.PIPE
  21. )

  23. # Give server time to start
  24. time.sleep(2)

  26. try:
  27. # Create and connect agent
  28. print("🤖 Starting AI Agent...")
  29. agent = FeedbackAgent()

  31. # Create workflow engine
  32. workflow = FeedbackWorkflow(agent)
  33. scheduler = WorkflowScheduler(workflow)

  35. # Schedule some workflows
  36. scheduler.schedule_daily(9, 0, "daily_analysis")
  37. scheduler.schedule_on_event("negative_feedback", "negative_feedback_alert")
  38. scheduler.schedule_daily(17, 0, "weekly_report")

  40. # Run the agent
  41. await agent.connect_to_server()

  43. finally:
  44. # Clean up
  45. server_process.terminate()
  46. print("\n👋 System shutdown complete")

  48. if __name__ == "__main__":
  49. print("🎯 Complete AI Feedback System Starting...\n")
  50. asyncio.run(run_complete_system())
python
第5步：测试您的AI系统
让我们测试完整系统：
  1. # 确保您的.env文件包含您的OpenAI API密钥
  2. python run_system.py
bash
尝试这些交互：
提供反馈：
"Hi, I'm Sarah. I loved the new latte recipe! 5 stars!"
"The service was terrible today. My order took 30 minutes."
检查分析：
输入：analyze
提问：
"What's the overall feedback trend?"
"How many people gave feedback today?"
我们实现的关键概念
1. Agent-服务器通信
Agent连接到MCP服务器并可以：
读取资源（反馈数据）
调用工具（收集反馈、分析情感）
智能处理结果
2. LLM集成
我们使用GPT-3.5来：
理解自然语言
从对话中提取信息
生成洞察和建议
创建个性化响应
3. 自动化工作流
我们的系统可以：
运行定时分析
响应事件（负面反馈）
自动生成报告
4. 智能决策制定
Agent：
分类用户意图
选择适当的行动
提供上下文响应
下一步是什么？
在我们的最后一部分（第5部分）中，我们将：
使用真实数据库添加数据持久性
实现高级功能（多语言支持、语音输入）
为生产使用部署系统
探索扩展策略
您现在已经构建了一个功能性的AI系统！您将如何扩展它以满足您的业务需求？哪些其他工作流会有帮助？
挑战：扩展您的系统
尝试添加这些功能：
负面反馈的电子邮件通知
情感趋势可视化
客户响应模板
多位置支持
在评论中分享您的扩展！
技术要点总结
AI Agent的核心功能
智能交互：使用LLM理解用户意图
自动反馈收集：从对话中提取结构化信息
趋势分析：生成业务洞察和建议
工作流自动化：定时任务和事件触发
系统架构特点
模块化设计：Agent、服务器、工作流引擎分离
异步处理：支持并发操作和实时响应
智能决策：基于LLM的意图识别和响应生成
可扩展性：易于添加新功能和工作流
实际应用价值
客户服务自动化：24/7智能客服支持
反馈管理：自动收集、分析和响应客户反馈
业务洞察：实时趋势分析和改进建议
工作流优化：自动化重复任务和报告生成
开发最佳实践
环境配置：使用.env文件管理API密钥
错误处理：完善的异常处理机制
日志记录：详细的系统运行日志
测试验证：完整的系统集成测试
通过这个教程，您已经掌握了构建完整AI系统的核心技能，可以开始创建自己的智能应用了！