Fix google_search.py

fix/extracted content properties

app/agent/__init__.py

@@ -1,5 +1,6 @@
 from app.agent.base import BaseAgent
 from app.agent.browser import BrowserAgent
+from app.agent.cot import CoTAgent
 from app.agent.mcp import MCPAgent
 from app.agent.planning import PlanningAgent
 from app.agent.react import ReActAgent
@@ -10,6 +11,7 @@ from app.agent.toolcall import ToolCallAgent
 __all__ = [
     "BaseAgent",
     "BrowserAgent",
+    "CoTAgent",
     "PlanningAgent",
     "ReActAgent",
     "SWEAgent",

app/agent/cot.py

@@ -0,0 +1,47 @@
+from typing import Optional
+
+from pydantic import Field
+
+from app.agent.base import BaseAgent
+from app.llm import LLM
+from app.logger import logger
+from app.prompt.cot import NEXT_STEP_PROMPT, SYSTEM_PROMPT
+from app.schema import AgentState, Message
+
+
+class CoTAgent(BaseAgent):
+    """Chain of Thought Agent - Focuses on demonstrating the thinking process of large language models without executing tools"""
+
+    name: str = "cot"
+    description: str = "An agent that uses Chain of Thought reasoning"
+
+    system_prompt: str = SYSTEM_PROMPT
+    next_step_prompt: Optional[str] = NEXT_STEP_PROMPT
+
+    llm: LLM = Field(default_factory=LLM)
+
+    max_steps: int = 1  # CoT typically only needs one step to complete reasoning
+
+    async def step(self) -> str:
+        """Execute one step of chain of thought reasoning"""
+        logger.info(f"🧠 {self.name} is thinking...")
+
+        # If next_step_prompt exists and this isn't the first message, add it to user messages
+        if self.next_step_prompt and len(self.messages) > 1:
+            self.memory.add_message(Message.user_message(self.next_step_prompt))
+
+        # Use system prompt and user messages
+        response = await self.llm.ask(
+            messages=self.messages,
+            system_msgs=[Message.system_message(self.system_prompt)]
+            if self.system_prompt
+            else None,
+        )
+
+        # Record assistant's response
+        self.memory.add_message(Message.assistant_message(response))
+
+        # Set state to finished after completion
+        self.state = AgentState.FINISHED
+
+        return response

app/bedrock.py

@@ -0,0 +1,334 @@
+import json
+import sys
+import time
+import uuid
+from datetime import datetime
+from typing import Dict, List, Literal, Optional
+
+import boto3
+
+
+# Global variables to track the current tool use ID across function calls
+# Tmp solution
+CURRENT_TOOLUSE_ID = None
+
+
+# Class to handle OpenAI-style response formatting
+class OpenAIResponse:
+    def __init__(self, data):
+        # Recursively convert nested dicts and lists to OpenAIResponse objects
+        for key, value in data.items():
+            if isinstance(value, dict):
+                value = OpenAIResponse(value)
+            elif isinstance(value, list):
+                value = [
+                    OpenAIResponse(item) if isinstance(item, dict) else item
+                    for item in value
+                ]
+            setattr(self, key, value)
+
+    def model_dump(self, *args, **kwargs):
+        # Convert object to dict and add timestamp
+        data = self.__dict__
+        data["created_at"] = datetime.now().isoformat()
+        return data
+
+
+# Main client class for interacting with Amazon Bedrock
+class BedrockClient:
+    def __init__(self):
+        # Initialize Bedrock client, you need to configure AWS env first
+        try:
+            self.client = boto3.client("bedrock-runtime")
+            self.chat = Chat(self.client)
+        except Exception as e:
+            print(f"Error initializing Bedrock client: {e}")
+            sys.exit(1)
+
+
+# Chat interface class
+class Chat:
+    def __init__(self, client):
+        self.completions = ChatCompletions(client)
+
+
+# Core class handling chat completions functionality
+class ChatCompletions:
+    def __init__(self, client):
+        self.client = client
+
+    def _convert_openai_tools_to_bedrock_format(self, tools):
+        # Convert OpenAI function calling format to Bedrock tool format
+        bedrock_tools = []
+        for tool in tools:
+            if tool.get("type") == "function":
+                function = tool.get("function", {})
+                bedrock_tool = {
+                    "toolSpec": {
+                        "name": function.get("name", ""),
+                        "description": function.get("description", ""),
+                        "inputSchema": {
+                            "json": {
+                                "type": "object",
+                                "properties": function.get("parameters", {}).get(
+                                    "properties", {}
+                                ),
+                                "required": function.get("parameters", {}).get(
+                                    "required", []
+                                ),
+                            }
+                        },
+                    }
+                }
+                bedrock_tools.append(bedrock_tool)
+        return bedrock_tools
+
+    def _convert_openai_messages_to_bedrock_format(self, messages):
+        # Convert OpenAI message format to Bedrock message format
+        bedrock_messages = []
+        system_prompt = []
+        for message in messages:
+            if message.get("role") == "system":
+                system_prompt = [{"text": message.get("content")}]
+            elif message.get("role") == "user":
+                bedrock_message = {
+                    "role": message.get("role", "user"),
+                    "content": [{"text": message.get("content")}],
+                }
+                bedrock_messages.append(bedrock_message)
+            elif message.get("role") == "assistant":
+                bedrock_message = {
+                    "role": "assistant",
+                    "content": [{"text": message.get("content")}],
+                }
+                openai_tool_calls = message.get("tool_calls", [])
+                if openai_tool_calls:
+                    bedrock_tool_use = {
+                        "toolUseId": openai_tool_calls[0]["id"],
+                        "name": openai_tool_calls[0]["function"]["name"],
+                        "input": json.loads(
+                            openai_tool_calls[0]["function"]["arguments"]
+                        ),
+                    }
+                    bedrock_message["content"].append({"toolUse": bedrock_tool_use})
+                    global CURRENT_TOOLUSE_ID
+                    CURRENT_TOOLUSE_ID = openai_tool_calls[0]["id"]
+                bedrock_messages.append(bedrock_message)
+            elif message.get("role") == "tool":
+                bedrock_message = {
+                    "role": "user",
+                    "content": [
+                        {
+                            "toolResult": {
+                                "toolUseId": CURRENT_TOOLUSE_ID,
+                                "content": [{"text": message.get("content")}],
+                            }
+                        }
+                    ],
+                }
+                bedrock_messages.append(bedrock_message)
+            else:
+                raise ValueError(f"Invalid role: {message.get('role')}")
+        return system_prompt, bedrock_messages
+
+    def _convert_bedrock_response_to_openai_format(self, bedrock_response):
+        # Convert Bedrock response format to OpenAI format
+        content = ""
+        if bedrock_response.get("output", {}).get("message", {}).get("content"):
+            content_array = bedrock_response["output"]["message"]["content"]
+            content = "".join(item.get("text", "") for item in content_array)
+        if content == "":
+            content = "."
+
+        # Handle tool calls in response
+        openai_tool_calls = []
+        if bedrock_response.get("output", {}).get("message", {}).get("content"):
+            for content_item in bedrock_response["output"]["message"]["content"]:
+                if content_item.get("toolUse"):
+                    bedrock_tool_use = content_item["toolUse"]
+                    global CURRENT_TOOLUSE_ID
+                    CURRENT_TOOLUSE_ID = bedrock_tool_use["toolUseId"]
+                    openai_tool_call = {
+                        "id": CURRENT_TOOLUSE_ID,
+                        "type": "function",
+                        "function": {
+                            "name": bedrock_tool_use["name"],
+                            "arguments": json.dumps(bedrock_tool_use["input"]),
+                        },
+                    }
+                    openai_tool_calls.append(openai_tool_call)
+
+        # Construct final OpenAI format response
+        openai_format = {
+            "id": f"chatcmpl-{uuid.uuid4()}",
+            "created": int(time.time()),
+            "object": "chat.completion",
+            "system_fingerprint": None,
+            "choices": [
+                {
+                    "finish_reason": bedrock_response.get("stopReason", "end_turn"),
+                    "index": 0,
+                    "message": {
+                        "content": content,
+                        "role": bedrock_response.get("output", {})
+                        .get("message", {})
+                        .get("role", "assistant"),
+                        "tool_calls": openai_tool_calls
+                        if openai_tool_calls != []
+                        else None,
+                        "function_call": None,
+                    },
+                }
+            ],
+            "usage": {
+                "completion_tokens": bedrock_response.get("usage", {}).get(
+                    "outputTokens", 0
+                ),
+                "prompt_tokens": bedrock_response.get("usage", {}).get(
+                    "inputTokens", 0
+                ),
+                "total_tokens": bedrock_response.get("usage", {}).get("totalTokens", 0),
+            },
+        }
+        return OpenAIResponse(openai_format)
+
+    async def _invoke_bedrock(
+        self,
+        model: str,
+        messages: List[Dict[str, str]],
+        max_tokens: int,
+        temperature: float,
+        tools: Optional[List[dict]] = None,
+        tool_choice: Literal["none", "auto", "required"] = "auto",
+        **kwargs,
+    ) -> OpenAIResponse:
+        # Non-streaming invocation of Bedrock model
+        (
+            system_prompt,
+            bedrock_messages,
+        ) = self._convert_openai_messages_to_bedrock_format(messages)
+        response = self.client.converse(
+            modelId=model,
+            system=system_prompt,
+            messages=bedrock_messages,
+            inferenceConfig={"temperature": temperature, "maxTokens": max_tokens},
+            toolConfig={"tools": tools} if tools else None,
+        )
+        openai_response = self._convert_bedrock_response_to_openai_format(response)
+        return openai_response
+
+    async def _invoke_bedrock_stream(
+        self,
+        model: str,
+        messages: List[Dict[str, str]],
+        max_tokens: int,
+        temperature: float,
+        tools: Optional[List[dict]] = None,
+        tool_choice: Literal["none", "auto", "required"] = "auto",
+        **kwargs,
+    ) -> OpenAIResponse:
+        # Streaming invocation of Bedrock model
+        (
+            system_prompt,
+            bedrock_messages,
+        ) = self._convert_openai_messages_to_bedrock_format(messages)
+        response = self.client.converse_stream(
+            modelId=model,
+            system=system_prompt,
+            messages=bedrock_messages,
+            inferenceConfig={"temperature": temperature, "maxTokens": max_tokens},
+            toolConfig={"tools": tools} if tools else None,
+        )
+
+        # Initialize response structure
+        bedrock_response = {
+            "output": {"message": {"role": "", "content": []}},
+            "stopReason": "",
+            "usage": {},
+            "metrics": {},
+        }
+        bedrock_response_text = ""
+        bedrock_response_tool_input = ""
+
+        # Process streaming response
+        stream = response.get("stream")
+        if stream:
+            for event in stream:
+                if event.get("messageStart", {}).get("role"):
+                    bedrock_response["output"]["message"]["role"] = event[
+                        "messageStart"
+                    ]["role"]
+                if event.get("contentBlockDelta", {}).get("delta", {}).get("text"):
+                    bedrock_response_text += event["contentBlockDelta"]["delta"]["text"]
+                    print(
+                        event["contentBlockDelta"]["delta"]["text"], end="", flush=True
+                    )
+                if event.get("contentBlockStop", {}).get("contentBlockIndex") == 0:
+                    bedrock_response["output"]["message"]["content"].append(
+                        {"text": bedrock_response_text}
+                    )
+                if event.get("contentBlockStart", {}).get("start", {}).get("toolUse"):
+                    bedrock_tool_use = event["contentBlockStart"]["start"]["toolUse"]
+                    tool_use = {
+                        "toolUseId": bedrock_tool_use["toolUseId"],
+                        "name": bedrock_tool_use["name"],
+                    }
+                    bedrock_response["output"]["message"]["content"].append(
+                        {"toolUse": tool_use}
+                    )
+                    global CURRENT_TOOLUSE_ID
+                    CURRENT_TOOLUSE_ID = bedrock_tool_use["toolUseId"]
+                if event.get("contentBlockDelta", {}).get("delta", {}).get("toolUse"):
+                    bedrock_response_tool_input += event["contentBlockDelta"]["delta"][
+                        "toolUse"
+                    ]["input"]
+                    print(
+                        event["contentBlockDelta"]["delta"]["toolUse"]["input"],
+                        end="",
+                        flush=True,
+                    )
+                if event.get("contentBlockStop", {}).get("contentBlockIndex") == 1:
+                    bedrock_response["output"]["message"]["content"][1]["toolUse"][
+                        "input"
+                    ] = json.loads(bedrock_response_tool_input)
+        print()
+        openai_response = self._convert_bedrock_response_to_openai_format(
+            bedrock_response
+        )
+        return openai_response
+
+    def create(
+        self,
+        model: str,
+        messages: List[Dict[str, str]],
+        max_tokens: int,
+        temperature: float,
+        stream: Optional[bool] = True,
+        tools: Optional[List[dict]] = None,
+        tool_choice: Literal["none", "auto", "required"] = "auto",
+        **kwargs,
+    ) -> OpenAIResponse:
+        # Main entry point for chat completion
+        bedrock_tools = []
+        if tools is not None:
+            bedrock_tools = self._convert_openai_tools_to_bedrock_format(tools)
+        if stream:
+            return self._invoke_bedrock_stream(
+                model,
+                messages,
+                max_tokens,
+                temperature,
+                bedrock_tools,
+                tool_choice,
+                **kwargs,
+            )
+        else:
+            return self._invoke_bedrock(
+                model,
+                messages,
+                max_tokens,
+                temperature,
+                bedrock_tools,
+                tool_choice,
+                **kwargs,
+            )

app/config.py

@@ -37,6 +37,30 @@ class ProxySettings(BaseModel):
 
 class SearchSettings(BaseModel):
     engine: str = Field(default="Google", description="Search engine the llm to use")
+    fallback_engines: List[str] = Field(
+        default_factory=lambda: ["DuckDuckGo", "Baidu"],
+        description="Fallback search engines to try if the primary engine fails",
+    )
+    retry_delay: int = Field(
+        default=60,
+        description="Seconds to wait before retrying all engines again after they all fail",
+    )
+    max_retries: int = Field(
+        default=3,
+        description="Maximum number of times to retry all engines when all fail",
+    )
+    api_key: Optional[str] = Field(
+        None,
+        description="API key for the search engine's official API (currently used for Google)",
+    )
+    cx: Optional[str] = Field(
+        None,
+        description="Custom Search Engine ID for search APIs that require it (currently used for Google)",
+    )
+    use_fallback: bool = Field(
+        True,
+        description="Whether to fall back to web scraping when the API fails or is not configured",
+    )
 
 
 class BrowserSettings(BaseModel):

app/flow/base.py

@@ -1,5 +1,4 @@
 from abc import ABC, abstractmethod
-from enum import Enum
 from typing import Dict, List, Optional, Union
 
 from pydantic import BaseModel
@@ -7,10 +6,6 @@ from pydantic import BaseModel
 from app.agent.base import BaseAgent
 
 
-class FlowType(str, Enum):
-    PLANNING = "planning"
-
-
 class BaseFlow(BaseModel, ABC):
     """Base class for execution flows supporting multiple agents"""
 
@@ -60,32 +55,3 @@ class BaseFlow(BaseModel, ABC):
     @abstractmethod
     async def execute(self, input_text: str) -> str:
         """Execute the flow with given input"""
-
-
-class PlanStepStatus(str, Enum):
-    """Enum class defining possible statuses of a plan step"""
-
-    NOT_STARTED = "not_started"
-    IN_PROGRESS = "in_progress"
-    COMPLETED = "completed"
-    BLOCKED = "blocked"
-
-    @classmethod
-    def get_all_statuses(cls) -> list[str]:
-        """Return a list of all possible step status values"""
-        return [status.value for status in cls]
-
-    @classmethod
-    def get_active_statuses(cls) -> list[str]:
-        """Return a list of values representing active statuses (not started or in progress)"""
-        return [cls.NOT_STARTED.value, cls.IN_PROGRESS.value]
-
-    @classmethod
-    def get_status_marks(cls) -> Dict[str, str]:
-        """Return a mapping of statuses to their marker symbols"""
-        return {
-            cls.COMPLETED.value: "[✓]",
-            cls.IN_PROGRESS.value: "[→]",
-            cls.BLOCKED.value: "[!]",
-            cls.NOT_STARTED.value: "[ ]",
-        }

app/flow/flow_factory.py

@@ -1,10 +1,15 @@
+from enum import Enum
 from typing import Dict, List, Union
 
 from app.agent.base import BaseAgent
-from app.flow.base import BaseFlow, FlowType
+from app.flow.base import BaseFlow
 from app.flow.planning import PlanningFlow
 
 
+class FlowType(str, Enum):
+    PLANNING = "planning"
+
+
 class FlowFactory:
     """Factory for creating different types of flows with support for multiple agents"""
 

app/flow/planning.py

@@ -1,17 +1,47 @@
 import json
 import time
+from enum import Enum
 from typing import Dict, List, Optional, Union
 
 from pydantic import Field
 
 from app.agent.base import BaseAgent
-from app.flow.base import BaseFlow, PlanStepStatus
+from app.flow.base import BaseFlow
 from app.llm import LLM
 from app.logger import logger
 from app.schema import AgentState, Message, ToolChoice
 from app.tool import PlanningTool
 
 
+class PlanStepStatus(str, Enum):
+    """Enum class defining possible statuses of a plan step"""
+
+    NOT_STARTED = "not_started"
+    IN_PROGRESS = "in_progress"
+    COMPLETED = "completed"
+    BLOCKED = "blocked"
+
+    @classmethod
+    def get_all_statuses(cls) -> list[str]:
+        """Return a list of all possible step status values"""
+        return [status.value for status in cls]
+
+    @classmethod
+    def get_active_statuses(cls) -> list[str]:
+        """Return a list of values representing active statuses (not started or in progress)"""
+        return [cls.NOT_STARTED.value, cls.IN_PROGRESS.value]
+
+    @classmethod
+    def get_status_marks(cls) -> Dict[str, str]:
+        """Return a mapping of statuses to their marker symbols"""
+        return {
+            cls.COMPLETED.value: "[✓]",
+            cls.IN_PROGRESS.value: "[→]",
+            cls.BLOCKED.value: "[!]",
+            cls.NOT_STARTED.value: "[ ]",
+        }
+
+
 class PlanningFlow(BaseFlow):
     """A flow that manages planning and execution of tasks using agents."""
 

app/llm.py

@@ -18,6 +18,7 @@ from tenacity import (
     wait_random_exponential,
 )
 
+from app.bedrock import BedrockClient
 from app.config import LLMSettings, config
 from app.exceptions import TokenLimitExceeded
 from app.logger import logger  # Assuming a logger is set up in your app
@@ -225,6 +226,8 @@ class LLM:
                     api_key=self.api_key,
                     api_version=self.api_version,
                 )
+            elif self.api_type == "aws":
+                self.client = BedrockClient()
             else:
                 self.client = AsyncOpenAI(api_key=self.api_key, base_url=self.base_url)
 

app/mcp/server.py

@@ -1,30 +1,19 @@
+import logging
+import sys
+
+
+logging.basicConfig(level=logging.INFO, handlers=[logging.StreamHandler(sys.stderr)])
+
 import argparse
 import asyncio
 import atexit
 import json
-import logging
-import os
-import sys
 from inspect import Parameter, Signature
 from typing import Any, Dict, Optional
 
 from mcp.server.fastmcp import FastMCP
 
-
-# Add directories to Python path (needed for proper importing)
-current_dir = os.path.dirname(os.path.abspath(__file__))
-parent_dir = os.path.dirname(current_dir)
-root_dir = os.path.dirname(parent_dir)
-sys.path.insert(0, parent_dir)
-sys.path.insert(0, current_dir)
-sys.path.insert(0, root_dir)
-
-# Configure logging (using the same format as original)
-logging.basicConfig(
-    level=logging.INFO, format="%(asctime)s - %(name)s - %(levelname)s - %(message)s"
-)
-logger = logging.getLogger("mcp-server")
-
+from app.logger import logger
 from app.tool.base import BaseTool
 from app.tool.bash import Bash
 from app.tool.browser_use_tool import BrowserUseTool
@@ -45,11 +34,6 @@ class MCPServer:
         self.tools["editor"] = StrReplaceEditor()
         self.tools["terminate"] = Terminate()
 
-        from app.logger import logger as app_logger
-
-        global logger
-        logger = app_logger
-
     def register_tool(self, tool: BaseTool, method_name: Optional[str] = None) -> None:
         """Register a tool with parameter validation and documentation."""
         tool_name = method_name or tool.name

app/prompt/cot.py

@@ -0,0 +1,15 @@
+SYSTEM_PROMPT = """You are an assistant focused on Chain of Thought reasoning. For each question, please follow these steps:
+
+1. Break down the problem: Divide complex problems into smaller, more manageable parts
+2. Think step by step: Think through each part in detail, showing your reasoning process
+3. Synthesize conclusions: Integrate the thinking from each part into a complete solution
+4. Provide an answer: Give a final concise answer
+
+Your response should follow this format:
+Thinking: [Detailed thought process, including problem decomposition, reasoning for each step, and analysis]
+Answer: [Final answer based on the thought process, clear and concise]
+
+Remember, the thinking process is more important than the final answer, as it demonstrates how you reached your conclusion.
+"""
+
+NEXT_STEP_PROMPT = "Please continue your thinking based on the conversation above. If you've reached a conclusion, provide your final answer."

app/tool/browser_use_tool.py

@@ -448,6 +448,22 @@ Page content:
                                         "extracted_content": {
                                             "type": "object",
                                             "description": "The content extracted from the page according to the goal",
+                                            "properties": {
+                                                "text": {
+                                                    "type": "string",
+                                                    "description": "Text content extracted from the page",
+                                                },
+                                                "metadata": {
+                                                    "type": "object",
+                                                    "description": "Additional metadata about the extracted content",
+                                                    "properties": {
+                                                        "source": {
+                                                            "type": "string",
+                                                            "description": "Source of the extracted content",
+                                                        }
+                                                    },
+                                                },
+                                            },
                                         }
                                     },
                                     "required": ["extracted_content"],

app/tool/search/google_search.py

@@ -1,9 +1,156 @@
+from typing import List
+
+import requests
 from googlesearch import search
 
+from app.config import config
+from app.logger import logger
 from app.tool.search.base import WebSearchEngine
 
 
 class GoogleSearchEngine(WebSearchEngine):
-    def perform_search(self, query, num_results=10, *args, **kwargs):
-        """Google search engine."""
-        return search(query, num_results=num_results)
+    def perform_search(
+        self, query: str, num_results: int = 10, *args, **kwargs
+    ) -> List[str]:
+        """
+        Google search engine using the official Google Custom Search API when configured,
+        falling back to web scraping if not configured or if the API call fails.
+
+        Args:
+            query (str): The search query to submit to the search engine.
+            num_results (int, optional): The number of search results to return. Default is 10.
+            *args: Additional positional arguments.
+            **kwargs: Additional keyword arguments.
+
+        Returns:
+            List[str]: A list of URLs matching the search query.
+        """
+        # Check for API configuration in the search settings
+        search_config = getattr(config, "search_config", None)
+        api_key = getattr(search_config, "api_key", None) if search_config else None
+        cx = getattr(search_config, "cx", None) if search_config else None
+        use_fallback = (
+            getattr(search_config, "use_fallback", True) if search_config else True
+        )
+
+        # If API is configured, try using the Google Search API
+        if api_key and cx:
+            try:
+                logger.info("Using Google Custom Search API for search")
+                return self._api_search(query, api_key, cx, num_results)
+            except requests.RequestException as e:
+                # More specific error handling for HTTP-related errors
+                status_code = (
+                    getattr(e.response, "status_code", None)
+                    if hasattr(e, "response")
+                    else None
+                )
+                if status_code == 429:
+                    logger.warning("Google API rate limit exceeded")
+                elif status_code and 400 <= status_code < 500:
+                    logger.warning(
+                        f"Google API client error: {e} (status code: {status_code})"
+                    )
+                elif status_code and 500 <= status_code < 600:
+                    logger.warning(
+                        f"Google API server error: {e} (status code: {status_code})"
+                    )
+                else:
+                    logger.warning(f"Google API request error: {e}")
+
+                if not use_fallback:
+                    logger.warning(
+                        "Fallback to scraping is disabled. Returning empty results."
+                    )
+                    return []
+                logger.info("Falling back to web scraping search")
+            except Exception as e:
+                # General error handling for other types of exceptions
+                logger.warning(f"Google API error: {e}")
+                if not use_fallback:
+                    logger.warning(
+                        "Fallback to scraping is disabled. Returning empty results."
+                    )
+                    return []
+                logger.info("Falling back to web scraping search")
+
+        # Use web scraping if API is not configured or if API call failed and fallback is enabled
+        return self._scraping_search(query, num_results)
+
+    @staticmethod
+    def _api_search(
+        query: str, api_key: str, cx: str, num_results: int = 10
+    ) -> List[str]:
+        """
+        Perform a search using Google's Custom Search JSON API.
+
+        Args:
+            query (str): The search query.
+            api_key (str): The API key for Google Custom Search.
+            cx (str): The Custom Search Engine ID.
+            num_results (int, optional): The number of results to return. Default is 10.
+
+        Returns:
+            List[str]: A list of URLs matching the search query.
+
+        Raises:
+            requests.RequestException: If there's an issue with the HTTP request.
+            ValueError: If the response cannot be parsed as JSON.
+        """
+        base_url = "https://www.googleapis.com/customsearch/v1"
+        results = []
+
+        # API allows max 10 results per request, so we need to paginate
+        for start_index in range(
+            1, min(num_results + 1, 101), 10
+        ):  # Google API limits to 100 results max
+            params = {
+                "q": query,
+                "key": api_key,
+                "cx": cx,
+                "start": start_index,
+                "num": min(
+                    10, num_results - len(results)
+                ),  # Can't request more than 10 at once
+            }
+
+            response = requests.get(base_url, params=params, timeout=10)  # Add timeout
+            response.raise_for_status()  # Raise exception for 4XX/5XX responses
+            data = response.json()
+
+            if "items" in data:
+                for item in data["items"]:
+                    if "link" in item:
+                        results.append(item["link"])
+                        if len(results) >= num_results:
+                            return results
+            else:
+                # No more results or empty result set
+                if (
+                    "searchInformation" in data
+                    and "totalResults" in data["searchInformation"]
+                ):
+                    logger.info(
+                        f"Total results: {data['searchInformation']['totalResults']}"
+                    )
+                break
+
+        return results
+
+    @staticmethod
+    def _scraping_search(query: str, num_results: int = 10) -> List[str]:
+        """
+        Perform a search using web scraping as a fallback method.
+
+        Args:
+            query (str): The search query.
+            num_results (int, optional): The number of results to return. Default is 10.
+
+        Returns:
+            List[str]: A list of URLs matching the search query.
+        """
+        try:
+            return list(search(query, num_results=num_results))
+        except Exception as e:
+            logger.warning(f"Web scraping search failed: {e}")
+            return []

app/tool/web_search.py

@@ -4,6 +4,7 @@ from typing import List
 from tenacity import retry, stop_after_attempt, wait_exponential
 
 from app.config import config
+from app.logger import logger
 from app.tool.base import BaseTool
 from app.tool.search import (
     BaiduSearchEngine,
@@ -44,6 +45,8 @@ class WebSearch(BaseTool):
     async def execute(self, query: str, num_results: int = 10) -> List[str]:
         """
         Execute a Web search and return a list of URLs.
+        Tries engines in order based on configuration, falling back if an engine fails with errors.
+        If all engines fail, it will wait and retry up to the configured number of times.
 
         Args:
             query (str): The search query to submit to the search engine.
@@ -52,37 +55,109 @@ class WebSearch(BaseTool):
         Returns:
             List[str]: A list of URLs matching the search query.
         """
+        # Get retry settings from config
+        retry_delay = 60  # Default to 60 seconds
+        max_retries = 3  # Default to 3 retries
+
+        if config.search_config:
+            retry_delay = getattr(config.search_config, "retry_delay", 60)
+            max_retries = getattr(config.search_config, "max_retries", 3)
+
+        # Try searching with retries when all engines fail
+        for retry_count in range(
+            max_retries + 1
+        ):  # +1 because first try is not a retry
+            links = await self._try_all_engines(query, num_results)
+            if links:
+                return links
+
+            if retry_count < max_retries:
+                # All engines failed, wait and retry
+                logger.warning(
+                    f"All search engines failed. Waiting {retry_delay} seconds before retry {retry_count + 1}/{max_retries}..."
+                )
+                await asyncio.sleep(retry_delay)
+            else:
+                logger.error(
+                    f"All search engines failed after {max_retries} retries. Giving up."
+                )
+
+        return []
+
+    async def _try_all_engines(self, query: str, num_results: int) -> List[str]:
+        """
+        Try all search engines in the configured order.
+
+        Args:
+            query (str): The search query to submit to the search engine.
+            num_results (int): The number of search results to return.
+
+        Returns:
+            List[str]: A list of URLs matching the search query, or empty list if all engines fail.
+        """
         engine_order = self._get_engine_order()
+        failed_engines = []
+
         for engine_name in engine_order:
             engine = self._search_engine[engine_name]
             try:
+                logger.info(f"🔎 Attempting search with {engine_name.capitalize()}...")
                 links = await self._perform_search_with_engine(
                     engine, query, num_results
                 )
                 if links:
+                    if failed_engines:
+                        logger.info(
+                            f"Search successful with {engine_name.capitalize()} after trying: {', '.join(failed_engines)}"
+                        )
                     return links
             except Exception as e:
-                print(f"Search engine '{engine_name}' failed with error: {e}")
+                failed_engines.append(engine_name.capitalize())
+                is_rate_limit = "429" in str(e) or "Too Many Requests" in str(e)
+
+                if is_rate_limit:
+                    logger.warning(
+                        f"⚠️ {engine_name.capitalize()} search engine rate limit exceeded, trying next engine..."
+                    )
+                else:
+                    logger.warning(
+                        f"⚠️ {engine_name.capitalize()} search failed with error: {e}"
+                    )
+
+        if failed_engines:
+            logger.error(f"All search engines failed: {', '.join(failed_engines)}")
         return []
 
     def _get_engine_order(self) -> List[str]:
         """
         Determines the order in which to try search engines.
-        Preferred engine is first (based on configuration), followed by the remaining engines.
+        Preferred engine is first (based on configuration), followed by fallback engines,
+        and then the remaining engines.
 
         Returns:
             List[str]: Ordered list of search engine names.
         """
         preferred = "google"
-        if config.search_config and config.search_config.engine:
-            preferred = config.search_config.engine.lower()
+        fallbacks = []
+
+        if config.search_config:
+            if config.search_config.engine:
+                preferred = config.search_config.engine.lower()
+            if config.search_config.fallback_engines:
+                fallbacks = [
+                    engine.lower() for engine in config.search_config.fallback_engines
+                ]
 
         engine_order = []
+        # Add preferred engine first
         if preferred in self._search_engine:
             engine_order.append(preferred)
-        for key in self._search_engine:
-            if key not in engine_order:
-                engine_order.append(key)
+
+        # Add configured fallback engines in order
+        for fallback in fallbacks:
+            if fallback in self._search_engine and fallback not in engine_order:
+                engine_order.append(fallback)
+
         return engine_order
 
     @retry(

config/config.example.toml

@@ -6,6 +6,14 @@ api_key = "YOUR_API_KEY"                    # Your API key
 max_tokens = 8192                           # Maximum number of tokens in the response
 temperature = 0.0                           # Controls randomness
 
+# [llm] # Amazon Bedrock
+# api_type = "aws"                                       # Required
+# model = "us.anthropic.claude-3-7-sonnet-20250219-v1:0" # Bedrock supported modelID
+# base_url = "bedrock-runtime.us-west-2.amazonaws.com"   # Not used now
+# max_tokens = 8192
+# temperature = 1.0
+# api_key = "bear"                                       # Required but not used for Bedrock
+
 # [llm] #AZURE OPENAI:
 # api_type= 'azure'
 # model = "YOUR_MODEL_NAME" #"gpt-4o-mini"
@@ -65,6 +73,21 @@ temperature = 0.0                           # Controls randomness for vision mod
 # [search]
 # Search engine for agent to use. Default is "Google", can be set to "Baidu" or "DuckDuckGo".
 #engine = "Google"
+# Fallback engine order. Default is ["DuckDuckGo", "Baidu"] - will try in this order after primary engine fails.
+#fallback_engines = ["DuckDuckGo", "Baidu"]
+# Seconds to wait before retrying all engines again when they all fail due to rate limits. Default is 60.
+#retry_delay = 60
+# Maximum number of times to retry all engines when all fail. Default is 3.
+#max_retries = 3
+# API key for the search engine's official API (currently used for Google)
+# For Google, create an API key at https://console.cloud.google.com/apis/credentials
+#api_key = ""
+# Custom Search Engine ID for search APIs that require it (currently used for Google)
+# For Google, create a Custom Search Engine at https://programmablesearchengine.google.com/
+#cx = ""
+# Whether to fall back to web scraping when the API fails or is not configured. Default is true.
+#use_fallback = true
+
 
 ## Sandbox configuration
 #[sandbox]

requirements.txt

@@ -7,6 +7,7 @@ numpy
 datasets~=3.2.0
 fastapi~=0.115.11
 tiktoken~=0.9.0
+requests~=2.31.0
 
 html2text~=2024.2.26
 gymnasium~=1.0.0
@@ -31,3 +32,5 @@ pytest-asyncio~=0.25.3
 mcp~=1.4.1
 httpx>=0.27.0
 tomli>=2.0.0
+
+boto3~=1.37.16

run_flow.py

@@ -2,8 +2,7 @@ import asyncio
 import time
 
 from app.agent.manus import Manus
-from app.flow.base import FlowType
-from app.flow.flow_factory import FlowFactory
+from app.flow.flow_factory import FlowFactory, FlowType
 from app.logger import logger
 
 

run_mcp.py

@@ -13,10 +13,14 @@ class MCPRunner:
 
     def __init__(self):
         self.root_path = config.root_path
-        self.server_script = self.root_path / "app" / "mcp" / "server.py"
+        self.server_reference = "app.mcp.server"
         self.agent = MCPAgent()
 
-    async def initialize(self, connection_type: str, server_url: str = None) -> None:
+    async def initialize(
+        self,
+        connection_type: str,
+        server_url: str | None = None,
+    ) -> None:
         """Initialize the MCP agent with the appropriate connection."""
         logger.info(f"Initializing MCPAgent with {connection_type} connection...")
 
@@ -24,7 +28,7 @@ class MCPRunner:
             await self.agent.initialize(
                 connection_type="stdio",
                 command=sys.executable,
-                args=[str(self.server_script)],
+                args=["-m", self.server_reference],
             )
         else:  # sse
             await self.agent.initialize(connection_type="sse", server_url=server_url)
@@ -47,9 +51,14 @@ class MCPRunner:
 
     async def run_default(self) -> None:
         """Run the agent in default mode."""
-        await self.agent.run(
-            "Hello, what tools are available to me? Terminate after you have listed the tools."
-        )
+        prompt = input("Enter your prompt: ")
+        if not prompt.strip():
+            logger.warning("Empty prompt provided.")
+            return
+
+        logger.warning("Processing your request...")
+        await self.agent.run(prompt)
+        logger.info("Request processing completed.")
 
     async def cleanup(self) -> None:
         """Clean up agent resources."""

run_mcp_server.py

@@ -0,0 +1,11 @@
+# coding: utf-8
+# A shortcut to launch OpenManus MCP server, where its introduction also solves other import issues.
+from app.mcp.server import MCPServer, parse_args
+
+
+if __name__ == "__main__":
+    args = parse_args()
+
+    # Create and run server (maintaining original flow)
+    server = MCPServer()
+    server.run(transport=args.transport)