Platform Task Example

import asyncio
import os
from datetime import datetime
from pathlib import Path
from pydantic import BaseModel, Field
from typing import List

from minitap.mobile_use.sdk import Agent
from minitap.mobile_use.sdk.types import PlatformTaskRequest, DevicePlatform
from minitap.mobile_use.sdk.builders import Builders
from minitap.mobile_use.sdk.types.exceptions import PlatformServiceError


# Define output structures
class Email(BaseModel):
    sender: str = Field(..., description="Email sender name/address")
    subject: str = Field(..., description="Email subject line")
    is_unread: bool = Field(..., description="Whether email is unread")


class EmailSummary(BaseModel):
    total_emails: int = Field(..., description="Total number of emails")
    unread_count: int = Field(..., description="Number of unread emails")
    recent_emails: List[Email] = Field(
        default_factory=list,
        description="List of recent emails"
    )


class MeetingInfo(BaseModel):
    title: str = Field(..., description="Meeting title")
    date: str = Field(..., description="Meeting date")
    time: str = Field(..., description="Meeting time")
    attendees: List[str] = Field(..., description="List of attendees")
    confirmed: bool = Field(..., description="Whether meeting was confirmed")


async def main():
    """
    Main execution demonstrating Minitap Platform usage.
    
    This example shows:
    1. Using platform-configured tasks
    2. Switching between different LLM profiles
    3. Structured output with type safety
    4. Error handling for platform errors
    """
    
    # Verify API key is configured
    if not os.getenv("MINITAP_API_KEY"):
        print("❌ Error: MINITAP_API_KEY environment variable not set")
        print("   Get your API key from: https://platform.mobile-use.ai/api-keys")
        return
    
    print("🚀 Starting platform task example...\n")
    
    # Configure agent with specific device
    config = (
        Builders.AgentConfig
        .for_device(platform=DevicePlatform.ANDROID, device_id="emulator-5554")
        .build()
    )
    
    agent = Agent(config=config)
    
    try:
        # Initialize agent
        print("📱 Initializing agent...")
        if not await agent.init():
            print("❌ Failed to initialize agent")
            return
        print("✅ Agent initialized\n")
        
        # ===================================================================
        # Task 1: Email Summary (using "default" profile)
        # ===================================================================
        print("=" * 60)
        print("TASK 1: Email Summary")
        print("=" * 60)
        print("Profile: default")
        print("Task: email-summary (configured on platform)")
        print()
        
        try:
            email_result = await agent.run_task(
                request=PlatformTaskRequest[EmailSummary](
                    task="email-summary",
                    profile="default",
                    # Local trace recording (in addition to platform)
                    record_trace=True,
                    trace_path=Path(f"/tmp/platform-traces/{datetime.now().strftime('%Y%m%d_%H%M%S')}")
                )
            )
            
            if email_result:
                print("✅ Email summary retrieved:")
                print(f"   Total emails: {email_result.total_emails}")
                print(f"   Unread: {email_result.unread_count}")
                print(f"\n   Recent emails:")
                for i, email in enumerate(email_result.recent_emails[:5], 1):
                    status = "📬" if email.is_unread else "📭"
                    print(f"   {i}. {status} {email.sender}: {email.subject}")
            else:
                print("⚠️  No result returned")
        
        except PlatformServiceError as e:
            print(f"❌ Platform error: {e}")
            print("   Check that 'email-summary' task exists on platform")
        except Exception as e:
            print(f"❌ Unexpected error: {e}")
        
        print()
        
        # ===================================================================
        # Task 2: Schedule Meeting (using "accurate" profile)
        # ===================================================================
        print("=" * 60)
        print("TASK 2: Schedule Meeting")
        print("=" * 60)
        print("Profile: accurate (more powerful models)")
        print("Task: schedule-meeting (configured on platform)")
        print()
        
        try:
            meeting_result = await agent.run_task(
                request=PlatformTaskRequest[MeetingInfo](
                    task="schedule-meeting",
                    profile="accurate",  # Use more powerful profile
                    record_trace=True,
                    trace_path=Path(f"/tmp/platform-traces/{datetime.now().strftime('%Y%m%d_%H%M%S')}")
                )
            )
            
            if meeting_result:
                print("✅ Meeting scheduled:")
                print(f"   Title: {meeting_result.title}")
                print(f"   Date: {meeting_result.date}")
                print(f"   Time: {meeting_result.time}")
                print(f"   Attendees: {', '.join(meeting_result.attendees)}")
                print(f"   Confirmed: {'✅' if meeting_result.confirmed else '⏳'}")
            else:
                print("⚠️  No result returned")
        
        except PlatformServiceError as e:
            print(f"❌ Platform error: {e}")
            print("   Check that 'schedule-meeting' task exists on platform")
        except Exception as e:
            print(f"❌ Unexpected error: {e}")
        
        print()
        
        # ===================================================================
        # Summary
        # ===================================================================
        print("=" * 60)
        print("📊 Summary")
        print("=" * 60)
        print("✓ View task runs at: https://platform.mobile-use.ai/task-runs")
        print("✓ See execution timelines, screenshots, and agent thoughts")
        print("✓ Analyze performance and debug failures")
        print()
    
    except Exception as e:
        print(f"❌ Fatal error: {e}")
    
    finally:
        print("🧹 Cleaning up...")
        await agent.clean()
        print("✅ Done!\n")


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