"""MCP server for Filmladder movie listings and recommendations."""
from __future__ import annotations
import asyncio
from datetime import date, time
from typing import Any
from mcp.server import Server
from mcp.server.stdio import stdio_server
from mcp.types import Tool, TextContent
from src.models import Movie, Showtime
from src.recommender import async_recommend_movies
from src.scraper import ScrapingError, fetch_amsterdam_cinemas
app = Server("filmladder-mcp")
@app.list_tools()
async def handle_list_tools() -> list[Tool]:
"""List available MCP tools."""
return [
Tool(
name="list_movies",
description="List all movies playing in Amsterdam cinemas, optionally filtered by date",
inputSchema={
"type": "object",
"properties": {
"date": {
"type": "string",
"format": "date",
"description": "Optional date filter (YYYY-MM-DD format). If not provided, shows all movies.",
},
},
},
),
Tool(
name="get_showtimes",
description="Get all showtimes for a specific movie (fuzzy matching on title)",
inputSchema={
"type": "object",
"properties": {
"movie_title": {
"type": "string",
"description": "Title of the movie to find showtimes for",
},
},
"required": ["movie_title"],
},
),
Tool(
name="list_cinema_movies",
description="List all movies playing at a specific cinema",
inputSchema={
"type": "object",
"properties": {
"cinema_name": {
"type": "string",
"description": "Name of the cinema",
},
},
"required": ["cinema_name"],
},
),
Tool(
name="recommend_movies",
description="Recommend movies based on rating, preferred showtimes, cinemas, and date",
inputSchema={
"type": "object",
"properties": {
"min_rating": {
"type": "number",
"description": "Minimum rating threshold (default: 0.0)",
"default": 0.0,
},
"preferred_times": {
"type": "array",
"items": {"type": "string", "format": "time"},
"description": "Preferred showtimes (HH:MM format)",
},
"preferred_cinemas": {
"type": "array",
"items": {"type": "string"},
"description": "Preferred cinema names",
},
"date": {
"type": "string",
"format": "date",
"description": "Target date for recommendations (YYYY-MM-DD format)",
},
},
},
),
]
@app.call_tool()
async def handle_call_tool(name: str, arguments: dict[str, Any] | None) -> list[TextContent]:
"""Handle tool calls."""
if arguments is None:
arguments = {}
try:
if name == "list_movies":
date_str = arguments.get("date")
target_date = date.fromisoformat(date_str) if date_str else None
movies = await list_movies(target_date)
return [TextContent(type="text", text=format_movies_response(movies))]
elif name == "get_showtimes":
movie_title = arguments.get("movie_title", "")
if not movie_title:
raise ValueError("movie_title is required")
showtimes = await get_showtimes(movie_title)
return [TextContent(type="text", text=format_showtimes_response(showtimes))]
elif name == "list_cinema_movies":
cinema_name = arguments.get("cinema_name", "")
if not cinema_name:
raise ValueError("cinema_name is required")
movies = await list_cinema_movies(cinema_name)
return [TextContent(type="text", text=format_movies_response(movies))]
elif name == "recommend_movies":
min_rating = arguments.get("min_rating", 0.0)
preferred_times_str = arguments.get("preferred_times", [])
preferred_times = (
[time.fromisoformat(t) for t in preferred_times_str] if preferred_times_str else None
)
preferred_cinemas = arguments.get("preferred_cinemas")
date_str = arguments.get("date")
target_date = date.fromisoformat(date_str) if date_str else None
movies = await async_recommend_movies(
min_rating=min_rating,
preferred_times=preferred_times,
preferred_cinemas=preferred_cinemas,
target_date=target_date,
)
return [TextContent(type="text", text=format_movies_response(movies))]
else:
raise ValueError(f"Unknown tool: {name}")
except ScrapingError as e:
raise ValueError(f"Scraping error: {e}") from e
except Exception as e:
raise ValueError(f"Error executing tool {name}: {e}") from e
async def list_movies(target_date: date | None = None) -> list[Movie]:
"""List all movies playing in Amsterdam, optionally filtered by date."""
cinemas = await fetch_amsterdam_cinemas()
# Collect all unique movies
movie_map: dict[str, Movie] = {}
for cinema in cinemas:
for movie in cinema.movies:
if movie.title not in movie_map:
movie_map[movie.title] = Movie(
title=movie.title,
rating=movie.rating,
showtimes=[],
)
# Merge showtimes
movie_map[movie.title].showtimes.extend(movie.showtimes)
# Filter by date if specified
if target_date:
for movie in movie_map.values():
movie.showtimes = [st for st in movie.showtimes if st.showtime_date == target_date]
# Remove movies with no showtimes after filtering
movies = [m for m in movie_map.values() if m.showtimes]
# Sort by rating (highest first)
movies.sort(key=lambda m: m.rating if m.rating is not None else 0.0, reverse=True)
return movies
async def get_showtimes(movie_title: str) -> list[Showtime]:
"""Get showtimes for a specific movie (fuzzy matching)."""
cinemas = await fetch_amsterdam_cinemas()
# Fuzzy match: case-insensitive partial match
movie_title_lower = movie_title.lower()
all_showtimes: list[Showtime] = []
for cinema in cinemas:
for movie in cinema.movies:
if movie_title_lower in movie.title.lower() or movie.title.lower() in movie_title_lower:
all_showtimes.extend(movie.showtimes)
# Sort by date, then time
all_showtimes.sort(key=lambda st: (st.showtime_date, st.showtime_time))
return all_showtimes
async def list_cinema_movies(cinema_name: str) -> list[Movie]:
"""List movies playing at a specific cinema."""
cinemas = await fetch_amsterdam_cinemas()
# Fuzzy match cinema name
cinema_name_lower = cinema_name.lower()
for cinema in cinemas:
if cinema_name_lower in cinema.name.lower() or cinema.name.lower() in cinema_name_lower:
# Sort movies by rating
movies = sorted(
cinema.movies,
key=lambda m: m.rating if m.rating is not None else 0.0,
reverse=True,
)
return movies
return []
def format_movies_response(movies: list[Movie]) -> str:
"""Format movies list as text response."""
if not movies:
return "No movies found."
lines = [f"Found {len(movies)} movie(s):\n"]
for movie in movies:
rating_str = f" ({movie.rating}★)" if movie.rating else ""
lines.append(f"- {movie.title}{rating_str}")
if movie.showtimes:
lines.append(f" Showtimes: {len(movie.showtimes)} available")
# Show first few showtimes
for st in movie.showtimes[:3]:
lines.append(f" • {st.cinema_name}: {st.showtime_date} at {st.showtime_time}")
if len(movie.showtimes) > 3:
lines.append(f" ... and {len(movie.showtimes) - 3} more")
return "\n".join(lines)
def format_showtimes_response(showtimes: list[Showtime]) -> str:
"""Format showtimes list as text response."""
if not showtimes:
return "No showtimes found."
lines = [f"Found {len(showtimes)} showtime(s):\n"]
current_movie = ""
for st in showtimes:
if st.movie_title != current_movie:
current_movie = st.movie_title
lines.append(f"\n{current_movie}:")
lines.append(f" • {st.cinema_name}: {st.showtime_date} at {st.showtime_time}")
return "\n".join(lines)
async def main() -> None:
"""Main entry point for the MCP server."""
async with stdio_server() as (read_stream, write_stream):
await app.run(read_stream, write_stream, app.create_initialization_options())
if __name__ == "__main__":
asyncio.run(main())
