test_task_direct.py

#!/usr/bin/env python3
"""
Direct task execution test - bypassing conversation loop
"""

import asyncio
import sys
import os
from pathlib import Path

# Add current directory to path
sys.path.insert(0, str(Path(__file__).parent))

from python_agent.agent import create_agent_from_config_file

async def test_direct_task():
    """Test the tutorialspoint screenshot task directly"""
    print("=== Direct Task Execution Test ===")
    
    # Create agent
    agent = create_agent_from_config_file('python_agent/mcp_config.yaml')
    
    try:
        # Start agent
        await agent.start()
        print("✅ Agent started successfully")
        
        # Get status
        status = await agent.get_status()
        print(f"✅ MCP tools loaded: {status['tools']['mcp']}")
        print(f"✅ Connected servers: {status['mcp_servers']['connected']}")
        
        # Create directory
        os.makedirs('temp_images', exist_ok=True)
        
        # Step 1: Navigate to the URL
        print("\n📍 Step 1: Navigating to tutorialspoint...")
        nav_tool = agent.tools.get('mcp_browser_navigate')
        if nav_tool:
            result = await nav_tool.execute({
                'url': 'https://www.tutorialspoint.com/parallel_algorithm/graph_algorithm.htm'
            })
            print(f"Navigation result: success={result.success}")
            if not result.success:
                print(f"❌ Navigation error: {result.error}")
                return False
            else:
                print("✅ Successfully navigated to the page!")
        else:
            print("❌ Navigation tool not found")
            return False
        
        # Wait for page to load
        print("⏳ Waiting for page to load...")
        await asyncio.sleep(3)
        
        # Step 2: Take screenshot
        print("\n📸 Step 2: Taking screenshot...")
        screenshot_tool = agent.tools.get('mcp_browser_take_screenshot')
        if screenshot_tool:
            result = await screenshot_tool.execute({
                'filename': 'temp_images/graph_algorithm.png'
            })
            print(f"Screenshot result: success={result.success}")
            if result.success:
                print("✅ Screenshot taken successfully!")
                print(f"Output: {result.output[:200]}..." if len(result.output) > 200 else result.output)
                
                # Check if file exists in current directory
                screenshot_files = []
                for root, dirs, files in os.walk('.'):
                    for file in files:
                        if 'graph_algorithm' in file and file.endswith(('.png', '.jpg', '.jpeg')):
                            screenshot_files.append(os.path.join(root, file))
                
                if screenshot_files:
                    print(f"✅ Found screenshot files: {screenshot_files}")
                    for file_path in screenshot_files:
                        size = os.path.getsize(file_path)
                        print(f"  - {file_path}: {size} bytes")
                else:
                    print("⚠️  No screenshot file found in expected location")
                    
                return True
            else:
                print(f"❌ Screenshot error: {result.error}")
                return False
        else:
            print("❌ Screenshot tool not found")
            return False
            
    finally:
        # Stop agent
        await agent.stop()
        print("🔚 Agent stopped")

if __name__ == "__main__":
    success = asyncio.run(test_direct_task())
    if success:
        print("\n🎉 Task completed successfully!")
    else:
        print("\n❌ Task failed!")