"""
IMDB MCP Server - Cloud-runnable JSON-RPC Service
This service provides IMDB data access via JSON-RPC protocol.
LLMs can query this service to get movie and person information.
"""
import os
import json
from typing import Any
from fastapi import FastAPI, Request, HTTPException
from fastapi.responses import JSONResponse, HTMLResponse
from pydantic import BaseModel
from imdb_backend import IMDBBackend
app = FastAPI(
title="IMDB MCP Server",
description="JSON-RPC service for querying IMDB data",
version="1.0.0"
)
backend = IMDBBackend()
# JSON-RPC Request/Response Models
class JSONRPCRequest(BaseModel):
jsonrpc: str = "2.0"
method: str
params: dict = {}
id: int | str | None = None
class JSONRPCResponse(BaseModel):
jsonrpc: str = "2.0"
result: Any = None
error: dict | None = None
id: int | str | None = None
# Available methods and their descriptions
METHODS = {
"search_movies": {
"description": "Search for movies by title",
"params": {
"query": {"type": "string", "required": True, "description": "Movie title to search for"},
"limit": {"type": "integer", "required": False, "default": 10, "description": "Maximum results to return"}
},
"returns": "List of movies with id, title, and year"
},
"get_movie_details": {
"description": "Get full details for a movie by its IMDB ID",
"params": {
"movie_id": {"type": "string", "required": True, "description": "IMDB movie ID (e.g., '0133093' for The Matrix)"}
},
"returns": "Movie details including title, year, rating, plot, genres, directors, writers, runtime, countries, languages"
},
"search_people": {
"description": "Search for actors, directors, or other film industry people by name",
"params": {
"query": {"type": "string", "required": True, "description": "Person name to search for"},
"limit": {"type": "integer", "required": False, "default": 10, "description": "Maximum results to return"}
},
"returns": "List of people with id and name"
},
"get_person_details": {
"description": "Get full details for a person including their filmography",
"params": {
"person_id": {"type": "string", "required": True, "description": "IMDB person ID (e.g., '0000206' for Keanu Reeves)"}
},
"returns": "Person details including name, birth_date, birth_place, bio, and filmography by role"
},
"get_top_250_movies": {
"description": "Get the IMDB Top 250 movies list",
"params": {
"limit": {"type": "integer", "required": False, "default": 25, "description": "Maximum movies to return"}
},
"returns": "List of top-rated movies with rank, id, title, year, and rating"
},
"get_movie_cast": {
"description": "Get the cast of a movie",
"params": {
"movie_id": {"type": "string", "required": True, "description": "IMDB movie ID"},
"limit": {"type": "integer", "required": False, "default": 20, "description": "Maximum cast members to return"}
},
"returns": "List of cast members with id, name, and role"
}
}
def execute_method(method: str, params: dict) -> Any:
"""Execute a JSON-RPC method and return the result."""
if method == "search_movies":
return backend.search_movies(
query=params.get("query", ""),
limit=params.get("limit", 10)
)
elif method == "get_movie_details":
return backend.get_movie_details(movie_id=params.get("movie_id", ""))
elif method == "search_people":
return backend.search_people(
query=params.get("query", ""),
limit=params.get("limit", 10)
)
elif method == "get_person_details":
return backend.get_person_details(person_id=params.get("person_id", ""))
elif method == "get_top_250_movies":
return backend.get_top_250_movies(limit=params.get("limit", 25))
elif method == "get_movie_cast":
return backend.get_movie_cast(
movie_id=params.get("movie_id", ""),
limit=params.get("limit", 20)
)
else:
raise ValueError(f"Unknown method: {method}")
@app.get("/", response_class=HTMLResponse)
async def root():
"""
Service discovery endpoint.
Returns HTML documentation explaining the service and available methods.
"""
html_content = """
<!DOCTYPE html>
<html>
<head>
<title>IMDB MCP Server</title>
<style>
body { font-family: -apple-system, BlinkMacSystemFont, 'Segoe UI', Roboto, sans-serif;
max-width: 900px; margin: 50px auto; padding: 20px; line-height: 1.6; }
h1 { color: #f5c518; }
h2 { color: #333; border-bottom: 2px solid #f5c518; padding-bottom: 10px; }
.method { background: #f8f9fa; padding: 20px; margin: 20px 0; border-radius: 8px; border-left: 4px solid #f5c518; }
.method h3 { margin-top: 0; color: #0066cc; }
code { background: #e9ecef; padding: 2px 6px; border-radius: 4px; font-size: 14px; }
pre { background: #2d2d2d; color: #f8f8f2; padding: 15px; border-radius: 8px; overflow-x: auto; }
.param { margin: 5px 0; }
.required { color: #dc3545; font-weight: bold; }
.optional { color: #6c757d; }
</style>
</head>
<body>
<h1>IMDB MCP Server</h1>
<p>This is a JSON-RPC service that provides access to IMDB movie and person data.
LLMs can query this service to retrieve information about movies, actors, directors, and more.</p>
<h2>How to Use</h2>
<p>Send a POST request to <code>/jsonrpc</code> with a JSON-RPC 2.0 formatted body:</p>
<pre>{
"jsonrpc": "2.0",
"method": "search_movies",
"params": {"query": "The Matrix", "limit": 5},
"id": 1
}</pre>
<h2>MCP Server Configuration</h2>
<p>Add this server to your LLM's MCP configuration:</p>
<pre>{
"mcpServers": {
"imdb": {
"url": "YOUR_DEPLOYED_URL/mcp",
"transport": "http"
}
}
}</pre>
<h2>Available Methods</h2>
"""
for method_name, method_info in METHODS.items():
html_content += f"""
<div class="method">
<h3>{method_name}</h3>
<p>{method_info['description']}</p>
<h4>Parameters:</h4>
"""
for param_name, param_info in method_info['params'].items():
req_class = "required" if param_info.get('required') else "optional"
req_text = "(required)" if param_info.get('required') else f"(optional, default: {param_info.get('default')})"
html_content += f"""
<div class="param">
<code>{param_name}</code>: {param_info['type']} - {param_info['description']}
<span class="{req_class}">{req_text}</span>
</div>
"""
html_content += f"""
<h4>Returns:</h4>
<p>{method_info['returns']}</p>
</div>
"""
html_content += """
<h2>Response Format</h2>
<p>Successful responses return:</p>
<pre>{
"jsonrpc": "2.0",
"result": { ... },
"id": 1
}</pre>
<p>Error responses return:</p>
<pre>{
"jsonrpc": "2.0",
"error": {"code": -32600, "message": "Invalid Request"},
"id": 1
}</pre>
</body>
</html>
"""
return HTMLResponse(content=html_content)
@app.get("/info")
async def info():
"""
Machine-readable service information endpoint.
Returns JSON describing the service and available methods.
"""
return {
"service": "IMDB MCP Server",
"version": "1.0.0",
"description": "JSON-RPC service for querying IMDB movie and person data. "
"Use the /jsonrpc endpoint to send JSON-RPC 2.0 requests.",
"endpoints": {
"/": "HTML documentation (human-readable)",
"/info": "Service information (JSON)",
"/jsonrpc": "JSON-RPC 2.0 endpoint",
"/mcp": "MCP protocol endpoint",
"/health": "Health check endpoint"
},
"methods": METHODS,
"usage_example": {
"endpoint": "POST /jsonrpc",
"body": {
"jsonrpc": "2.0",
"method": "search_movies",
"params": {"query": "Inception", "limit": 5},
"id": 1
}
}
}
@app.get("/health")
async def health():
"""Health check endpoint for cloud deployments."""
return {"status": "healthy", "service": "imdb-mcp"}
@app.post("/jsonrpc")
async def jsonrpc_endpoint(request: JSONRPCRequest):
"""
JSON-RPC 2.0 endpoint for executing IMDB queries.
Send requests with:
- jsonrpc: "2.0"
- method: One of the available methods
- params: Method parameters as an object
- id: Request identifier
"""
try:
if request.method not in METHODS:
return JSONResponse(content={
"jsonrpc": "2.0",
"error": {
"code": -32601,
"message": f"Method not found: {request.method}",
"data": {"available_methods": list(METHODS.keys())}
},
"id": request.id
})
result = execute_method(request.method, request.params)
return {
"jsonrpc": "2.0",
"result": result,
"id": request.id
}
except Exception as e:
return JSONResponse(content={
"jsonrpc": "2.0",
"error": {
"code": -32603,
"message": f"Internal error: {str(e)}"
},
"id": request.id
})
@app.post("/mcp")
async def mcp_endpoint(request: Request):
"""
MCP protocol endpoint for LLM tool calling.
Handles MCP JSON-RPC requests for tool discovery and execution.
"""
try:
body = await request.json()
method = body.get("method", "")
params = body.get("params", {})
request_id = body.get("id")
# Handle MCP protocol methods
if method == "initialize":
return {
"jsonrpc": "2.0",
"result": {
"protocolVersion": "2024-11-05",
"capabilities": {
"tools": {"listChanged": False}
},
"serverInfo": {
"name": "imdb-mcp",
"version": "1.0.0"
}
},
"id": request_id
}
elif method == "tools/list":
tools = []
for tool_name, tool_info in METHODS.items():
properties = {}
required = []
for param_name, param_info in tool_info["params"].items():
properties[param_name] = {
"type": param_info["type"],
"description": param_info["description"]
}
if param_info.get("required"):
required.append(param_name)
tools.append({
"name": tool_name,
"description": tool_info["description"],
"inputSchema": {
"type": "object",
"properties": properties,
"required": required
}
})
return {
"jsonrpc": "2.0",
"result": {"tools": tools},
"id": request_id
}
elif method == "tools/call":
tool_name = params.get("name", "")
tool_args = params.get("arguments", {})
if tool_name not in METHODS:
return JSONResponse(content={
"jsonrpc": "2.0",
"error": {
"code": -32601,
"message": f"Tool not found: {tool_name}"
},
"id": request_id
})
result = execute_method(tool_name, tool_args)
return {
"jsonrpc": "2.0",
"result": {
"content": [
{
"type": "text",
"text": json.dumps(result, indent=2)
}
]
},
"id": request_id
}
else:
return JSONResponse(content={
"jsonrpc": "2.0",
"error": {
"code": -32601,
"message": f"Method not found: {method}"
},
"id": request_id
})
except Exception as e:
return JSONResponse(content={
"jsonrpc": "2.0",
"error": {
"code": -32603,
"message": f"Internal error: {str(e)}"
},
"id": body.get("id") if 'body' in dir() else None
})
if __name__ == "__main__":
import uvicorn
port = int(os.environ.get("PORT", 8080))
uvicorn.run(app, host="0.0.0.0", port=port)
