test_browser_visible.py

#!/usr/bin/env python3
"""
测试浏览器可见性设置 - 确保 browser_navigate 能打开可见的浏览器窗口
"""

import asyncio
import os
import sys
from pathlib import Path

# Add the python_agent to the path
sys.path.insert(0, str(Path(__file__).parent / "python_agent"))

from python_agent.core.config import AgentConfig, ApiConfiguration, ApiProviderType
from python_agent.agent import Agent

async def test_browser_visibility():
    """测试浏览器可见性"""
    print("=== 测试浏览器可见性设置 ===\n")
    
    # 创建专门的 MCP 配置，强制非 headless 模式
    mcp_servers = {
        "playwright": {
            "enabled": True,
            "command": "/opt/homebrew/bin/npx",
            "args": ["@playwright/mcp@latest"],
            "description": "Browser automation with visible window",
            "env": {
                "PLAYWRIGHT_HEADLESS": "false",  # 强制非 headless
                "PLAYWRIGHT_BROWSER": "chromium"
            }
        }
    }
    
    # 检查环境变量
    api_key = os.getenv("ANTHROPIC_API_KEY")
    if not api_key:
        print("❌ 未找到 ANTHROPIC_API_KEY")
        print("请设置: export ANTHROPIC_API_KEY='your-key-here'")
        return
    
    # 创建代理配置
    config = AgentConfig(
        api_config=ApiConfiguration(
            provider=ApiProviderType.ANTHROPIC,
            api_key=api_key,
            model_id="claude-3-sonnet-20240229"
        ),
        mcp_servers=mcp_servers,
        show_tool_output=True,
        tool_output_style="full"
    )
    
    agent = Agent(config)
    
    try:
        print("🚀 启动代理...")
        await agent.start()
        
        print("⏱️  等待 MCP 服务器连接...")
        await asyncio.sleep(3)
        
        # 检查 MCP 连接状态
        status = await agent.get_status()
        mcp_status = status.get("mcp_servers", {})
        print(f"📊 MCP 状态: 已连接 {mcp_status.get('connected', 0)}/{mcp_status.get('total', 0)} 服务器")
        
        if mcp_status.get('connected', 0) == 0:
            print("❌ 没有 MCP 服务器连接，无法测试浏览器")
            return
        
        # 获取工具信息
        tools_info = await agent.get_tools_info()
        browser_tools = [name for name in tools_info['tools'].keys() if 'browser' in name or 'navigate' in name]
        print(f"🔧 可用的浏览器工具: {browser_tools}")
        
        if not browser_tools:
            print("❌ 未找到浏览器工具")
            return
        
        # 测试浏览器导航（应该会打开可见的浏览器窗口）
        print("\n🌐 测试浏览器导航...")
        print("注意观察是否有浏览器窗口打开！")
        
        test_message = """请使用浏览器工具导航到 https://www.example.com，
        请确保浏览器窗口是可见的，而不是 headless 模式。
        然后告诉我你看到了什么内容。"""
        
        print(f"📝 发送测试消息: {test_message}")
        
        async for result in agent.execute_task(test_message):
            result_type = result.get("type")
            
            if result_type == "reasoning":
                print(f"🧠 思考: {result.get('content', '')[:100]}...")
            elif result_type == "tool_use":
                tool_name = result.get("name")
                print(f"🔧 使用工具: {tool_name}")
                if 'browser' in tool_name or 'navigate' in tool_name:
                    print("   ⚠️  如果这是浏览器工具，你现在应该看到浏览器窗口打开！")
            elif result_type == "tool_result":
                tool_name = result.get("name")
                success = result.get("success")
                print(f"✅ 工具结果 {tool_name}: {'成功' if success else '失败'}")
                if not success and 'browser' in tool_name:
                    error = result.get("error", "未知错误")
                    print(f"   ❌ 错误: {error}")
            elif result_type == "assistant_message":
                content = result.get("content", "")
                print(f"🤖 助手回复: {content}")
            elif result_type == "error":
                print(f"❌ 错误: {result.get('error')}")
                
    except Exception as e:
        print(f"❌ 测试失败: {e}")
    finally:
        print("\n🔄 关闭代理...")
        await agent.stop()

def check_playwright_installation():
    """检查 Playwright 安装"""
    print("=== 检查 Playwright 安装 ===\n")
    
    # 检查 npx 是否可用
    import subprocess
    try:
        result = subprocess.run(["npx", "--version"], capture_output=True, text=True)
        if result.returncode == 0:
            print(f"✅ npx 版本: {result.stdout.strip()}")
        else:
            print("❌ npx 不可用")
            return False
    except FileNotFoundError:
        print("❌ 未找到 npx")
        return False
    
    # 检查 @playwright/mcp 是否可用
    try:
        result = subprocess.run(["npx", "@playwright/mcp@latest", "--help"], 
                              capture_output=True, text=True, timeout=10)
        if result.returncode == 0:
            print("✅ @playwright/mcp 可用")
        else:
            print("❌ @playwright/mcp 不可用")
            print("运行: npm install -g @playwright/mcp")
            return False
    except subprocess.TimeoutExpired:
        print("⏰ @playwright/mcp 响应超时")
        return False
    except Exception as e:
        print(f"❌ 检查 @playwright/mcp 时出错: {e}")
        return False
    
    return True

async def main():
    """主函数"""
    print("🔍 浏览器可见性测试\n")
    
    # 检查 Playwright 安装
    if not check_playwright_installation():
        print("\n❌ Playwright 环境检查失败")
        return
    
    print("\n" + "="*50)
    
    # 测试浏览器可见性
    await test_browser_visibility()
    
    print("\n" + "="*50)
    print("📋 测试总结:")
    print("1. 如果你看到了浏览器窗口打开，说明配置正确")
    print("2. 如果没有看到浏览器窗口，可能是以下原因:")
    print("   - MCP 服务器使用 headless 模式")
    print("   - 浏览器工具执行失败")
    print("   - MCP 服务器连接问题")
    print("3. 检查日志中的详细错误信息")

if __name__ == "__main__":
    asyncio.run(main())