Chrome DevTools MCP 实战指南：浏览器自动化解决方案

{
  "mcpServers": {
    "chrome-devtools": {
      "command": "npx",
      "args": ["-y", "chrome-devtools-mcp@latest"]
    }
  }
}

使用 chrome mcp 工具，
1. 打开 www.baidu.com 
2. 搜索 ApFramework.com
3. 点击第一个链接

# 创建新项目并初始化虚拟环境
uv init chrome-devtools-mcp-samples
cd chrome-devtools-mcp-samples

# 激活虚拟环境
source .venv/bin/activate

# 安装核心依赖包
uv add python-dotenv langchain-deepseek langgraph langchain langchain-mcp-adapters

# DeepSeek API 配置（或其他兼容的 LLM 服务）
DEEPSEEK_API_KEY=your_api_key_here

# 可选：其他 LLM 服务配置
QWEN_API_KEY=your_qwen_api_key
QWEN_BASE_URL=https://dashscope.aliyuncs.com/compatible-mode/v1

{
    "command": "npx",
    "args": [
        "-y", 
        "chrome-devtools-mcp@latest", 
        "--browserUrl=http://127.0.0.1:9222"
    ],
    "transport": "stdio"
}

"""
LangGraph 与 Chrome DevTools MCP 集成示例

这个示例展示了如何使用 LangGraph 调用 chrome-devtools-mcp 服务器
来实现浏览器自动化任务。
"""

from langchain_mcp_adapters.client import MultiServerMCPClient
from langchain.chat_models import init_chat_model
from langgraph.graph import StateGraph, MessagesState, START, END
from langgraph.prebuilt import ToolNode

import os
import asyncio
from dotenv import load_dotenv

# 加载.env文件中的环境变量
load_dotenv()

class ChromeDevToolsAgent:
    """Chrome DevTools 自动化代理"""
    
    def __init__(self):
        # 初始化模型
        self.model = init_chat_model(
            "qwen3-max",  # 使用千问模型
            model_provider="openai",  # 使用 OpenAI 兼容接口
            api_key=os.environ.get("QWEN_API_KEY"),
            base_url=os.environ.get("QWEN_BASE_URL"),
            # 优化模型参数
            temperature=0.1,  # 降低随机性，提高一致性
            max_tokens=4096,  # 增加输出长度
            top_p=0.8,  # 控制采样范围
            frequency_penalty=0.1,  # 减少重复
            presence_penalty=0.1,  # 鼓励多样性
        )
        print("✅ 大模型初始化完成")
        
        # 为模型添加系统提示词，提升任务执行能力
        self.system_prompt = """你是一个专业的浏览器自动化助手，具备以下能力：

            1. 精确理解任务：仔细分析用户的每个步骤要求
            2. 智能网页操作：能够准确识别页面元素并执行相应操作
            3. 内容提取专家：擅长从网页中提取和整理有价值的信息
            4. 逻辑推理：能够根据页面内容做出合理的判断和选择

            ## 重要的工具使用规则：

            ### 必须遵循的操作顺序：
            1. 导航到页面后，必须先调用 `mcp_chrome__devtools_take_snapshot` 获取页面快照
            2. 任何页面操作前，都要先调用 `mcp_chrome__devtools_take_snapshot` 确保有最新的页面状态
            3. 点击或填写表单后，再次调用 `mcp_chrome__devtools_take_snapshot` 查看页面变化

            ### 工具调用流程：导航页面 → take_snapshot → 分析页面 → 执行操作 → take_snapshot → 继续下一步

            ### 错误处理：
            - 如果遇到 "No snapshot found" 错误，立即调用 `mcp_chrome__devtools_take_snapshot`
            - 如果找不到元素，先调用 `mcp_chrome__devtools_take_snapshot` 重新获取页面状态
            - 每次操作失败后，都要重新获取快照再重试

            ### 执行原则：
            - 每个操作前先观察页面状态（通过快照）
            - 优先使用最可靠的元素选择器
            - 遇到问题时提供清晰的错误说明
            - 确保操作的准确性和完整性

            请严格按照用户的步骤要求执行任务，并始终记住先获取页面快照！
        """
        
        # 设置 MCP 客户端 - 连接到 chrome-devtools-mcp
        self.client = MultiServerMCPClient({
            "chrome-devtools": {
                # chrome-devtools-mcp 通过 npx 直接调用，使用 stdio 传输
                # --browserUrl 连接到已运行的Chrome实例，避免每次启动新的浏览器进程
                "command": "npx",
                "args": [
                    "-y", 
                    "chrome-devtools-mcp@latest", 
                    "--browserUrl=http://127.0.0.1:9222"
                ],
                "transport": "stdio",
            }
        })
        
        self.graph = None
        self.tools = None
    
    async def initialize(self):
        """初始化代理，获取工具并构建图"""
        print("\n=== 初始化 Chrome DevTools 代理 ===")
        
        # 获取 MCP 服务器的工具
        print("正在获取 Chrome DevTools MCP 工具...")
        try:
            self.tools = await self.client.get_tools()
            print(f"成功获取 {len(self.tools)} 个工具:")
            for tool in self.tools:
                print(f"  - {tool.name}: {tool.description}")
        except Exception as e:
            print(f"获取工具失败: {e}")
            print("请确保 chrome-devtools-mcp 服务器正在运行")
            print("启动命令: npx -y chrome-devtools-mcp@latest")
            raise
        
        # 将工具绑定到模型
        print("\n绑定工具到模型...")
        self.model_with_tools = self.model.bind_tools(self.tools)
        print(f"已将 {len(self.tools)} 个工具绑定到模型")
        
        # 构建 LangGraph 工作流
        self._build_graph()
        print("LangGraph 工作流构建完成")
    
    def _build_graph(self):
        """构建 LangGraph 工作流"""
        # 创建工具节点
        tool_node = ToolNode(self.tools)
        
        def should_continue(state: MessagesState):
            """决定是否继续执行工具"""
            messages = state["messages"]
            last_message = messages[-1]
            if last_message.tool_calls:
                return "tools"
            return END
        
        async def enhanced_tool_node(state: MessagesState):
            """增强的工具节点，处理快照状态管理"""
            messages = state["messages"]
            last_message = messages[-1]
            
            if hasattr(last_message, 'tool_calls') and last_message.tool_calls:
                print(f"\n=== 执行工具调用 ===")
                
                # 检查是否需要先获取快照
                needs_snapshot = False
                for tool_call in last_message.tool_calls:
                    tool_name = tool_call.get('name', '')
                    if tool_name in ['mcp_chrome__devtools_click', 'mcp_chrome__devtools_fill', 
                                   'mcp_chrome__devtools_hover', 'mcp_chrome__devtools_drag']:
                        needs_snapshot = True
                        break
                
                # 如果需要快照但没有先调用take_snapshot，自动添加
                if needs_snapshot:
                    snapshot_exists = False
                    for msg in reversed(messages[-5:]):  # 检查最近5条消息
                        if hasattr(msg, 'content') and msg.content and 'Page content' in str(msg.content):
                            snapshot_exists = True
                            break
                    
                    if not snapshot_exists:
                        print("⚠️  检测到需要快照但未找到，自动获取快照...")
                        # 创建快照工具调用
                        from langchain_core.messages import AIMessage
                        snapshot_call = AIMessage(
                            content="",
                            tool_calls=[{
                                "name": "mcp_chrome__devtools_take_snapshot",
                                "args": {},
                                "id": "auto_snapshot"
                            }]
                        )
                        # 先执行快照
                        snapshot_result = await tool_node.ainvoke({"messages": messages + [snapshot_call]})
                        messages.extend(snapshot_result["messages"])
            
            # 执行原始工具调用
            result = await tool_node.ainvoke(state)
            return result
        
        async def call_model(state: MessagesState):
            """调用模型"""
            messages = state["messages"]
            print(f"\n=== 调用模型 (消息数: {len(messages)}) ===")
            
            if messages:
                last_msg = messages[-1]
                if hasattr(last_msg, 'content') and last_msg.content:
                    # 检查是否有工具调用错误
                    if "ToolException" in str(last_msg.content) or "No snapshot found" in str(last_msg.content):
                        print(f"⚠️  检测到工具错误: {last_msg.content[:200]}...")
                        # 添加更强的错误处理指导
                        error_guidance = """
                            检测到页面快照错误！请严格按照以下步骤处理：

                            1. 🔍 立即调用 mcp_chrome__devtools_take_snapshot 获取当前页面快照
                            2. 📋 仔细分析快照内容，确认页面元素的uid
                            3. 🎯 使用正确的uid进行后续操作
                            4. ⚠️  绝对不要在没有快照的情况下尝试点击或填充元素

                            记住：每次页面导航或操作后都需要重新获取快照！
                        """
                        messages.append({"role": "system", "content": error_guidance})
                    else:
                        print(f"最新消息: {last_msg.content[:500]}...")
            
            try:
                response = await self.model_with_tools.ainvoke(messages)
                
                if hasattr(response, 'tool_calls') and response.tool_calls:
                    print(f"✅ 模型请求调用 {len(response.tool_calls)} 个工具:")
                    for i, tool_call in enumerate(response.tool_calls):
                        print(f"  {i+1}. {tool_call['name']}")
                elif hasattr(response, 'content'):
                    print(f"💬 模型响应: {response.content[:500]}...")
                
                return {"messages": [response]}
                
            except Exception as e:
                print(f"❌ 模型调用失败: {e}")
                error_response = {
                    "role": "assistant", 
                    "content": f"模型调用出错: {str(e)}，请重试或检查配置。"
                }
                return {"messages": [error_response]}
        
        # 构建图
        builder = StateGraph(MessagesState)
        builder.add_node("call_model", call_model)
        builder.add_node("tools", enhanced_tool_node)  # 使用增强的工具节点
        
        builder.add_edge(START, "call_model")
        builder.add_conditional_edges("call_model", should_continue)
        builder.add_edge("tools", "call_model")
        
        self.graph = builder.compile()
    
    async def execute_task(self, task_description: str):
        """执行浏览器自动化任务"""
        if not self.graph:
            raise RuntimeError("代理未初始化，请先调用 initialize()")
        
        print(f"\n=== 执行任务: {task_description} ===")
        
        try:
            # 构建包含系统提示词的完整消息
            messages = [
                {"role": "system", "content": self.system_prompt},
                {"role": "user", "content": task_description}
            ]
            
            response = await self.graph.ainvoke({
                "messages": messages
            })
            
            print("\n=== 任务执行完成 ===")
            
            # 提取最终响应
            final_message = response["messages"][-1]
            if hasattr(final_message, 'content'):
                return final_message.content
            else:
                return str(final_message)
                
        except Exception as e:
            print(f"任务执行失败: {e}")
            raise
    
    async def close(self):
        """关闭客户端连接"""
        # MultiServerMCPClient 不需要显式关闭
        # 它会在垃圾回收时自动清理资源
        pass


async def main():
    """主函数 - 演示 Chrome DevTools 自动化"""
    agent = ChromeDevToolsAgent()
    
    try:
        # 初始化代理
        await agent.initialize()
        
        # 示例任务列表
        tasks = [
            """
            (1)打开 chrome 浏览器并导航到 https://apframework.com/
            (2)理解当前页面内容，跳转到 About 页面
            (3)然后翻到文章最后，获取完整的文章内容并输出
            """
        ]
        
        print("\n" + "="*50)
        print("Chrome DevTools 自动化任务演示")
        print("="*50)
        
        for i, task in enumerate(tasks, 1):
            print(f"\n--- 任务 {i}: {task} ---")
            try:
                result = await agent.execute_task(task)
                print(f"结果: {result}")
            except Exception as e:
                print(f"任务失败: {e}")
                # 继续执行下一个任务
                continue
            
            # 在任务之间稍作停顿
            await asyncio.sleep(1)
    
    except Exception as e:
        print(f"程序执行出错: {e}")
    
    finally:
        # 清理资源
        await agent.close()
        print("\n程序执行完成")


if __name__ == "__main__":
    # 运行主程序
    asyncio.run(main())

#!/usr/bin/env python3
"""
Chrome持久化服务管理器
管理独立的Chrome实例，确保浏览器在MCP调用间保持运行
"""

import subprocess
import time
import signal
import sys
import os
import requests
import json
from pathlib import Path

class ChromePersistentService:
    """Chrome持久化服务管理器"""
    
    def __init__(self, port=9222, user_data_dir="/tmp/chrome-persistent-session"):
        self.port = port
        self.user_data_dir = user_data_dir
        self.process = None
        self.pid_file = Path("/tmp/chrome-persistent.pid")
    
    def is_chrome_running(self):
        """检查Chrome是否在指定端口运行"""
        try:
            response = requests.get(f"http://127.0.0.1:{self.port}/json/version", timeout=2)
            return response.status_code == 200
        except:
            return False
    
    def get_chrome_info(self):
        """获取Chrome实例信息"""
        try:
            response = requests.get(f"http://127.0.0.1:{self.port}/json/version", timeout=2)
            if response.status_code == 200:
                return response.json()
        except:
            pass
        return None
    
    def start_chrome(self):
        """启动Chrome实例"""
        if self.is_chrome_running():
            print(f"✅ Chrome已在端口{self.port}运行")
            return True
        
        print(f"🚀 启动Chrome实例 (端口: {self.port})")
        
        # Chrome启动命令
        chrome_cmd = [
            "/Applications/Google Chrome.app/Contents/MacOS/Google Chrome",
            f"--remote-debugging-port={self.port}",
            f"--user-data-dir={self.user_data_dir}",
            "--no-first-run",
            "--no-default-browser-check",
            "--disable-background-timer-throttling",
            "--disable-backgrounding-occluded-windows",
            "--disable-renderer-backgrounding"
        ]
        
        try:
            # 启动Chrome进程
            self.process = subprocess.Popen(
                chrome_cmd,
                stdout=subprocess.DEVNULL,
                stderr=subprocess.DEVNULL,
                preexec_fn=os.setsid  # 创建新的进程组
            )
            
            # 保存PID
            with open(self.pid_file, 'w') as f:
                f.write(str(self.process.pid))
            
            # 等待Chrome启动
            for i in range(10):
                time.sleep(1)
                if self.is_chrome_running():
                    print(f"✅ Chrome启动成功 (PID: {self.process.pid})")
                    return True
                print(f"⏳ 等待Chrome启动... ({i+1}/10)")
            
            print("❌ Chrome启动超时")
            return False
            
        except Exception as e:
            print(f"❌ Chrome启动失败: {e}")
            return False
    
    def stop_chrome(self):
        """停止Chrome实例"""
        print("🛑 停止Chrome实例...")
        
        # 尝试从PID文件读取进程ID
        if self.pid_file.exists():
            try:
                with open(self.pid_file, 'r') as f:
                    pid = int(f.read().strip())
                
                # 终止进程组
                os.killpg(os.getpgid(pid), signal.SIGTERM)
                time.sleep(2)
                
                # 如果还在运行，强制终止
                try:
                    os.killpg(os.getpgid(pid), signal.SIGKILL)
                except:
                    pass
                
                self.pid_file.unlink()
                print("✅ Chrome已停止")
                
            except Exception as e:
                print(f"⚠️  停止Chrome时出错: {e}")
        
        # 清理进程引用
        if self.process:
            try:
                self.process.terminate()
                self.process.wait(timeout=5)
            except:
                try:
                    self.process.kill()
                except:
                    pass
            self.process = None
    
    def restart_chrome(self):
        """重启Chrome实例"""
        print("🔄 重启Chrome实例...")
        self.stop_chrome()
        time.sleep(2)
        return self.start_chrome()
    
    def status(self):
        """显示Chrome状态"""
        print("📊 Chrome持久化服务状态")
        print("=" * 40)
        
        if self.is_chrome_running():
            info = self.get_chrome_info()
            print(f"✅ 状态: 运行中")
            print(f"🌐 端口: {self.port}")
            print(f"📁 数据目录: {self.user_data_dir}")
            
            if info:
                print(f"🔖 版本: {info.get('Browser', 'Unknown')}")
                print(f"🆔 WebSocket: {info.get('webSocketDebuggerUrl', 'Unknown')}")
            
            # 检查PID文件
            if self.pid_file.exists():
                with open(self.pid_file, 'r') as f:
                    pid = f.read().strip()
                print(f"🆔 PID: {pid}")
        else:
            print("❌ 状态: 未运行")
            print(f"🌐 端口: {self.port}")
            print(f"📁 数据目录: {self.user_data_dir}")

def main():
    """主函数"""
    service = ChromePersistentService()
    
    if len(sys.argv) < 2:
        print("Chrome持久化服务管理器")
        print("=" * 30)
        print("用法:")
        print("  python chrome_persistent_service.py start   # 启动Chrome")
        print("  python chrome_persistent_service.py stop    # 停止Chrome")
        print("  python chrome_persistent_service.py restart # 重启Chrome")
        print("  python chrome_persistent_service.py status  # 查看状态")
        return
    
    command = sys.argv[1].lower()
    
    if command == "start":
        service.start_chrome()
    elif command == "stop":
        service.stop_chrome()
    elif command == "restart":
        service.restart_chrome()
    elif command == "status":
        service.status()
    else:
        print(f"❌ 未知命令: {command}")
        print("可用命令: start, stop, restart, status")

if __name__ == "__main__":
    # 处理Ctrl+C信号
    def signal_handler(sig, frame):
        print("\n👋 收到退出信号...")
        service = ChromePersistentService()
        service.stop_chrome()
        sys.exit(0)
    
    signal.signal(signal.SIGINT, signal_handler)
### 服务启动与运行

#### 启动 Chrome 持久化服务

首先启动 Chrome 持久化服务，确保浏览器实例在后台稳定运行：

```bash
python chrome_persistent_service.py start

uv run chrome_devtools_langgraph.py