HyperCluster/Fara-BrowserUse
5
1"""
2FARA Backend Server for HuggingFace Space
3Provides WebSocket communication and REST API for the React frontend
4"""
5
6import asyncio
7import base64
8import logging
9import os
10
11# Import FARA components
12import sys
13import tempfile
14import uuid
15from datetime import datetime
16from typing import Dict, Optional
17
18import httpx
19from fastapi import FastAPI, WebSocket, WebSocketDisconnect
20from fastapi.middleware.cors import CORSMiddleware
21from fastapi.responses import JSONResponse
22from playwright._impl._errors import TargetClosedError
23
24sys.path.insert(0, "/app")
25from fara import FaraAgent
26from fara.browser.browser_bb import BrowserBB
27
28# Configure logging
29logging.basicConfig(level=logging.INFO)
30logger = logging.getLogger(__name__)
31
32# Modal trace storage configuration
33MODAL_TRACE_STORAGE_URL = os.environ.get("MODAL_TRACE_STORAGE_URL", "")
34MODAL_TOKEN_ID = os.environ.get("MODAL_TOKEN_ID", "")
35MODAL_TOKEN_SECRET = os.environ.get("MODAL_TOKEN_SECRET", "")
36
37# Modal vLLM endpoint configuration (from environment variables for HF Spaces)
38# Includes proxy auth headers for authenticated Modal endpoints
39ENDPOINT_CONFIG = {
40 "model": os.environ.get("FARA_MODEL_NAME", "microsoft/Fara-7B"),
41 "base_url": os.environ.get("FARA_ENDPOINT_URL"),
42 "api_key": os.environ.get("FARA_API_KEY", "not-needed"),
43 "default_headers": {
44 "Modal-Key": MODAL_TOKEN_ID,
45 "Modal-Secret": MODAL_TOKEN_SECRET,
46 }
47 if MODAL_TOKEN_ID and MODAL_TOKEN_SECRET
48 else None,
49}
50
51# Available models (for the frontend dropdown)
52AVAILABLE_MODELS = ["microsoft/Fara-7B"]
53
54app = FastAPI(title="FARA Backend")
55
56# CORS middleware
57app.add_middleware(
58 CORSMiddleware,
59 allow_origins=["*"],
60 allow_credentials=True,
61 allow_methods=["*"],
62 allow_headers=["*"],
63)
64
65# Store active connections and their sessions
66active_connections: Dict[str, WebSocket] = {}
67active_sessions: Dict[str, "FaraSession"] = {}
68
69
70class FaraSession:
71 """Manages a single FARA agent session"""
72
73 def __init__(self, trace_id: str, websocket: WebSocket):
74 self.trace_id = trace_id
75 self.websocket = websocket
76 self.agent: Optional[FaraAgent] = None
77 self.browser_manager: Optional[BrowserBB] = None
78 self.screenshots_dir: Optional[str] = None
79 self.is_running = False
80 self.should_stop = False
81 self.step_count = 0
82 self.start_time: Optional[datetime] = None
83 self.total_input_tokens = 0
84 self.total_output_tokens = 0
85
86 async def initialize(self, start_page: str = "https://www.bing.com/"):
87 """Initialize the browser and agent"""
88 # Create temp directory for screenshots
89 self.screenshots_dir = tempfile.mkdtemp(prefix="fara_screenshots_")
90
91 # Initialize browser manager (headless for HF Space)
92 self.browser_manager = BrowserBB(
93 headless=True,
94 viewport_height=900,
95 viewport_width=1440,
96 page_script_path=None,
97 browser_channel="chromium",
98 browser_data_dir=None,
99 downloads_folder=self.screenshots_dir,
100 to_resize_viewport=True,
101 single_tab_mode=True,
102 animate_actions=False,
103 use_browser_base=False,
104 logger=logger,
105 )
106
107 self.agent = FaraAgent(
108 browser_manager=self.browser_manager,
109 client_config=ENDPOINT_CONFIG,
110 start_page=start_page,
111 downloads_folder=self.screenshots_dir,
112 save_screenshots=True,
113 max_rounds=50,
114 )
115
116 await self.agent.initialize()
117 return True
118
119 async def send_event(self, event: dict):
120 """Send event to the connected WebSocket"""
121 try:
122 await self.websocket.send_json(event)
123 except Exception as e:
124 logger.error(f"Error sending event: {e}")
125
126 async def get_screenshot_base64(self) -> Optional[str]:
127 """Get the current browser screenshot as base64"""
128 if self.agent:
129 try:
130 # Get the current active page from the browser context
131 page = self._get_active_page()
132 if page:
133 screenshot_bytes = (
134 await self.agent._playwright_controller.get_screenshot(page)
135 )
136 return f"data:image/png;base64,{base64.b64encode(screenshot_bytes).decode()}"
137 except TargetClosedError:
138 logger.warning(
139 "Page closed while getting screenshot, attempting recovery..."
140 )
141 page = self._get_active_page()
142 if page:
143 try:
144 screenshot_bytes = (
145 await self.agent._playwright_controller.get_screenshot(page)
146 )
147 return f"data:image/png;base64,{base64.b64encode(screenshot_bytes).decode()}"
148 except Exception as e:
149 logger.error(f"Recovery screenshot failed: {e}")
150 except Exception as e:
151 logger.error(f"Error getting screenshot: {e}")
152 return None
153
154 def _get_active_page(self):
155 """Get the currently active page from the browser context"""
156 if (
157 self.agent
158 and self.agent.browser_manager
159 and self.agent.browser_manager._context
160 ):
161 pages = self.agent.browser_manager._context.pages
162 if pages:
163 # Return the last (most recent) page, or the one marked as active
164 return pages[-1]
165 return self.agent._page if self.agent else None
166
167 async def run_task(self, instruction: str, model_id: str):
168 """Run a task and stream results via WebSocket"""
169 self.is_running = True
170 self.should_stop = False
171 self.step_count = 0
172 self.start_time = datetime.now()
173 self.total_input_tokens = 0
174 self.total_output_tokens = 0
175
176 try:
177 # Send agent_start event
178 await self.send_event(
179 {
180 "type": "agent_start",
181 "agentTrace": {
182 "id": self.trace_id,
183 "instruction": instruction,
184 "modelId": model_id,
185 "timestamp": self.start_time.isoformat(),
186 "isRunning": True,
187 "traceMetadata": {
188 "traceId": self.trace_id,
189 "inputTokensUsed": 0,
190 "outputTokensUsed": 0,
191 "duration": 0,
192 "numberOfSteps": 0,
193 "maxSteps": 50,
194 "completed": False,
195 },
196 },
197 }
198 )
199
200 # Initialize agent
201 await self.initialize()
202
203 # Get initial screenshot
204 initial_screenshot = await self.get_screenshot_base64()
205
206 # Run the agent with custom loop to stream progress
207 await self._run_agent_with_streaming(instruction)
208
209 except Exception as e:
210 logger.exception("Error running agent task")
211 await self.send_event({"type": "agent_error", "error": str(e)})
212 finally:
213 self.is_running = False
214 await self.close()
215
216 async def _run_agent_with_streaming(self, user_message: str):
217 """Run the agent and stream each step to the frontend"""
218 agent = self.agent
219
220 # Initialize if not already done
221 await agent.initialize()
222 assert agent._page is not None, "Page should be initialized"
223
224 # Get initial screenshot
225 scaled_screenshot = await agent._get_scaled_screenshot()
226
227 if agent.save_screenshots:
228 await agent._playwright_controller.get_screenshot(
229 agent._page,
230 path=os.path.join(
231 agent.downloads_folder, f"screenshot{agent._num_actions}.png"
232 ),
233 )
234
235 # Add user message to chat history
236 from fara.types import ImageObj, UserMessage
237
238 agent._chat_history.append(
239 UserMessage(
240 content=[ImageObj.from_pil(scaled_screenshot), user_message],
241 is_original=True,
242 )
243 )
244
245 final_answer = "<no_answer>"
246 is_stop_action = False
247
248 for i in range(agent.max_rounds):
249 if self.should_stop:
250 # User requested stop
251 await self.send_event(
252 {
253 "type": "agent_complete",
254 "traceMetadata": self._get_metadata(),
255 "final_state": "stopped",
256 }
257 )
258 return
259
260 is_first_round = i == 0
261 step_start_time = datetime.now()
262
263 # Wait for captcha if needed
264 if not agent.browser_manager._captcha_event.is_set():
265 logger.info("Waiting 60s for captcha to finish...")
266 captcha_solved = await agent.wait_for_captcha_with_timeout(60)
267 if (
268 not captcha_solved
269 and not agent.browser_manager._captcha_event.is_set()
270 ):
271 raise RuntimeError("Captcha timed out")
272
273 try:
274 # Generate model response
275 function_call, raw_response = await agent.generate_model_call(
276 is_first_round, scaled_screenshot if is_first_round else None
277 )
278
279 # Parse response
280 thoughts, action_dict = agent._parse_thoughts_and_action(raw_response)
281 action_args = action_dict.get("arguments", {})
282 action = action_args["action"]
283
284 logger.info(
285 f"\nThought #{i + 1}: {thoughts}\nAction #{i + 1}: {action}"
286 )
287
288 # Execute action with recovery for page changes
289 try:
290 (
291 is_stop_action,
292 new_screenshot,
293 action_description,
294 ) = await agent.execute_action(function_call)
295 except TargetClosedError as e:
296 logger.warning(
297 "Page closed during action execution, attempting recovery..."
298 )
299 # Try to recover the page reference
300 new_page = self._get_active_page()
301 if new_page and new_page != agent._page:
302 logger.info("Recovered with new active page")
303 agent._page = new_page
304 # Wait for the page to stabilize
305 await asyncio.sleep(1)
306 action_description = (
307 "Action completed (page navigation occurred)"
308 )
309 is_stop_action = False
310 new_screenshot = None
311 else:
312 raise e
313
314 # Sync the agent's page reference with the active page
315 active_page = self._get_active_page()
316 if active_page and active_page != agent._page:
317 logger.info("Updating agent page reference to active page")
318 agent._page = active_page
319
320 # Get screenshot for this step
321 screenshot_base64 = await self.get_screenshot_base64()
322
323 except TargetClosedError as e:
324 logger.error(f"Unrecoverable page error: {e}")
325 await self.send_event(
326 {
327 "type": "agent_error",
328 "error": f"Browser page closed unexpectedly: {str(e)}",
329 }
330 )
331 return
332 except Exception as e:
333 logger.exception(f"Error in agent step {i + 1}")
334 await self.send_event({"type": "agent_error", "error": str(e)})
335 return
336
337 # Calculate step duration and tokens (estimated)
338 step_duration = (datetime.now() - step_start_time).total_seconds()
339 step_input_tokens = 1000 # Estimated
340 step_output_tokens = len(raw_response) // 4 # Rough estimate
341
342 self.total_input_tokens += step_input_tokens
343 self.total_output_tokens += step_output_tokens
344 self.step_count += 1
345
346 # Create step object
347 step = {
348 "stepId": str(uuid.uuid4()),
349 "traceId": self.trace_id,
350 "stepNumber": self.step_count,
351 "thought": thoughts,
352 "actions": [
353 {
354 "function_name": action,
355 "description": action_description,
356 "parameters": action_args,
357 }
358 ],
359 "image": screenshot_base64,
360 "duration": step_duration,
361 "inputTokensUsed": step_input_tokens,
362 "outputTokensUsed": step_output_tokens,
363 "timestamp": datetime.now().isoformat(),
364 }
365
366 # Send progress event
367 await self.send_event(
368 {
369 "type": "agent_progress",
370 "agentStep": step,
371 "traceMetadata": self._get_metadata(),
372 }
373 )
374
375 if is_stop_action:
376 final_answer = thoughts
377 break
378
379 # Send completion event
380 final_state = "success" if is_stop_action else "max_steps_reached"
381 await self.send_event(
382 {
383 "type": "agent_complete",
384 "traceMetadata": self._get_metadata(completed=True),
385 "final_state": final_state,
386 }
387 )
388
389 def _get_metadata(self, completed: bool = False) -> dict:
390 """Get current trace metadata"""
391 duration = 0
392 if self.start_time:
393 duration = (datetime.now() - self.start_time).total_seconds()
394
395 return {
396 "traceId": self.trace_id,
397 "inputTokensUsed": self.total_input_tokens,
398 "outputTokensUsed": self.total_output_tokens,
399 "duration": duration,
400 "numberOfSteps": self.step_count,
401 "maxSteps": 50,
402 "completed": completed,
403 }
404
405 async def stop(self):
406 """Request the agent to stop"""
407 self.should_stop = True
408
409 async def close(self):
410 """Clean up resources"""
411 if self.agent:
412 try:
413 await self.agent.close()
414 except Exception as e:
415 logger.error(f"Error closing agent: {e}")
416 self.agent = None
417 self.browser_manager = None
418
419 if self.screenshots_dir and os.path.exists(self.screenshots_dir):
420 import shutil
421
422 try:
423 shutil.rmtree(self.screenshots_dir)
424 except Exception as e:
425 logger.error(f"Error cleaning up screenshots: {e}")
426 self.screenshots_dir = None
427
428
429@app.get("/api/models")
430async def get_models():
431 """Return available models"""
432 return JSONResponse(content=AVAILABLE_MODELS)
433
434
435@app.post("/api/traces")
436async def store_trace(trace_data: dict):
437 """
438 Store a task trace by forwarding to the Modal trace storage endpoint.
439 This keeps Modal credentials on the server side.
440 """
441 if not MODAL_TRACE_STORAGE_URL:
442 logger.warning("Modal trace storage URL not configured")
443 return JSONResponse(
444 status_code=503,
445 content={"success": False, "error": "Trace storage not configured"},
446 )
447
448 if not MODAL_TOKEN_ID or not MODAL_TOKEN_SECRET:
449 logger.warning("Modal proxy auth credentials not configured")
450 return JSONResponse(
451 status_code=503,
452 content={"success": False, "error": "Modal auth not configured"},
453 )
454
455 try:
456 async with httpx.AsyncClient(timeout=30.0) as client:
457 response = await client.post(
458 MODAL_TRACE_STORAGE_URL,
459 json=trace_data,
460 headers={
461 "Content-Type": "application/json",
462 "Modal-Key": MODAL_TOKEN_ID,
463 "Modal-Secret": MODAL_TOKEN_SECRET,
464 },
465 )
466
467 if response.status_code == 200:
468 result = response.json()
469 logger.info(
470 f"Trace stored successfully: {result.get('trace_id', 'unknown')}"
471 )
472 return JSONResponse(content=result)
473 else:
474 error_text = response.text
475 logger.error(
476 f"Failed to store trace: {response.status_code} - {error_text}"
477 )
478 return JSONResponse(
479 status_code=response.status_code,
480 content={
481 "success": False,
482 "error": f"Modal API error: {error_text}",
483 },
484 )
485 except httpx.TimeoutException:
486 logger.error("Timeout storing trace to Modal")
487 return JSONResponse(
488 status_code=504,
489 content={"success": False, "error": "Timeout connecting to trace storage"},
490 )
491 except Exception as e:
492 logger.exception("Error storing trace")
493 return JSONResponse(
494 status_code=500, content={"success": False, "error": str(e)}
495 )
496
497
498@app.get("/api/random-question")
499async def get_random_question():
500 """Return a random example question"""
501 questions = [
502 "Search for the latest news about AI agents",
503 "Find the weather forecast for San Francisco",
504 "Go to GitHub and search for 'computer use agent'",
505 "Find the top trending repositories on GitHub today",
506 "Search for Python tutorials on YouTube",
507 "Look up the current stock price of Microsoft",
508 "Find the schedule for upcoming SpaceX launches",
509 "Search for healthy breakfast recipes",
510 ]
511 import random
512
513 return JSONResponse(content={"question": random.choice(questions)})
514
515
516@app.websocket("/ws")
517async def websocket_endpoint(websocket: WebSocket):
518 """WebSocket endpoint for real-time communication"""
519 await websocket.accept()
520
521 # Generate a unique connection ID
522 connection_id = str(uuid.uuid4())
523 active_connections[connection_id] = websocket
524
525 # Send heartbeat with the connection ID (used as trace ID base)
526 trace_id = str(uuid.uuid4())
527 await websocket.send_json(
528 {"type": "heartbeat", "uuid": trace_id, "timestamp": datetime.now().isoformat()}
529 )
530
531 try:
532 while True:
533 # Wait for messages from the client
534 data = await websocket.receive_json()
535 message_type = data.get("type")
536
537 if message_type == "user_task":
538 # Extract task details
539 trace = data.get("trace", {})
540 trace_id = trace.get("id", str(uuid.uuid4()))
541 instruction = trace.get("instruction", "")
542 model_id = trace.get("modelId", "microsoft/Fara-7B")
543
544 # Create and start session
545 session = FaraSession(trace_id, websocket)
546 active_sessions[trace_id] = session
547
548 # Run the task in the background
549 asyncio.create_task(session.run_task(instruction, model_id))
550
551 elif message_type == "stop_task":
552 # Stop the running task
553 trace_id = data.get("trace_id")
554 if trace_id and trace_id in active_sessions:
555 await active_sessions[trace_id].stop()
556
557 elif message_type == "ping":
558 await websocket.send_json({"type": "pong"})
559
560 except WebSocketDisconnect:
561 logger.info(f"WebSocket disconnected: {connection_id}")
562 except Exception as e:
563 logger.exception(f"WebSocket error: {e}")
564 finally:
565 # Clean up
566 if connection_id in active_connections:
567 del active_connections[connection_id]
568
569 # Clean up any sessions for this connection
570 sessions_to_remove = []
571 for trace_id, session in active_sessions.items():
572 if session.websocket == websocket:
573 await session.close()
574 sessions_to_remove.append(trace_id)
575 for trace_id in sessions_to_remove:
576 del active_sessions[trace_id]
577
578
579@app.get("/api/health")
580async def health_check():
581 """Health check endpoint"""
582 return {"status": "healthy"}
583
584
585if __name__ == "__main__":
586 import uvicorn
587
588 uvicorn.run(app, host="0.0.0.0", port=8000)
589