begon impl for support of api endpoint

ae/core/notification_manager.py

@@ -0,0 +1,70 @@
+from collections.abc import Callable
+from queue import Empty
+from queue import Queue
+
+
+class NotificationManager:
+    """
+    NotificationManager handles the dispatching of notifications to registered listeners.
+
+    Attributes:
+        notification_queue (Queue): A queue to hold notifications.
+        listeners (list[Callable[[dict[str, str]], None]]): A list of listener callbacks to notify.
+    """
+
+    def __init__(self):
+        """
+        Initialize the NotificationManager with an empty queue and no listeners.
+        """
+        self.notification_queue = Queue() # type: ignore
+        self.listeners: list[Callable[[dict[str, str]], None]] = []
+
+    def notify(self, message: str, message_type: str) -> None:
+        """
+        Notify all registered listeners with a message and its type.
+
+        Args:
+            message (str): The message to notify.
+            message_type (str): The type of the message.
+        """
+        notification = {
+            "message": message,
+            "type": message_type,
+        }
+
+        if self.listeners:
+            self.notification_queue.put(notification) # type: ignore
+            for listener in self.listeners:
+                listener(notification)
+        else:
+            print(f"No listeners available, discarding message: {notification}")
+
+    def get_next_notification(self) -> dict[str, str] | None:
+        """
+        Get the next notification from the queue, if available.
+
+        Returns:
+            dict[str, str] | None: The next notification, or None if the queue is empty.
+        """
+        try:
+            return self.notification_queue.get_nowait() # type: ignore
+        except Empty:
+            return None
+
+    def register_listener(self, listener: Callable[[dict[str, str]], None]) -> None:
+        """
+        Register a new listener to receive notifications.
+
+        Args:
+            listener (Callable[[dict[str, str]], None]): The listener callback to register.
+        """
+        self.listeners.append(listener)
+
+    def unregister_listener(self, listener: Callable[[dict[str, str]], None]) -> None:
+        """
+        Unregister a listener from receiving notifications.
+
+        Args:
+            listener (Callable[[dict[str, str]], None]): The listener callback to unregister.
+        """
+        self.listeners.remove(listener)

ae/core/playwright_manager.py

@@ -8,6 +8,7 @@
 from playwright.async_api import Page
 from playwright.async_api import Playwright
 
+from ae.core.notification_manager import NotificationManager
 from ae.core.ui_manager import UIManager
 from ae.utils.dom_mutation_observer import dom_mutation_change_detected
 from ae.utils.dom_mutation_observer import handle_navigation_for_mutation_observer
@@ -62,6 +63,7 @@ def __init__(self, browser_type: str = "chromium", headless: bool = False, gui_i
         self.browser_type = browser_type
         self.isheadless = headless
         self.__initialized = True
+        self.notification_manager = NotificationManager()
         self.user_response_event = asyncio.Event()
         if gui_input_mode:
             self.ui_manager: UIManager = UIManager()
@@ -318,6 +320,8 @@ async def notify_user(self, message: str, message_type: MessageType = MessageTyp
         except Exception as e:
             logger.error(f"Failed to notify user with message \"{message}\". However, most likey this will work itself out after the page loads: {e}")
 
+        self.notification_manager.notify(message, message_type.value)
+
     async def highlight_element(self, selector: str, add_highlight: bool):
         try:
             page: Page = await self.get_current_page()

ae/server/api_routes.py

@@ -0,0 +1,133 @@
+import asyncio
+import json
+import logging
+import os
+
+import uvicorn
+from fastapi import FastAPI
+from fastapi import Request
+from fastapi.middleware.cors import CORSMiddleware
+from fastapi.responses import StreamingResponse
+from pydantic import BaseModel
+from pydantic import Field
+
+import ae.core.playwright_manager as browserManager
+from ae.core.autogen_wrapper import AutogenWrapper
+from ae.utils.ui_messagetype import MessageType
+
+browser_manager = browserManager.PlaywrightManager(headless=False)
+
+APP_VERSION = "1.0.0"
+APP_NAME = "Agent-E Web API"
+API_PREFIX = "/api"
+IS_DEBUG = False
+HOST = os.getenv("HOST", "0.0.0.0")
+PORT = int(os.getenv("PORT", 8080))
+WORKERS = 1
+
+class CommandQueryModel(BaseModel):
+    command: str = Field(..., description="The command related to web navigation to execute.") # Required field with description
+
+
+def get_app() -> FastAPI:
+    '''Starts the Application'''
+    fast_app = FastAPI(
+        title=APP_NAME,
+        version=APP_VERSION,
+        debug=IS_DEBUG)
+
+    fast_app.add_middleware(
+        CORSMiddleware,
+        allow_origins=["*"],
+        allow_credentials=True,
+        allow_methods=["*"],
+        allow_headers=["*"])
+
+    return fast_app
+
+app = get_app()
+
+@app.on_event("startup") # type: ignore
+async def startup_event():
+    """
+    Startup event handler to initialize browser manager asynchronously.
+    """
+    await browser_manager.async_initialize()
+
+
+@app.post("/execute_task", description="Execute a given command related to web navigation and return the result.")
+async def execute_task(request: Request,
+                       query_model: CommandQueryModel):
+
+    return StreamingResponse(run_task(query_model.command, browser_manager), media_type='application/json')
+
+
+def run_task(command: str, playwright_manager: browserManager.PlaywrightManager):
+    """
+    Run the task to process the command and generate events.
+
+    Args:
+        command (str): The command to execute.
+        playwright_manager (PlaywrightManager): The manager handling browser interactions and notifications.
+
+    Yields:
+        str: JSON-encoded string representing a notification.
+    """
+    async def event_generator():
+        # Start the process command task
+        task = asyncio.create_task(process_command(command, playwright_manager))
+
+        while not task.done():
+            notification = playwright_manager.notification_manager.get_next_notification()
+            if notification:
+                yield f"{json.dumps(notification)}\n"
+            await asyncio.sleep(0.1)
+
+        # Once the task is done, yield any remaining notifications
+        while True:
+            notification = playwright_manager.notification_manager.get_next_notification()
+            if notification:
+                yield f"{json.dumps(notification)}\n"
+            else:
+                break
+
+        # Ensure the task is awaited to propagate any exceptions
+        await task
+
+    return event_generator()
+
+
+async def process_command(command: str, playwright_manager: browserManager.PlaywrightManager):
+    """
+    Process the command and send notifications.
+
+    Args:
+        command (str): The command to process.
+        playwright_manager (PlaywrightManager): The manager handling browser interactions and notifications.
+    """
+    current_url = await playwright_manager.get_current_url()
+    await playwright_manager.notify_user("Processing command", MessageType.INFO)
+
+    ag = await AutogenWrapper.create()
+    command_exec_result = await ag.process_command(command, current_url)
+
+    # TODO: See how to extract the actual final answer from here
+    final_answer = "Final answer"
+
+    # Notify about the completion of the command
+    await playwright_manager.notify_user("Command completed", MessageType.INFO)
+
+    await playwright_manager.notify_user(final_answer, MessageType.FINAL)
+
+# Configure logging
+logging.basicConfig(level=logging.INFO, format="%(asctime)s - %(name)s - %(levelname)s - %(message)s")
+logger = logging.getLogger("uvicorn")
+
+
+if __name__ == "__main__":
+    logger.info('**********Application Started**********')
+    uvicorn.run(
+        "main:app",
+        host=HOST,
+        port=PORT,
+        workers=WORKERS, reload=IS_DEBUG, log_level="info")

ae/utils/ui_messagetype.py

@@ -7,5 +7,6 @@ class MessageType(Enum):
     STEP = "step"
     ACTION ="action"
     ANSWER = "answer"
-    QUESTION= "question"
+    QUESTION = "question"
     INFO = "info"
+    FINAL = "final"

pyproject.toml

@@ -23,7 +23,9 @@ dependencies = [
     "pydantic==2.6.2",
     "python-dotenv==1.0.0",
     "tabulate==0.9.0",
-    "nest-asyncio==1.6.0"
+    "nest-asyncio==1.6.0",
+    "fastapi==0.111.1",
+    "uvicorn==0.30.3"
 ]
 
 [project.optional-dependencies]