puppeteer_mcp_server.py•3.75 kB
#!/usr/bin/env python3
import logging
import sys
import os
from os import getenv
from contextlib import asynccontextmanager
from typing import AsyncIterator, Dict
from mcp.server.fastmcp import FastMCP, Context
from mcp.types import TextContent, Tool, Resource, ResourceContents
from pyppeteer import launch
# Logging configuration
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)
# Global variable to store the browser
_browser = None
# Server lifespan function
@asynccontextmanager
async def server_lifespan(ctx: Context) -> AsyncIterator[None]:
"""Manages the MCP server lifecycle."""
global _browser
if os.getenv('DOCKER_CONTAINER') is not None:
default_args = [
"--no-sandbox", # Necessário para containers Docker
"--disable-setuid-sandbox",
"--disable-dev-shm-usage", # Resolve problemas com /dev/shm em containers
"--disable-accelerated-2d-canvas",
"--disable-gpu", # Desativa aceleração de GPU
"--window-size=1920,1080", # Define tamanho da janela
]
_browser = await launch(executablePath=os.getenv('BROWSER_EXECUTABLE_PATH'), args=default_args)
elif os.getenv('DOCKER_CONTAINER') is None:
_browser = await launch(executablePath=os.getenv('BROWSER_EXECUTABLE_PATH'), args=sys.argv[1:])
logger.info("Server started - browser launched")
try:
yield
finally:
await _browser.close()
logger.info("Server stopped - browser closed")
# Create the MCP server
mcp = FastMCP("pyppeteer_server", lifespan=server_lifespan)
@mcp.tool()
async def pyppeteer_navigation(url: str):
"""
Navigates to the URL specified in the parameter
"""
global _browser
page = await _browser.newPage()
await page.goto(url)
return {"status": "success", "url": url}
@mcp.tool()
async def take_screenshot(name: str):
"""
Takes a screenshot of the current page and saves it with the specified name
"""
global _browser
pages = await _browser.pages()
page = pages[-1] # gets the current page
await page.screenshot({'path': name})
return {"status": "success", "file": name}
@mcp.tool()
async def click_element(selector: str, by: str = "css"):
"""
Clicks an element on the page using the specified selector
This selector can be css, xpath, or class_name
"""
global _browser
pages = await _browser.pages()
page = pages[-1]
if by == "css":
await page.click(selector)
elif by == "xpath":
elements = await page.xpath(selector)
if elements:
await elements[0].click()
elif by == "class_name":
await page.evaluate(f'document.getElementsByClassName("{selector}")[0].click()')
return {"status": "success", "selector": selector, "by": by}
@mcp.tool()
async def exit_server():
"""
Stops the server
"""
return {"status": "Server will exit"}
# Define the resource
# Define the resource
@mcp.resource("page://current")
async def current_page() -> ResourceContents:
"""
Returns the URL of the current page
"""
global _browser
pages = await _browser.pages()
if not pages:
return ResourceContents(content=TextContent(text="No open pages"), mime_type="text/plain")
page = pages[-1]
url = await page.evaluate('window.location.href')
# Remova o parâmetro 'body' ou defina-o adequadamente
return ResourceContents(content=TextContent(text=url), mime_type="text/plain")
def main():
"""Main entry point for the MCP Puppeteer server."""
mcp.run()
# Run the server
if __name__ == "__main__":
main()