""" Browser Use REST API Server Простой HTTP API для управления браузером из n8n """ import os import asyncio from typing import Optional from fastapi import FastAPI, HTTPException from pydantic import BaseModel from browser_use import Agent, Browser, ChatOpenAI from browser_use.agent.views import AgentHistoryList from browser_use.browser.context import BrowserContextConfig, BrowserConfig import uvicorn from fastapi.responses import HTMLResponse # Настройки Remote Debugging REMOTE_DEBUGGING_PORT = int(os.getenv("REMOTE_DEBUGGING_PORT", "9222")) REMOTE_DEBUGGING_ENABLED = os.getenv("REMOTE_DEBUGGING_ENABLED", "false").lower() == "true" app = FastAPI(title="Browser Use API", version="1.0.0") # Модели запросов class TaskRequest(BaseModel): task: str max_steps: Optional[int] = 20 model: Optional[str] = DEFAULT_MODEL class TaskResponse(BaseModel): task_id: str status: str result: Optional[str] = None error: Optional[str] = None # Хранилище задач tasks = {} @app.post("/api/tasks", response_model=TaskResponse) async def create_task(request: TaskRequest): """Создать новую задачу браузера""" import uuid task_id = str(uuid.uuid4())[:8] tasks[task_id] = { "status": "pending", "task": request.task, "result": None } # Запускаем задачу в фоне asyncio.create_task(run_browser_task(task_id, request.task, request.max_steps, request.model)) return TaskResponse( task_id=task_id, status="pending" ) async def run_browser_task(task_id: str, task: str, max_steps: int, model: str): """Выполнение задачи браузера""" browser = None try: tasks[task_id]["status"] = "running" # Настройка LLM через OpenRouter api_key = os.getenv("OPENROUTER_API_KEY", "") if not api_key: raise ValueError("OPENROUTER_API_KEY not set") # Используем ChatOpenAI с base_url для OpenRouter llm = ChatOpenAI( model=model, base_url="https://openrouter.ai/api/v1", api_key=api_key ) # Настройки браузера browser_config = BrowserConfig( user_data_dir="/app/browser-data", profile_directory="Default", headless=True, ) # Добавляем remote debugging если включено if REMOTE_DEBUGGING_ENABLED: browser_config.extra_chromium_args = [ f"--remote-debugging-port={REMOTE_DEBUGGING_PORT}", "--remote-debugging-address=0.0.0.0" ] browser = Browser(config=browser_config) # Создание агента agent = Agent( task=task, llm=llm, browser=browser, ) # Выполнение задачи history: AgentHistoryList = await agent.run(max_steps=max_steps) # Получение результата result = history.final_result() tasks[task_id]["status"] = "completed" tasks[task_id]["result"] = result except Exception as e: tasks[task_id]["status"] = "failed" tasks[task_id]["error"] = str(e) finally: # Закрытие браузера (правильный способ) if browser: try: # browser_use сам управляет закрытием pass except: pass @app.get("/api/tasks/{task_id}", response_model=TaskResponse) async def get_task(task_id: str): """Получить статус/результат задачи""" if task_id not in tasks: raise HTTPException(status_code=404, detail="Task not found") task = tasks[task_id] return TaskResponse( task_id=task_id, status=task["status"], result=task.get("result"), error=task.get("error") ) @app.get("/api/tasks") async def list_tasks(): """Получить список всех задач""" return { "tasks": [ { "task_id": task_id, "status": task["status"], "task": task.get("task", ""), "result": task.get("result"), "error": task.get("error") } for task_id, task in tasks.items() ] } @app.get("/api/health") async def health_check(): """Проверка здоровья сервера""" return { "status": "healthy", "service": "browser-use-api", "version": "1.0.0" } # Web UI - Главная страница WEB_UI_HTML = """ Browser Use API

Browser Use API

REST API для автоматизации браузера с помощью AI

Быстрые ссылки

Открыть Dashboard

API Endpoints

Пример использования

curl -X POST http://localhost:8383/api/tasks \\
  -H "Content-Type: application/json" \\
  -d '{"task": "Перейди на example.com", "max_steps": 20}'
""" DASHBOARD_HTML = """ Browser Use Dashboard

Browser Use Dashboard

Создать задачу

Задачи

""" @app.get("/", response_class=HTMLResponse) async def root(): """Главная страница""" return HTMLResponse(content=WEB_UI_HTML) @app.get("/dashboard", response_class=HTMLResponse) async def dashboard(): """Dashboard""" return HTMLResponse(content=DASHBOARD_HTML) if __name__ == "__main__": uvicorn.run(app, host="0.0.0.0", port=8383)