#!/usr/bin/env python3
"""
Moviesda MCP Server
A Model Context Protocol server for searching Tamil movies from moviesda16.com
"""
import asyncio
import sqlite3
from datetime import datetime
from typing import Optional
import mcp.types as types
from mcp.server import Server
from mcp.server.stdio import stdio_server
import httpx
from bs4 import BeautifulSoup
# Database setup
DB_FILE = "moviesda.db"
def init_database():
"""Initialize SQLite database with movies table"""
conn = sqlite3.connect(DB_FILE)
cursor = conn.cursor()
cursor.execute("""
CREATE TABLE IF NOT EXISTS movies (
id INTEGER PRIMARY KEY AUTOINCREMENT,
title TEXT NOT NULL,
year INTEGER,
url TEXT NOT NULL UNIQUE,
category TEXT,
added_date TEXT DEFAULT CURRENT_TIMESTAMP
)
""")
# Create index for faster searching
cursor.execute("""
CREATE INDEX IF NOT EXISTS idx_title ON movies(title)
""")
conn.commit()
conn.close()
def search_movies(query: str, limit: int = 10) -> list:
"""Search movies in database by title"""
conn = sqlite3.connect(DB_FILE)
cursor = conn.cursor()
# Search using LIKE for partial matches
cursor.execute("""
SELECT title, year, url, category
FROM movies
WHERE title LIKE ?
ORDER BY added_date DESC
LIMIT ?
""", (f"%{query}%", limit))
results = cursor.fetchall()
conn.close()
return [
{
"title": row[0],
"year": row[1],
"url": f"https://moviesda16.com{row[2]}",
"category": row[3]
}
for row in results
]
def add_movie(title: str, year: Optional[int], url: str, category: str = "Tamil") -> bool:
"""Add a movie to the database"""
conn = sqlite3.connect(DB_FILE)
cursor = conn.cursor()
try:
cursor.execute("""
INSERT OR IGNORE INTO movies (title, year, url, category)
VALUES (?, ?, ?, ?)
""", (title, year, url, category))
conn.commit()
success = cursor.rowcount > 0
conn.close()
return success
except Exception as e:
conn.close()
raise e
async def scrape_movies_page(url: str = "https://moviesda16.com/tamil-2026-movies/") -> list:
"""Scrape movies from moviesda page"""
async with httpx.AsyncClient(timeout=30.0) as client:
try:
response = await client.get(url)
response.raise_for_status()
soup = BeautifulSoup(response.text, 'html.parser')
movies = []
# Find all movie divs
movie_divs = soup.find_all('div', class_='f')
for div in movie_divs:
link = div.find('a')
if link:
href = link.get('href')
title_text = link.text.strip()
# Extract year from title (e.g., "Movie Name (2026)")
year = None
if '(' in title_text and ')' in title_text:
year_str = title_text[title_text.rfind('(')+1:title_text.rfind(')')]
try:
year = int(year_str)
title = title_text[:title_text.rfind('(')].strip()
except ValueError:
title = title_text
else:
title = title_text
movies.append({
'title': title,
'year': year,
'url': href,
'category': 'Tamil'
})
return movies
except Exception as e:
return []
async def update_database_from_web():
"""Scrape and update database with latest movies"""
movies = await scrape_movies_page()
added_count = 0
for movie in movies:
try:
if add_movie(movie['title'], movie['year'], movie['url'], movie['category']):
added_count += 1
except Exception:
continue
return added_count
# Create MCP server instance
app = Server("moviesda-server")
@app.list_tools()
async def list_tools() -> list[types.Tool]:
"""List available tools"""
return [
types.Tool(
name="search_movie",
description="Search for Tamil movies by title. Returns movie name, year, and download link.",
inputSchema={
"type": "object",
"properties": {
"query": {
"type": "string",
"description": "Movie title to search for"
},
"limit": {
"type": "integer",
"description": "Maximum number of results to return (default: 10)",
"default": 10
}
},
"required": ["query"]
}
),
types.Tool(
name="add_movie",
description="Manually add a movie to the database",
inputSchema={
"type": "object",
"properties": {
"title": {
"type": "string",
"description": "Movie title"
},
"year": {
"type": "integer",
"description": "Release year"
},
"url": {
"type": "string",
"description": "Movie URL path (e.g., /movie-name-2026/)"
},
"category": {
"type": "string",
"description": "Movie category (default: Tamil)",
"default": "Tamil"
}
},
"required": ["title", "url"]
}
),
types.Tool(
name="update_movies",
description="Scrape and update the database with latest movies from moviesda16.com",
inputSchema={
"type": "object",
"properties": {}
}
)
]
@app.call_tool()
async def call_tool(name: str, arguments: dict) -> list[types.TextContent]:
"""Handle tool calls"""
if name == "search_movie":
query = arguments.get("query", "")
limit = arguments.get("limit", 10)
results = search_movies(query, limit)
if not results:
return [types.TextContent(
type="text",
text=f"No movies found matching '{query}'"
)]
# Format results
output = f"Found {len(results)} movie(s) matching '{query}':\n\n"
for i, movie in enumerate(results, 1):
output += f"{i}. {movie['title']}"
if movie['year']:
output += f" ({movie['year']})"
output += f"\n 🔗 Link: {movie['url']}\n\n"
return [types.TextContent(type="text", text=output)]
elif name == "add_movie":
title = arguments.get("title")
year = arguments.get("year")
url = arguments.get("url")
category = arguments.get("category", "Tamil")
success = add_movie(title, year, url, category)
if success:
message = f"✅ Successfully added '{title}' to the database"
else:
message = f"⚠️ Movie '{title}' already exists in the database"
return [types.TextContent(type="text", text=message)]
elif name == "update_movies":
added_count = await update_database_from_web()
message = f"✅ Database updated! Added {added_count} new movies from moviesda16.com"
return [types.TextContent(type="text", text=message)]
else:
raise ValueError(f"Unknown tool: {name}")
async def main():
"""Run the MCP server"""
# Initialize database
init_database()
# Run 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())
