#!/usr/bin/env python3
"""
YouTube MCP Server for VS Code.
Communicates via stdio with JSON-RPC protocol.
Provides tools for searching and playing songs from YouTube.
"""
import json
import logging
import os
import sys
import warnings
from typing import Any
# Suprimir warning de pkg_resources
warnings.filterwarnings("ignore", category=DeprecationWarning, module="pkg_resources")
from dotenv import load_dotenv
# Add the package to the path
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
from youtube_mcp.server import YouTubeMCPServer
from youtube_mcp.logger import setup_logger
# Set up logger
logger = setup_logger(__name__, level=logging.DEBUG)
class MCPServer:
"""Model Context Protocol server for YouTube."""
def __init__(self, api_key: str) -> None:
"""Initialize MCP server."""
self.youtube_server = YouTubeMCPServer(api_key)
def process_request(self, request: dict[str, Any]) -> dict[str, Any]:
"""Process a JSON-RPC request."""
method = request.get("method", "")
params = request.get("params", {})
request_id = request.get("id")
if method == "initialize":
return self.handle_initialize(request)
elif method == "tools/list":
return self.handle_list_tools(request)
elif method == "tools/call":
return self.handle_tool_call(request, params)
else:
return self._error_response(request_id, f"Unknown method: {method}")
def handle_initialize(self, request: dict[str, Any]) -> dict[str, Any]:
"""Handle MCP initialize message."""
return {
"jsonrpc": "2.0",
"id": request.get("id"),
"result": {
"protocolVersion": "2024-11-05",
"capabilities": {"tools": {}},
"serverInfo": {
"name": "youtube-mcp",
"version": "0.1.0"
}
}
}
def handle_list_tools(self, request: dict[str, Any]) -> dict[str, Any]:
"""List available tools."""
tools = self.youtube_server.get_tools()
return {
"jsonrpc": "2.0",
"id": request.get("id"),
"result": {
"tools": tools
}
}
def handle_tool_call(
self, request: dict[str, Any], params: dict[str, Any]
) -> dict[str, Any]:
"""Handle tools/call request."""
try:
tool_name = params.get("name", "")
tool_params = params.get("arguments", {})
# Log the request
logger.debug("Tool call: %s with params: %s", tool_name, tool_params)
if tool_name == "search_youtube":
result = self.youtube_server.handle_search_youtube(**tool_params)
# Format response like postgres does
if result.get("success"):
text = json.dumps(result, indent=2, ensure_ascii=False)
else:
text = f"Error: {result.get('error', 'Unknown error')}"
return {
"jsonrpc": "2.0",
"id": request.get("id"),
"result": {
"content": [{"type": "text", "text": text}]
}
}
elif tool_name == "play_youtube":
result = self.youtube_server.handle_play_youtube(**tool_params)
# Format response like postgres does
if result.get("success"):
text = json.dumps(result, indent=2, ensure_ascii=False)
else:
text = f"Error: {result.get('error', 'Unknown error')}"
return {
"jsonrpc": "2.0",
"id": request.get("id"),
"result": {
"content": [{"type": "text", "text": text}]
}
}
else:
return self._error_response(
request.get("id"), f"Unknown tool: {tool_name}"
)
except Exception as e: # pylint: disable=broad-except
logger.error("Exception in handle_tool_call: %s", str(e))
import traceback
logger.error(traceback.format_exc())
return {
"jsonrpc": "2.0",
"id": request.get("id"),
"result": {
"content": [{"type": "text", "text": f"Error: {str(e)}"}]
}
}
def _error_response(self, request_id: Any, error: str) -> dict[str, Any]:
"""Create an error response."""
return {
"jsonrpc": "2.0",
"id": request_id,
"error": {"code": -32603, "message": error},
}
def main() -> None:
"""Main server loop."""
load_dotenv()
api_key = os.getenv("YOUTUBE_API_KEY")
if not api_key:
raise ValueError(
"YOUTUBE_API_KEY environment variable is required. "
"Please set it in your .env file or as an environment variable."
)
server = MCPServer(api_key)
try:
while True:
line = sys.stdin.readline()
if not line:
break
line = line.strip()
if not line:
continue
try:
request = json.loads(line)
response = server.process_request(request)
sys.stdout.write(json.dumps(response) + "\n")
sys.stdout.flush()
except json.JSONDecodeError:
pass
except (KeyError, ValueError, TypeError) as e:
logger.error("Error processing request: %s", str(e))
error_response = {
"jsonrpc": "2.0",
"id": None,
"error": {"code": -32700, "message": str(e)},
}
sys.stdout.write(json.dumps(error_response) + "\n")
sys.stdout.flush()
except KeyboardInterrupt:
logger.info("Server interrupted by user")
sys.exit(0)
except Exception as e: # pylint: disable=broad-except
logger.error("Unexpected error in server loop: %s", str(e))
error_response = {
"jsonrpc": "2.0",
"id": None,
"error": {"code": -32603, "message": str(e)},
}
sys.stdout.write(json.dumps(error_response) + "\n")
sys.stdout.flush()
if __name__ == "__main__":
main()
