MCP server source
This commit is contained in:
+12
@@ -0,0 +1,12 @@
|
||||
FROM python:3.13-slim
|
||||
WORKDIR /app
|
||||
RUN apt-get update && apt-get install -y --no-install-recommends \
|
||||
curl ca-certificates && rm -rf /var/lib/apt/lists/*
|
||||
COPY requirements.txt .
|
||||
RUN pip install --no-cache-dir -r requirements.txt
|
||||
RUN python3 -m playwright install chromium
|
||||
RUN python3 -m playwright install-deps chromium
|
||||
COPY server.py ./
|
||||
ENV PYTHONUNBUFFERED=1
|
||||
EXPOSE 8901
|
||||
CMD ["python3", "server.py"]
|
||||
@@ -0,0 +1,3 @@
|
||||
fastmcp>=3.0.0
|
||||
playwright>=1.45.0
|
||||
httpx>=0.27.0
|
||||
@@ -0,0 +1,97 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Browser MCP Server — Web page rendering, screenshots, JS execution via Playwright."""
|
||||
from fastmcp import FastMCP
|
||||
from playwright.async_api import async_playwright
|
||||
import base64, asyncio, os
|
||||
|
||||
BROWSER_WS = os.getenv('BROWSER_WS', 'ws://browserless:3000')
|
||||
mcp = FastMCP('Browser', version='1.0.0')
|
||||
|
||||
_browser = None
|
||||
|
||||
async def get_browser():
|
||||
global _browser
|
||||
if _browser is None:
|
||||
pw = await async_playwright().start()
|
||||
_browser = await pw.chromium.connect_over_cdp(BROWSER_WS)
|
||||
return _browser
|
||||
|
||||
@mcp.tool(description="Navigate to a URL and return the page's text content + title")
|
||||
async def browse(url: str) -> str:
|
||||
"""Navigate to URL, return title and visible text."""
|
||||
browser = await get_browser()
|
||||
page = await browser.new_page()
|
||||
try:
|
||||
await page.goto(url, wait_until='networkidle', timeout=30000)
|
||||
title = await page.title()
|
||||
text = await page.evaluate('''() => document.body?.innerText || ''')
|
||||
result = f"Title: {title}\n\n{text[:10000]}"
|
||||
if len(text) > 10000:
|
||||
result += f"\n\n... (truncated, {len(text)} total chars)"
|
||||
return result
|
||||
finally:
|
||||
await page.close()
|
||||
|
||||
@mcp.tool(description="Take a screenshot of a webpage, return as base64 PNG")
|
||||
async def screenshot(url: str) -> str:
|
||||
"""Take full-page screenshot, return as base64 data URL."""
|
||||
browser = await get_browser()
|
||||
page = await browser.new_page()
|
||||
try:
|
||||
await page.goto(url, wait_until='networkidle', timeout=30000)
|
||||
data = await page.screenshot(full_page=True, type='png')
|
||||
title = await page.title()
|
||||
b64 = base64.b64encode(data).decode()
|
||||
return f"data:image/png;base64,{b64}"
|
||||
finally:
|
||||
await page.close()
|
||||
|
||||
@mcp.tool(description="Execute JavaScript on a webpage and return the result")
|
||||
async def execute_js(url: str, code: str) -> str:
|
||||
"""Navigate to URL, run JS, return result."""
|
||||
browser = await get_browser()
|
||||
page = await browser.new_page()
|
||||
try:
|
||||
await page.goto(url, wait_until='networkidle', timeout=30000)
|
||||
result = await page.evaluate(code)
|
||||
return str(result)[:5000]
|
||||
finally:
|
||||
await page.close()
|
||||
|
||||
@mcp.tool(description="Extract specific content from a page using a CSS selector")
|
||||
async def extract_element(url: str, selector: str) -> str:
|
||||
"""Extract text from elements matching a CSS selector."""
|
||||
browser = await get_browser()
|
||||
page = await browser.new_page()
|
||||
try:
|
||||
await page.goto(url, wait_until='networkidle', timeout=30000)
|
||||
elements = await page.query_selector_all(selector)
|
||||
texts = []
|
||||
for el in elements:
|
||||
t = await el.inner_text()
|
||||
if t.strip():
|
||||
texts.append(t.strip())
|
||||
return f"Found {len(texts)} elements matching '{selector}':\n" + "\n---\n".join(texts[:10])
|
||||
finally:
|
||||
await page.close()
|
||||
|
||||
@mcp.tool(description="Click an element on a page by text or selector and return the result")
|
||||
async def click(url: str, target: str, is_selector: bool = False) -> str:
|
||||
"""Navigate, click element by text content or CSS selector, return new page content."""
|
||||
browser = await get_browser()
|
||||
page = await browser.new_page()
|
||||
try:
|
||||
await page.goto(url, wait_until='networkidle', timeout=30000)
|
||||
if is_selector:
|
||||
await page.click(target)
|
||||
else:
|
||||
await page.click(f'text={target}')
|
||||
await page.wait_for_load_state('networkidle', timeout=10000)
|
||||
title = await page.title()
|
||||
text = await page.evaluate('''() => document.body?.innerText || ''')
|
||||
return f"Title: {title}\nURL: {page.url}\n\n{text[:8000]}"
|
||||
finally:
|
||||
await page.close()
|
||||
|
||||
if __name__ == '__main__':
|
||||
mcp.run(transport='http', host='0.0.0.0', port=8901, path='/mcp')
|
||||
Reference in New Issue
Block a user