import os from typing import Dict, List, Optional from pydantic import Field from app.agent.toolcall import ToolCallAgent from app.config import config from app.logger import logger from app.prompt.manus import NEXT_STEP_PROMPT, SYSTEM_PROMPT from app.schema import Message from app.tool import Terminate, ToolCollection from app.tool.ask_human import AskHuman from app.tool.mcp import MCPClients, MCPClientTool from app.tool.python_execute import PythonExecute from app.tool.str_replace_editor import StrReplaceEditor _BROWSER_USE_SERVER_ID = "browser_use" _BROWSER_USE_COMMAND = "uvx" _BROWSER_USE_ARGS = ["browser-use", "--cli-mcp"] _BROWSER_USE_ENV_VARS = ( "BROWSER_USE_API_KEY", "BROWSER_USE_CLOUD_API_URL", "BU_BROWSER_ID", "BU_CDP_URL", "BU_CDP_WS", "BU_NAME", ) _BROWSER_USE_TRANSPORT_INSTRUCTIONS = """\ Browser Use CLI 3.0 is exposed here as MCP tools. When the Browser Use skill shows `browser-use <<'PY'`, pass the Python body to `browser_exec` instead. Use `browser_screenshot` when visual inspection is needed. Both tools use the same persistent browser-harness session as CLI 3.0. """ def _browser_use_env() -> Dict[str, str]: return {name: value for name in _BROWSER_USE_ENV_VARS if (value := os.getenv(name))} class Manus(ToolCallAgent): """A versatile general-purpose agent with support for both local and MCP tools.""" name: str = "Manus" description: str = "A versatile agent that can solve various tasks using multiple tools including MCP-based tools" system_prompt: str = SYSTEM_PROMPT.format(directory=config.workspace_root) next_step_prompt: str = NEXT_STEP_PROMPT max_observe: int = 10000 max_steps: int = 20 # MCP clients for remote tool access mcp_clients: MCPClients = Field(default_factory=MCPClients) # Add general-purpose tools to the tool collection available_tools: ToolCollection = Field( default_factory=lambda: ToolCollection( PythonExecute(), StrReplaceEditor(), AskHuman(), Terminate(), ) ) special_tool_names: list[str] = Field(default_factory=lambda: [Terminate().name]) # Track connected MCP servers connected_servers: Dict[str, str] = Field( default_factory=dict ) # server_id -> url/command mcp_instruction_servers: set[str] = Field(default_factory=set, exclude=True) _initialized: bool = False @classmethod async def create(cls, **kwargs) -> "Manus": """Factory method to create and properly initialize a Manus instance.""" instance = cls(**kwargs) await instance.initialize_mcp_servers() instance._initialized = True return instance async def initialize_mcp_servers(self) -> None: """Initialize connections to configured MCP servers.""" if _BROWSER_USE_SERVER_ID not in config.mcp_config.servers and os.getenv( "OPENMANUS_DISABLE_BROWSER_USE", "" ).lower() not in {"1", "true", "yes"}: try: await self.connect_mcp_server( _BROWSER_USE_COMMAND, _BROWSER_USE_SERVER_ID, use_stdio=True, stdio_args=_BROWSER_USE_ARGS, tool_name_prefix=False, stdio_env=_browser_use_env(), ) logger.info("Connected to Browser Use CLI 3.0 through MCP") except Exception as e: logger.error(f"Failed to connect to Browser Use CLI 3.0: {e}") for server_id, server_config in config.mcp_config.servers.items(): try: if server_config.type == "sse": if server_config.url: await self.connect_mcp_server(server_config.url, server_id) logger.info( f"Connected to MCP server {server_id} at {server_config.url}" ) elif server_config.type == "stdio": if server_config.command: await self.connect_mcp_server( server_config.command, server_id, use_stdio=True, stdio_args=server_config.args, tool_name_prefix=server_id != _BROWSER_USE_SERVER_ID, stdio_env=( _browser_use_env() if server_id == _BROWSER_USE_SERVER_ID else None ), ) logger.info( f"Connected to MCP server {server_id} using command {server_config.command}" ) except Exception as e: logger.error(f"Failed to connect to MCP server {server_id}: {e}") async def connect_mcp_server( self, server_url: str, server_id: str = "", use_stdio: bool = False, stdio_args: Optional[List[str]] = None, tool_name_prefix: bool = True, stdio_env: Optional[Dict[str, str]] = None, ) -> None: """Connect to an MCP server and add its tools.""" if use_stdio: await self.mcp_clients.connect_stdio( server_url, stdio_args or [], server_id, tool_name_prefix=tool_name_prefix, env=stdio_env, ) self.connected_servers[server_id or server_url] = server_url else: await self.mcp_clients.connect_sse(server_url, server_id) self.connected_servers[server_id or server_url] = server_url # Update available tools with only the new tools from this server new_tools = [ tool for tool in self.mcp_clients.tools if tool.server_id == server_id ] self.available_tools.add_tools(*new_tools) resolved_server_id = server_id or server_url instructions = self.mcp_clients.server_instructions.get(resolved_server_id) if instructions and resolved_server_id not in self.mcp_instruction_servers: transport_instructions = ( f"{_BROWSER_USE_TRANSPORT_INSTRUCTIONS}\n" if resolved_server_id == _BROWSER_USE_SERVER_ID else "" ) self.memory.add_message( Message.system_message( f"{transport_instructions}MCP server instructions:\n{instructions}" ) ) self.mcp_instruction_servers.add(resolved_server_id) async def disconnect_mcp_server(self, server_id: str = "") -> None: """Disconnect from an MCP server and remove its tools.""" await self.mcp_clients.disconnect(server_id) if server_id: self.connected_servers.pop(server_id, None) else: self.connected_servers.clear() # Rebuild available tools without the disconnected server's tools base_tools = [ tool for tool in self.available_tools.tools if not isinstance(tool, MCPClientTool) ] self.available_tools = ToolCollection(*base_tools) self.available_tools.add_tools(*self.mcp_clients.tools) async def cleanup(self): """Clean up Manus agent resources.""" # Disconnect from all MCP servers only if we were initialized if self._initialized: await self.disconnect_mcp_server() self._initialized = False async def think(self) -> bool: """Process current state and decide next actions with appropriate context.""" if not self._initialized: await self.initialize_mcp_servers() self._initialized = True return await super().think()