在当今快速发展的AI领域，将大型语言模型(LLM)与实际应用场景相结合已成为提升生产力的关键。然而，LLM本身存在局限性——它们无法直接与现实世界交互、操作应用程序或执行复杂的工作流。这就是为什么我们需要像Playwright MCP这样的工具来弥合这一差距。

本文将深入探讨如何利用Playwright MCP（Model Context Protocol）与LLM协同工作，构建能够处理复杂任务的工作流和智能AI代理。

什么是Playwright MCP？
Playwright MCP是一个基于Model Context Protocol的桥接工具，它将强大的浏览器自动化框架Playwright与LLM连接起来。MCP协议允许LLM访问外部工具和资源，而Playwright则提供了跨浏览器的自动化能力。

核心组件
Playwright: Microsoft开发的跨浏览器自动化工具，支持Chromium、Firefox和WebKit
MCP Server: 处理LLM与Playwright之间的通信
LLM接口: 提供自然语言理解和任务规划能力
环境设置与安装
prerequisites
Node.js 16+
Python 3.8+
访问LLM API（如OpenAI GPT、Claude等）
安装步骤

安装Playwright

npm install playwright
npx playwright install

安装MCP相关依赖

pip install mcp-client playwright-async

克隆Playwright MCP仓库

git clone https://github.com/your-repo/playwright-mcp.git
cd playwright-mcp
基础配置

config.py

import os
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client

class PlaywrightMCPConfig:
def init(self):
self.browser_type = "chromium"# chromium, firefox, webkit
self.headless = False
self.timeout = 30000
self.llm_api_key = os.getenv("LLM_API_KEY")

def get_server_params(self): return StdioServerParameters( command="node", args=["path/to/playwright-mcp-server.js"] ) 

构建基础工作流

1. 初始化连接
   import asyncio
   from mcp.client.stdio import stdio_client
   from mcp import ClientSession
   from config import PlaywrightMCPConfig

class PlaywrightMCPClient:
def init(self, config: PlaywrightMCPConfig):
self.config = config
self.session = None

asyncdef connect(self): server_params = self.config.get_server_params() asyncwith stdio_client(server_params) as (read, write): asyncwith ClientSession(read, write) as session: self.session = session # 初始化会话 await session.initialize() return self 

1. 基本网页操作
   class WebAutomationWorkflow:
   def init(self, mcp_client):

    self.client = mcp_client 

   asyncdef navigate_to_page(self, url: str):

    """导航到指定页面""" result = await self.client.session.call_tool( "navigate", {"url": url} ) return result 

   asyncdef fill_form(self, selector: str, value: str):

    """填写表单""" result = await self.client.session.call_tool( "fill", {"selector": selector, "value": value} ) return result 

   asyncdef click_element(self, selector: str):

    """点击元素""" result = await self.client.session.call_tool( "click", {"selector": selector} ) return result 

   asyncdef extract_text(self, selector: str):

    """提取文本内容""" result = await self.client.session.call_tool( "get_text", {"selector": selector} ) return result 

   集成LLM创建智能工作流

2. LLM任务规划器
   import openai
   from typing import List, Dict, Any

class LLMTaskPlanner:
def init(self, api_key: str):
self.client = openai.OpenAI(api_key=api_key)

def plan_workflow(self, user_request: str) -> List[Dict[str, Any]]: """使用LLM解析用户请求并生成工作流步骤""" prompt = f""" 根据以下用户请求，生成一个详细的Playwright自动化工作流。 用户请求: {user_request} 请以JSON格式返回步骤列表，每个步骤包含: - action: 操作类型 (navigate, click, fill, extract, wait, etc.) - parameters: 操作参数 - description: 步骤描述 只返回JSON格式的结果。 """ response = self.client.chat.completions.create( model="gpt-4", messages=[{"role": "user", "content": prompt}], temperature=0.1 ) return self._parse_response(response.choices[0].message.content) def _parse_response(self, response: str) -> List[Dict[str, Any]]: """解析LLM响应为结构化工作流""" import json try: # 清理响应并提取JSON cleaned_response = response.strip() if"```json"in cleaned_response: cleaned_response = cleaned_response.split("```json")[1].split("```")[0] elif"```"in cleaned_response: cleaned_response = cleaned_response.split("```")[1] return json.loads(cleaned_response) except Exception as e: print(f"解析LLM响应失败: {e}") return [] 

1. 智能工作流执行器
   class IntelligentWorkflowExecutor:
   def init(self, mcp_client, llm_planner):

    self.mcp_client = mcp_client self.planner = llm_planner self.automation = WebAutomationWorkflow(mcp_client) 

   asyncdef execute_user_request(self, user_request: str):

    """执行用户自然语言请求的完整工作流""" print(f"处理用户请求: {user_request}") # 1. 使用LLM规划工作流 workflow_steps = self.planner.plan_workflow(user_request) print(f"生成的工作流步骤: {len(workflow_steps)}步") # 2. 执行工作流 results = [] for i, step in enumerate(workflow_steps, 1): print(f"执行步骤 {i}: {step['description']}") try: result = await self._execute_step(step) results.append({ "step": i, "description": step["description"], "result": result, "status": "success" }) except Exception as e: results.append({ "step": i, "description": step["description"], "error": str(e), "status": "failed" }) print(f"步骤 {i} 执行失败: {e}") break return results 

   asyncdef _execute_step(self, step: Dict[str, Any]):

    """执行单个工作流步骤""" action = step["action"] params = step["parameters"] if action == "navigate": returnawait self.automation.navigate_to_page(params["url"]) elif action == "click": returnawait self.automation.click_element(params["selector"]) elif action == "fill": returnawait self.automation.fill_form(params["selector"], params["value"]) elif action == "extract": returnawait self.automation.extract_text(params["selector"]) elif action == "wait": await asyncio.sleep(params.get("seconds", 2)) return"等待完成" else: raise ValueError(f"未知操作: {action}") 

   高级应用：构建AI智能体

2. 自适应智能体
   class AdaptiveAIAgent:
   def init(self, mcp_client, llm_planner, executor):

    self.mcp_client = mcp_client self.planner = llm_planner self.executor = executor self.conversation_history = [] 

   asyncdef process_request(self, user_input: str, context: Dict = None):

    """处理用户输入并执行相应操作""" # 添加上下文到对话历史 self.conversation_history.append({"role": "user", "content": user_input}) # 分析用户意图 intent = await self._analyze_intent(user_input, context) if intent["type"] == "automation": # 执行自动化工作流 results = await self.executor.execute_user_request(user_input) # 生成自然语言总结 summary = await self._generate_summary(user_input, results) self.conversation_history.append({ "role": "assistant", "content": summary }) return { "type": "automation", "results": results, "summary": summary } elif intent["type"] == "query": # 处理查询请求 response = await self._handle_query(user_input) return { "type": "query", "response": response } 

   asyncdef _analyze_intent(self, user_input: str, context: Dict) -> Dict:

    """使用LLM分析用户意图""" # 简化的意图分析实现 automation_keywords = ["打开", "点击", "填写", "导航", "提取", "自动化"] if any(keyword in user_input for keyword in automation_keywords): return {"type": "automation", "confidence": 0.9} else: return {"type": "query", "confidence": 0.7} 

   asyncdef _generate_summary(self, request: str, results: List) -> str:

    """生成工作流执行总结""" success_steps = [r for r in results if r["status"] == "success"] returnf""" 已完成您的要求: {request} 执行统计: - 总步骤数: {len(results)} - 成功步骤: {len(success_steps)} - 失败步骤: {len(results) - len(success_steps)} {'所有步骤均成功完成!' if len(success_steps) == len(results) else '部分步骤执行失败，请检查错误信息。'} """ 

3. 复杂工作流示例：电商数据采集
   class EcommerceDataAgent:
   def init(self, base_agent):

    self.agent = base_agent 

   asyncdef collect_product_data(self, product_url: str, data_points: List[str]):

    """采集电商产品数据""" workflow_request = f""" 请执行以下电商数据采集任务: 1. 导航到产品页面: {product_url} 2. 提取产品标题 3. 提取产品价格 4. 提取产品评分 5. 提取产品描述 6. 提取客户评论数量 """ # 执行数据采集 results = await self.agent.process_request(workflow_request) # 数据清洗和结构化 structured_data = await self._structure_product_data(results) return structured_data 

   asyncdef _structure_product_data(self, raw_results: Dict) -> Dict:

    """将采集的数据结构化""" # 实现数据解析和结构化逻辑 structured = {} for result in raw_results.get("results", []): if"result"in result and result["result"]: # 解析提取的数据 text_content = result["result"].get("content", "") # 根据步骤描述识别数据类型 if"标题"in result["description"]: structured["title"] = self._clean_text(text_content) elif"价格"in result["description"]: structured["price"] = self._extract_price(text_content) elif"评分"in result["description"]: structured["rating"] = self._extract_rating(text_content) return structured 

   def _clean_text(self, text: str) -> str:

    """清理文本数据""" return text.strip() if text else"" 

   def _extract_price(self, text: str) -> float:

    """提取价格信息""" import re matches = re.findall(r'[\d.,]+', text) return float(matches[0].replace(',', '')) if matches else0.0 

   错误处理与优化

4. 鲁棒性增强
   class RobustWorkflowExecutor(IntelligentWorkflowExecutor):
   asyncdef execute_with_retry(self, user_request: str, max_retries: int = 3):

    """带重试机制的工作流执行""" for attempt in range(max_retries): try: results = await self.execute_user_request(user_request) # 检查是否有失败步骤 failed_steps = [r for r in results if r["status"] == "failed"] ifnot failed_steps: return results print(f"第 {attempt + 1} 次尝试，{len(failed_steps)} 个步骤失败") # 最后一次尝试仍然失败，抛出异常 if attempt == max_retries - 1: raise Exception(f"工作流执行失败，{len(failed_steps)} 个步骤未完成") except Exception as e: print(f"第 {attempt + 1} 次尝试失败: {e}") if attempt == max_retries - 1: raise await asyncio.sleep(2) # 重试前等待 return [] 

   asyncdef _execute_step_with_fallback(self, step: Dict):

    """带备用方案的步骤执行""" try: returnawait self._execute_step(step) except Exception as e: print(f"步骤执行失败: {e}，尝试备用方案") # 实现备用执行逻辑 if step["action"] == "click": # 尝试不同的选择器 returnawait self._try_alternative_selectors(step) elif step["action"] == "extract": # 尝试不同的数据提取方法 returnawait self._try_alternative_extraction(step) else: raise 

5. 性能监控
   import time
   from dataclasses import dataclass
   from typing import List

@dataclass
class PerformanceMetrics:
total_steps: int
successful_steps: int
failed_steps: int
total_time: float
average_step_time: float

class PerformanceMonitor:
def init(self):
self.metrics_history: List[PerformanceMetrics] = []

def start_execution(self): self.start_time = time.time() self.step_times = [] def record_step(self, success: bool, step_time: float): self.step_times.append(step_time) def end_execution(self, total_steps: int, successful_steps: int): total_time = time.time() - self.start_time avg_time = sum(self.step_times) / len(self.step_times) if self.step_times else0 metrics = PerformanceMetrics( total_steps=total_steps, successful_steps=successful_steps, failed_steps=total_steps - successful_steps, total_time=total_time, average_step_time=avg_time ) self.metrics_history.append(metrics) return metrics 

实际应用场景

1. 自动化测试智能体
   class TestingAutomationAgent:
   def init(self, base_agent):

    self.agent = base_agent 

   async def run_e2e_test(self, test_scenario: str):

    """执行端到端测试""" test_request = f""" 执行以下端到端测试场景: {test_scenario} 包括: 1. 导航到测试页面 2. 执行测试步骤 3. 验证预期结果 4. 生成测试报告 """ return await self.agent.process_request(test_request) 

2. 数据监控智能体
   class MonitoringAgent:
   def init(self, base_agent, alert_thresholds: Dict):

    self.agent = base_agent self.thresholds = alert_thresholds 

   asyncdef monitor_website(self, url: str, check_interval: int = 3600):

    """定期监控网站状态""" whileTrue: try: status = await self._check_website_status(url) ifnot status["is_healthy"]: await self._send_alert(f"网站异常: {status['issues']}") except Exception as e: await self._send_alert(f"监控检查失败: {e}") await asyncio.sleep(check_interval) 

   asyncdef _check_website_status(self, url: str) -> Dict:

    """检查网站健康状态""" check_request = f""" 检查网站健康状况: 1. 访问 {url} 2. 检查页面加载时间 3. 验证关键功能是否正常 4. 检查错误信息 """ results = await self.agent.process_request(check_request) return self._analyze_health_status(results) 

   结论
   通过结合Playwright MCP和LLM，我们能够构建强大的AI智能体和工作流系统，这些系统能够：

理解自然语言指令并转化为具体操作
自动化复杂业务流程，减少人工干预
自适应处理异常情况，提高系统鲁棒性
持续学习和优化执行策略
这种技术组合为自动化测试、数据采集、监控警报等场景提供了全新的解决方案。随着AI技术的不断发展，这种模式将在更多领域展现其价值，推动企业数字化转型和智能化升级。

Playwright MCP与LLM的结合只是AI驱动自动化的开始，这个领域的发展潜力无限，值得我们持续关注和探索。