MCP Browser Use Server
by JovaniPink
- src
- mcp_browser_use
# -*- coding: utf-8 -*-
import asyncio
import os
import sys
import traceback
import logging
from typing import Optional
from browser_use import BrowserConfig
from browser_use.browser.context import BrowserContextConfig
from fastmcp import FastMCP
from mcp.types import TextContent
from mcp_browser_use.agent.custom_agent import CustomAgent
from mcp_browser_use.browser.custom_browser import CustomBrowser
from mcp_browser_use.controller.custom_controller import CustomController
from mcp_browser_use.utils import utils
from mcp_browser_use.utils.agent_state import AgentState
# Configure logging for the entire module
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
# Global references for single "running agent" approach
_global_agent: Optional[CustomAgent] = None
_global_browser: Optional[CustomBrowser] = None
from browser_use.browser.context import BrowserContext
_global_browser_context: Optional[BrowserContext] = None
_global_agent_state: AgentState = AgentState()
app = FastMCP("mcp_browser_use")
async def _cleanup_browser_resources() -> None:
"""
Safely clean up browser resources and reset global references.
Any exceptions are logged and swallowed to ensure we attempt
as much cleanup as possible.
Need to add types.
"""
global _global_browser, _global_agent_state, _global_browser_context, _global_agent
try:
if _global_agent_state:
try:
await _global_agent_state.request_stop()
except Exception as stop_error:
logger.warning("Error stopping agent state: %s", stop_error)
if _global_browser_context:
try:
await _global_browser_context.close()
except Exception as context_error:
logger.warning("Error closing browser context: %s", context_error)
if _global_browser:
try:
await _global_browser.close()
except Exception as browser_error:
logger.warning("Error closing browser: %s", browser_error)
except Exception as e:
# Log the error, but don't re-raise
logger.error("Unexpected error during browser cleanup: %s", e)
finally:
# Reset global variables
_global_browser = None
_global_browser_context = None
_global_agent_state = AgentState()
_global_agent = None
@app.tool()
async def run_browser_agent(task: str, add_infos: str = "") -> str:
"""
This is the entrypoint for running a browser-based agent.
:param task: The main instruction or goal for the agent.
:param add_infos: Additional information or context for the agent.
:return: The final result string from the agent run.
"""
global _global_agent, _global_browser, _global_browser_context, _global_agent_state
try:
# Clear any previous agent stop signals
_global_agent_state.clear_stop()
# Read environment variables with defaults and parse carefully
# Fallback to defaults if parsing fails.
model_provider = os.getenv("MCP_MODEL_PROVIDER", "anthropic")
model_name = os.getenv("MCP_MODEL_NAME", "claude-3-5-sonnet-20241022")
def safe_float(env_var: str, default: float) -> float:
"""Safely parse a float from an environment variable."""
try:
return float(os.getenv(env_var, str(default)))
except ValueError:
logger.warning(f"Invalid float for {env_var}, using default={default}")
return default
def safe_int(env_var: str, default: int) -> int:
"""Safely parse an int from an environment variable."""
try:
return int(os.getenv(env_var, str(default)))
except ValueError:
logger.warning(f"Invalid int for {env_var}, using default={default}")
return default
# Get environment variables with defaults
temperature = safe_float("MCP_TEMPERATURE", 0.3)
max_steps = safe_int("MCP_MAX_STEPS", 30)
use_vision = os.getenv("MCP_USE_VISION", "true").lower() == "true"
max_actions_per_step = safe_int("MCP_MAX_ACTIONS_PER_STEP", 5)
tool_call_in_content = (
os.getenv("MCP_TOOL_CALL_IN_CONTENT", "true").lower() == "true"
)
# Prepare LLM
llm = utils.get_llm_model(
provider=model_provider, model_name=model_name, temperature=temperature
)
# Get path to the Chrome/Chromium binary if provided
chrome_path = os.getenv("CHROME_PATH") or None
# Create or reuse the global browser instance
if not _global_browser:
_global_browser = CustomBrowser(
config=BrowserConfig(
headless=False,
disable_security=False,
chrome_instance_path=chrome_path,
extra_chromium_args=[],
wss_url=None,
proxy=None,
)
)
# Create or reuse the global browser context
if not _global_browser_context:
_global_browser_context = await _global_browser.new_context(
config=BrowserContextConfig(
trace_path=None, save_recording_path=None, no_viewport=False
)
)
# Create controller and agent
controller = CustomController()
_global_agent = CustomAgent(
task=task,
add_infos=add_infos,
use_vision=use_vision,
llm=llm,
browser=_global_browser,
browser_context=_global_browser_context,
controller=controller,
max_actions_per_step=max_actions_per_step,
tool_call_in_content=tool_call_in_content,
agent_state=_global_agent_state,
)
# Run agent
history = await _global_agent.run(max_steps=max_steps)
# Extract final result from the agent's history
final_result = history.final_result()
if not final_result:
final_result = f"No final result. Possibly incomplete. {history}"
return final_result
except Exception as e:
logger.error("run-browser-agent error: %s", str(e))
raise ValueError(f"run-browser-agent error: {e}\n{traceback.format_exc()}")
finally:
# Always ensure cleanup, even if no error.
await _cleanup_browser_resources()
def main() -> None:
"""
Entry point for running the FastMCP application.
Handles server start and final resource cleanup.
"""
try:
app.run()
except Exception as e:
logger.error("Error running MCP server: %s\n%s", e, traceback.format_exc())
finally:
# Use a separate event loop to ensure cleanup is awaited properly
try:
loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)
loop.run_until_complete(_cleanup_browser_resources())
loop.close()
except Exception as cleanup_error:
logger.error("Cleanup error: %s", cleanup_error)
if __name__ == "__main__":
main()