"""
FastMCP server instance for ThePornDB MCP Service.
Implements the MCP protocol using FastMCP framework with token validation
on startup via lifespan management.
"""
import asyncio
import logging
from collections.abc import AsyncIterator
from contextlib import asynccontextmanager
from dataclasses import dataclass
from mcp.server.fastmcp import FastMCP
from mcp.server.session import ServerSession
from mcp.server.fastmcp import Context
from mcp.types import CallToolResult, TextContent
from .config import Config, parse_args
from .api import ThePornDBAPI, APIError
logger = logging.getLogger(__name__)
@dataclass
class AppContext:
"""
Application context with validated dependencies.
Attributes:
config: Application configuration
api: ThePornDB API client
"""
config: Config
api: ThePornDBAPI
@asynccontextmanager
async def lifespan(server: FastMCP) -> AsyncIterator[AppContext]:
"""
Manage application lifecycle with token validation.
Validates the API token on startup and fails fast if invalid.
Args:
server: FastMCP server instance
Yields:
AppContext with validated configuration and API client
Raises:
SystemExit: If token validation fails (prevents server from starting)
"""
# Parse CLI arguments
args = parse_args()
config = Config(args)
logger.info("ThePornDB MCP Service starting...")
# Create API client
api = ThePornDBAPI(config)
# Validate token on startup (fail fast if invalid)
if config.args.token: # Only validate if user provided a token
logger.info("Validating API token...")
if not api.validate_token():
logger.error("Token validation failed - server cannot start")
raise SystemExit(
"Failed to start: Invalid API token. "
"Please check your token and try again. "
"Get your token at https://theporndb.net"
)
else:
logger.info("No token provided - using default public token")
try:
# Provide context to tools
yield AppContext(config=config, api=api)
finally:
# Cleanup on shutdown
logger.info("ThePornDB MCP Service shutting down...")
def create_server() -> FastMCP:
"""
Create and configure the FastMCP server instance.
Returns:
Configured FastMCP server
"""
server = FastMCP(
name="ThePornDB",
instructions=(
"ThePornDB MCP Service - Search and retrieve detailed information "
"about adult video content including scenes, movies, JAV, and performers. "
"Provides tools for searching content, retrieving details, and looking up performer information."
),
lifespan=lifespan
)
return server
def register_tools(server: FastMCP) -> None:
"""
Register all MCP tools with the server.
Args:
server: FastMCP server instance
"""
from .tools import search, details, performers
# Import and register search tools
server.tool()(search.search_scenes)
server.tool()(search.search_movies)
server.tool()(search.search_jav)
# Import and register detail tools
server.tool()(details.get_content_details)
# Import and register performer tools
server.tool()(performers.search_performers)
server.tool()(performers.get_performer_details)
logger.info("All tools registered successfully")
def main():
"""
Main entry point for the MCP server.
Parses CLI arguments, creates server, registers tools, and runs.
"""
# Parse CLI arguments early for logging setup
args = parse_args()
# Create server with lifespan
server = create_server()
# Register all tools
register_tools(server)
# Run server with appropriate transport
logger.info(f"Starting MCP server with {args.transport} transport...")
if args.transport == "streamable-http":
# Production mode: HTTP transport with stateless mode
server.run(
transport="streamable-http",
stateless_http=True,
json_response=True
)
else:
# Development mode: stdio transport
server.run()
if __name__ == "__main__":
main()
