movies_database.py•7.79 kB
"""
MySQL-MCP Example: Movie Database
This example shows how to use MySQL-MCP with a movie database.
It includes setup instructions for creating the sample database.
Setup:
1. Make sure you have MySQL running
2. Execute the SQL in setup_movies_db.sql to create the sample database
3. Update the .env file with your MySQL credentials
4. Run this example:
fastmcp dev examples/movies_database.py
"""
from fastmcp import FastMCP, Context
import mysql.connector
from mysql.connector import Error
import os
from dotenv import load_dotenv
from typing import Dict, Any, List, Optional
# Load environment variables
load_dotenv()
# Initialize FastMCP app
mcp = FastMCP(
"Movies Database",
description="Explore and query a movies database",
dependencies=["mysql-connector-python", "python-dotenv"]
)
# Database connection configuration
DB_CONFIG = {
"host": os.getenv("MYSQL_HOST", "localhost"),
"port": int(os.getenv("MYSQL_PORT", "3306")),
"user": os.getenv("MYSQL_USER", "root"),
"password": os.getenv("MYSQL_PASSWORD", ""),
"database": "movies" # This example uses a fixed database
}
def get_connection():
"""Create a MySQL connection"""
try:
conn = mysql.connector.connect(**DB_CONFIG)
return conn
except Error as e:
raise Exception(f"Database connection error: {e}")
@mcp.tool()
def search_movies(title_keyword: str = "", genre: Optional[str] = None, min_rating: Optional[float] = None) -> Dict[str, Any]:
"""Search for movies by title keyword, genre, and/or minimum rating"""
conn = get_connection()
cursor = conn.cursor(dictionary=True)
# Build the query with parameters
query = "SELECT m.id, m.title, m.release_year, m.rating, GROUP_CONCAT(g.name) as genres "
query += "FROM movies m "
query += "LEFT JOIN movie_genres mg ON m.id = mg.movie_id "
query += "LEFT JOIN genres g ON mg.genre_id = g.id "
query += "WHERE 1=1 "
params = []
if title_keyword:
query += "AND m.title LIKE %s "
params.append(f"%{title_keyword}%")
if genre:
query += "AND g.name = %s "
params.append(genre)
if min_rating is not None:
query += "AND m.rating >= %s "
params.append(min_rating)
query += "GROUP BY m.id "
query += "ORDER BY m.rating DESC "
query += "LIMIT 20"
try:
cursor.execute(query, params)
results = cursor.fetchall()
return {
"movies": results,
"count": len(results)
}
except Error as e:
raise Exception(f"Search error: {e}")
finally:
cursor.close()
conn.close()
@mcp.tool()
def get_movie_details(movie_id: int) -> Dict[str, Any]:
"""Get detailed information about a movie by ID"""
conn = get_connection()
movie_cursor = conn.cursor(dictionary=True)
cast_cursor = conn.cursor(dictionary=True)
try:
# Get movie details
movie_cursor.execute("""
SELECT m.*, GROUP_CONCAT(DISTINCT g.name) as genres
FROM movies m
LEFT JOIN movie_genres mg ON m.id = mg.movie_id
LEFT JOIN genres g ON mg.genre_id = g.id
WHERE m.id = %s
GROUP BY m.id
""", (movie_id,))
movie = movie_cursor.fetchone()
if not movie:
raise ValueError(f"Movie with ID {movie_id} not found")
# Get cast information
cast_cursor.execute("""
SELECT a.name, c.character_name, c.role
FROM movie_cast c
JOIN actors a ON c.actor_id = a.id
WHERE c.movie_id = %s
ORDER BY c.role
""", (movie_id,))
cast = cast_cursor.fetchall()
return {
"movie": movie,
"cast": cast
}
except Error as e:
raise Exception(f"Query error: {e}")
finally:
movie_cursor.close()
cast_cursor.close()
conn.close()
@mcp.tool()
def list_genres() -> Dict[str, Any]:
"""List all movie genres in the database"""
conn = get_connection()
cursor = conn.cursor(dictionary=True)
try:
cursor.execute("""
SELECT g.name, COUNT(mg.movie_id) as movie_count
FROM genres g
LEFT JOIN movie_genres mg ON g.id = mg.genre_id
GROUP BY g.id
ORDER BY movie_count DESC
""")
genres = cursor.fetchall()
return {
"genres": genres
}
except Error as e:
raise Exception(f"Query error: {e}")
finally:
cursor.close()
conn.close()
@mcp.tool()
def top_rated_movies(limit: int = 10, genre: Optional[str] = None) -> Dict[str, Any]:
"""Get the top rated movies, optionally filtered by genre"""
conn = get_connection()
cursor = conn.cursor(dictionary=True)
# Safety limit
safe_limit = min(limit, 50)
try:
query = """
SELECT m.id, m.title, m.release_year, m.rating, GROUP_CONCAT(g.name) as genres
FROM movies m
LEFT JOIN movie_genres mg ON m.id = mg.movie_id
LEFT JOIN genres g ON mg.genre_id = g.id
"""
params = []
if genre:
query += "WHERE g.name = %s "
params.append(genre)
query += """
GROUP BY m.id
HAVING m.rating IS NOT NULL
ORDER BY m.rating DESC
LIMIT %s
"""
params.append(safe_limit)
cursor.execute(query, params)
movies = cursor.fetchall()
return {
"movies": movies,
"count": len(movies)
}
except Error as e:
raise Exception(f"Query error: {e}")
finally:
cursor.close()
conn.close()
@mcp.resource("schema://movies")
def get_movies_schema() -> str:
"""Get the database schema for the movies database"""
conn = get_connection()
cursor = conn.cursor()
schema = []
try:
# Get all tables
cursor.execute("SHOW TABLES")
tables = [table[0] for table in cursor.fetchall()]
# Get CREATE TABLE statements for each table
for table in tables:
cursor.execute(f"SHOW CREATE TABLE {table}")
create_stmt = cursor.fetchone()[1]
schema.append(create_stmt)
return "\n\n".join(schema)
except Error as e:
raise Exception(f"Error getting schema: {e}")
finally:
cursor.close()
conn.close()
@mcp.prompt()
def movie_recommendation() -> str:
"""Recommend movies based on user preferences"""
return """I'd like you to recommend some movies for me to watch.
Before making recommendations:
1. Use the list_genres tool to see what genres are available
2. Use the schema://movies resource to understand the database structure
3. Use the search_movies tool to find movies matching my criteria
4. Use the get_movie_details tool to get more information about specific movies
Then provide personalized recommendations with brief explanations of why you chose each movie.
"""
@mcp.prompt()
def analyze_movie_trends() -> str:
"""Help Claude analyze trends in the movie database"""
return """I'd like you to analyze trends in the movie database.
Some interesting aspects to explore:
1. How movie ratings vary by genre
2. Whether certain actors appear more frequently in highly-rated movies
3. If there are trends in movie releases over time
4. Which genres are most popular in different time periods
Use the available tools to query the database and provide interesting insights backed by data.
"""
if __name__ == "__main__":
# Run the server directly
mcp.run()