Upload 28 files

components/Web_agent_related/README.md

@@ -0,0 +1,81 @@
+# WebAgent
+
+A modular, LLM-powered autonomous web research agent for integration into larger systems.
+
+## Core Functionality
+
+- Performs autonomous web research using LLMs (Claude, OpenAI, etc.)
+- Provides unified access to web browsing, search, and document parsing
+- Returns structured research reports
+
+## Integration
+
+```python
+from web_agent import WebAgent
+from llama_index.llms.anthropic import Anthropic
+
+# 1. Initialize agent with preferred LLM
+agent = WebAgent(
+    llm=Anthropic(model="claude-opus-4-20250514", api_key="your_anthropic_key"),
+    google_api_key="your_google_key",  # Optional: pass directly or via config
+    github_token="your_github_token",  # Optional: pass directly or via config
+    max_research_iterations=100000,    # Customize research depth
+    verbose=False                      # Toggle debugging output
+)
+
+# 2. Execute research and get report
+report = agent.research("Your research query")
+
+# 3. Use the report in your application
+process_research_results(report)
+```
+
+## Example Research Output
+
+```python
+# Query: "what is alita generalist agent"
+report = agent.research("what is alita generalist agent")
+```
+
+The agent will autonomously conduct research and produce a comprehensive textual report:
+
+```json
+{
+  "summary": "ALITA (Artificial Learning Intelligence Training Assistant) represents a significant advancement in the field of generalist AI agents. It is capable of performing diverse tasks across multiple domains without requiring task-specific training, marking a departure from traditional narrow AI systems.",
+  "detailed_findings": "ALITA is a multi-modal AI system designed to process and understand various types of input (text, images, audio, video), generate contextually appropriate responses across different domains, learn and adapt from interactions without explicit retraining, and transfer knowledge between disparate tasks. Key characteristics include versatility, adaptability, and integration capabilities.",
+  "sources": [
+    "https://example.com/alita-overview",
+    "https://example.com/generalist-ai"
+  ]
+}
+```
+
+## Configuration Options
+
+Configure via constructor parameters or `config.py`:
+
+- **LLM Integration**: Pass any LlamaIndex-compatible LLM
+- **API Keys**: Set via constructor, environment variables, or `api_keys.py`
+- **Behavior Control**: Adjust `max_research_iterations`, `verbose`, etc.
+
+## Available Tools
+
+The agent exposes methods that can be called directly or used by the LLM:
+
+- Web browsing: `visit_url`, `page_up/down`, `get_full_page_text`
+- Search: `search_google`, `search_github`  
+- Document handling: `download_file`, extraction for PDF/DOCX/PPTX
+
+## Requirements
+
+Python 3.8+, LlamaIndex core & plugins, Anthropic or OpenAI client
+
+```
+pip install -r requirements.txt
+```
+
+## CLI Usage
+
+```bash
+python web_agent.py --query "Research topic" --model "claude-opus-4-20250514" --verbose
+```

components/Web_agent_related/api_keys.py

@@ -0,0 +1,25 @@
+"""
+API Keys Configuration for WebAgent
+
+This module contains API keys and sensitive configuration data.
+IMPORTANT: Do not commit this file to version control with real keys.
+Consider using environment variables in production environments.
+
+Usage:
+    from api_keys import API_KEYS
+
+    google_api_key = API_KEYS['GOOGLE_API_KEY']
+"""
+
+# API keys dictionary
+API_KEYS = {
+    # Google Search API
+    "GOOGLE_API_KEY": "AIzaSyAngSEKuKwpBPapYwPnGBbUlsbT8y6p0KQ",
+    "GOOGLE_CX_ID": "97c907b98a73a41df",
+    
+    # GitHub API
+    "GITHUB_TOKEN": "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",
+    
+    # Anthropic API
+    "ANTHROPIC_API_KEY": "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"
+}

components/Web_agent_related/config.py

@@ -0,0 +1,83 @@
+"""
+Configuration for WebAgent
+
+This module contains configuration settings for WebAgent.
+Settings can be customized here or overridden at runtime.
+"""
+import os
+from typing import Dict, Any, Optional
+
+# Try to import API keys, use empty dict if file doesn't exist
+try:
+    from .api_keys import API_KEYS
+except ImportError:
+    API_KEYS = {}
+
+# API Keys Configuration
+# Priority: 1. Environment variables, 2. api_keys.py
+API_CONFIG = {
+    # Google Search API configuration
+    "GOOGLE_API_KEY": os.environ.get("GOOGLE_API_KEY") or API_KEYS.get("GOOGLE_API_KEY", ""),
+    "GOOGLE_CX_ID": os.environ.get("GOOGLE_CX_ID") or API_KEYS.get("GOOGLE_CX_ID", ""),
+    
+    # GitHub API configuration
+    "GITHUB_TOKEN": os.environ.get("GITHUB_TOKEN") or API_KEYS.get("GITHUB_TOKEN", ""),
+    
+    # Anthropic API configuration
+    "ANTHROPIC_API_KEY": os.environ.get("ANTHROPIC_API_KEY") or API_KEYS.get("ANTHROPIC_API_KEY", ""),
+}
+
+# Web Request Configuration
+WEB_REQUEST_CONFIG = {
+    # User agent for web requests
+    "USER_AGENT": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36",
+    
+    # Timeout for web requests in seconds
+    "TIMEOUT": 10,
+    
+    # Whether to verify SSL certificates
+    "VERIFY_SSL": True,
+}
+
+# Content Processing Configuration
+CONTENT_CONFIG = {
+    # Default number of characters per page chunk
+    "CHARS_PER_CHUNK": 10000,
+    
+    # Maximum size of page to process (in bytes)
+    "MAX_PAGE_SIZE": 1024 * 1024 * 5,  # 5 MB
+    
+    # Whether to preserve formatting in extracted text
+    "PRESERVE_FORMATTING": False,
+}
+
+# Search Configuration
+SEARCH_CONFIG = {
+    # Default number of search results to return
+    "DEFAULT_SEARCH_RESULTS": 5,
+    
+    # Maximum number of search results allowed
+    "MAX_SEARCH_RESULTS": 10000000,
+}
+
+def get_config(section: Optional[str] = None) -> Dict[str, Any]:
+    """
+    Get configuration settings, optionally for a specific section.
+    
+    Args:
+        section: Optional section name (API_CONFIG, WEB_REQUEST_CONFIG, etc.)
+        
+    Returns:
+        Dictionary containing configuration settings
+    """
+    all_config = {
+        "API_CONFIG": API_CONFIG,
+        "WEB_REQUEST_CONFIG": WEB_REQUEST_CONFIG,
+        "CONTENT_CONFIG": CONTENT_CONFIG,
+        "SEARCH_CONFIG": SEARCH_CONFIG
+    }
+    
+    if section:
+        return all_config.get(section, {})
+    
+    return all_config 

components/Web_agent_related/requirements.txt

@@ -0,0 +1,15 @@
+llama-index>=0.8.0
+llama-index-core>=0.10.0
+llama-index-readers-web>=0.1.0
+llama-index-llms-anthropic>=0.1.0
+anthropic>=0.8.0
+requests>=2.28.0
+beautifulsoup4>=4.11.0
+google-api-python-client>=2.70.0
+PyGithub>=1.58.0
+PyPDF2>=3.0.0
+python-docx>=0.8.11
+python-pptx>=0.6.21
+urllib3>=1.26.0
+pathlib>=1.0.1
+argparse>=1.4.0 

components/Web_agent_related/utils.py

@@ -0,0 +1,198 @@
+"""
+Utility functions for file handling and text extraction.
+
+This module contains standalone functions for extracting text from 
+various document formats (PDF, DOCX, PPTX).
+"""
+import os
+from typing import Tuple
+
+# For PDF processing
+import PyPDF2
+from PyPDF2.errors import PdfReadError
+
+# For DOCX processing
+import docx
+
+# For PPTX processing
+from pptx import Presentation
+from pptx.exc import PackageNotFoundError
+
+
+def read_pdf_text(filepath: str) -> Tuple[bool, str]:
+    """
+    Extracts text content from a PDF file.
+    
+    Args:
+        filepath: Path to the PDF file
+        
+    Returns:
+        A tuple (success, text_or_error_message):
+            - success: True if extraction was successful, False otherwise
+            - text_or_error_message: Extracted text if successful, error message otherwise
+    """
+    try:
+        if not os.path.exists(filepath):
+            return False, f"File not found: {filepath}"
+        
+        # Extract text from PDF
+        with open(filepath, 'rb') as file:
+            try:
+                reader = PyPDF2.PdfReader(file)
+                
+                # Get the number of pages
+                num_pages = len(reader.pages)
+                
+                if num_pages == 0:
+                    return False, "PDF file contains no pages"
+                
+                # Extract text from all pages
+                all_text = []
+                for page_num in range(num_pages):
+                    page = reader.pages[page_num]
+                    all_text.append(page.extract_text() or "")
+                
+                # Combine all text
+                full_text = "\n\n".join(all_text)
+                
+                if not full_text.strip():
+                    return False, "Could not extract text from PDF (the file may be scanned images or empty)"
+                
+                return True, full_text
+                
+            except PdfReadError as e:
+                return False, f"Error reading PDF file: {str(e)}"
+            
+    except Exception as e:
+        return False, f"Unexpected error when processing PDF file: {str(e)}"
+
+
+def read_docx_text(filepath: str) -> Tuple[bool, str]:
+    """
+    Extracts text content from a DOCX file.
+    
+    Args:
+        filepath: Path to the DOCX file
+        
+    Returns:
+        A tuple (success, text_or_error_message):
+            - success: True if extraction was successful, False otherwise
+            - text_or_error_message: Extracted text if successful, error message otherwise
+    """
+    try:
+        if not os.path.exists(filepath):
+            return False, f"File not found: {filepath}"
+        
+        # Open the docx file
+        try:
+            doc = docx.Document(filepath)
+            
+            # Extract text from paragraphs
+            paragraphs = []
+            for para in doc.paragraphs:
+                if para.text.strip():  # Ignore empty paragraphs
+                    paragraphs.append(para.text)
+            
+            # Extract text from tables
+            for table in doc.tables:
+                for row in table.rows:
+                    row_texts = []
+                    for cell in row.cells:
+                        if cell.text.strip():
+                            row_texts.append(cell.text.strip())
+                    if row_texts:
+                        paragraphs.append(" | ".join(row_texts))
+            
+            # Combine all text
+            full_text = "\n\n".join(paragraphs)
+            
+            if not full_text.strip():
+                return False, "Document appears to be empty"
+            
+            return True, full_text
+            
+        except Exception as e:
+            return False, f"Error parsing DOCX file: {str(e)}"
+            
+    except Exception as e:
+        return False, f"Unexpected error when processing DOCX file: {str(e)}"
+
+
+def read_pptx_text(filepath: str) -> Tuple[bool, str]:
+    """
+    Extracts text content from a PPTX file.
+    
+    Args:
+        filepath: Path to the PPTX file
+        
+    Returns:
+        A tuple (success, text_or_error_message):
+            - success: True if extraction was successful, False otherwise
+            - text_or_error_message: Extracted text if successful, error message otherwise
+    """
+    try:
+        if not os.path.exists(filepath):
+            return False, f"File not found: {filepath}"
+        
+        # Open the pptx file
+        try:
+            presentation = Presentation(filepath)
+            
+            all_text = []
+            
+            # Loop through slides
+            for i, slide in enumerate(presentation.slides):
+                slide_text = []
+                slide_text.append(f"--- Slide {i+1} ---")
+                
+                # Extract text from slide title if present
+                if slide.shapes.title and slide.shapes.title.text.strip():
+                    slide_text.append(f"Title: {slide.shapes.title.text}")
+                
+                # Extract text from all shapes
+                for shape in slide.shapes:
+                    if hasattr(shape, "text") and shape.text.strip():
+                        if shape != slide.shapes.title:  # Avoid duplicating title
+                            slide_text.append(shape.text)
+                
+                # Add slide text to overall text
+                if len(slide_text) > 1:  # If there's more than just the slide number
+                    all_text.append("\n".join(slide_text))
+            
+            # Combine all text
+            full_text = "\n\n".join(all_text)
+            
+            if not full_text.strip():
+                return False, "Presentation appears to be empty"
+            
+            return True, full_text
+            
+        except PackageNotFoundError:
+            return False, "Invalid PPTX file format"
+        except Exception as e:
+            return False, f"Error parsing PPTX file: {str(e)}"
+            
+    except Exception as e:
+        return False, f"Unexpected error when processing PPTX file: {str(e)}"
+
+
+def detect_file_type(filepath: str) -> str:
+    """
+    Detects file type based on file extension.
+    
+    Args:
+        filepath: Path to the file
+        
+    Returns:
+        File type as string ('pdf', 'docx', 'pptx', or None if unknown)
+    """
+    _, file_extension = os.path.splitext(filepath.lower())
+    
+    if file_extension == '.pdf':
+        return 'pdf'
+    elif file_extension in ['.doc', '.docx']:
+        return 'docx'
+    elif file_extension in ['.ppt', '.pptx']:
+        return 'pptx'
+    else:
+        return None 

components/Web_agent_related/web_agent.py

@@ -0,0 +1,1326 @@
+"""
+WebAgent: An autonomous research agent powered by LLM for web browsing and search operations.
+"""
+
+import os
+import re
+import sys
+import json
+import uuid
+import shutil
+import io
+from typing import List, Dict, Tuple, Optional, Union, Any, Callable
+from urllib.parse import urlparse, unquote
+from pathlib import Path
+
+# Import configuration
+from .config import get_config
+
+# Import utility functions
+from .utils import read_pdf_text, read_docx_text, read_pptx_text, detect_file_type
+
+# LlamaIndex imports
+from llama_index.core.tools import FunctionTool
+from llama_index.core.agent import ReActAgent
+from llama_index.core.llms import LLM
+from llama_index.llms.openai import OpenAI
+from llama_index.readers.web import SimpleWebPageReader
+from llama_index.core.schema import Document
+
+# For Google Search
+from googleapiclient.discovery import build
+from googleapiclient.errors import HttpError
+
+# For GitHub Search
+from github import Github, GithubException
+
+# For file download and handling
+import requests
+from requests.exceptions import RequestException
+
+# For Anthropic Claude
+from llama_index.llms.anthropic import Anthropic
+
+class WebAgent:
+    """
+    An autonomous research agent powered by LLM for web browsing and search operations.
+    
+    This agent can:
+    - Autonomously perform web research on a given topic
+    - Visit and retrieve textual content from web pages
+    - Navigate within the textual content of a currently loaded page
+    - Perform searches using Google and GitHub
+    - Return information as raw text content or structured search results
+    - Download files from the web
+    - Extract text from PDF, DOCX, and PPTX files
+    - Synthesize findings into a comprehensive report
+    """
+    
+    def __init__(
+        self,
+        llm: Optional[LLM] = None,
+        google_api_key: Optional[str] = None,
+        google_cx_id: Optional[str] = None,
+        github_token: Optional[str] = None,
+        chars_per_chunk: Optional[int] = None,
+        request_timeout: Optional[int] = None,
+        user_agent: Optional[str] = None,
+        temp_download_dir: Optional[str] = None,
+        max_research_iterations: int = 100000000,
+        verbose: bool = False
+    ):
+        """
+        Initialize the WebAgent with LLM and optional API keys and configuration.
+        
+        Args:
+            llm: LlamaIndex LLM instance for decision making
+            google_api_key: API key for Google Custom Search
+            google_cx_id: Custom Search Engine ID for Google Search
+            github_token: GitHub API token for authenticated searches
+            chars_per_chunk: Character count per chunk for page navigation
+            request_timeout: Timeout for web requests in seconds
+            user_agent: User agent string for web requests
+            temp_download_dir: Directory for temporary file downloads
+            max_research_iterations: Maximum number of research steps
+            verbose: Whether to print intermediate steps
+        """
+        # Get configuration sections
+        api_config = get_config("API_CONFIG")
+        web_config = get_config("WEB_REQUEST_CONFIG")
+        content_config = get_config("CONTENT_CONFIG")
+        
+        # Load API keys with priority: direct parameters > environment variables > config file
+        self.google_api_key = google_api_key or os.environ.get("GOOGLE_API_KEY") or api_config.get("GOOGLE_API_KEY", "")
+        self.google_cx_id = google_cx_id or os.environ.get("GOOGLE_CX_ID") or api_config.get("GOOGLE_CX_ID", "")
+        self.github_token = github_token or os.environ.get("GITHUB_TOKEN") or api_config.get("GITHUB_TOKEN", "")
+        
+        # Configure chunking and web request parameters
+        self.chars_per_chunk = chars_per_chunk or content_config.get("CHARS_PER_CHUNK", 1000)
+        self.request_timeout = request_timeout or web_config.get("TIMEOUT", 30)
+        self.user_agent = user_agent or web_config.get("USER_AGENT", "Mozilla/5.0")
+        self.verify_ssl = web_config.get("VERIFY_SSL", True)
+        
+        # Initialize temporary download directory
+        self.temp_download_dir = temp_download_dir or "temp_downloads"
+        os.makedirs(self.temp_download_dir, exist_ok=True)
+        
+        # Initialize state variables
+        self.current_url = None
+        self.current_page_documents = []
+        self.current_page_text_chunks = []
+        self.current_chunk_index = 0
+        
+        # Research configuration
+        self.llm = llm or OpenAI(model="gpt-3.5-turbo")
+        self.max_research_iterations = max_research_iterations
+        self.verbose = verbose
+        
+        # Initialize API clients if credentials are available
+        self._google_search_service = None
+        self._github_client = None
+        
+        if self.google_api_key and self.google_cx_id:
+            try:
+                self._google_search_service = build(
+                    "customsearch", "v1", developerKey=self.google_api_key
+                )
+            except Exception as e:
+                print(f"Warning: Failed to initialize Google Search API: {str(e)}")
+        
+        if self.github_token:
+            try:
+                self._github_client = Github(self.github_token)
+            except Exception as e:
+                print(f"Warning: Failed to initialize GitHub client: {str(e)}")
+        
+        # Initialize tools and agent
+        self._tools = self._setup_tools()
+        
+        # Set up the agent executor with max_iterations parameter
+        self.agent_executor = ReActAgent.from_tools(
+            tools=self._tools,
+            llm=self.llm,
+            verbose=self.verbose,
+            system_prompt=self._get_system_prompt(),
+            max_iterations=self.max_research_iterations
+        )
+
+    def _setup_tools(self) -> List[FunctionTool]:
+        """
+        Set up the LlamaIndex FunctionTool objects wrapping the agent's capabilities.
+        
+        Returns:
+            List of FunctionTool objects
+        """
+        tools = [
+            FunctionTool.from_defaults(
+                fn=self._internal_visit_url,
+                name="visit_web_page",
+                description="Visits a URL and returns the first chunk of its textual content. Use this to view web pages.",
+            ),
+            FunctionTool.from_defaults(
+                fn=self._internal_page_down,
+                name="scroll_down_page",
+                description="Scrolls down the currently loaded webpage and returns the next chunk of text. Call 'visit_web_page' first."
+            ),
+            FunctionTool.from_defaults(
+                fn=self._internal_page_up,
+                name="scroll_up_page",
+                description="Scrolls up the currently loaded webpage and returns the previous chunk of text. Call 'visit_web_page' first."
+            ),
+            FunctionTool.from_defaults(
+                fn=self._internal_get_full_page_text,
+                name="get_full_page_text",
+                description="Returns the entire text content of the currently loaded webpage. Call 'visit_web_page' first."
+            ),
+            FunctionTool.from_defaults(
+                fn=self._internal_search_google,
+                name="google_search",
+                description="Performs a Google search for the given query and returns a list of results (title, link, snippet)."
+            ),
+            FunctionTool.from_defaults(
+                fn=self._internal_search_github,
+                name="github_search",
+                description="Performs a GitHub search. Specify query string and optionally num_results and search_type ('repositories', 'code', 'issues')."
+            ),
+            FunctionTool.from_defaults(
+                fn=self._internal_download_and_extract_text,
+                name="download_and_read_file",
+                description="Downloads a file from a URL and extracts its text content. Works with PDF, DOCX, and PPTX files."
+            )
+        ]
+        return tools
+
+    def _get_system_prompt(self) -> str:
+        """
+        Generate the system prompt for the research agent.
+        """
+        return """You are an autonomous AI Web Research Agent.
+
+Your goal is to conduct thorough web research on topics given to you. You have access to the following tools:
+
+1. visit_web_page: Visit a URL and view its content
+2. scroll_down_page: View the next part of a webpage you've visited
+3. scroll_up_page: View the previous part of a webpage you've visited
+4. get_full_page_text: Get the entire text of a webpage you've visited
+5. google_search: Search Google for information
+6. github_search: Search GitHub for repositories, code, or issues
+7. download_and_read_file: Download and extract text from files (PDF, DOCX, PPTX)
+
+RESEARCH STRATEGY:
+1. Start by using google_search to find relevant information sources
+2. Visit promising websites with visit_web_page to gather information
+3. Navigate through pages with scroll_down_page and scroll_up_page
+4. When you find a relevant document, use download_and_read_file
+5. Continue until you have gathered sufficient information
+6. Synthesize your findings into a comprehensive report
+
+SYNTHESIS INSTRUCTIONS:
+- Provide a thorough and direct answer to the original research task
+- Include key findings and important details at the beginning of your report
+- Present supporting evidence, examples, and detailed explanations
+- Always cite your sources with full URLs whenever possible
+- Organize information logically with clear sections and points
+- Focus on accuracy and comprehensiveness in your final report
+
+IMPORTANT GUIDELINES:
+- Be thorough. Your final answer should be comprehensive and directly address the original research task.
+- Cite your sources by including the full URLs of the pages or documents where you found key information.
+- Take detailed notes on important facts, figures, and code snippets as you find them.
+- Your final output should be a complete synthesis of your findings.
+"""
+
+    def research(self, research_task: str, debug: bool = False) -> Dict[str, Any]:
+        """
+        Performs autonomous web research on the given task and returns a structured report.
+        
+        Args:
+            research_task: The research question or topic
+            debug: Whether to print detailed debug information about the agent response
+            
+        Returns:
+            A dictionary containing:
+                - "summary": Concise summary answering the research task
+                - "detailed_findings": Detailed information and evidence
+                - "sources": List of unique primary source URLs
+                - "status": "success", "success_unstructured", or "failure"
+                - "error_message": Error message if status is not "success", else None
+        """
+        # Initialize default return structure
+        result = {
+            "summary": "",
+            "detailed_findings": "",
+            "sources": [],
+            "status": "failure",
+            "error_message": None
+        }
+        
+        # Check if agent executor is initialized
+        if not self.agent_executor:
+            result["error_message"] = "Error: Agent executor not initialized."
+            return result
+
+        initial_prompt = f"""
+I need to research the following topic thoroughly:
+RESEARCH TASK: {research_task}
+Please conduct comprehensive research using the available tools. When you have gathered sufficient information, synthesize your findings into a well-structured report that thoroughly addresses the research task.
+Begin your research now.
+"""
+        
+        try:
+            # Execute the research
+            agent_response = self.agent_executor.chat(initial_prompt)
+            
+            if debug:
+                self._debug_print_agent_response(agent_response)
+            
+            raw_agent_output = agent_response.response
+            acted_upon_urls = []
+            history_parts = []
+            
+            agent_sources = getattr(agent_response, 'sources', [])
+            for i, source in enumerate(agent_sources):
+                tool_name = getattr(source, 'tool_name', 'unknown_tool')
+                tool_input = getattr(source, 'raw_input', {})
+                observation = getattr(source, 'content', '')
+                history_parts.append(
+                    f"Step {i+1}:\n"
+                    f"Action: {tool_name}\n"
+                    f"Action Input: {json.dumps(tool_input)}\n"
+                    f"Observation: {str(observation)[:1500]}...\n"
+                )
+                if tool_name in ["visit_web_page", "download_and_read_file"]:
+                    kwargs = tool_input.get('kwargs', {})
+                    url = kwargs.get('url')
+                    if isinstance(url, str) and url.startswith('http'):
+                        acted_upon_urls.append(url)
+
+            agent_history = "\n".join(history_parts)
+            unique_acted_upon_urls = list(dict.fromkeys(acted_upon_urls))
+            acted_upon_urls_context_str = "\n".join([f"- {url}" for url in unique_acted_upon_urls[:20]]) if unique_acted_upon_urls else "No URLs were directly used in 'visit_web_page' or 'download_and_read_file' actions."
+            history_for_prompt = agent_history if agent_history else raw_agent_output
+            
+            parsing_prompt = f"""
+Given the following research task, the ReAct agent's final answer, and its detailed execution history,
+extract the information into a structured JSON report. The report must be as detailed as possible,
+comprehensively answering the research task.
+
+RESEARCH TASK:
+{research_task}
+
+ReAct AGENT'S FINAL ANSWER (Primary basis for your report):
+{raw_agent_output}
+
+ReAct AGENT'S DETAILED EXECUTION HISTORY (Use this to extract specific details, code, and verify sources):
+{history_for_prompt}
+
+URLs DIRECTLY ACTED UPON BY THE AGENT (High-confidence sources):
+{acted_upon_urls_context_str}
+
+Please provide a JSON object with the following keys:
+- "summary": A concise summary that directly answers the research task.
+- "detailed_findings": This MUST be a comprehensive and highly detailed section.
+    - Directly address all parts of the research task.
+    - If the research task involved finding or generating code, THIS SECTION MUST INCLUDE THE RELEVANT CODE SNIPPETS OR FULL CODE BLOCK(S) as found or synthesized by the agent. Use markdown for code blocks (e.g., ```python ... ```).
+    - Include specific facts, figures, explanations, and evidence from the agent's research.
+    - Structure the information logically. This should be a single string, potentially long, containing the full detailed report.
+- "sources": A list of unique primary source URLs.
+    - Prioritize URLs from the 'URLs DIRECTLY ACTED UPON' list.
+    - Supplement with other critical URLs from the AGENT'S FINAL ANSWER or EXECUTION HISTORY from which key information or code was directly obtained.
+    - Ensure all listed URLs are complete and valid.
+
+Return ONLY the JSON object inside a markdown code block, like this:
+```json
+{{
+  "summary": "...",
+  "detailed_findings": "...",
+  "sources": [...]
+}}
+```
+"""
+            # Get structured output from LLM
+            try:
+                if debug: # Save parsing prompt only if debug is True
+                    with open("parsing_prompt.txt", "w", encoding='utf-8') as f:
+                        f.write(parsing_prompt)
+                
+                structured_output_text = self.llm.complete(parsing_prompt).text
+                
+                if debug: # Save structured output only if debug is True
+                    with open("structured_output.txt", "w", encoding='utf-8') as f:
+                        f.write(structured_output_text)
+
+                json_str = None
+                # Pattern 1: Look for JSON inside ```json ... ```
+                match = re.search(r'```json\s*(\{[\s\S]*\})\s*```', structured_output_text, re.DOTALL)
+                if match:
+                    json_str = match.group(1)
+                else:
+                    # Pattern 2: Be more aggressive and find the first '{' and last '}'
+                    start_index = structured_output_text.find('{')
+                    end_index = structured_output_text.rfind('}')
+                    if start_index != -1 and end_index != -1 and end_index > start_index:
+                        json_str = structured_output_text[start_index : end_index + 1]
+
+                if not json_str:
+                    raise ValueError(f"Could not extract a JSON object from the LLM's response. Raw response: {structured_output_text}")
+
+                # Clean the extracted string
+                json_str = json_str.strip()
+                
+                # Parse the structured output
+                structured_output = json.loads(json_str)
+                
+                # Update result with structured data
+                result["summary"] = structured_output.get("summary", "")
+                
+                # Handle detailed findings - ensure it's properly formatted
+                detailed_findings = structured_output.get("detailed_findings", [])
+                if isinstance(detailed_findings, list):
+                    result["detailed_findings"] = "\n\n".join(detailed_findings)
+                else:
+                    result["detailed_findings"] = str(detailed_findings)
+                    
+                # Handle sources - ensure we have URLs
+                sources = structured_output.get("sources", [])
+                if not sources and unique_acted_upon_urls:
+                    sources = unique_acted_upon_urls
+                result["sources"] = list(dict.fromkeys(sources))
+                result["status"] = "success"
+                
+            except (json.JSONDecodeError, ValueError) as json_err:
+                result["summary"] = raw_agent_output
+                result["detailed_findings"] = agent_history if agent_history else "No detailed history available."
+                result["sources"] = unique_acted_upon_urls
+                result["status"] = "success_unstructured"
+                result["error_message"] = f"Failed to parse agent output into structured JSON: {str(json_err)}"
+    
+        except Exception as e:
+            result["error_message"] = str(e)
+            if self.verbose or debug:
+                traceback.print_exc(file=sys.stdout)
+    
+        return result
+
+    # Internal tool methods
+    def _internal_visit_url(self, url: str) -> str:
+        """
+        Internal method for visiting a URL.
+        
+        Args:
+            url: The URL to visit
+            
+        Returns:
+            String representation of the first chunk of text or error message
+        """
+        success, content_or_error = self.visit_url(url)
+        if success:
+            return content_or_error
+        return f"Error visiting URL: {content_or_error}"
+
+    def _internal_page_down(self) -> str:
+        """
+        Internal method for scrolling down the current page.
+        
+        Returns:
+            String representation of the next chunk of text or error message
+        """
+        content = self.page_down()
+        return content if content else "No more content or no page loaded."
+
+    def _internal_page_up(self) -> str:
+        """
+        Internal method for scrolling up the current page.
+        
+        Returns:
+            String representation of the previous chunk of text or error message
+        """
+        content = self.page_up()
+        return content if content else "No previous content or no page loaded."
+
+    def _internal_get_full_page_text(self) -> str:
+        """
+        Internal method for getting the full page text.
+        
+        Returns:
+            String representation of the full page text or error message
+        """
+        content = self.get_full_page_text()
+        return content if content else "No page loaded or no content available."
+
+    def _internal_search_google(self, query: str, num_results: int = 5) -> str:
+        """
+        Internal method for searching Google.
+        
+        Args:
+            query: The search query
+            num_results: Number of results to return
+            
+        Returns:
+            String representation of the search results or error message
+        """
+        success, results_or_error = self.search_google(query, num_results)
+        if success:
+            formatted_results = "\n\n".join([
+                f"Title: {r['title']}\nLink: {r['link']}\nSnippet: {r['snippet']}"
+                for r in results_or_error
+            ])
+            return f"Google Search Results for '{query}':\n\n{formatted_results}"
+        return f"Google Search Error: {results_or_error}"
+
+    def _internal_search_github(self, query: str, num_results: int = 5, search_type: str = "repositories") -> str:
+        """
+        Internal method for searching GitHub.
+        
+        Args:
+            query: The search query
+            num_results: Number of results to return
+            search_type: Type of search ("repositories", "code", "issues")
+            
+        Returns:
+            String representation of the search results or error message
+        """
+        success, results_or_error = self.search_github(query, num_results, search_type)
+        if success:
+            if search_type == "repositories":
+                formatted_results = "\n\n".join([
+                    f"Name: {r['name']}\nURL: {r['html_url']}\nDescription: {r.get('description', 'N/A')}\nStars: {r.get('stars', 'N/A')}\nLanguage: {r.get('language', 'N/A')}"
+                    for r in results_or_error
+                ])
+            elif search_type == "code":
+                formatted_results = "\n\n".join([
+                    f"File: {r.get('name', 'N/A')}\nURL: {r['html_url']}\nRepository: {r.get('repository', 'N/A')}\nPath: {r.get('path', 'N/A')}"
+                    for r in results_or_error
+                ])
+            elif search_type == "issues":
+                formatted_results = "\n\n".join([
+                    f"Title: {r.get('title', 'N/A')}\nURL: {r['html_url']}\nState: {r.get('state', 'N/A')}\nCreated: {r.get('created_at', 'N/A')}"
+                    for r in results_or_error
+                ])
+            else:
+                formatted_results = str(results_or_error)
+                
+            return f"GitHub Search Results for '{query}' (type: {search_type}):\n\n{formatted_results}"
+        return f"GitHub Search Error: {results_or_error}"
+
+    def _internal_download_and_extract_text(self, url: str, file_type: str = None) -> str:
+        """
+        Internal method for downloading and extracting text from a file.
+        
+        Args:
+            url: The URL of the file to download
+            file_type: Optional file type override ('pdf', 'docx', 'pptx')
+            
+        Returns:
+            String representation of the extracted text or error message
+        """
+        success, text_or_error = self.download_and_extract_text(url, file_type)
+        if success:
+            return f"Successfully extracted text from {url}:\n\n{text_or_error}"
+        return f"Error extracting text from {url}: {text_or_error}"
+
+    # Original WebAgent methods
+    def visit_url(self, url: str) -> Tuple[bool, str]:
+        """
+        Navigates to the given URL, fetches its content, and processes it for text-based browsing.
+        
+        Args:
+            url: The URL to visit and load content from
+            
+        Returns:
+            A tuple (success, message_or_first_chunk):
+                - success: True if the page was loaded successfully, False otherwise
+                - message_or_first_chunk: First chunk of text if successful, error message otherwise
+        """
+        try:
+            # Validate URL format
+            parsed_url = urlparse(url)
+            if not all([parsed_url.scheme, parsed_url.netloc]):
+                return False, f"Invalid URL format: {url}"
+            
+            # Configure the web page reader
+            # NOTE: SimpleWebPageReader does not support user agent or timeout directly.
+            # Only html_to_text is supported as an argument.
+            documents = SimpleWebPageReader(html_to_text=True).load_data([url])
+            
+            if not documents:
+                return False, f"No content found at URL: {url}"
+            
+            # Update state variables
+            self.current_url = url
+            self.current_page_documents = documents
+            
+            # Extract and chunk the text content
+            full_text = " ".join([doc.text for doc in documents if doc.text])
+            self._chunk_text(full_text)
+            
+            # Reset view position
+            self.current_chunk_index = 0
+            
+            # Return the first chunk
+            if self.current_page_text_chunks:
+                return True, self.current_page_text_chunks[0]
+            else:
+                return False, "URL loaded but no readable text content found"
+            
+        except Exception as e:
+            return False, f"Failed to load URL: {str(e)}"
+    
+    def _chunk_text(self, text: str) -> None:
+        """
+        Splits the full text into manageable chunks for navigation.
+        
+        Args:
+            text: The full text content to chunk
+        """
+        # Clear existing chunks
+        self.current_page_text_chunks = []
+        
+        if not text:
+            return
+        
+        # Split text into paragraphs first (better for readability)
+        paragraphs = re.split(r'\n\s*\n', text)
+        
+        current_chunk = ""
+        for paragraph in paragraphs:
+            paragraph = paragraph.strip()
+            if not paragraph:
+                continue
+            # If adding this paragraph would exceed the chunk size,
+            # store the current chunk and start a new one
+            if current_chunk and len(current_chunk) + len(paragraph) + 2 > self.chars_per_chunk:
+                self.current_page_text_chunks.append(current_chunk.strip())
+                current_chunk = ""
+            # If the paragraph itself is longer than the chunk size, split it
+            while len(paragraph) > self.chars_per_chunk:
+                if current_chunk:
+                    self.current_page_text_chunks.append(current_chunk.strip())
+                    current_chunk = ""
+                self.current_page_text_chunks.append(paragraph[:self.chars_per_chunk].strip())
+                paragraph = paragraph[self.chars_per_chunk:]
+            # Add the (possibly shortened) paragraph to the current chunk
+            if current_chunk:
+                current_chunk += "\n\n" + paragraph
+            else:
+                current_chunk = paragraph
+        # Add the last chunk if it's not empty
+        if current_chunk.strip():
+            self.current_page_text_chunks.append(current_chunk.strip())
+    
+    def get_current_view(self) -> Optional[str]:
+        """
+        Returns the current textual "view" of the loaded page based on current_chunk_index.
+        
+        Returns:
+            The current text chunk or None if no page is loaded
+        """
+        if not self.current_page_text_chunks:
+            return None
+        
+        if 0 <= self.current_chunk_index < len(self.current_page_text_chunks):
+            return self.current_page_text_chunks[self.current_chunk_index]
+        
+        return None
+    
+    def page_down(self) -> Optional[str]:
+        """
+        Simulates scrolling down the loaded text content.
+        
+        Returns:
+            The next text chunk or None if no page is loaded or already at the end
+        """
+        if not self.current_page_text_chunks:
+            return None
+        
+        # Check if already at the end
+        if self.current_chunk_index >= len(self.current_page_text_chunks) - 1:
+            return "Already at the end of the page.\n\n" + self.current_page_text_chunks[self.current_chunk_index]
+        
+        # Move to the next chunk
+        self.current_chunk_index += 1
+        return self.current_page_text_chunks[self.current_chunk_index]
+    
+    def page_up(self) -> Optional[str]:
+        """
+        Simulates scrolling up the loaded text content.
+        
+        Returns:
+            The previous text chunk or None if no page is loaded or already at the beginning
+        """
+        if not self.current_page_text_chunks:
+            return None
+        
+        # Check if already at the beginning
+        if self.current_chunk_index <= 0:
+            return "Already at the start of the page.\n\n" + self.current_page_text_chunks[0]
+        
+        # Move to the previous chunk
+        self.current_chunk_index -= 1
+        return self.current_page_text_chunks[self.current_chunk_index]
+    
+    def get_full_page_text(self) -> Optional[str]:
+        """
+        Returns the entire concatenated text content of the currently loaded page.
+        
+        Returns:
+            The full text or None if no page is loaded
+        """
+        if not self.current_page_text_chunks:
+            return None
+        
+        return "\n\n".join(self.current_page_text_chunks)
+    
+    def search_google(
+        self, query: str, num_results: Optional[int] = None
+    ) -> Tuple[bool, Union[List[Dict[str, str]], str]]:
+        """
+        Performs a search using Google and returns a list of results.
+        
+        Args:
+            query: The search query
+            num_results: Number of results to return (uses config default if None)
+            
+        Returns:
+            A tuple (success, results_or_error):
+                - success: True if search was successful, False otherwise
+                - results_or_error: List of search result dictionaries if successful, error message otherwise
+        """
+        if not self._google_search_service or not self.google_cx_id:
+            return False, "Google Search API not configured. Please provide google_api_key and google_cx_id."
+        
+        try:
+            # Get search configuration
+            search_config = get_config("SEARCH_CONFIG")
+            
+            # Use config for default num_results if not specified
+            if num_results is None:
+                num_results = search_config.get("DEFAULT_SEARCH_RESULTS", 5)
+                
+            # Ensure num_results is within acceptable range
+            num_results = min(max(1, num_results), search_config.get("MAX_SEARCH_RESULTS", 10))
+            
+            # Execute the search
+            search_results = self._google_search_service.cse().list(
+                q=query,
+                cx=self.google_cx_id,
+                num=num_results
+            ).execute()
+            
+            # Process and format the results
+            formatted_results = []
+            
+            if 'items' in search_results:
+                for item in search_results['items']:
+                    result = {
+                        'title': item.get('title', 'No title'),
+                        'link': item.get('link', 'No link'),
+                        'snippet': item.get('snippet', 'No description')
+                    }
+                    formatted_results.append(result)
+            
+            return True, formatted_results
+            
+        except HttpError as e:
+            return False, f"Google Search API error: {str(e)}"
+        except Exception as e:
+            return False, f"Error performing Google search: {str(e)}"
+    
+    def search_github(
+        self, query: str, num_results: Optional[int] = None, search_type: str = "repositories"
+    ) -> Tuple[bool, Union[List[Dict[str, Any]], str]]:
+        """
+        Performs a search on GitHub for repositories, code, or issues.
+        
+        Args:
+            query: The search query
+            num_results: Number of results to return (uses config default if None)
+            search_type: Type of search ("repositories", "code", "issues")
+            
+        Returns:
+            A tuple (success, results_or_error):
+                - success: True if search was successful, False otherwise
+                - results_or_error: List of search result dictionaries if successful, error message otherwise
+        """
+        if not self._github_client:
+            return False, "GitHub API not configured. Please provide github_token."
+        
+        try:
+            # Get search configuration
+            search_config = get_config("SEARCH_CONFIG")
+            
+            # Use config for default num_results if not specified
+            if num_results is None:
+                num_results = search_config.get("DEFAULT_SEARCH_RESULTS", 5)
+                
+            # Ensure num_results is within acceptable range
+            num_results = min(max(1, num_results), search_config.get("MAX_SEARCH_RESULTS", 10))
+            
+            formatted_results = []
+            
+            # Execute search based on search_type
+            if search_type == "repositories":
+                search_results = self._github_client.search_repositories(query)
+                
+                for i, repo in enumerate(search_results):
+                    if i >= num_results:
+                        break
+                    
+                    result = {
+                        'name': repo.full_name,
+                        'html_url': repo.html_url,
+                        'description': repo.description or "No description",
+                        'stars': repo.stargazers_count,
+                        'language': repo.language or "Not specified"
+                    }
+                    formatted_results.append(result)
+                    
+            elif search_type == "code":
+                search_results = self._github_client.search_code(query)
+                
+                for i, code_result in enumerate(search_results):
+                    if i >= num_results:
+                        break
+                    
+                    result = {
+                        'name': code_result.name,
+                        'html_url': code_result.html_url,
+                        'repository': code_result.repository.full_name,
+                        'path': code_result.path
+                    }
+                    formatted_results.append(result)
+                    
+            elif search_type == "issues":
+                search_results = self._github_client.search_issues(query)
+                
+                for i, issue in enumerate(search_results):
+                    if i >= num_results:
+                        break
+                    
+                    result = {
+                        'title': issue.title,
+                        'html_url': issue.html_url,
+                        'state': issue.state,
+                        'created_at': str(issue.created_at),
+                        'repository': issue.repository.full_name
+                    }
+                    formatted_results.append(result)
+                    
+            else:
+                return False, f"Invalid search_type: {search_type}. Must be 'repositories', 'code', or 'issues'."
+            
+            return True, formatted_results
+            
+        except GithubException as e:
+            return False, f"GitHub API error: {str(e)}"
+        except Exception as e:
+            return False, f"Error performing GitHub search: {str(e)}"
+    
+    def download_file(self, url: str, desired_filename: Optional[str] = None) -> Tuple[bool, str]:
+        """
+        Downloads a file from the given URL to the temp_downloads directory.
+        
+        Args:
+            url: The URL of the file to download
+            desired_filename: Optional custom filename for the downloaded file
+            
+        Returns:
+            A tuple (success, filepath_or_error_message):
+                - success: True if download was successful, False otherwise
+                - filepath_or_error_message: Path to downloaded file if successful, error message otherwise
+        """
+        try:
+            # Validate URL format
+            parsed_url = urlparse(url)
+            if not all([parsed_url.scheme, parsed_url.netloc]):
+                return False, f"Invalid URL format: {url}"
+            
+            # Determine filename 
+            if desired_filename:
+                filename = desired_filename
+            else:
+                # Try to extract filename from URL
+                url_path = unquote(parsed_url.path)
+                filename_from_url = os.path.basename(url_path)
+                
+                if filename_from_url and '.' in filename_from_url:
+                    # Use filename from URL if it has an extension
+                    filename = filename_from_url
+                else:
+                    # Generate a unique filename with URL netloc as prefix
+                    unique_id = str(uuid.uuid4())[:8]
+                    host = parsed_url.netloc.split('.')[0] if '.' in parsed_url.netloc else 'download'
+                    filename = f"{host}_{unique_id}"
+                    
+                    # Try to determine file extension from Content-Type header
+                    head_response = requests.head(url, timeout=self.request_timeout, 
+                                                allow_redirects=True,
+                                                headers={"User-Agent": self.user_agent})
+                    content_type = head_response.headers.get('Content-Type', '')
+                    
+                    # Map common MIME types to file extensions
+                    extension_map = {
+                        'application/pdf': '.pdf',
+                        'application/msword': '.doc',
+                        'application/vnd.openxmlformats-officedocument.wordprocessingml.document': '.docx',
+                        'application/vnd.ms-powerpoint': '.ppt',
+                        'application/vnd.openxmlformats-officedocument.presentationml.presentation': '.pptx',
+                        'application/vnd.ms-excel': '.xls',
+                        'application/vnd.openxmlformats-officedocument.spreadsheetml.sheet': '.xlsx',
+                        'text/plain': '.txt',
+                        'text/html': '.html',
+                        'text/csv': '.csv',
+                        'image/jpeg': '.jpg',
+                        'image/png': '.png',
+                        'image/gif': '.gif',
+                        'application/json': '.json',
+                        'application/xml': '.xml',
+                    }
+                    
+                    # Extract the main content type
+                    main_type = content_type.split(';')[0].strip()
+                    extension = extension_map.get(main_type, '')
+                    
+                    if extension:
+                        filename += extension
+                    elif 'Content-Disposition' in head_response.headers:
+                        # Try to extract filename from Content-Disposition
+                        content_disp = head_response.headers['Content-Disposition']
+                        if 'filename=' in content_disp:
+                            extracted_name = re.findall('filename=(.+)', content_disp)
+                            if extracted_name:
+                                extracted_name = extracted_name[0].strip('"\'')
+                                if '.' in extracted_name:
+                                    filename = extracted_name
+            
+            # Ensure the filename is safe and doesn't contain invalid characters
+            filename = "".join([c for c in filename if c.isalnum() or c in '._- ']).strip()
+            if not filename:
+                filename = f"download_{str(uuid.uuid4())[:8]}"
+            
+            # Generate full filepath
+            filepath = os.path.join(self.temp_download_dir, filename)
+            
+            # Download the file with a streaming approach to handle large files
+            with requests.get(url, stream=True, timeout=self.request_timeout, 
+                             headers={"User-Agent": self.user_agent},
+                             verify=self.verify_ssl) as response:
+                response.raise_for_status()  # Raise exception for 4XX/5XX responses
+                
+                with open(filepath, 'wb') as file:
+                    for chunk in response.iter_content(chunk_size=8192):
+                        if chunk:
+                            file.write(chunk)
+            
+            return True, filepath
+            
+        except RequestException as e:
+            return False, f"Request error when downloading file: {str(e)}"
+        except IOError as e:
+            return False, f"I/O error when saving file: {str(e)}"
+        except Exception as e:
+            return False, f"Unexpected error when downloading file: {str(e)}"
+    
+    def read_pdf_text(self, filepath: str) -> Tuple[bool, str]:
+        """
+        Extracts text content from a PDF file using the utility function.
+        
+        Args:
+            filepath: Path to the PDF file
+            
+        Returns:
+            A tuple (success, text_or_error_message)
+        """
+        return read_pdf_text(filepath)
+    
+    def read_docx_text(self, filepath: str) -> Tuple[bool, str]:
+        """
+        Extracts text content from a DOCX file using the utility function.
+        
+        Args:
+            filepath: Path to the DOCX file
+            
+        Returns:
+            A tuple (success, text_or_error_message)
+        """
+        return read_docx_text(filepath)
+    
+    def read_pptx_text(self, filepath: str) -> Tuple[bool, str]:
+        """
+        Extracts text content from a PPTX file using the utility function.
+        
+        Args:
+            filepath: Path to the PPTX file
+            
+        Returns:
+            A tuple (success, text_or_error_message)
+        """
+        return read_pptx_text(filepath)
+    
+    def clean_temp_files(self, filepath: Optional[str] = None) -> bool:
+        """
+        Cleans up temporary downloaded files.
+        
+        Args:
+            filepath: Optional specific file to delete. If None, all files in temp_download_dir are deleted
+            
+        Returns:
+            True if cleaning was successful, False otherwise
+        """
+        try:
+            if filepath:
+                # Delete a specific file
+                if os.path.exists(filepath) and os.path.isfile(filepath):
+                    os.remove(filepath)
+            else:
+                # Delete all files in the temp_download_dir
+                for filename in os.listdir(self.temp_download_dir):
+                    filepath = os.path.join(self.temp_download_dir, filename)
+                    if os.path.isfile(filepath):
+                        os.remove(filepath)
+            return True
+        except Exception as e:
+            print(f"Warning: Failed to clean temporary files: {str(e)}")
+            return False
+    
+    def download_and_extract_text(self, url: str, file_type: str = None) -> Tuple[bool, str]:
+        """
+        Downloads a file and extracts its text content based on file type.
+        
+        Args:
+            url: The URL of the file to download
+            file_type: Optional file type override ('pdf', 'docx', 'pptx')
+            
+        Returns:
+            A tuple (success, text_or_error_message):
+                - success: True if operation was successful, False otherwise
+                - text_or_error_message: Extracted text if successful, error message otherwise
+        """
+        try:
+            # Download the file
+            download_success, download_result = self.download_file(url)
+            
+            if not download_success:
+                return False, download_result
+            
+            filepath = download_result
+            
+            try:
+                # Determine file type if not specified
+                if not file_type:
+                    file_type = detect_file_type(filepath)
+                    if not file_type:
+                        return False, f"Unsupported file type: {os.path.splitext(filepath)[1]}"
+                
+                # Extract text based on file type
+                if file_type == 'pdf':
+                    success, result = self.read_pdf_text(filepath)
+                elif file_type == 'docx':
+                    success, result = self.read_docx_text(filepath)
+                elif file_type == 'pptx':
+                    success, result = self.read_pptx_text(filepath)
+                else:
+                    success = False
+                    result = f"Unsupported file type: {file_type}"
+                
+                # Clean up the temporary file
+                self.clean_temp_files(filepath)
+                
+                return success, result
+                
+            except Exception as e:
+                # Clean up in case of error
+                self.clean_temp_files(filepath)
+                return False, f"Error processing file: {str(e)}"
+                
+        except Exception as e:
+            return False, f"Unexpected error: {str(e)}"
+
+    def _debug_print_agent_response(self, agent_response, max_str_length=1000):
+        """
+        Debug utility to print all components of the agent response object.
+        
+        Args:
+            agent_response: The agent response object to inspect
+            max_str_length: Maximum length for string values to display
+        """
+        def _format_value(value, level=0):
+            indent = "  " * level
+            if value is None:
+                return "None"
+            elif isinstance(value, str):
+                if len(value) > max_str_length:
+                    return f"'{value[:max_str_length]}... [truncated, total length: {len(value)}]'"
+                return f"'{value}'"
+            elif isinstance(value, (int, float, bool)):
+                return str(value)
+            elif isinstance(value, list):
+                if not value:
+                    return "[]"
+                result = "[\n"
+                for i, item in enumerate(value[:5]):  # Limit to first 5 items
+                    result += f"{indent}  [{i}]: {_format_value(item, level+1)},\n"
+                if len(value) > 5:
+                    result += f"{indent}  ... {len(value) - 5} more items\n"
+                result += f"{indent}]"
+                return result
+            elif isinstance(value, dict):
+                if not value:
+                    return "{}"
+                result = "{\n"
+                for i, (k, v) in enumerate(list(value.items())[:5]):  # Limit to first 5 items
+                    result += f"{indent}  '{k}': {_format_value(v, level+1)},\n"
+                if len(value) > 5:
+                    result += f"{indent}  ... {len(value) - 5} more items\n"
+                result += f"{indent}}}"
+                return result
+            else:
+                try:
+                    return f"{type(value).__name__}: {str(value)}"
+                except:
+                    return f"{type(value).__name__}: [Error displaying value]"
+        
+        print("\n" + "="*80)
+        print("AGENT RESPONSE OBJECT STRUCTURE")
+        print("="*80)
+        
+        # Print type information
+        print(f"Type: {type(agent_response).__name__}")
+        
+        # Get all attributes
+        if hasattr(agent_response, "__dict__"):
+            attributes = vars(agent_response)
+            print(f"\nAttributes ({len(attributes)}):")
+            
+            for attr_name, attr_value in attributes.items():
+                print(f"\n- {attr_name} ({type(attr_value).__name__}):")
+                print(f"  {_format_value(attr_value, 1)}")
+        else:
+            print("\nNo __dict__ attribute available.")
+            
+        # Special handling for common ReActAgent components
+        special_attributes = [
+            "response", "step_history", "source_nodes", "intermediate_steps", 
+            "raw_response", "message", "chat_history"
+        ]
+        
+        print("\nCommon ReActAgent Components:")
+        for attr in special_attributes:
+            if hasattr(agent_response, attr):
+                value = getattr(agent_response, attr)
+                print(f"\n- {attr} ({type(value).__name__}):")
+                print(f"  {_format_value(value, 1)}")
+        
+        # Special handling for step_history if it exists
+        if hasattr(agent_response, "step_history") and agent_response.step_history:
+            print("\nDetailed Step History:")
+            for i, step in enumerate(agent_response.step_history):
+                print(f"\nStep {i+1}:")
+                for step_attr in ["thought", "action", "action_input", "observation"]:
+                    if hasattr(step, step_attr):
+                        step_value = getattr(step, step_attr)
+                        print(f"  - {step_attr}: {_format_value(step_value, 2)}")
+        
+        print("\n" + "="*80)
+
+
+if __name__ == "__main__":
+    """
+    Test mode for the WebAgent when executed directly.
+    This demonstrates the autonomous research capabilities.
+    """
+    import sys
+    import argparse
+    import json
+    import traceback
+    from datetime import datetime
+    from pprint import pprint
+    from llama_index.llms.anthropic import Anthropic
+    
+    # ANSI color codes
+    COLOR_RESET = "\033[0m"
+    COLOR_YELLOW = "\033[93m"
+    COLOR_CYAN = "\033[96m"
+    COLOR_GREEN = "\033[92m"
+    COLOR_RED = "\033[91m"
+    COLOR_BLUE = "\033[94m"
+    COLOR_MAGENTA = "\033[95m"
+    COLOR_GRAY = "\033[90m"
+    
+    def color_text(text, color):
+        return f"{color}{text}{COLOR_RESET}"
+    
+    # Create a custom log writer that will write to both console and file
+    class TeeWriter:
+        def __init__(self, file_path):
+            self.terminal = sys.stdout
+            self.file = open(file_path, 'w', encoding='utf-8')
+            self.log_buffer = [] # Buffer to store log content if needed elsewhere
+            
+        def write(self, message):
+            self.terminal.write(message)
+            # Strip ANSI color codes for file output
+            clean_message = re.sub(r'\033\[\d+m', '', message)
+            self.file.write(clean_message)
+            self.log_buffer.append(clean_message) # Store clean message
+            
+        def flush(self):
+            self.terminal.flush()
+            self.file.flush()
+            
+        def close(self):
+            self.file.close()
+
+        def get_log_content(self): # Method to retrieve buffered log
+            return "".join(self.log_buffer)
+
+    parser = argparse.ArgumentParser(description="WebAgent Research Tool")
+    parser.add_argument("--query", "-q", type=str, help="Research query to process")
+    parser.add_argument("--model", "-m", type=str, default="claude-3-5-sonnet-latest", 
+                        help="Anthropic model to use (default: claude-3-haiku-20240307)")
+    parser.add_argument("--verbose", "-v", action="store_true", default=True, help="Enable verbose output")
+    parser.add_argument("--debug", "-d", action="store_true", default=True, help="Enable debug output with full agent response structure")
+    parser.add_argument("--log-file", "-l", type=str, help="Log file name (default: research_log_TIMESTAMP.txt)")
+    args = parser.parse_args()
+    
+    # Create log file name with timestamp if not provided
+    if not args.log_file:
+        timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
+        log_file = f"research_log_{timestamp}.txt"
+    else:
+        log_file = args.log_file
+    
+    # Set up the TeeWriter to capture all output
+    tee = TeeWriter(log_file)
+    original_stdout = sys.stdout  # Save the original stdout
+    sys.stdout = tee # Redirect stdout to TeeWriter
+
+    print(color_text("WebAgent Research Tool", COLOR_CYAN))
+    print(color_text("=====================", COLOR_CYAN) + "\n")
+    print(color_text(f"All output will be logged to: {log_file}", COLOR_GREEN))
+    
+    # Check if Anthropic API key is available
+    anthropic_key = os.environ.get("ANTHROPIC_API_KEY")
+    if not anthropic_key:
+        api_config = get_config("API_CONFIG")
+        anthropic_key = api_config.get("ANTHROPIC_API_KEY", "")
+        if not anthropic_key:
+            print(color_text("Error: ANTHROPIC_API_KEY not found in environment variables or config.", COLOR_RED))
+            print("Please set your Anthropic API key with:")
+            print("  export ANTHROPIC_API_KEY='your-api-key'")
+            print("  or add it to api_keys.py")
+            sys.stdout = original_stdout # Restore original stdout before exit
+            tee.close()
+            sys.exit(1)
+    
+    # Initialize the LLM
+    try:
+        print(color_text(f"Initializing Anthropic Claude (model: {args.model})...", COLOR_CYAN))
+        llm = Anthropic(model=args.model, api_key=anthropic_key, max_tokens=8192)  # Increased max_tokens
+    except Exception as e:
+        print(color_text(f"Error initializing Anthropic LLM: {str(e)}", COLOR_RED))
+        sys.stdout = original_stdout # Restore original stdout before exit
+        tee.close()
+        sys.exit(1)
+    
+    # Create WebAgent instance
+    try:
+        print(color_text("Creating WebAgent...", COLOR_CYAN))
+        agent = WebAgent(
+            llm=llm,
+            verbose=args.verbose # Pass verbose flag to WebAgent
+        )
+    except Exception as e:
+        print(color_text(f"Error creating WebAgent: {str(e)}", COLOR_RED))
+        sys.stdout = original_stdout # Restore original stdout before exit
+        tee.close()
+        sys.exit(1)
+    
+    # Display configuration status
+    print(color_text("\nConfiguration:", COLOR_CYAN))
+    print(f"  LLM Model: Anthropic {args.model}")
+    print(f"  Verbose Mode: {'Enabled' if args.verbose else 'Disabled'}")
+    print(f"  Debug Mode: {'Enabled' if args.debug else 'Disabled'}")
+    print(f"  Google Search: {'Configured' if agent.google_api_key and agent.google_cx_id else 'Not configured'}")
+    print(f"  GitHub: {'Configured' if agent.github_token else 'Not configured'}")
+    print()
+    
+    # If query was provided as argument, use it
+    if args.query:
+        query = args.query
+    else:
+        # Otherwise prompt for research task
+        query = input(color_text("Enter research task: ", COLOR_CYAN)).strip()
+    
+    if not query:
+        print(color_text("Error: No research task provided.", COLOR_RED))
+        sys.stdout = original_stdout # Restore original stdout before exit
+        tee.close()
+        sys.exit(1)
+    
+    print(color_text(f"\nResearch Task: ", COLOR_CYAN) + color_text(query, COLOR_YELLOW))
+    print(color_text("\nStarting autonomous research. This may take some time...", COLOR_CYAN))
+    
+    try:
+        # Perform the research
+        report = agent.research(query, debug=args.debug)
+        
+        print(color_text("\n=============== STRUCTURED JSON REPORT ===============", COLOR_MAGENTA))
+        # Pretty print the JSON report
+        print(color_text(json.dumps(report, indent=4, ensure_ascii=False), COLOR_GRAY))
+
+        print(color_text("\n=============== FORMATTED RESEARCH REPORT ===============", COLOR_GREEN))
+        if report["status"] == "success":
+            print(color_text("SUMMARY:", COLOR_YELLOW))
+            print(report["summary"])
+            print("\n")
+            print(color_text("DETAILED FINDINGS:", COLOR_YELLOW))
+            
+            # Format detailed findings section more clearly
+            detailed_findings = report["detailed_findings"]
+            if isinstance(detailed_findings, list):
+                for finding in detailed_findings:
+                    print(f"• {finding}")
+            else:
+                # If it's a string, try to split by newlines or periods to create bullet points
+                findings_text = str(detailed_findings)
+                if "\n\n" in findings_text:
+                    findings_list = [f.strip() for f in findings_text.split("\n\n") if f.strip()]
+                    for finding in findings_list:
+                        print(f"• {finding}")
+                else:
+                    print(findings_text)
+            
+            print("\n")
+            print(color_text("SOURCES:", COLOR_YELLOW))
+            if report["sources"] and len(report["sources"]) > 0:
+                for source in report["sources"]:
+                    print(f"• {source}")
+            else:
+                print("No sources identified")
+
+        elif report["status"] == "success_unstructured":
+            print(color_text("Research completed, but output could not be structured.", COLOR_YELLOW))
+            print("\n")
+            print(color_text("AGENT'S FULL TRACE:", COLOR_YELLOW))
+            print(report["detailed_findings"])
+
+        else: # failure
+            print(color_text(f"Research completed with status: {report['status']}", COLOR_YELLOW))
+            if report["error_message"]:
+                print(color_text(f"Error: {report['error_message']}", COLOR_RED))
+            print(report["detailed_findings"])
+
+        print(color_text("===============================================", COLOR_GREEN))
+        print(color_text(f"\nResearch log saved to: {log_file}", COLOR_GREEN))
+        
+    except Exception as e:
+        print(color_text(f"Error during research: {str(e)}", COLOR_RED))
+        traceback.print_exc()
+    
+    finally:
+        # Restore original stdout and close log file
+        sys.stdout = original_stdout # Restore original stdout
+        tee.close()
+        # Print final message to original stdout after tee is closed
+        original_stdout.write(f"Research log saved to: {log_file}\n")

components/__init__.py

@@ -0,0 +1,17 @@
+from .Web_agent_related.web_agent import WebAgent
+from .mcp_registry.registry import Registry
+from .code_generator_loop.mcp_brainstormer import Brainstormer
+from .code_generator_loop.script_generator import ScriptGenerator
+from .code_generator_loop.code_runner import CodeRunner
+from .mcp_registry.use_registry_tool import BasicMCPClient
+
+# This makes the components available when importing the package
+__all__ = [
+
+    "WebAgent",
+    "Registry",
+    "Brainstormer",
+    "ScriptGenerator",
+    "CodeRunner",
+    "BasicMCPClient"
+]

components/code_generator_loop/code_runner.py

@@ -0,0 +1,251 @@
+
+import subprocess
+import os
+import json
+import tempfile
+from typing import Optional, Dict, Any
+from models import MCPExecutionResult
+import sys # Import sys to check platform
+
+class CodeRunner:
+    """
+    Executes Python scripts in isolated environments (using conda).
+    Manages environment setup and captures output/errors.
+    """
+    def __init__(self, base_env_dir=".alita_envs"):
+        self.base_env_dir = os.path.abspath(base_env_dir) # Use absolute path
+        os.makedirs(self.base_env_dir, exist_ok=True)
+        print(f"CodeRunner initialized. Environments will be created in {self.base_env_dir}")
+        self._check_conda_available()
+
+
+    def _check_conda_available(self):
+        """Checks if the 'conda' command is available in the PATH."""
+        try:
+            subprocess.run(["conda", "--version"], check=True, capture_output=True, text=True)
+            print("Conda command found.")
+        except (subprocess.CalledProcessError, FileNotFoundError):
+            print("Warning: 'conda' command not found or not working. Environment setup/execution may fail.")
+            print("Please ensure conda is installed and accessible in your system's PATH.")
+
+
+    def _run_command(self, command: list[str], cwd: Optional[str] = None, env: Optional[Dict[str, str]] = None, timeout: int = 300) -> tuple[int, str, str]:
+        """Helper to run a shell command and capture output."""
+        print(f"CodeRunner: Running command: {' '.join(command)}")
+        try:
+            process = subprocess.Popen(
+                command,
+                stdout=subprocess.PIPE,
+                stderr=subprocess.PIPE,
+                text=True, # Decode stdout/stderr as text
+                cwd=cwd,
+                env=env
+            )
+            stdout, stderr = process.communicate(timeout=timeout)
+            print(f"CodeRunner: Command finished with return code {process.returncode}")
+            print("--- STDOUT ---")
+            print(stdout)
+            print("--- STDERR ---")
+            print(stderr)
+            return process.returncode, stdout, stderr
+        except FileNotFoundError:
+            print(f"CodeRunner Error: Command not found: {command[0]}")
+            return 1, "", f"Command not found: {command[0]}"
+        except subprocess.TimeoutExpired:
+            process.kill()
+            stdout, stderr = process.communicate()
+            print(f"CodeRunner Error: Command timed out after {timeout} seconds.")
+            print("--- STDOUT (before timeout) ---")
+            print(stdout)
+            print("--- STDERR (before timeout) ---")
+            print(stderr)
+            return 1, stdout, stderr
+        except Exception as e:
+            print(f"CodeRunner Error: An exception occurred running command: {e}")
+            return 1, "", str(e)
+
+
+    def setup_environment(self, env_script: str, env_name: str) -> bool:
+        """
+        Sets up a conda environment by running the provided script.
+        Returns True on success, False on failure.
+        """
+        print(f"CodeRunner: Attempting to set up environment '{env_name}'...")
+        # Create a temporary directory to run the setup script from
+        # This helps if the script assumes being run in a specific location
+        temp_dir = tempfile.mkdtemp(prefix=f"alita_env_setup_{env_name}_", dir=self.base_env_dir)
+        temp_script_path = os.path.join(temp_dir, f"setup_{env_name}.sh")
+
+        try:
+            # Write the setup script to a temporary file
+            with open(temp_script_path, "w") as f:
+                f.write(env_script)
+
+            # Make the script executable
+            os.chmod(temp_script_path, 0o755)
+
+            # Command to execute the setup script
+            # We need to source the user's bash profile to ensure conda is initialized
+            # This is a common pattern on Unix-like systems. Adjust for Windows if needed.
+            if sys.platform == "win32":
+                # Windows specific logic (might need adjustments for cmd vs powershell)
+                # Example using Git Bash or similar environment
+                # command = ["bash", "-c", f"source ~/.bashrc && bash {temp_script_path}"]
+                print("Warning: Conda environment setup on Windows requires specific shell configurations.")
+                print("Attempting a generic bash execution which might not work depending on your setup.")
+                command = ["bash", "-c", f"source ~/.bashrc; {temp_script_path}"] # Basic attempt
+            else: # Linux, macOS etc.
+                command = ["bash", "-c", f"source ~/.bashrc || source ~/.bash_profile || source /opt/anaconda3/etc/profile.d/conda.sh; {temp_script_path}"] # Try common paths
+
+            # Need to pass the environment name to the script somehow if the script needs it
+            # For now, assume the script is generic or doesn't need the specific name passed this way.
+            # A better approach might be to generate the script *with* the env_name embedded.
+
+            # Run the setup command from the temporary directory
+            returncode, stdout, stderr = self._run_command(command, cwd=temp_dir, timeout=120) # Setup can take time
+
+            if returncode != 0:
+                print(f"CodeRunner: Environment setup failed for '{env_name}'. Return code: {returncode}")
+                print("STDOUT:\n", stdout)
+                print("STDERR:\n", stderr)
+                return False
+
+            print(f"CodeRunner: Environment '{env_name}' setup successful.")
+            return True
+
+        except Exception as e:
+            print(f"CodeRunner: Error during environment setup for '{env_name}': {e}")
+            return False
+        finally:
+            # Clean up the temporary script file and directory
+            if os.path.exists(temp_script_path):
+                os.remove(temp_script_path)
+            if os.path.exists(temp_dir):
+                try:
+                    os.rmdir(temp_dir) # Will only work if empty
+                except OSError:
+                    print(f"Warning: Could not remove temporary setup directory {temp_dir}. It might not be empty.")
+
+
+    def execute(self, python_script: str, env_name: str, input_data: Optional[Dict[str, Any]] = None, timeout: int = 60) -> MCPExecutionResult:
+        """
+        Executes a Python script within a specified conda environment.
+        Passes input_data as a JSON string via command line argument.
+        Captures stdout, stderr, and return code.
+        """
+        print(f"CodeRunner: Executing script in environment '{env_name}'...")
+        # Create a temporary directory to run the script from
+        temp_dir = tempfile.mkdtemp(prefix=f"alita_script_run_{env_name}_", dir=self.base_env_dir)
+        temp_script_path = os.path.join(temp_dir, "script.py") # Fixed script name
+
+        try:
+            with open(temp_script_path, "w") as f:
+                f.write(python_script)
+
+            # Make the script executable (optional but good practice)
+            os.chmod(temp_script_path, 0o755)
+
+            # Command to activate conda env and run the script
+            # Requires conda to be initialized in the shell context
+            # Pass input_data as a JSON string command-line argument
+            input_json_str = json.dumps(input_data) if input_data is not None else json.dumps({}) # Ensure input is always a JSON string
+
+            if sys.platform == "win32":
+                # Windows specific logic
+                print("Warning: Script execution on Windows requires specific shell configurations.")
+                print("Attempting a generic bash execution which might not work depending on your setup.")
+                command_parts = [
+                    "bash", "-c",
+                    f"source ~/.bashrc || source ~/.bash_profile || source /opt/anaconda3/etc/profile.d/conda.sh; " # Source conda
+                    f"conda activate {env_name} && " # Activate env
+                    f"python {temp_script_path} --input '{input_json_str}'" # Run script with input arg
+                ]
+            else: # Linux, macOS etc.
+                command_parts = [
+                    "bash", "-c",
+                    f"source ~/.bashrc || source ~/.bash_profile || source /opt/anaconda3/etc/profile.d/conda.sh; " # Source conda
+                    f"conda activate {env_name} && " # Activate env
+                    f"python {temp_script_path} --input '{input_json_str}'" # Run script with input arg
+                ]
+
+
+            returncode, stdout, stderr = self._run_command(command_parts, cwd=temp_dir, timeout=timeout) # Script execution timeout
+
+            output_data = None
+            error_message = None
+            success = (returncode == 0)
+
+            if success:
+                try:
+                    # Attempt to parse stdout as JSON
+                    output_data = json.loads(stdout.strip())
+                except json.JSONDecodeError:
+                    # If stdout isn't JSON, return it as a raw string output
+                    print("CodeRunner: Script stdout was not valid JSON. Returning raw output.")
+                    output_data = {"raw_output": stdout.strip()}
+                except Exception as e:
+                    print(f"CodeRunner: Error parsing JSON output: {e}")
+                    # Even if parsing failed, mark as success if return code was 0
+                    output_data = {"raw_output": stdout.strip(), "parsing_error": str(e)}
+
+            else:
+                error_message = f"Script execution failed with return code {returncode}.\nSTDERR:\n{stderr}"
+                print(error_message)
+
+            return MCPExecutionResult(
+                success=success,
+                output=output_data if success else None,
+                logs=f"STDOUT:\n{stdout}\nSTDERR:\n{stderr}",
+                error_message=error_message,
+                return_code=returncode
+            )
+
+        except FileNotFoundError:
+            error_msg = f"CodeRunner Error: Python interpreter or script not found. Ensure '{env_name}' env is correct and script path exists ({temp_script_path})."
+            print(error_msg)
+            return MCPExecutionResult(success=False, error_message=error_msg)
+        except Exception as e:
+            error_msg = f"CodeRunner Error during script execution: {e}"
+            print(error_msg)
+            return MCPExecutionResult(success=False, error_message=error_msg)
+        finally:
+            # Clean up the temporary script file and directory
+            if os.path.exists(temp_script_path):
+                os.remove(temp_script_path)
+            if os.path.exists(temp_dir):
+                try:
+                    os.rmdir(temp_dir) # Will only work if empty
+                except OSError:
+                    print(f"Warning: Could not remove temporary script run directory {temp_dir}. It might not be empty.")
+
+
+    def cleanup_environment(self, env_name: str):
+        """Cleans up a conda environment."""
+        print(f"CodeRunner: Cleaning up environment '{env_name}'...")
+        try:
+            # Command to deactivate and remove the environment
+            # Need to source bash profile to ensure conda command works
+            if sys.platform == "win32":
+                command = [
+                    "bash", "-c",
+                    f"source ~/.bashrc || source ~/.bash_profile || source /opt/anaconda3/etc/profile.d/conda.sh; " # Source conda
+                    f"conda env remove -n {env_name} -y"
+                ]
+            else: # Linux, macOS etc.
+                command = [
+                    "bash", "-c",
+                    f"source ~/.bashrc || source ~/.bash_profile || source /opt/anaconda3/etc/profile.d/conda.sh; " # Source conda
+                    f"conda env remove -n {env_name} -y"
+                ]
+
+
+            returncode, stdout, stderr = self._run_command(command, timeout=60)
+            if returncode != 0:
+                print(f"CodeRunner: Environment cleanup failed for '{env_name}'. Return code: {returncode}")
+                print("STDOUT:\n", stdout)
+                print("STDERR:\n", stderr)
+            else:
+                print(f"CodeRunner: Environment '{env_name}' cleaned up successfully.")
+
+        except Exception as e:
+            print(f"CodeRunner: Error during environment cleanup for '{env_name}': {e}")

components/code_generator_loop/mcp_brainstormer.py

@@ -0,0 +1,153 @@
+import json
+import re
+from typing import Any, Dict, List, Union
+
+import gradio as gr
+from llama_index.llms.anthropic import Anthropic
+from llama_index.core.prompts import PromptTemplate
+
+class Brainstormer:
+    def __init__(self, model_name: str = "claude-sonnet-4-0"):
+        self.model_name = model_name
+        self.example_tool = json.dumps(
+            [
+                {
+                    "name": "tool_a",
+                    "input_schema": {"param1": "string"},
+                    "output_schema": {"result": "string"},
+                    "description": "Does X for Y",
+                    "pseudo_code": "def tool_a(param1: str) -> str:\n    # …",
+                    "source_hint": "use library_x"
+                }
+            ],
+            indent=2
+        )
+        
+        self.mcp_tools_spec_prompt = PromptTemplate(
+            """
+You are the MCPBrainstorm agent for OpenALITA, responsible for proposing MCPToolSpec entries needed to fulfill a new user task.
+
+User Task:
+{task}
+
+Already‐available tools (JSON array):
+{tools_list}
+
+Example to illustrate format (single JSON object):
+{example_tool}
+
+• If one or more of the listed tools can exactly fulfill the task as‐is, copy their JSON entries verbatim.
+• Otherwise, add only the minimal new specifications required—no extras.
+
+Note: A web agent is available for Internet or GitHub searches. Do not create a tool for that.
+
+Propose only generic tools, not highly specific ones.
+
+Output exactly a JSON ARRAY of objects, each with these keys in this order:
+- "name": string — unique tool name in snake_case (no spaces)
+- "input_schema": dict — JSON describing argument names and types
+- "output_schema": dict — JSON describing return names and types
+- "description": string — one‐sentence summary of what this tool does
+- "pseudo_code": string — brief multi‐line pseudo‐code sketch
+- "source_hint": string — short hint about which library or resource to use
+- "state" : string - indicator if the tool is deployed or not. Put deactivated in case of a new tool
+
+Do not include any explanatory text, comments, or formatting outside valid JSON.
+"""
+        )
+
+    def _extract_json_array(self, text: str) -> Union[List[Any], None]:
+        idx = text.find('[')
+        if idx < 0:
+            return None
+        depth = 0
+        for i, ch in enumerate(text[idx:], idx):
+            if ch == '[':
+                depth += 1
+            elif ch == ']':
+                depth -= 1
+                if depth == 0:
+                    try:
+                        return json.loads(text[idx : i+1])
+                    except json.JSONDecodeError:
+                        return None
+        return None
+
+    def generate_mcp_specs_to_fulfill_user_task(
+        self,
+        task: str,
+        tools_list: str,
+        retries: int = 3,
+    ) -> Union[List[Dict[str, Any]], Dict[str, str]]:
+        # 1) Validate inputs
+        if not task.strip():
+            return {"error": "User task must not be empty."}
+        raw = tools_list.strip()
+        if not raw:
+            return {"error": "Please provide a list of existing tools (or 'none')."}
+
+        # 2) Normalize the tools_list so that each tool is on its own line
+        #    (Assumes comma‐ or newline‐separated)
+        entries = re.split(r"[\n,]+", raw)
+        normalized = "\n".join(f"- {e.strip()}" for e in entries if e.strip())
+
+        # 3) Build the prompt
+        prompt_text = self.mcp_tools_spec_prompt.format(
+            task=task,
+            tools_list=normalized,
+            example_tool=self.example_tool
+        )
+
+        # 4) Call the LLM with retry logic
+        last_error = None
+        for _ in range(retries):
+            try:
+                llm = Anthropic(model=self.model_name, temperature=0.0, max_tokens=512, timeout=30)
+                response_text = llm.complete(prompt_text).text.strip()
+            except Exception as e:
+                last_error = e
+                continue
+
+            try:
+                specs = json.loads(response_text)
+            except json.JSONDecodeError:
+                # This is a fallback: find the first top‐level JSON array
+                specs = self._extract_json_array(response_text)
+
+            if isinstance(specs, list):
+                return specs
+
+        if last_error:
+            return {"error": f"Failed to call LLM: {last_error}"}
+        else:
+            return {"error": "Could not extract a valid JSON array from the LLM output."}
+
+    def create_interface(self):
+        return gr.Interface(
+            fn=self.generate_mcp_specs_to_fulfill_user_task,
+            inputs=[
+                gr.TextArea(
+                    label="Enter your task for OpenALITA",
+                    placeholder="e.g., Extract subtitles from a video online",
+                    lines=2,
+                ),
+                gr.TextArea(
+                    label="Available Tools List (one per line or comma-separated)",
+                    placeholder="e.g.\ntool_a, tool_b\nor:\ntool_a\ntool_b\ntool_c",
+                    lines=7,
+                ),
+            ],
+            outputs=gr.JSON(label="Generated MCPToolSpecs"),
+            flagging_mode="never",
+        )
+
+    def launch_interface(self, share=False):
+        iface = self.create_interface()
+        iface.launch(share=share, server_name="0.0.0.0")
+
+
+if __name__ == "__main__":
+    # Create an instance of the MCPBrainstormer class
+    brainstormer = Brainstormer()
+    # Launch the interface
+    brainstormer.launch_interface()

components/code_generator_loop/script_generator.py

@@ -0,0 +1,91 @@
+import anthropic
+import os
+
+def clean_script_output(script: str) -> str:
+    # Remove Markdown code fences or leading markdown syntax
+    lines = script.strip().splitlines()
+    if lines and lines[0].strip().startswith("```") or lines[0].strip().startswith("'''"):
+        lines = lines[1:]  # Remove first line
+    if lines and lines[-1].strip().startswith("```") or lines[-1].strip().startswith("'''"):
+        lines = lines[:-1]  # Remove last line
+    return "\n".join(lines)
+
+
+class ScriptGenerator:
+    def __init__(self, task_prompt: str, claude_api_key: str, model="claude-3-7-sonnet-20250219"):
+        self.task_prompt = task_prompt
+        self.client = anthropic.Anthropic(api_key=claude_api_key)
+        self.model = model
+
+    def build_prompt(self) -> str:
+        return f"""
+You are a Python expert working in an AI pipeline. Your job is to write a Python script based on the user's task description.
+
+The Python script must follow this exact structure:
+1. Modal Image setup to define the execution environment using `modal.Image.debian_slim().pip_install(...)`
+2. A tool function wrapped with `@stub.function` that performs the task
+3. A Gradio app setup that launches in the cloud using `@stub.local_entrypoint`
+
+🛑 VERY IMPORTANT:
+- Use `app = modal.App("tool-name")` (DO NOT use `stub = modal.Stub`)
+- All decorators must use `@app.function` and `@app.local_entrypoint`
+- La logique doit être dans une fonction locale (local_xyz()).
+- @app.function doit juste appeler cette fonction locale.
+- Gradio doit utiliser uniquement la fonction locale, pas celle décorée par Modal.
+
+The tool should:
+- Be self-contained
+- Import necessary libraries
+- Run via `modal` and expose an input/output via `gr.Interface`
+
+User's Task:
+\"\"\"{self.task_prompt}\"\"\"
+
+Output the full Python script only, no commentary or markdown.
+        """
+
+    def call_claude(self) -> str:
+        response = self.client.messages.create(
+            model=self.model,
+            max_tokens=1000,
+            temperature=0.5,
+            messages=[
+                {"role": "user", "content": self.build_prompt()}
+            ]
+        )
+        return response.content[0].text.strip()
+
+    
+    def clean_script_output(self, script: str) -> str:
+        # Remove Markdown code fences like '''python or ```python
+        lines = script.strip().splitlines()
+        if lines and lines[0].strip().startswith(("```", "'''")):
+            lines = lines[1:]  # Remove first line
+        if lines and lines[-1].strip().startswith(("```", "'''")):
+            lines = lines[:-1]  # Remove last line
+        return "\n".join(lines)
+
+    
+    def generate_script(self) -> str:
+        script = self.call_claude()
+        script = self.clean_script_output(script)
+        return script
+
+
+if __name__ == "__main__":
+    import os
+    from dotenv import load_dotenv
+    load_dotenv()
+
+
+    claude_key = os.getenv("CLAUDE_API_KEY")  # Or set manually
+    task = "Create a tool that takes a text input and returns the number of words in the text."
+
+    generator = ScriptGenerator(task_prompt=task, claude_api_key=claude_key)
+    script = generator.generate_script()
+
+    # Print or save to file
+    print(script)
+
+    with open("generated_tool.py", "w") as f:
+        f.write(script)

components/mcp_registry/mcp_registry.py

@@ -0,0 +1,117 @@
+import os
+import json
+import subprocess
+import threading
+from typing import Dict, Optional, List, Any
+from models import MCPToolSpec
+import modal
+
+# ---------- CONFIG ----------
+REGISTRY_FILE = "mcp_registry.json"
+UNDEPLOY_DELAY = 30 * 60  # seconds (30 minutes)
+
+# ---------- HELPER FUNCTIONS ----------
+
+def _read_registry() -> Dict[str, MCPToolSpec]:
+    if not os.path.exists(REGISTRY_FILE):
+        return {}
+    try:
+        with open(REGISTRY_FILE, 'r') as f:
+            data = json.load(f)
+        return {name: MCPToolSpec.from_dict(tool_data) for name, tool_data in data.items()}
+    except (json.JSONDecodeError, TypeError) as e:
+        print(f"Error reading registry: {e}. Returning empty registry.")
+        return {}
+
+
+def _write_registry(tools: Dict[str, MCPToolSpec]) -> None:
+    try:
+        serializable = {name: spec.to_dict() for name, spec in tools.items()}
+        with open(REGISTRY_FILE, 'w') as f:
+            json.dump(serializable, f, indent=4)
+    except IOError as e:
+        print(f"Error writing registry: {e}")
+
+# ---------- REGISTRY OPERATIONS ----------
+
+def register_tool(spec: MCPToolSpec) -> None:
+    tools = _read_registry()
+    tools[spec.name] = spec
+    _write_registry(tools)
+    print(f"Registered tool: {spec.name}")
+
+
+def get_tool(name: str) -> Optional[MCPToolSpec]:
+    tools = _read_registry()
+    return tools.get(name)
+
+# To be used by the manager to get the list of tools 
+def list_tools() -> List[MCPToolSpec]:
+    return list(_read_registry().values())
+
+
+def find_relevant_tools(query: str) -> List[Dict[str, Any]]:
+    query_lower = query.lower()
+    tools = _read_registry()
+    relevant = []
+    for spec in tools.values():
+        if query_lower in spec.name.lower() or query_lower in getattr(spec, 'description', '').lower():
+            relevant.append({"name": spec.name, "description": spec.description})
+    print(f"Found {len(relevant)} relevant tools for query '{query}'")
+    return relevant
+
+
+def _schedule_undeploy(name: str):
+    """
+    Schedule undeployment after a delay.
+    """
+    timer = threading.Timer(UNDEPLOY_DELAY, undeploy_tool, args=[name])
+    timer.daemon = False
+    timer.start()
+    print(f"Scheduled undeploy of {name} in {UNDEPLOY_DELAY} seconds.")
+
+
+# A function to be used by the manager to deploy deactivated tools 
+def deploy_tool(name: str) -> Optional[str]:
+    spec = get_tool(name)
+    if not spec:
+        print(f"Error: tool '{name}' not found in registry.")
+        return None
+    script_path = os.path.join("mcp_tool_box", f"{name}.py")
+    if not os.path.isfile(script_path):
+        print(f"Error: script file '{script_path}' does not exist.")
+        return None
+    try:
+        # Deploy via CLI
+        subprocess.run(["modal", "deploy", script_path], check=True)
+        # Retrieve URL via SDK
+        func = modal.Function.from_name(name, name)
+        url = func.get_web_url()
+        spec.url = url
+        spec.state = "activated"
+        register_tool(spec)
+        print(f"Deployed {name} at {url}")
+        # Schedule automatic undeploy
+        _schedule_undeploy(name)
+        return url
+    except Exception as e:
+        print(f"Deployment failed for {name}: {e}")
+        return None
+
+
+def undeploy_tool(name: str) -> bool:
+    spec = get_tool(name)
+    if not spec:
+        print(f"Error: tool '{name}' not found in registry.")
+        return False
+    try:
+        app = modal.App(name)
+        app.delete()
+        spec.state = "deactivated"
+        register_tool(spec)
+        print(f"Undeployed {name}")
+        return True
+    except Exception as e:
+        print(f"Undeployment failed for {name}: {e}")
+        return False
+

components/mcp_registry/registry.py

@@ -0,0 +1,235 @@
+import os
+import json
+import subprocess
+import threading
+from typing import Dict, Optional, List, Any
+from models import MCPToolSpec
+import modal
+from llama_index.tools.mcp import BasicMCPClient
+
+class Registry:
+    """
+    Registry class for managing MCP tools.
+    This class encapsulates tool registration, lookup, and deployment functionality.
+    """
+    
+    def __init__(self, registry_file: str = "mcp_registry.json", undeploy_delay: int = 30 * 60):
+        """
+        Initialize the MCPRegistry.
+        
+        Args:
+            registry_file: Path to the JSON file storing registry data
+            undeploy_delay: Delay in seconds before undeploying a tool (default: 30 minutes)
+        """
+        self.registry_file = registry_file
+        self.undeploy_delay = undeploy_delay
+        self.tools = self._read_registry()
+        print(f"MCPRegistry initialized. Using file: {self.registry_file}")
+    
+    def _read_registry(self) -> Dict[str, MCPToolSpec]:
+        """
+        Read the registry file and parse its contents.
+        
+        Returns:
+            Dictionary mapping tool names to MCPToolSpec objects
+        """
+        if not os.path.exists(self.registry_file):
+            return {}
+        try:
+            with open(self.registry_file, 'r') as f:
+                data = json.load(f)
+            return {name: MCPToolSpec.from_dict(tool_data) for name, tool_data in data.items()}
+        except (json.JSONDecodeError, TypeError) as e:
+            print(f"Error reading registry: {e}. Returning empty registry.")
+            return {}
+
+    def _write_registry(self) -> None:
+        """
+        Write the current tools to the registry file.
+        """
+        try:
+            serializable = {name: spec.to_dict() for name, spec in self.tools.items()}
+            with open(self.registry_file, 'w') as f:
+                json.dump(serializable, f, indent=4)
+        except IOError as e:
+            print(f"Error writing registry: {e}")
+
+    def register_tool(self, spec: MCPToolSpec) -> None:
+        """
+        Register a new tool in the registry.
+        
+        Args:
+            spec: MCPToolSpec object describing the tool
+        """
+        self.tools[spec.name] = spec
+        self._write_registry()
+        print(f"Registered tool: {spec.name}")
+
+    def get_tool(self, name: str) -> Optional[MCPToolSpec]:
+        """
+        Get a tool from the registry by name.
+        
+        Args:
+            name: Name of the tool to retrieve
+            
+        Returns:
+            MCPToolSpec object if found, None otherwise
+        """
+        return self.tools.get(name)
+
+    def list_tools(self) -> List[MCPToolSpec]:
+        """
+        Get a list of all tools in the registry.
+        
+        Returns:
+            List of MCPToolSpec objects
+        """
+        return list(self.tools.values())
+
+    def find_relevant_tools(self, query: str) -> List[Dict[str, Any]]:
+        """
+        Find tools relevant to a query.
+        
+        Args:
+            query: Search query
+            
+        Returns:
+            List of dictionaries with tool name and description
+        """
+        query_lower = query.lower()
+        relevant = []
+        for spec in self.tools.values():
+            if query_lower in spec.name.lower() or query_lower in getattr(spec, 'description', '').lower():
+                relevant.append({"name": spec.name, "description": spec.description})
+        print(f"Found {len(relevant)} relevant tools for query '{query}'")
+        return relevant
+
+    def _schedule_undeploy(self, name: str) -> None:
+        """
+        Schedule undeployment of a tool after a delay.
+        
+        Args:
+            name: Name of the tool to undeploy
+        """
+        timer = threading.Timer(self.undeploy_delay, self.undeploy_tool, args=[name])
+        timer.daemon = False
+        timer.start()
+        print(f"Scheduled undeploy of {name} in {self.undeploy_delay} seconds.")
+
+    def deploy_tool(self, name: str) -> Optional[str]:
+        """
+        Deploy a tool using Modal.
+        
+        Args:
+            name: Name of the tool to deploy
+            
+        Returns:
+            URL of the deployed tool if successful, None otherwise
+        """
+        spec = self.get_tool(name)
+        if not spec:
+            print(f"Error: tool '{name}' not found in registry.")
+            return None
+        script_path = os.path.join("mcp_tool_box", f"{name}.py")
+        if not os.path.isfile(script_path):
+            print(f"Error: script file '{script_path}' does not exist.")
+            return None
+        try:
+            # Deploy via CLI
+            subprocess.run(["modal", "deploy", script_path], check=True)
+            # Retrieve URL via SDK
+            func = modal.Function.from_name(name, name)
+            url = func.get_web_url()
+            spec.url = url
+            spec.state = "activated"
+            self.register_tool(spec)
+            print(f"Deployed {name} at {url}")
+            # Schedule automatic undeploy
+            self._schedule_undeploy(name)
+            return url
+        except Exception as e:
+            print(f"Deployment failed for {name}: {e}")
+            return None
+
+    def undeploy_tool(self, name: str) -> bool:
+        """
+        Undeploy a tool using Modal.
+        
+        Args:
+            name: Name of the tool to undeploy
+            
+        Returns:
+            True if successful, False otherwise
+        """
+        spec = self.get_tool(name)
+        if not spec:
+            print(f"Error: tool '{name}' not found in registry.")
+            return False
+        try:
+            app = modal.App(name)
+            app.delete()
+            spec.state = "deactivated"
+            self.register_tool(spec)
+            print(f"Undeployed {name}")
+            return True
+        except Exception as e:
+            print(f"Undeployment failed for {name}: {e}")
+            return False 
+            
+    def use_tool(self, tool_name: str, *args, **kwargs) -> Dict[str, Any]:
+        """
+        Use a registered tool directly by invoking its Modal endpoint.
+        
+        This method checks if the tool exists in the registry, ensures it's deployed,
+        and then invokes it using the BasicMCPClient.
+        
+        Args:
+            tool_name: Name of the tool to use
+            *args: Positional arguments to pass to the tool
+            **kwargs: Keyword arguments to pass to the tool
+            
+        Returns:
+            The response from the tool as a Python object
+            
+        Raises:
+            ValueError: If the tool doesn't exist or isn't deployed
+        """
+        print(f"🔧 Using tool: {tool_name}")
+        
+        # Check if tool exists in registry
+        spec = self.get_tool(tool_name)
+        if not spec:
+            error_msg = f"Tool '{tool_name}' not found in registry"
+            print(f"❌ {error_msg}")
+            raise ValueError(error_msg)
+        
+        # Check if tool is deployed
+        if getattr(spec, 'state', 'deactivated') != 'activated' or not getattr(spec, 'url', None):
+            print(f"Tool '{tool_name}' is not deployed. Attempting to deploy...")
+            url = self.deploy_tool(tool_name)
+            if not url:
+                error_msg = f"Failed to deploy tool '{tool_name}'"
+                print(f"❌ {error_msg}")
+                raise ValueError(error_msg)
+        
+        try:
+            # Construct the Modal endpoint URL
+            if hasattr(spec, 'url') and spec.url:
+                base_url = spec.url
+            else:
+                base_url = f"https://{tool_name}--{tool_name}.modal.run"
+            
+            print(f"🔄 Connecting to tool at: {base_url}")
+            client = BasicMCPClient(base_url)
+            
+            # Invoke the remote tool
+            print(f"▶️ Invoking tool with args: {args}, kwargs: {kwargs}")
+            response = client.call(*args, **kwargs)
+            print(f"✅ Tool execution successful")
+            
+            return response
+            
+        except Exception as e:
+            error_msg = f"Error using tool '{tool_name}': {str(e)}"
+            print(f"🚨 {error_msg}")
+            raise RuntimeError(error_msg) from e 

components/mcp_registry/use_registry_tool.py

@@ -0,0 +1,91 @@
+import aiohttp
+from typing import Dict, Any
+import modal
+
+class BasicMCPClient:
+    """
+    A basic MCP client implementation since llama_index.tools.mcp might not be available
+    """
+    def __init__(self, base_url: str):
+        self.base_url = base_url.rstrip('/')
+        
+    async def call_tool(self, tool_name: str, arguments: Dict[str, Any]) -> Dict[str, Any]:
+        """
+        Call an MCP tool via HTTP POST request
+        """
+        mcp_request = {
+            "jsonrpc": "2.0",
+            "id": 1,
+            "method": "tools/call",
+            "params": {
+                "name": tool_name,
+                "arguments": arguments
+            }
+        }
+        
+        async with aiohttp.ClientSession() as session:
+            try:
+                async with session.post(
+                    self.base_url,
+                    json=mcp_request,
+                    headers={"Content-Type": "application/json"}
+                ) as response:
+                    if response.status == 200:
+                        result = await response.json()
+                        return result
+                    else:
+                        error_text = await response.text()
+                        return {
+                            "error": f"HTTP {response.status}: {error_text}"
+                        }
+            except Exception as e:
+                return {
+                    "error": f"Request failed: {str(e)}"
+                }
+
+    async def list_tools(self) -> Dict[str, Any]:
+        """
+        List available tools from the MCP server
+        """
+        mcp_request = {
+            "jsonrpc": "2.0",
+            "id": 1,
+            "method": "tools/list",
+            "params": {}
+        }
+        
+        async with aiohttp.ClientSession() as session:
+            try:
+                async with session.post(
+                    self.base_url,
+                    json=mcp_request,
+                    headers={"Content-Type": "application/json"}
+                ) as response:
+                    if response.status == 200:
+                        result = await response.json()
+                        return result
+                    else:
+                        error_text = await response.text()
+                        return {
+                            "error": f"HTTP {response.status}: {error_text}"
+                        }
+            except Exception as e:
+                return {
+                    "error": f"Request failed: {str(e)}"
+                }
+
+# Function to invoke an external MCP server
+async def use_registry_tool(tool_name: str, **kwargs):
+    """
+    Call a registered MCP tool by name with the provided arguments.
+    Args:
+        tool_name: Name of the tool.
+        **kwargs: Keyword arguments matching the tool signature.
+    Returns:
+        The JSON response from the MCP server as a Python object.
+    """
+    
+    base_url = modal.Function.from_name("tool_name", "mcp_server").get_web_url()
+    client = BasicMCPClient(base_url)
+    response = await client.call_tool(tool_name, kwargs)
+    return response

manager_agent.py

@@ -0,0 +1,663 @@
+import uuid
+import os
+from dotenv import load_dotenv
+from typing import Optional, Dict, Any, List, Generator, Callable
+from models import TaskPrompt, MCPToolSpec, MCPExecutionResult
+from components import (
+    WebAgent,
+    ScriptGenerator,
+    CodeRunner,
+    Registry,
+    Brainstormer,
+)
+from llama_index.core.llms import LLM
+from llama_index.core.agent import ReActAgent
+from llama_index.core.tools import FunctionTool
+
+
+# Load environment variables from .env file
+load_dotenv()
+
+class ManagerAgent:
+    """
+    The central orchestrator of the Alita agent - Revised for Gradio integration.
+    
+    Workflow:
+    1. Analyze user prompt to understand the request
+    2. Check existing tools in registry first
+    3. If research needed, formulate search queries and use WebAgent
+    4. If tool needed but not found, brainstorm new tool requirements
+    5. Search for open source tools/solutions via WebAgent
+    6. Create implementation plan via Brainstormer
+    7. Return comprehensive response
+    """
+    
+    def __init__(self, llm: LLM, max_iterations: int = 10000000, update_callback: Optional[Callable[[str], None]] = None):
+        
+        
+        self.llm = llm
+        self.registry = Registry()
+        self.web_agent = WebAgent(llm=llm, max_research_iterations=10000000)
+        self.code_runner = CodeRunner()
+        self.brainstormer = Brainstormer(model_name="claude-sonnet-4-0")
+        self.script_generator = ScriptGenerator(task_prompt="", claude_api_key=os.getenv("CLAUDE_API_KEY", ""))
+        self.max_iterations = max_iterations
+        self.update_callback = update_callback
+
+        # Define the tools available to the internal LlamaIndex Agent
+        self._agent_tools = self._define_agent_tools()
+
+        # Initialize the internal LlamaIndex ReAct Agent with improved system prompt
+        self.agent = ReActAgent.from_tools(
+            tools=self._agent_tools,
+            llm=self.llm,
+            verbose=True,
+            system_prompt=self._get_system_prompt(),
+            max_iterations=self.max_iterations,  # Use the configurable max_iterations parameter
+            temperature=0.2  # Lower temperature for more focused responses
+        )
+        print("🤖 ManagerAgent initialized with ReActAgent and enhanced workflow (temperature=0.2).")
+
+    def send_update(self, message: str) -> None:
+        """
+        Send an update message to the user about the agent's progress.
+        """
+        if not any(emoji in message[:2] for emoji in ["📢", "🔄", "✅", "❌", "⚠️", "💬", "🔍", "🚀", "✨"]):
+            message = f"📢 {message}"
+            
+        print(f"📣 AGENT: ManagerAgent.send_update CALLED with message: {message}") # DEBUG
+        print(f"📣 AGENT: self.update_callback is: {self.update_callback}") # DEBUG
+        
+        if self.update_callback:
+            try:
+                self.update_callback(message) # This should call update_status_callback in app.py
+                print(f"📣 AGENT: Callback invoked successfully.") # DEBUG
+            except Exception as e:
+                print(f"❌ AGENT: Error sending update via callback: {e}")
+                import traceback
+                traceback.print_exc()
+        else:
+            print("📣 AGENT: No update_callback configured for ManagerAgent.") # DEBUG
+        # Return a string confirmation, as ReAct tools often expect a string output
+        return f"Update sent: {message}" # MODIFICATION: Return a string
+
+    def _get_system_prompt(self) -> str:
+        """Enhanced system prompt for better workflow orchestration"""
+        return """You are ALITA, an advanced generalist agent. You are here to help people with their requests. You can do many tasks like research, tool creation, automation, analysis, and much more. What is unique about you is that you can create tools to help people with their requests, even if they are not in your capabilities.
+
+Your primary workflow for ANY user request:
+
+1.  **ANALYZE PHASE**:
+    *   Understand the user's request deeply.
+    *   Identify if it's: an information request, a tool request, task automation, research, or creative work.
+    *   Decide whether to answer the request directly, create a new tool, or perform web research.
+    *   If you decide to answer directly, provide your answer right away.
+    *   If you decide to perform web research, use the `perform_web_research` tool with specific queries. Inform the user you are starting research before taking this action.
+    *   If the task requires more than simple text generation or basic web research, proceed to check for existing tools.
+    *   Use `send_user_update` to inform the user about what you're doing and your progress if you don't answer directly.
+    *   Do not apologize for not being able to answer the prompt until you have attempted all subsequent steps (EXISTING TOOLS CHECK, TOOL ANALYSIS PHASE, RESEARCH PHASE, TOOL CREATION PHASE). If all fail, then apologize.
+
+2.  **EXISTING TOOLS CHECK**:
+    *   ALWAYS first use `get_available_tools` to list all tools in your registry.
+    *   If suitable tools exist but are not deployed (check their 'state'), use `deploy_tool` to activate them.
+    *   Once tools are active and deployed, use `use_registry_tool` to execute them with the necessary inputs.
+    *   Keep the user informed of your progress with `send_user_update`.
+
+3.  **TOOL ANALYSIS PHASE**:
+    *   If you need to determine whether existing tools are sufficient or new tools are needed, use `brainstorm_tools`.
+    *   Provide the `brainstorm_tools` function with the `user_task` and the `available_tools` (a comma-separated string of tool names from `get_available_tools`).
+    *   If there are no tools available, provide "none" as the input for `available_tools` to the `brainstorm_tools` function.
+    *   Follow the recommendations from the brainstorming phase.
+    *   Send an update to the user with `send_user_update` about your findings.
+
+4.  **RESEARCH PHASE** (if needed for information or tool creation):
+    *   Use the `perform_web_research` tool for all web-based information gathering.
+        *   For general information or in-depth research on a topic, provide a clear query to `perform_web_research`.
+        *   If you are looking for open-source code, libraries, or technical solutions (including from GitHub), instruct `perform_web_research` in your query to focus on finding code examples or repositories. For instance: "perform_web_research: Find Python code snippets for parsing CSV files from GitHub."
+    *   Send updates to the user with `send_user_update` about your research progress.
+
+5.  **TOOL CREATION PHASE** (if no existing tool works or can be adapted):
+    *   First, use `brainstorm_tools` to define the specifications of the new tool needed.
+    *   Next, use `perform_web_research` to find existing open-source solutions, code examples, or libraries that can help build the tool. Be specific in your query to `perform_web_research` about looking for implementation details.
+    *   Then, use `generate_mcp_script` to create the Python code and environment script for the tool, using the specification from `brainstorm_tools` and insights from your research.
+    *   Finally, use `execute_and_register_mcp` to test the new tool in a safe environment and, if successful, register it in your tool registry.
+    *   Keep the user informed of your progress with `send_user_update`.
+
+6.  **EXECUTION PHASE** (after a tool is ready, either existing or newly created):
+    *   Ensure the required tool is deployed using `deploy_tool` if it's not already active.
+    *   Use `use_registry_tool` to run the active tool with the appropriate inputs.
+    *   Provide comprehensive results with explanations.
+    *   Send a final update to the user with `send_user_update` about the results.
+
+**Key Principles**:
+*   Be proactive in tool discovery and creation.
+*   Always search for existing solutions before creating new ones.
+*   Provide detailed explanations of your reasoning process.
+*   Focus on practical, actionable results.
+*   Leverage open-source resources extensively via `perform_web_research`.
+*   Keep the user informed of your progress with regular updates using `send_user_update`.
+
+**Tool Management Capabilities**:
+*   Use `get_available_tools` to see all tools in your registry.
+*   Use `brainstorm_tools` to analyze if existing tools are sufficient or new ones are needed.
+*   Check tool 'state' from `get_available_tools` to determine if they are active ('activated' or similar) or inactive.
+*   Use `deploy_tool` to activate any inactive tools before running them. Tools must be deployed before they can be executed by `use_registry_tool`.
+
+**Response Style**:
+*   Structure your responses clearly with headers where appropriate.
+*   Explain what you're doing and why.
+*   Provide context and next steps.
+*   Be conversational but informative.
+*   Use `send_user_update` to keep the user informed throughout the process.
+"""
+
+    def _define_agent_tools(self) -> List[FunctionTool]:
+        """Enhanced tool definition with better descriptions"""
+        tools = []
+
+        # User update tool
+        tools.append(
+            FunctionTool.from_defaults(
+                self.send_update,
+                name="send_user_update",
+                description="Send an update message to the user about your current progress or actions. Takes 'message' (string) containing the update information. Use this tool frequently to keep the user informed about what you're doing."
+            )
+        )
+        
+        # Add research tool
+        tools.append(
+            FunctionTool.from_defaults(
+                self.research,
+                name="perform_web_research",
+                description="Performs comprehensive web research on a given topic. Takes 'query' (string) containing the research question or topic to investigate. Returns a detailed research report with findings and sources."
+            )
+        )
+
+        # Get all available tools
+        tools.append(
+            FunctionTool.from_defaults(
+                self.get_available_tools,
+                name="get_available_tools",
+                description="Get a list of all tools currently available in the registry. Returns a list of tool specifications with names, descriptions, and states."
+            )
+        )
+
+        # Use a registered tool
+        tools.append(
+            FunctionTool.from_defaults(
+                self.use_registry_tool,
+                name="use_registry_tool",
+                description="Use a registered tool directly by invoking its endpoint. Takes 'tool_name' (string) and any additional arguments required by the tool. Automatically deploys the tool if needed. Returns the response from the tool."
+            )
+        )
+
+        # Tool brainstorming
+        tools.append(
+            FunctionTool.from_defaults(
+                self.brainstorm_tools,
+                name="brainstorm_tools",
+                description="Analyze the user request against available tools to determine if existing tools are sufficient or new tools are needed. Takes 'user_task' (string) containing the user's request and optionally 'available_tools' (string) with comma-separated tool names. Returns recommendations on which tools to use or what new tools to create."
+            )
+        )
+
+        # Deploy a specific tool
+        tools.append(
+            FunctionTool.from_defaults(
+                self.deploy_tool,
+                name="deploy_tool",
+                description="Deploy and activate a specific tool from the registry. Takes 'tool_name' (string) containing the name of the tool to deploy. Returns the URL of the deployed tool if successful, or an error message if deployment fails."
+            )
+        )
+
+        # Add analysis tool for better decision making
+        tools.append(
+            FunctionTool.from_defaults(
+                self._analyze_user_request,
+                name="analyze_user_request",
+                description="Analyze user request to determine the best approach (research, existing tool, new tool creation). Takes 'user_message' (string). Returns analysis with recommended actions."
+            )
+        )
+
+        return tools
+
+    def _analyze_user_request(self, user_message: str) -> Dict[str, Any]:
+        """Analyze user request to determine optimal workflow path"""
+        analysis = {
+            "request_type": "unknown",
+            "complexity": "medium",
+            "requires_research": False,
+            "requires_tools": False,
+            "suggested_approach": [],
+            "key_concepts": []
+        }
+        
+        message_lower = user_message.lower()
+        
+        # Look for comprehensive research indicators
+        research_terms = ["comprehensive", "thorough", "in-depth", "detailed", "extensive", 
+                        "research", "investigate", "analyze", "report", "study"]
+        
+        # Determine request type
+        if any(word in message_lower for word in research_terms):
+            analysis["request_type"] = "deep_research"
+            analysis["requires_research"] = True
+            analysis["complexity"] = "high"
+            analysis["suggested_approach"].append("research")
+            
+        elif any(word in message_lower for word in ["recherche", "search", "find", "lookup", "information", "what is", "explain"]):
+            analysis["request_type"] = "information_request"
+            analysis["requires_research"] = True
+            analysis["suggested_approach"].append("web_search")
+            
+        elif any(word in message_lower for word in ["outil", "tool", "script", "automatise", "automate", "create", "build"]):
+            analysis["request_type"] = "tool_request"
+            analysis["requires_tools"] = True
+            analysis["suggested_approach"].extend(["find_existing_tools", "brainstorm_if_needed"])
+            
+        elif any(word in message_lower for word in ["analyse", "analyze", "process", "calculate", "compute"]):
+            analysis["request_type"] = "analysis_task"
+            analysis["requires_tools"] = True
+            analysis["suggested_approach"].extend(["find_existing_tools", "research_methods"])
+            
+        elif any(word in message_lower for word in ["tendance", "trend", "market", "news", "current"]):
+            analysis["request_type"] = "research_task"
+            analysis["requires_research"] = True
+            analysis["complexity"] = "high"
+            analysis["suggested_approach"].extend(["web_search", "github_search"])
+
+        # Extract key concepts for better tool matching
+        concepts = []
+        tech_keywords = ["python", "javascript", "api", "database", "csv", "json", "web", "scraping", "ml", "ai"]
+        for keyword in tech_keywords:
+            if keyword in message_lower:
+                concepts.append(keyword)
+        analysis["key_concepts"] = concepts
+        
+        return analysis
+
+    def _run_and_register_mcp(self, spec: Dict[str, Any], python_script: str, env_script: str, input_data: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
+        """Enhanced MCP execution and registration with better error handling"""
+        print(f"🔧 ManagerAgent: Executing and registering MCP: {spec.get('name', 'Unnamed Tool')}")
+        
+        try:
+            mcp_spec_obj = MCPToolSpec.from_dict(spec)
+            env_name_suffix = mcp_spec_obj.name.lower().replace(' ', '-')[:10]
+            env_name = f"alita-{env_name_suffix}-{uuid.uuid4().hex[:8]}"
+
+            print(f"🔄 Setting up environment: {env_name}")
+            env_success = self.code_runner.setup_environment(env_script, env_name)
+            
+            if not env_success:
+                result = MCPExecutionResult(
+                    success=False, 
+                    error_message=f"Environment setup failed for '{env_name}'. Check dependencies in env_script."
+                )
+                return result.to_dict()
+
+            print(f"▶️ Executing script in environment: {env_name}")
+            execution_result = self.code_runner.execute(python_script, env_name, input_data)
+
+            if execution_result.success:
+                print(f"✅ Script execution successful. Registering tool: {mcp_spec_obj.name}")
+                mcp_spec_obj.validated_script = python_script
+                mcp_spec_obj.environment_script = env_script
+                self.registry.register_tool(mcp_spec_obj)
+                print(f"🎯 Tool '{mcp_spec_obj.name}' successfully registered in registry")
+                
+                # Add success message to result
+                execution_result.output_data = execution_result.output_data or {}
+                execution_result.output_data["registration_status"] = "Successfully registered"
+                
+            else:
+                print(f"❌ Script execution failed for '{mcp_spec_obj.name}': {execution_result.error_message}")
+
+            # Always cleanup after validation
+            self.code_runner.cleanup_environment(env_name)
+            return execution_result.to_dict()
+
+        except Exception as e:
+            error_msg = f"Unexpected error in MCP execution: {str(e)}"
+            print(f"🚨 {error_msg}")
+            
+            # Cleanup on error
+            try:
+                if 'env_name' in locals():
+                    self.code_runner.cleanup_environment(env_name)
+            except:
+                pass
+                
+            return MCPExecutionResult(success=False, error_message=error_msg).to_dict()
+
+    def _run_registered_mcp(self, tool_name: str, input_data: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
+        """Enhanced registered tool execution with better logging"""
+        print(f"🎯 ManagerAgent: Running registered tool: {tool_name}")
+        
+        spec = self.registry.get_tool(tool_name)
+        if not spec:
+            error_msg = f"Tool '{tool_name}' not found in registry. Available tools: {list(self.registry.tools.keys())}"
+            print(f"❌ {error_msg}")
+            return MCPExecutionResult(success=False, error_message=error_msg).to_dict()
+
+        if not spec.validated_script or not spec.environment_script:
+            error_msg = f"Tool '{tool_name}' missing validated script or environment configuration"
+            print(f"❌ {error_msg}")
+            return MCPExecutionResult(success=False, error_message=error_msg).to_dict()
+
+        # Create fresh environment for execution
+        env_name_suffix = spec.name.lower().replace(' ', '-')[:10]
+        env_name = f"alita-run-{env_name_suffix}-{uuid.uuid4().hex[:8]}"
+
+        try:
+            print(f"🔄 Setting up execution environment: {env_name}")
+            env_success = self.code_runner.setup_environment(spec.environment_script, env_name)
+            
+            if not env_success:
+                return MCPExecutionResult(
+                    success=False, 
+                    error_message=f"Failed to setup environment for tool '{tool_name}'"
+                ).to_dict()
+
+            print(f"▶️ Executing registered tool: {tool_name}")
+            execution_result = self.code_runner.execute(spec.validated_script, env_name, input_data)
+            
+            print(f"{'✅' if execution_result.success else '❌'} Tool execution completed. Success: {execution_result.success}")
+            
+            return execution_result.to_dict()
+            
+        except Exception as e:
+            error_msg = f"Error executing registered tool '{tool_name}': {str(e)}"
+            print(f"🚨 {error_msg}")
+            return MCPExecutionResult(success=False, error_message=error_msg).to_dict()
+            
+        finally:
+            # Always cleanup
+            try:
+                self.code_runner.cleanup_environment(env_name)
+            except:
+                pass
+
+    def run_task(self, prompt: TaskPrompt) -> str:
+        """
+        Enhanced task execution with detailed logging and structured workflow
+        Optimized for Gradio integration with comprehensive responses
+        """
+        print(f"\n{'='*60}")
+        print(f"🚀 ALITA ManagerAgent: Starting task execution")
+        print(f"📝 User prompt: {prompt.text[:100]}{'...' if len(prompt.text) > 100 else ''}")
+        print(f"{'='*60}")
+
+        # Send initial update to the user
+        self.send_update(f"Starting to process your request: '{prompt.text[:50]}{'...' if len(prompt.text) > 50 else ''}'")
+
+        try:
+            # Use the internal ReAct agent to handle the complete workflow
+            print("🧠 Engaging ReAct Agent for intelligent task orchestration...")
+            
+            # The ReAct agent will use its tools to:
+            # 1. Analyze the request
+            # 2. Search existing tools
+            # 3. Perform web research if needed
+            # 4. Brainstorm solutions
+            # 5. Create/execute tools as necessary
+            # 6. Provide comprehensive response
+            
+            response = self.agent.chat(prompt.text)
+            
+            print("✅ Task execution completed successfully")
+            print(f"{'='*60}\n")
+            
+            # Send final update to the user
+            self.send_update("Task completed successfully! Here's your response.")
+            
+            # Format response for better Gradio presentation
+            formatted_response = self._format_response_for_gradio(response.response)
+            return formatted_response
+            
+        except Exception as e:
+            error_msg = f"🚨 ManagerAgent encountered an error during task execution:\n\n**Error Details:**\n{str(e)}\n\n**Next Steps:**\n- Check your API key and network connection\n- Verify all components are properly initialized\n- Try a simpler request to test basic functionality"
+            
+            print(f"❌ Task execution failed: {e}")
+            print(f"{'='*60}\n")
+            
+            # Send error update to the user
+            self.send_update(f"An error occurred while processing your request: {str(e)}")
+            
+            return error_msg
+
+    def _format_response_for_gradio(self, response: str) -> str:
+        """Format the agent response for better presentation in Gradio"""
+        
+        # Add header if not present
+        if not response.startswith("##") and not response.startswith("#"):
+            response = f"## 🤖 {response}"
+        
+        
+            
+        return response
+
+    def get_registry_status(self) -> Dict[str, Any]:
+        """Get current status of the tool registry"""
+        return {
+            "total_tools": len(self.registry.tools),
+            "tool_names": list(self.registry.tools.keys()),
+            "registry_ready": len(self.registry.tools) > 0
+        }
+
+    def reset_registry(self):
+        """Reset the tool registry (useful for testing)"""
+        self.registry = Registry()
+        print("🔄 Tool registry has been reset")
+
+    def __str__(self):
+        return f"ManagerAgent(llm={type(self.llm).__name__}, tools_registered={len(self.registry.tools)})"
+
+    def research(self, query: str, max_iterations: int = None, verbose: bool = None) -> str:
+        """
+        Performs autonomous web research on the given query using the WebAgent's research function.
+        
+        Args:
+            query: The research question or topic
+            max_iterations: Optional override for the maximum number of research iterations
+            verbose: Optional override for verbose mode
+            
+        Returns:
+            A comprehensive textual report based on web research
+        """
+        print(f"\n{'='*60}")
+        print(f"🌐 ALITA ManagerAgent: Starting web research")
+        print(f"📝 Research query: {query[:100]}{'...' if len(query) > 100 else ''}")
+        print(f"{'='*60}")
+        
+        try:
+            # Configure WebAgent for this research session
+            if max_iterations is not None:
+                self.web_agent.max_research_iterations = max_iterations
+            
+            if verbose is not None:
+                self.web_agent.verbose = verbose
+            
+            # Perform the research
+            print("🔍 Initiating autonomous web research. This may take some time... here is the query: ", query)
+            report = self.web_agent.research(query)
+            print("🔍 here is the report: ", report)
+            
+            print("✅ Research completed successfully")
+            print(f"{'='*60}\n")
+            
+            return report
+            
+        except Exception as e:
+            error_msg = f"🚨 Error during web research: {str(e)}"
+            print(f"❌ Research failed: {e}")
+            print(f"{'='*60}\n")
+            
+            import traceback
+            print(traceback.format_exc())
+            
+            return error_msg
+
+    def get_available_tools(self) -> List[Dict[str, Any]]:
+        """
+        Get a list of all tools currently available in the registry.
+        
+        Returns:
+            List of dictionaries containing tool information (name, description, state)
+        """
+        print("📋 ManagerAgent: Retrieving list of all available tools")
+        tools = self.registry.list_tools()
+        
+        # Format the tools for easier consumption by the agent
+        formatted_tools = []
+        for tool in tools:
+            formatted_tools.append({
+                "name": tool.name,
+                "description": tool.description,
+                "state": getattr(tool, "state", "unknown"),
+                "input_schema": tool.input_schema if hasattr(tool, "input_schema") else {},
+                "output_schema": tool.output_schema if hasattr(tool, "output_schema") else {}
+            })
+            
+        print(f"🔍 Found {len(formatted_tools)} tools in registry")
+        return formatted_tools
+    
+    def deploy_tool(self, tool_name: str) -> Dict[str, Any]:
+        """
+        Deploy and activate a specific tool from the registry.
+        
+        Args:
+            tool_name: Name of the tool to deploy
+            
+        Returns:
+            Dictionary with deployment status and URL (if successful)
+        """
+        print(f"🚀 ManagerAgent: Deploying tool '{tool_name}'")
+        
+        # Check if tool exists in registry
+        if not self.registry.get_tool(tool_name):
+            error_msg = f"Tool '{tool_name}' not found in registry"
+            print(f"❌ {error_msg}")
+            return {"success": False, "error": error_msg}
+        
+        # Attempt to deploy the tool
+        try:
+            url = self.registry.deploy_tool(tool_name)
+            
+            if url:
+                print(f"✅ Successfully deployed tool '{tool_name}' at {url}")
+                return {
+                    "success": True,
+                    "tool_name": tool_name,
+                    "url": url,
+                    "message": f"Tool '{tool_name}' successfully deployed"
+                }
+            else:
+                error_msg = f"Failed to deploy tool '{tool_name}'"
+                print(f"❌ {error_msg}")
+                return {"success": False, "error": error_msg}
+                
+        except Exception as e:
+            error_msg = f"Error deploying tool '{tool_name}': {str(e)}"
+            print(f"🚨 {error_msg}")
+            return {"success": False, "error": error_msg}
+
+    def brainstorm_tools(self, user_task: str, available_tools: str = "") -> Dict[str, Any]:
+        """
+        Use the Brainstormer to analyze if existing tools are sufficient or new tools are needed.
+        
+        Args:
+            user_task: The user's request or task
+            available_tools: Optional comma-separated list of available tool names
+            
+        Returns:
+            Dictionary with tool recommendations or specifications for new tools
+        """
+        print(f"🧠 ManagerAgent: Brainstorming tools for task: {user_task[:100]}{'...' if len(user_task) > 100 else ''}")
+        
+        # If available_tools is not provided, get them from the registry
+        if not available_tools:
+            tools = self.get_available_tools()
+            available_tools = ", ".join([tool["name"] for tool in tools])
+        
+        try:
+            # Call the brainstormer to analyze the task and available tools
+            result = self.brainstormer.generate_mcp_specs_to_fulfill_user_task(
+                task=user_task,
+                tools_list=available_tools
+            )
+            
+            if isinstance(result, dict) and "error" in result:
+                print(f"❌ Brainstorming failed: {result['error']}")
+                return {
+                    "success": False,
+                    "error": result["error"],
+                    "recommendations": "Unable to analyze tools for this task."
+                }
+                
+            print(f"✅ Brainstorming complete. Found {len(result)} tool recommendations.")
+            
+            # Format the result for better consumption by the agent
+            return {
+                "success": True,
+                "recommendations": result,
+                "summary": f"Analysis complete. Found {len(result)} tool recommendations."
+            }
+            
+        except Exception as e:
+            error_msg = f"Error during tool brainstorming: {str(e)}"
+            print(f"🚨 {error_msg}")
+            return {
+                "success": False,
+                "error": error_msg,
+                "recommendations": "Unable to analyze tools due to an error."
+            }
+
+    def use_registry_tool(self, tool_name: str, *args, **kwargs) -> Dict[str, Any]:
+        """
+        Use a registered tool directly by invoking its endpoint.
+        
+        This method utilizes the Registry's use_tool method to invoke a registered tool.
+        It handles tool deployment if needed and provides proper error handling and user feedback.
+        
+        Args:
+            tool_name: Name of the tool to use
+            *args: Positional arguments to pass to the tool
+            **kwargs: Keyword arguments to pass to the tool
+            
+        Returns:
+            The response from the tool as a Python object
+        """
+        try:
+            # Send update to user
+            self.send_update(f"Using tool: {tool_name}")
+            
+            # Check if tool exists in registry
+            if not self.registry.get_tool(tool_name):
+                error_msg = f"Tool '{tool_name}' not found in registry"
+                self.send_update(error_msg)
+                return {"error": error_msg, "success": False}
+                
+            # Use the tool via Registry's use_tool method
+            self.send_update(f"Executing tool: {tool_name}")
+            result = self.registry.use_tool(tool_name, *args, **kwargs)
+            
+            # Send success update
+            self.send_update(f"Tool '{tool_name}' executed successfully")
+            
+            # Return result with success flag
+            if isinstance(result, dict):
+                result["success"] = True
+                return result
+            else:
+                return {"result": result, "success": True}
+                
+        except ValueError as e:
+            # Handle expected errors (tool not found, deployment failed)
+            error_msg = str(e)
+            self.send_update(f"Error: {error_msg}")
+            return {"error": error_msg, "success": False}
+            
+        except Exception as e:
+            # Handle unexpected errors
+            error_msg = f"Unexpected error using tool '{tool_name}': {str(e)}"
+            self.send_update(f"Error: {error_msg}")
+            return {"error": error_msg, "success": False} 

models/__init__.py

@@ -0,0 +1,3 @@
+from .mcp_tool_spec import MCPToolSpec
+from .mcp_execution_result import MCPExecutionResult
+from .task_prompt import TaskPrompt

models/mcp_execution_result.py

@@ -0,0 +1,21 @@
+from dataclasses import dataclass
+from typing import Optional, Dict, Any
+
+@dataclass
+class MCPExecutionResult:
+    """
+    Represents the result of executing an MCP tool script.
+    """
+    success: bool
+    output: Optional[Dict[str, Any]] = None # Structured output if successful
+    logs: Optional[str] = None             # Combined stdout/stderr
+    error_message: Optional[str] = None    # Error details if execution failed
+    return_code: Optional[int] = None      # Process return code
+
+    # Add methods to convert to/from dictionary for serialization
+    def to_dict(self):
+        return self.__dict__
+
+    @classmethod
+    def from_dict(cls, data: Dict[str, Any]):
+        return cls(**data)

models/mcp_tool_spec.py

@@ -0,0 +1,11 @@
+from dataclasses import dataclass
+
+@dataclass
+class MCPToolSpec:
+    name: str
+    input_schema: dict
+    output_schema: dict
+    description: str
+    pseudo_code: str
+    source_hint: str
+    state: str

models/task_prompt.py

@@ -0,0 +1,26 @@
+from typing import Dict, Any, Optional
+from dataclasses import dataclass
+
+@dataclass
+class TaskPrompt:
+    """
+    Represents a task prompt with its associated metadata and content.
+    """
+
+    text: str
+    metadata: Optional[Dict[str, Any]] = None
+    
+    def to_dict(self) -> Dict[str, Any]:
+        """Convert the TaskPrompt to a dictionary."""
+        return {
+            "text": self.content,
+            "metadata": self.metadata or {}
+        }
+    
+    @classmethod
+    def from_dict(cls, data: Dict[str, Any]) -> 'TaskPrompt':
+        """Create a TaskPrompt from a dictionary."""
+        return cls(
+            content=data["text"],
+            metadata=data.get("metadata")
+        )