Agent/examples/browser-test/diagnose_cdp.py

#!/usr/bin/env python3
"""
CDP连接诊断脚本
"""
import asyncio
import sys
from pathlib import Path

# 添加项目根目录到路径
sys.path.insert(0, str(Path(__file__).parent.parent.parent))

from agent.tools.builtin.browser.baseClass import init_browser_session, cleanup_browser_session

async def diagnose():
    """诊断browser-use的CDP连接问题"""
    print("=" * 60)
    print("Browser-Use CDP 连接诊断")
    print("=" * 60)

    try:
        print("\n步骤1: 初始化浏览器会话...")
        print("  - 浏览器类型: local")
        print("  - Headless: False")

        session, tools = await init_browser_session(
            browser_type="local",
            headless=False
        )

        print("✅ 浏览器会话初始化成功")

        print("\n步骤2: 测试基本导航...")
        result = await tools.navigate(
            url="https://www.baidu.com",
            browser_session=session
        )

        print(f"✅ 导航成功")
        print(f"  - 结果: {result.extracted_content[:100] if result.extracted_content else 'N/A'}")

        print("\n步骤3: 获取页面URL...")
        current_url = await session.get_current_page_url()
        print(f"✅ 当前URL: {current_url}")

        print("\n步骤4: 清理浏览器会话...")
        await cleanup_browser_session()
        print("✅ 清理完成")

        print("\n" + "=" * 60)
        print("✅ 所有测试通过！browser-use工作正常")
        print("=" * 60)
        return True

    except Exception as e:
        print(f"\n❌ 错误: {type(e).__name__}")
        print(f"   消息: {str(e)}")

        import traceback
        print("\n完整堆栈:")
        traceback.print_exc()

        print("\n" + "=" * 60)
        print("诊断建议:")
        print("=" * 60)

        error_msg = str(e).lower()
        if "json" in error_msg or "expecting value" in error_msg:
            print("1. CDP endpoint返回空响应")
            print("   - 可能Chrome启动失败")
            print("   - 尝试: 手动启动Chrome并检查是否正常")
            print("   - 尝试: 使用headless=True模式")
        elif "connection" in error_msg or "websocket" in error_msg:
            print("1. WebSocket连接失败")
            print("   - 可能防火墙阻止了连接")
            print("   - 尝试: 检查防火墙设置")
        elif "timeout" in error_msg:
            print("1. 连接超时")
            print("   - Chrome启动太慢")
            print("   - 尝试: 增加超时时间")
        else:
            print("1. 未知错误")
            print("   - 查看上面的完整堆栈信息")

        return False

if __name__ == "__main__":
    success = asyncio.run(diagnose())
    sys.exit(0 if success else 1)