Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 0 additions & 1 deletion src/animawatch/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -6,4 +6,3 @@

__version__ = "0.2.0"
__all__ = ["server", "browser", "vision", "config"]

39 changes: 27 additions & 12 deletions src/animawatch/browser.py
Original file line number Diff line number Diff line change
@@ -1,31 +1,39 @@
"""Browser automation and video recording using Playwright for AnimaWatch."""

import asyncio
import os
import tempfile
from collections.abc import AsyncGenerator
from contextlib import asynccontextmanager
from pathlib import Path
from typing import AsyncGenerator
from typing import Any

from playwright.async_api import Browser, BrowserContext, Page, async_playwright
from playwright.async_api import (
Browser,
BrowserContext,
Page,
Playwright,
async_playwright,
)

from .config import settings


class BrowserRecorder:
"""Manages browser automation and video recording for AnimaWatch."""

def __init__(self):
self._playwright = None
def __init__(self) -> None:
self._playwright: Playwright | None = None
self._browser: Browser | None = None

async def start(self):
async def start(self) -> None:
"""Start the Playwright browser."""
self._playwright = await async_playwright().start()
self._browser = await self._playwright.chromium.launch(
headless=settings.browser_headless,
)

async def stop(self):
async def stop(self) -> None:
"""Stop the browser and cleanup."""
if self._browser:
await self._browser.close()
Expand All @@ -49,6 +57,9 @@ async def recording_context(

video_dir.mkdir(parents=True, exist_ok=True)

if self._browser is None:
raise RuntimeError("Browser not initialized")

context = await self._browser.new_context(
viewport=settings.video_size,
record_video_dir=str(video_dir),
Expand All @@ -60,17 +71,17 @@ async def recording_context(
try:
yield context, page, video_dir
finally:
# Get the video path before closing
# Ensure video is saved before closing
video = page.video
if video:
video_path = await video.path()
await video.path() # Wait for video to be saved

await context.close()

async def record_interaction(
self,
url: str,
actions: list[dict] | None = None,
actions: list[dict[str, Any]] | None = None,
wait_time: float = 3.0,
Comment thread
coderabbitai[bot] marked this conversation as resolved.
video_dir: Path | None = None,
) -> Path:
Expand Down Expand Up @@ -111,20 +122,25 @@ async def take_screenshot(self, url: str, full_page: bool = True) -> Path:
if not self._browser:
await self.start()

if self._browser is None:
raise RuntimeError("Browser not initialized")

context = await self._browser.new_context(viewport=settings.video_size)
page = await context.new_page()

try:
await page.goto(url, wait_until="networkidle")

screenshot_path = Path(tempfile.mktemp(suffix=".png"))
fd, tmp_path = tempfile.mkstemp(suffix=".png")
os.close(fd)
screenshot_path = Path(tmp_path)
await page.screenshot(path=str(screenshot_path), full_page=full_page)

return screenshot_path
finally:
await context.close()

async def _perform_action(self, page: Page, action: dict):
async def _perform_action(self, page: Page, action: dict[str, Any]) -> None:
"""Perform a single browser action."""
action_type = action.get("type", "")

Expand All @@ -151,4 +167,3 @@ async def _perform_action(self, page: Page, action: dict):
selector = action.get("selector")
if selector:
await page.hover(selector)

16 changes: 11 additions & 5 deletions src/animawatch/config.py
Original file line number Diff line number Diff line change
@@ -1,12 +1,19 @@
"""Configuration settings for AnimaWatch MCP Server."""

from pathlib import Path
from typing import Literal
from typing import Literal, TypedDict

from pydantic import Field
from pydantic_settings import BaseSettings, SettingsConfigDict


class ViewportSize(TypedDict):
"""Viewport size type compatible with Playwright."""

width: int
height: int


class Settings(BaseSettings):
"""Server configuration loaded from environment variables."""

Expand Down Expand Up @@ -61,11 +68,10 @@ class Settings(BaseSettings):
)

@property
def video_size(self) -> dict:
"""Return video size as dict for Playwright."""
return {"width": self.video_width, "height": self.video_height}
def video_size(self) -> ViewportSize:
"""Return video size as TypedDict for Playwright compatibility."""
return ViewportSize(width=self.video_width, height=self.video_height)


# Global settings instance
settings = Settings()

43 changes: 23 additions & 20 deletions src/animawatch/server.py
Original file line number Diff line number Diff line change
Expand Up @@ -8,12 +8,13 @@
- Sampling for server-side LLM requests
"""

import contextlib
import uuid
from collections.abc import AsyncIterator
from contextlib import asynccontextmanager
from dataclasses import dataclass, field
from pathlib import Path
from typing import Any
import uuid
from typing import Any, Literal

from mcp.server.fastmcp import Context, FastMCP
from mcp.server.fastmcp.utilities.types import Image
Expand All @@ -23,7 +24,6 @@
from .config import settings
from .vision import VisionProvider, get_vision_provider


# =============================================================================
# Application Context (Lifespan Management)
# =============================================================================
Expand Down Expand Up @@ -204,7 +204,7 @@ async def watch(
wait_time: float = 3.0,
focus: str = "all",
save_recording: bool = False,
ctx: Context[ServerSession, AppContext] = None,
ctx: Context[ServerSession, AppContext] | None = None,
) -> str:
"""Watch a web page and analyze animations for issues.

Expand All @@ -218,6 +218,8 @@ async def watch(
focus: Focus area for analysis (e.g., "modal animations", "scroll behavior")
save_recording: Whether to save the recording for later access via resources
"""
if ctx is None:
raise RuntimeError("Context is required")
app_ctx = ctx.request_context.lifespan_context
browser = app_ctx.browser
vision = app_ctx.vision
Expand All @@ -240,10 +242,8 @@ async def watch(
if save_recording:
app_ctx.recordings[result_id] = video_path
else:
try:
with contextlib.suppress(OSError):
video_path.unlink()
except Exception:
pass

app_ctx.analyses[result_id] = analysis

Expand All @@ -261,7 +261,7 @@ async def screenshot(
url: str,
full_page: bool = True,
focus: str = "layout, colors, typography, spacing",
ctx: Context[ServerSession, AppContext] = None,
ctx: Context[ServerSession, AppContext] | None = None,
) -> Image:
"""Take a screenshot and return it with analysis.

Expand All @@ -272,6 +272,8 @@ async def screenshot(
full_page: Capture full scrollable page or just viewport
focus: Aspects to focus analysis on
"""
if ctx is None:
raise RuntimeError("Context is required")
app_ctx = ctx.request_context.lifespan_context
browser = app_ctx.browser
vision = app_ctx.vision
Expand All @@ -290,10 +292,8 @@ async def screenshot(
with open(screenshot_path, "rb") as f:
image_data = f.read()

try:
with contextlib.suppress(OSError):
screenshot_path.unlink()
except Exception:
pass

return Image(data=image_data, format="png")

Expand All @@ -302,14 +302,16 @@ async def screenshot(
async def analyze_video(
video_path: str,
focus: str = "all",
ctx: Context[ServerSession, AppContext] = None,
ctx: Context[ServerSession, AppContext] | None = None,
) -> str:
"""Analyze an existing video file for animation issues.

Args:
video_path: Path to the video file
focus: Focus area for analysis
"""
if ctx is None:
raise RuntimeError("Context is required")
app_ctx = ctx.request_context.lifespan_context
vision = app_ctx.vision

Expand All @@ -332,7 +334,7 @@ async def record(
actions: list[dict[str, Any]] | None = None,
wait_time: float = 3.0,
output_dir: str | None = None,
ctx: Context[ServerSession, AppContext] = None,
ctx: Context[ServerSession, AppContext] | None = None,
) -> str:
"""Record a browser interaction without analysis.

Expand All @@ -344,6 +346,8 @@ async def record(
wait_time: Seconds to wait after actions
output_dir: Directory to save video (default: temp directory)
"""
if ctx is None:
raise RuntimeError("Context is required")
app_ctx = ctx.request_context.lifespan_context
browser = app_ctx.browser

Expand All @@ -370,7 +374,7 @@ async def record(
@mcp.tool()
async def check_accessibility(
url: str,
ctx: Context[ServerSession, AppContext] = None,
ctx: Context[ServerSession, AppContext] | None = None,
) -> str:
"""Check a page for visual accessibility issues.

Expand All @@ -379,6 +383,8 @@ async def check_accessibility(
Args:
url: URL to check
"""
if ctx is None:
raise RuntimeError("Context is required")
app_ctx = ctx.request_context.lifespan_context
browser = app_ctx.browser
vision = app_ctx.vision
Expand All @@ -388,10 +394,8 @@ async def check_accessibility(
prompt = accessibility_check()
analysis = await vision.analyze_image(screenshot_path, prompt)

try:
with contextlib.suppress(OSError):
screenshot_path.unlink()
except Exception:
pass

result_id = str(uuid.uuid4())[:8]
app_ctx.analyses[result_id] = analysis
Expand All @@ -404,12 +408,12 @@ async def check_accessibility(
# =============================================================================


def main():
def main() -> None:
"""Run the AnimaWatch MCP server."""
import sys

# Support both stdio (default) and streamable-http transports
transport = "stdio"
transport: Literal["stdio", "streamable-http"] = "stdio"
if "--http" in sys.argv:
transport = "streamable-http"

Expand All @@ -418,4 +422,3 @@ def main():

if __name__ == "__main__":
main()

Loading
Loading