MultiOn浏览器自动化Agent开发

import requestsimport time class MultiOnClient:    """MultiOn API 客户端示例"""        def __init__(self, api_key: str):        self.api_key = api_key        self.base_url = "https://api.multion.ai/v1"        self.headers = {            "X-MULTION-API-KEY": api_key,            "Content-Type": "application/json"        }        def create_session(self, url: str) -> str:        """创建新的浏览器会话，返回会话ID"""        response = requests.post(            f"{self.base_url}/session",            headers=self.headers,            json={"url": url}        )        response.raise_for_status()        return response.json()["session_id"]        def execute_task(        self,         session_id: str,         task: str,        mode: str = "standard"    ) -> dict:        """在指定会话中执行任务（同步模式）"""        response = requests.post(            f"{self.base_url}/session/{session_id}/task",            headers=self.headers,            json={                "task": task,                "mode": mode            }        )        response.raise_for_status()        return response.json()        def close_session(self, session_id: str):        """关闭会话"""        requests.delete(            f"{self.base_url}/session/{session_id}",            headers=self.headers        ) # 使用示例client = MultiOnClient("your-api-key-here")session_id = client.create_session("https://example.com")result = client.execute_task(    session_id,     "找到页面上的所有产品名称和价格，以JSON格式返回")print(f"提取结果: {result.get('extracted_data')}")client.close_session(session_id)

# -*- coding: utf-8 -*-"""MultiOn API 客户端演示展示：会话管理、任务执行、状态轮询、结果提取、错误处理"""import requestsimport timefrom typing import Optional, Dict, Listfrom dataclasses import dataclass, fieldimport json  @dataclassclass TaskResult:    """任务执行结果"""    success: bool    status: str    extracted_data: Dict = field(default_factory=dict)    message: str = ""    execution_time: float = 0.0  class MultiOnAPIClient:    """MultiOn API 客户端封装（模拟实现，用于演示调用流程）"""     def __init__(self, api_key: str, base_url: str = "https://api.multion.ai/v1"):        self.api_key = api_key        self.base_url = base_url        self.session = requests.Session()        self.session.headers.update({            "X-MULTION-API-KEY": api_key,            "Content-Type": "application/json"        })        self.active_sessions: Dict[str, Dict] = {}  # 活跃会话记录     def create_session(self, initial_url: str, **kwargs) -> str:        """创建新的浏览器会话                Args:            initial_url: 初始打开的网址            kwargs: 其他会话参数（如视口大小、用户代理等）                    Returns:            会话ID        """        payload = {"url": initial_url}        payload.update(kwargs)                try:            response = self.session.post(                f"{self.base_url}/session",                json=payload,                timeout=30            )            response.raise_for_status()            data = response.json()            session_id = data["session_id"]                        # 记录会话信息            self.active_sessions[session_id] = {                "url": initial_url,                "created_at": time.time(),                "status": "active"            }            print(f"✓ 会话创建成功: {session_id}")            return session_id                    except requests.exceptions.RequestException as e:            print(f"✗ 会话创建失败: {e}")            raise     def execute_task_sync(        self,        session_id: str,        task_description: str,        output_format: Optional[str] = None,        timeout: int = 120    ) -> TaskResult:        """同步执行任务（等待完成后返回结果）                Args:            session_id: 会话ID            task_description: 自然语言任务描述            output_format: 期望的输出格式（如json、table等）            timeout: 超时时间（秒）                    Returns:            TaskResult 对象        """        start_time = time.time()        payload = {"task": task_description}        if output_format:            payload["output_format"] = output_format         print(f"  执行任务: {task_description[:50]}...")                try:            response = self.session.post(                f"{self.base_url}/session/{session_id}/task",                json=payload,                timeout=timeout            )            response.raise_for_status()            data = response.json()                        result = TaskResult(                success=data.get("status") == "success",                status=data.get("status", "unknown"),                extracted_data=data.get("extracted_data", {}),                message=data.get("message", ""),                execution_time=time.time() - start_time            )                        status_icon = "✓" if result.success else "✗"            print(f"  {status_icon} 任务完成: {result.status} ({result.execution_time:.1f}s)")            return result                    except requests.exceptions.Timeout:            return TaskResult(                success=False,                status="timeout",                message=f"任务执行超时（{timeout}秒）"            )        except requests.exceptions.RequestException as e:            return TaskResult(                success=False,                status="error",                message=f"请求失败: {str(e)}"            )     def execute_task_async(self, session_id: str, task_description: str) -> str:        """异步执行任务（立即返回任务ID，后续轮询状态）                Returns:            任务ID        """        response = self.session.post(            f"{self.base_url}/session/{session_id}/task/async",            json={"task": task_description}        )        response.raise_for_status()        task_id = response.json()["task_id"]        print(f"  异步任务已提交: {task_id}")        return task_id     def get_task_status(self, session_id: str, task_id: str) -> Dict:        """查询异步任务状态"""        response = self.session.get(            f"{self.base_url}/session/{session_id}/task/{task_id}"        )        response.raise_for_status()        return response.json()     def wait_for_task(        self,         session_id: str,         task_id: str,         max_wait: int = 300,        poll_interval: int = 3    ) -> TaskResult:        """等待异步任务完成（轮询方式）"""        start_time = time.time()                while time.time() - start_time < max_wait:            status_data = self.get_task_status(session_id, task_id)            status = status_data.get("status")                        if status in ("success", "failed", "error"):                return TaskResult(                    success=status == "success",                    status=status,                    extracted_data=status_data.get("extracted_data", {}),                    message=status_data.get("message", ""),                    execution_time=time.time() - start_time                )                        print(f"  任务进行中... ({status}) 已等待{time.time() - start_time:.0f}s")            time.sleep(poll_interval)                return TaskResult(            success=False,            status="timeout",            message=f"等待任务超时（{max_wait}秒）"        )     def close_session(self, session_id: str) -> bool:        """关闭会话"""        try:            self.session.delete(                f"{self.base_url}/session/{session_id}",                timeout=10            )            if session_id in self.active_sessions:                del self.active_sessions[session_id]            print(f"✓ 会话已关闭: {session_id}")            return True        except requests.exceptions.RequestException as e:            print(f"✗ 会话关闭失败: {e}")            return False     def list_active_sessions(self) -> List[str]:        """列出当前活跃会话"""        return list(self.active_sessions.keys())  # ===================== 电商比价演示场景 ===================== def demo_ecommerce_price_comparison():    """演示：使用MultiOn API进行电商比价"""    print("=" * 60)    print("  MultiOn API 演示：电商比价场景")    print("=" * 60)        # 注意：实际使用时替换为真实API密钥    client = MultiOnAPIClient(api_key="your-api-key-here")        # 模拟演示（因为没有真实API密钥，展示调用流程）    print("\n【流程说明】")    print("  1. 创建浏览器会话")    print("  2. 依次访问各电商网站搜索商品")    print("  3. 提取商品名称、价格、评分等信息")    print("  4. 汇总比价结果，返回结构化数据")    print()        # 展示一个完整的多步骤任务示例    print("【示例：多步骤电商比价任务】")    task_example = '''    任务描述：    请在以下三个电商网站上搜索"无线蓝牙耳机"：    1. 京东 (jd.com)    2. 淘宝 (taobao.com)     3. 拼多多 (pinduoduo.com)        对每个网站，提取前5个商品的：    - 商品名称    - 价格    - 评分    - 销量        最后按价格从低到高排序，返回JSON格式结果。    '''    print(task_example)        print("【返回结果示例】")    sample_result = {        "results": [            {"platform": "拼多多", "product": "XX无线蓝牙耳机", "price": 99.0, "rating": 4.5, "sales": "10万+"},            {"platform": "淘宝", "product": "YY无线蓝牙耳机", "price": 129.0, "rating": 4.7, "sales": "5万+"},            {"platform": "京东", "product": "ZZ无线蓝牙耳机", "price": 149.0, "rating": 4.8, "sales": "3万+"},        ],        "cheapest": {"platform": "拼多多", "product": "XX无线蓝牙耳机", "price": 99.0}    }    print(json.dumps(sample_result, ensure_ascii=False, indent=2))        print(f"\n{'=' * 60}")    print("  核心要点")    print(f"{'=' * 60}")    print("  1. 会话管理：创建/查询/关闭，保持状态连续性")    print("  2. 同步调用：简单任务直接等待返回")    print("  3. 异步调用：复杂任务轮询状态不阻塞")    print("  4. 自然语言：用自然语言描述任务，AI自动规划执行")    print("  5. 结构化输出：按指定格式返回数据，直接可用")    print("  6. 错误处理：超时、网络异常、任务失败的优雅处理")    print(f"{'=' * 60}")  def demo_integration_patterns():    """演示：常见集成模式"""    print("\n" + "=" * 60)    print("  常见集成模式")    print("=" * 60)        patterns = [        {            "模式": "Agent工具调用",            "场景": "AI Agent需要网页操作能力时",            "方式": "将MultiOn封装为LangChain Tool，Agent按需调用",            "适用": "通用AI助手、自动化工作流"        },        {            "模式": "批量数据采集",            "场景": "需要从多个网站提取结构化数据",            "方式": "任务队列 + 异步调用 + 结果汇总",            "适用": "市场调研、竞品分析、价格监控"        },        {            "模式": "表单自动填写",            "场景": "需要重复填写各种在线表单",            "方式": "传入表单数据和URL，自动填写提交",            "适用": "报名、注册、订单提交"        },        {            "模式": "端到端测试",            "场景": "Web应用的自动化测试",            "方式": "用自然语言描述测试用例，自动执行验证",            "适用": "快速测试验证、QA辅助"        }    ]        for p in patterns:        print(f"\n  【{p['模式']}】")        print(f"    场景: {p['场景']}")        print(f"    方式: {p['方式']}")        print(f"    适用: {p['适用']}")        print(f"\n{'=' * 60}")  if __name__ == "__main__":    demo_ecommerce_price_comparison()    demo_integration_patterns()