""" Ultimate Web Agent – Fully Functional """ import os import json import asyncio import base64 import re import uuid import logging from fastapi import FastAPI, HTTPException from fastapi.middleware.cors import CORSMiddleware from fastapi.responses import HTMLResponse, JSONResponse from pydantic import BaseModel from playwright.async_api import async_playwright from groq import AsyncGroq import httpx logging.basicConfig(level=logging.INFO) logger = logging.getLogger("WebAgent") # ---------- API Keys ---------- GROQ_KEYS = [ "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", "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" ] OPENROUTER_KEYS = ["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"] # ---------- LLM Client ---------- class LLMClient: _idx = 0 _lock = asyncio.Lock() @classmethod async def chat(cls, messages, temp=0.1, max_tokens=1024): # Try Groq first try: async with cls._lock: key = GROQ_KEYS[cls._idx % len(GROQ_KEYS)] cls._idx += 1 client = AsyncGroq(api_key=key) completion = await client.chat.completions.create( model="llama-3.1-8b-instant", messages=messages, temperature=temp, max_tokens=max_tokens, response_format={"type": "json_object"} ) return completion.choices[0].message.content except Exception as e: logger.warning(f"Groq failed: {e}") # Fallback to OpenRouter try: async with cls._lock: key = OPENROUTER_KEYS[cls._idx % len(OPENROUTER_KEYS)] cls._idx += 1 async with httpx.AsyncClient(timeout=30) as client: resp = await client.post( "https://openrouter.ai/api/v1/chat/completions", headers={"Authorization": f"Bearer {key}", "Content-Type": "application/json"}, json={ "model": "meta-llama/llama-3.1-8b-instruct", "messages": messages, "temperature": temp, "max_tokens": max_tokens } ) resp.raise_for_status() data = resp.json() if "choices" in data and data["choices"]: return data["choices"][0]["message"]["content"] except Exception as e: logger.warning(f"OpenRouter failed: {e}") # Ultimate fallback return json.dumps({ "thought": "LLM unavailable, using fallback", "action": {"tool": "done", "params": {"result": "LLM error"}} }) # ---------- Tool Registry ---------- class Tools: def __init__(self, page, state): self.page = page self.state = state async def navigate(self, url): await self.page.goto(url, timeout=30000) return {"result": f"Navigated to {url}"} async def click(self, selector): await self.page.click(selector, timeout=5000) return {"result": f"Clicked {selector}"} async def fill(self, selector, value): await self.page.fill(selector, value, timeout=5000) return {"result": f"Filled {selector}"} async def get_text(self, selector): text = await self.page.text_content(selector, timeout=5000) return {"result": text.strip() if text else ""} async def screenshot(self): ss = await self.page.screenshot(full_page=False) return {"result": base64.b64encode(ss).decode()} async def done(self, result="Task completed."): self.state.done = True self.state.result = result return {"result": result} # ---------- Agent ---------- class Agent: def __init__(self, tools, state): self.tools = tools self.state = state self.history = [] async def run(self, task): self.state.task = task for step in range(8): # max 8 steps if self.state.done: break # Build context last_result = self.history[-1]['result'] if self.history else 'None' context = f"Task: {task}\nLast result: {last_result}" sys_prompt = ( "You are a web automation agent. Output JSON with 'thought' and 'action'. " "Tools: navigate, click, fill, get_text, screenshot, done. " "Example: {\"action\":{\"tool\":\"navigate\",\"params\":{\"url\":\"https://www.wikipedia.org\"}}}" ) raw = await LLMClient.chat([ {"role": "system", "content": sys_prompt}, {"role": "user", "content": context} ]) try: data = json.loads(raw) action = data.get("action", {"tool": "done", "params": {"result": "No action"}}) thought = data.get("thought", "") except: action = {"tool": "done", "params": {"result": "Invalid LLM output"}} thought = "Invalid output" tool = action.get("tool") params = action.get("params", {}) if tool == "done": self.state.done = True self.state.result = params.get("result", "Task completed") try: ss = await self.tools.screenshot() result = {"result": self.state.result, "screenshot": ss["result"]} except: result = {"result": self.state.result} else: try: result = await getattr(self.tools, tool)(**params) ss = await self.tools.screenshot() result["screenshot"] = ss["result"] except Exception as e: result = {"error": str(e)} self.history.append({ "thought": thought, "action": action, "result": result }) return { "history": self.history, "result": self.state.result, "steps": len(self.history) } # ---------- FastAPI ---------- app = FastAPI() app.add_middleware( CORSMiddleware, allow_origins=["*"], allow_credentials=True, allow_methods=["*"], allow_headers=["*"], ) sessions = {} status = {} async def run_agent(sid, task, headless): try: playwright = await async_playwright().start() browser = await playwright.chromium.launch(headless=headless) context = await browser.new_context(viewport={"width": 1280, "height": 720}) page = await context.new_page() await page.goto("about:blank") state = type('State', (), {'variables': {}, 'done': False, 'result': None, 'task': task})() tools = Tools(page, state) agent = Agent(tools, state) result = await agent.run(task) status[sid] = {"status": "completed", "result": result, "progress": 100} await browser.close() except Exception as e: status[sid] = {"status": "error", "error": str(e)} class StartRequest(BaseModel): task: str headless: bool = True @app.post("/start") async def start(req: StartRequest): sid = str(uuid.uuid4()) status[sid] = {"status": "running", "progress": 0} asyncio.create_task(run_agent(sid, req.task, req.headless)) return {"session_id": sid} @app.get("/status/{sid}") async def get_status(sid: str): if sid not in status: raise HTTPException(404, "Session not found") return JSONResponse(status[sid]) @app.post("/stop/{sid}") async def stop(sid: str): if sid in status: status[sid]["status"] = "stopped" return {"status": "stopped"} return {"status": "not found"} # ---------- UI ---------- HTML = """ Web Agent

🤖 Web Agent

Ready
×
""" @app.get("/", response_class=HTMLResponse) async def root(): return HTMLResponse(HTML) if __name__ == "__main__": import uvicorn uvicorn.run(app, host="0.0.0.0", port=7860)