One-MCP / tools /trakt.py
arcticaurora's picture
Update tools/trakt.py
4b6713b verified
from mcp.server.fastmcp import FastMCP
import requests
import os
from typing import Dict, List
mcp = FastMCP("Trakt")
@mcp.tool()
def get_trending_movies(limit: int = 10) -> Dict:
"""Get currently trending movies from Trakt.tv.
Args:
limit (int, optional): Maximum number of trending movies to return.
Must be between 1-50. Defaults to 10.
Values outside range are clamped to 1-50.
Returns:
Dict: Trending movies information containing:
- status (str): "success" or "error"
- count (int): Number of movies returned
- movies (List[Dict]): List of trending movies, each containing:
- title (str): Movie title
- year (int): Release year
- rating (float): Trakt user rating (0-10 scale)
- votes (int): Number of user votes
- watchers (int): Number of people currently watching
- error (str, optional): Error message if status is "error"
Examples:
get_trending_movies() returns top 10 trending movies
get_trending_movies(5) returns top 5 trending movies
get_trending_movies(100) returns 50 movies (clamped to maximum)
Note:
Uses Trakt.tv public API. No API key required for basic trending data.
For enhanced features, set TRAKT_API_KEY environment variable.
"""
if limit < 1 or limit > 50:
limit = max(1, min(50, limit))
try:
url = "https://api.trakt.tv/movies/trending"
headers = {
"Content-Type": "application/json",
"trakt-api-version": "2"
}
# Add API key if available for enhanced data
api_key = os.getenv("TRAKT_API_KEY")
if api_key:
headers["trakt-api-key"] = api_key
params = {"limit": limit}
response = requests.get(url, headers=headers, params=params, timeout=10)
response.raise_for_status()
data = response.json()
movies = []
for item in data:
movie = item.get("movie", {})
movies.append({
"title": movie.get("title", "Unknown Title"),
"year": movie.get("year", 0),
"rating": round(movie.get("rating", 0), 1),
"votes": movie.get("votes", 0),
"watchers": item.get("watchers", 0)
})
return {
"status": "success",
"count": len(movies),
"movies": movies
}
except requests.exceptions.RequestException as e:
return {
"status": "error",
"error": f"Failed to fetch trending movies: {str(e)}",
"message": "Check internet connection and try again"
}
@mcp.tool()
def get_popular_shows(limit: int = 10) -> Dict:
"""Get popular TV shows from Trakt.tv.
Args:
limit (int, optional): Maximum number of popular shows to return.
Must be between 1-50. Defaults to 10.
Values outside range are clamped to 1-50.
Returns:
Dict: Popular shows information containing:
- status (str): "success" or "error"
- count (int): Number of shows returned
- shows (List[Dict]): List of popular TV shows, each containing:
- title (str): Show title
- year (int): First air year
- rating (float): Trakt user rating (0-10 scale)
- votes (int): Number of user votes
- network (str): Original broadcast network
- error (str, optional): Error message if status is "error"
Examples:
get_popular_shows() returns top 10 popular shows
get_popular_shows(15) returns top 15 popular shows
get_popular_shows(0) returns 1 show (clamped to minimum)
Note:
Uses Trakt.tv public API. Enhanced data available with TRAKT_API_KEY.
"""
if limit < 1 or limit > 50:
limit = max(1, min(50, limit))
try:
url = "https://api.trakt.tv/shows/popular"
headers = {
"Content-Type": "application/json",
"trakt-api-version": "2"
}
# Add API key if available
api_key = os.getenv("TRAKT_API_KEY")
if api_key:
headers["trakt-api-key"] = api_key
params = {"limit": limit}
response = requests.get(url, headers=headers, params=params, timeout=10)
response.raise_for_status()
data = response.json()
shows = []
for show in data:
shows.append({
"title": show.get("title", "Unknown Title"),
"year": show.get("year", 0),
"rating": round(show.get("rating", 0), 1),
"votes": show.get("votes", 0),
"network": show.get("network", "Unknown Network")
})
return {
"status": "success",
"count": len(shows),
"shows": shows
}
except requests.exceptions.RequestException as e:
return {
"status": "error",
"error": f"Failed to fetch popular shows: {str(e)}",
"message": "Check internet connection and try again"
}
@mcp.tool()
def search_movies(query: str, limit: int = 5) -> Dict:
"""Search for movies by title on Trakt.tv.
Args:
query (str): Movie title or partial title to search for. Cannot be empty.
Examples: "Inception", "Star Wars", "Marvel", "Batman"
limit (int, optional): Maximum number of search results to return.
Must be between 1-20. Defaults to 5.
Values outside range are clamped to 1-20.
Returns:
Dict: Search results containing:
- status (str): "success" or "error"
- query (str): The search term that was used
- count (int): Number of movies found
- movies (List[Dict]): List of matching movies, each containing:
- title (str): Movie title
- year (int): Release year
- rating (float): Trakt user rating (0-10 scale)
- overview (str): Plot description (truncated to 200 characters)
- error (str, optional): Error message if status is "error"
Examples:
search_movies("Inception") finds movies matching "Inception"
search_movies("Star Wars", 3) finds 3 Star Wars related movies
search_movies("Marvel") finds Marvel movies with default limit of 5
search_movies("") returns error for empty query
Note:
Search is case-insensitive and matches partial titles.
Enhanced search results available with TRAKT_API_KEY.
"""
if not query or not query.strip():
return {
"status": "error",
"error": "Search query cannot be empty",
"message": "Please provide a movie title or partial title to search for"
}
if limit < 1 or limit > 20:
limit = max(1, min(20, limit))
try:
url = "https://api.trakt.tv/search/movie"
headers = {
"Content-Type": "application/json",
"trakt-api-version": "2"
}
# Add API key if available
api_key = os.getenv("TRAKT_API_KEY")
if api_key:
headers["trakt-api-key"] = api_key
params = {
"query": query.strip(),
"limit": limit
}
response = requests.get(url, headers=headers, params=params, timeout=10)
response.raise_for_status()
data = response.json()
movies = []
for item in data:
movie = item.get("movie", {})
overview = movie.get("overview", "No description available")
# Truncate overview to 200 characters
if len(overview) > 200:
overview = overview[:197] + "..."
movies.append({
"title": movie.get("title", "Unknown Title"),
"year": movie.get("year", 0),
"rating": round(movie.get("rating", 0), 1),
"overview": overview
})
return {
"status": "success",
"query": query.strip(),
"count": len(movies),
"movies": movies
}
except requests.exceptions.RequestException as e:
return {
"status": "error",
"query": query.strip(),
"error": f"Failed to search movies: {str(e)}",
"message": "Check internet connection and try again"
}
if __name__ == "__main__":
mcp.run()