MCP-Server mit Python III: Einbindung von Playwright

pip install playwright
playwright install chromium

#!/usr/bin/env python3
"""MCP-Server mit Playwright-Integration."""

import asyncio
import base64
import sys
from pathlib import Path

from mcp.server import Server
from mcp.server.stdio import stdio_server
from playwright.async_api import async_playwright, Browser, Page

mcp = Server("playwright-tools")

# Globaler Zustand
_browser: Browser | None = None
_page: Page | None = None
_playwright = None


async def get_page() -> Page:
    """Liefert die aktuelle Page, startet Browser falls nötig."""
    global _browser, _page, _playwright

    if _page is None or _page.is_closed():
        if _playwright is None:
            _playwright = await async_playwright().start()
        if _browser is None:
            _browser = await _playwright.chromium.launch(
                headless=True,  # False für Debugging
            )
        _page = await _browser.new_page()

    return _page


async def cleanup():
    """Räumt Browser-Ressourcen auf."""
    global _browser, _page, _playwright

    if _page and not _page.is_closed():
        await _page.close()
    if _browser:
        await _browser.close()
    if _playwright:
        await _playwright.stop()

    _page = None
    _browser = None
    _playwright = None


# =============================================================================
# Navigation
# =============================================================================

@mcp.tool()
async def navigate(url: str) -> str:
    """Navigiert zu einer URL. Wartet bis die Seite geladen ist."""
    page = await get_page()
    try:
        await page.goto(url, wait_until="domcontentloaded")
        return f"Navigiert zu: {page.url}"
    except Exception as e:
        return f"Fehler: {e}"


@mcp.tool()
async def current_url() -> str:
    """Gibt die aktuelle URL zurück."""
    page = await get_page()
    return page.url


@mcp.tool()
async def go_back() -> str:
    """Geht eine Seite zurück."""
    page = await get_page()
    await page.go_back()
    return f"Zurück zu: {page.url}"


# =============================================================================
# Seiten-Inhalt
# =============================================================================

@mcp.tool()
async def get_title() -> str:
    """Gibt den Seitentitel zurück."""
    page = await get_page()
    return await page.title()


@mcp.tool()
async def get_text(selector: str) -> str:
    """Liest den Text eines Elements aus."""
    page = await get_page()
    try:
        element = page.locator(selector)
        await element.wait_for(timeout=5000)
        return await element.inner_text()
    except Exception as e:
        return f"Fehler: {e}"


@mcp.tool()
async def get_all_texts(selector: str) -> str:
    """Liest den Text aller passenden Elemente aus."""
    page = await get_page()
    try:
        elements = page.locator(selector)
        count = await elements.count()
        if count == 0:
            return "(keine Elemente gefunden)"

        texts = []
        for i in range(count):
            text = await elements.nth(i).inner_text()
            texts.append(f"{i+1}. {text.strip()}")

        return "\n".join(texts)
    except Exception as e:
        return f"Fehler: {e}"


@mcp.tool()
async def get_page_content(max_length: int = 10000) -> str:
    """Gibt den sichtbaren Text der Seite zurück (für Kontext)."""
    page = await get_page()
    try:
        # Nur sichtbarer Text, kein HTML
        text = await page.inner_text("body")
        text = "\n".join(line.strip() for line in text.split("\n") if line.strip())

        if len(text) > max_length:
            text = text[:max_length] + "\n... (gekürzt)"

        return text
    except Exception as e:
        return f"Fehler: {e}"


@mcp.tool()
async def get_links() -> str:
    """Listet alle Links auf der Seite auf."""
    page = await get_page()
    try:
        links = await page.eval_on_selector_all(
            "a[href]",
            """elements => elements.map(e => ({
                text: e.innerText.trim().substring(0, 50),
                href: e.href
            })).filter(l => l.text && l.href)"""
        )

        if not links:
            return "(keine Links gefunden)"

        return "\n".join(f"- {l['text']}: {l['href']}" for l in links[:50])
    except Exception as e:
        return f"Fehler: {e}"


# =============================================================================
# Interaktion
# =============================================================================

@mcp.tool()
async def click(selector: str) -> str:
    """Klickt auf ein Element."""
    page = await get_page()
    try:
        await page.click(selector, timeout=5000)
        await page.wait_for_load_state("domcontentloaded")
        return f"Geklickt: {selector}"
    except Exception as e:
        return f"Fehler: {e}"


@mcp.tool()
async def fill(selector: str, text: str) -> str:
    """Füllt ein Eingabefeld aus."""
    page = await get_page()
    try:
        await page.fill(selector, text, timeout=5000)
        return f"Ausgefüllt: {selector}"
    except Exception as e:
        return f"Fehler: {e}"


@mcp.tool()
async def press(key: str) -> str:
    """Drückt eine Taste (z.B. 'Enter', 'Tab', 'Escape')."""
    page = await get_page()
    try:
        await page.keyboard.press(key)
        return f"Taste gedrückt: {key}"
    except Exception as e:
        return f"Fehler: {e}"


@mcp.tool()
async def select_option(selector: str, value: str) -> str:
    """Wählt eine Option in einem Dropdown aus."""
    page = await get_page()
    try:
        await page.select_option(selector, value, timeout=5000)
        return f"Ausgewählt: {value} in {selector}"
    except Exception as e:
        return f"Fehler: {e}"


# =============================================================================
# Screenshots
# =============================================================================

@mcp.tool()
async def screenshot(path: str = "screenshot.png", full_page: bool = False) -> str:
    """Macht einen Screenshot der aktuellen Seite."""
    page = await get_page()
    try:
        filepath = Path(path).expanduser()
        await page.screenshot(path=str(filepath), full_page=full_page)
        return f"Screenshot gespeichert: {filepath}"
    except Exception as e:
        return f"Fehler: {e}"


# =============================================================================
# Warten
# =============================================================================

@mcp.tool()
async def wait_for(selector: str, timeout: int = 10000) -> str:
    """Wartet bis ein Element sichtbar ist."""
    page = await get_page()
    try:
        await page.wait_for_selector(selector, timeout=timeout)
        return f"Element gefunden: {selector}"
    except Exception as e:
        return f"Fehler (Timeout?): {e}"


@mcp.tool()
async def wait_seconds(seconds: float) -> str:
    """Wartet eine bestimmte Zeit (für Debugging)."""
    await asyncio.sleep(seconds)
    return f"Gewartet: {seconds}s"


# =============================================================================
# Browser-Steuerung
# =============================================================================

@mcp.tool()
async def new_page() -> str:
    """Öffnet einen neuen Tab."""
    global _page
    if _browser is None:
        await get_page()  # Browser initialisieren

    _page = await _browser.new_page()
    return "Neuer Tab geöffnet"


@mcp.tool()
async def close_page() -> str:
    """Schließt den aktuellen Tab."""
    global _page
    if _page and not _page.is_closed():
        await _page.close()
        _page = None
        return "Tab geschlossen"
    return "Kein Tab offen"


@mcp.tool()
async def set_headless(headless: bool) -> str:
    """Startet Browser neu im headless/headed Modus."""
    await cleanup()
    global _browser, _playwright

    _playwright = await async_playwright().start()
    _browser = await _playwright.chromium.launch(headless=headless)

    mode = "headless" if headless else "headed"
    return f"Browser neu gestartet im {mode} Modus"


# =============================================================================
# Main
# =============================================================================

async def main():
    try:
        async with stdio_server() as (read, write):
            await mcp.run(read, write)
    finally:
        await cleanup()


if __name__ == "__main__":
    asyncio.run(main())

{
  "mcpServers": {
    "playwright": {
      "command": "python",
      "args": ["/pfad/zu/playwright_server.py"]
    }
  }
}

def test_mein_test(page: Page) -> None:
    page.goto(f"{BASE_URL}/about/")
    expect(page).to_have_title(re.compile(re.escape("Über mich")), timeout=5000)
    page.click("nav a[href='/kontakt/']")
    # ... weitere Schritte