Ultra MCP-SS

from fastapi import FastAPI, Depends, HTTPException, Response, Query   
from pydantic import BaseModel, Field
import os
import urllib.parse
import httpx
from fastapi_mcp import FastApiMCP
from dotenv import load_dotenv

app = FastAPI(title="ultra-mcp-ss")

load_dotenv()

YOUTUBE_API_KEY = os.environ.get("YOUTUBE_API_KEY")
YOUTUBE_API_URL = "https://www.googleapis.com/youtube/v3/search"


SMARTSCREEN_API_URL = "https://smartscreen.tv/api"
SMARTSCREEN_SERVICE_TOKEN = os.environ.get("SS_SERVICE_TOKEN")


async def search_youtube_video(query: str) -> str | None:
    """
    Searches YouTube for the most relevant video and returns its ID.
    Returns None if no video is found or an error occurs.
    """
    if not YOUTUBE_API_KEY:
        print("Error: YOUTUBE_API_KEY environment variable not set.")
        return None

    encoded_query = urllib.parse.quote_plus(query)
    params = {
        "part": "snippet",
        "q": encoded_query,
        "key": YOUTUBE_API_KEY,
        "type": "video",
        "maxResults": 1,
        "order": "relevance"
    }

    try:
        async with httpx.AsyncClient() as client:
            response = await client.get(YOUTUBE_API_URL, params=params)
            response.raise_for_status() # Raise an exception for 4xx or 5xx status codes
            
            data = response.json()

            if data.get("items"):
                # Safely get the videoId
                video_id = data["items"][0].get("id", {}).get("videoId")
                return video_id
            else:
                # No results found for the query
                return None

    except httpx.RequestError as exc:
        print(f"An error occurred while requesting {exc.request.url!r}: {exc}")
        return None
    except httpx.HTTPStatusError as exc:
        print(f"Error response {exc.response.status_code} while requesting {exc.request.url!r}.")
        print(f"Response body: {exc.response.text}")
        return None
    except Exception as e:
        print(f"An unexpected error occurred: {e}")
        return None


# Pydantic models for input
class DropRequest(BaseModel):
    display: str
    item: str  # the item to drop
    x: int = 0  # coordinate (if applicable)
    y: int = 0  # coordinate (if applicable)

class NotifyRequest(BaseModel):
    display: str
    message: str
    priority: str# = Field(default="normal")

class ToastRequest(BaseModel):
    display: str
    message: str
    heading: str
    icon: str
    transition: str
    duration: str = "5"

class MarqueeRequest(BaseModel):
    display: str
    message: str
    duration: str = "30"
    color: str
    size: str = "3"
    bgcolor: str

class TextRequest(BaseModel):
    display: str
    message: str
    duration: str = "30"
    color: str
    size: str = "3"
    bgcolor: str
    align: str
    frame: str
    animate: str
    aniduration: str = "2"

class AppRequest(BaseModel):
    display: str
    url: str
    duration: str = "0"
    frame: str

class TouchRequest(BaseModel):
    display: str
    option: str
    value: str# = ""

class StatusRequest(BaseModel):
    display: str
    option: str
    value: str# = ""

class DjRequest(BaseModel):
    display: str
    option: str
    value: str


class SmartScreenClient:
    """
    Client for calling the real SmartScreen API.
    
    It sends a POST request to the SmartScreen API endpoint with a payload containing:
      - 'to': { "ddn": "yview" } (using yview as the destination target)
      - 'data': (a command JSON as specified)
    
    It uses httpx in async mode.
    """
    def __init__(self):
        self.api_url = SMARTSCREEN_API_URL
        self.token = SMARTSCREEN_SERVICE_TOKEN

    async def drop(self, item: str, display: str, x: int, y: int) -> bool:
        # Construct the command payload for "drop"
        command = {
            "cmd": "drop",
            "type": "url",
            "src": [item],  # Use the passed URL instead of hardcoded one
            "duration": "0",    # 0 means play to end
            "frame": "main",
            "animate": "fade",
            "aniduration": "2",
            "pmode": "loop",
            "bgcolor": "white"
        }
        payload = {
            "to": {"name": display},
            "data": command
        }
        headers = {
            "Content-Type": "application/json",
            "Authorization": f"Bearer {self.token}"  # Use class variable instead of hardcoded token
        }
        async with httpx.AsyncClient() as client:
            response = await client.post(self.api_url, headers=headers, json=payload)
            # Check for HTTP success (200 OK)
            if response.status_code != 200:
                print(f"SmartScreen drop failed: {response.text}")
                return False
            result = response.json()
            if result.get("ErrCode", -1) != 0:
                print(f"SmartScreen drop error: {result.get('ErrMsg', 'Unknown error')}")
                return False
        return True

    async def notify(self, message: str, display: str, priority: str) -> bool:
        # Construct the command payload for "notify"
        command = {
            "cmd": "notify",
            "msg": message,
            "duration": "30",  # set default duration; could be a parameter in the future
            "color": "blue",
            "size": "3"
        }
        payload = {
            "to": {"name": display},
            "data": command
        }
        headers = {
            "Authorization": f"Bearer {self.token}"
        }
        async with httpx.AsyncClient() as client:
            response = await client.post(self.api_url, headers=headers, json=payload)
            if response.status_code != 200:
                print(f"SmartScreen notify failed: {response.text}")
                return False
            result = response.json()
            if result.get("ErrCode", -1) != 0:
                print(f"SmartScreen notify error: {result.get('ErrMsg', 'Unknown error')}")
                return False
        return True

    async def toast(self, message: str, display: str, heading: str, icon: str, transition: str, duration: str) -> bool:
        # Construct the command payload for "toast"
        command = {
            "cmd": "toast",
            "msg": message,
            "heading": heading,
            "icon": icon,
            "transition": transition,
            "duration": duration
        }
        payload = {
            "to": {"name": display},
            "data": command
        }
        headers = {
            "Authorization": f"Bearer {self.token}"
        }
        async with httpx.AsyncClient() as client:
            response = await client.post(self.api_url, headers=headers, json=payload)
            if response.status_code != 200:
                print(f"SmartScreen toast failed: {response.text}")
                return False
            result = response.json()
            if result.get("ErrCode", -1) != 0:
                print(f"SmartScreen toast error: {result.get('ErrMsg', 'Unknown error')}")
                return False
        return True

    async def marquee(self, message: str, display: str, duration: str, color: str, size: str, bgcolor: str) -> bool:
        # Construct the command payload for "marquee"
        command = {
            "cmd": "marquee",
            "msg": message,
            "duration": duration,
            "color": color,
            "size": size,
            "bgcolor": bgcolor
        }
        payload = {
            "to": {"name": display},
            "data": command
        }
        headers = {
            "Authorization": f"Bearer {self.token}"
        }
        async with httpx.AsyncClient() as client:
            response = await client.post(self.api_url, headers=headers, json=payload)
            if response.status_code != 200:
                print(f"SmartScreen marquee failed: {response.text}")
                return False
            result = response.json()
            if result.get("ErrCode", -1) != 0:
                print(f"SmartScreen marquee error: {result.get('ErrMsg', 'Unknown error')}")
                return False
        return True

    async def text(self, message: str, display: str, duration: str, color: str, size: str, 
                   bgcolor: str, align: str, frame: str, animate: str, aniduration: str) -> bool:
        # Construct the command payload for "text"
        command = {
            "cmd": "text",
            "msg": message,
            "duration": duration,
            "color": color,
            "size": size,
            "bgcolor": bgcolor,
            "align": align,
            "frame": frame,
            "animate": animate,
            "aniduration": aniduration
        }
        payload = {
            "to": {"name": display},
            "data": command
        }
        headers = {
            "Authorization": f"Bearer {self.token}"
        }
        async with httpx.AsyncClient() as client:
            response = await client.post(self.api_url, headers=headers, json=payload)
            if response.status_code != 200:
                print(f"SmartScreen text failed: {response.text}")
                return False
            result = response.json()
            if result.get("ErrCode", -1) != 0:
                print(f"SmartScreen text error: {result.get('ErrMsg', 'Unknown error')}")
                return False
        return True

    async def app(self, url: str, display: str, duration: str, frame: str) -> bool:
        # Construct the command payload for "app"
        command = {
            "cmd": "app",
            "url": url,
            "duration": duration,
            "frame": frame
        }
        payload = {
            "to": {"name": display},
            "data": command
        }
        headers = {
            "Authorization": f"Bearer {self.token}"
        }
        async with httpx.AsyncClient() as client:
            response = await client.post(self.api_url, headers=headers, json=payload)
            if response.status_code != 200:
                print(f"SmartScreen app failed: {response.text}")
                return False
            result = response.json()
            if result.get("ErrCode", -1) != 0:
                print(f"SmartScreen app error: {result.get('ErrMsg', 'Unknown error')}")
                return False
        return True

    async def touch(self, option: str, display: str, value: str) -> bool:
        # Construct the command payload for "touch"
        command = {
            "cmd": "touch",
            "option": option,
            "value": value
        }
        payload = {
            "to": {"name": display},
            "data": command
        }
        headers = {
            "Authorization": f"Bearer {self.token}"
        }
        async with httpx.AsyncClient() as client:
            response = await client.post(self.api_url, headers=headers, json=payload)
            if response.status_code != 200:
                print(f"SmartScreen touch failed: {response.text}")
                return False
            result = response.json()
            if result.get("ErrCode", -1) != 0:
                print(f"SmartScreen touch error: {result.get('ErrMsg', 'Unknown error')}")
                return False
        return True

    async def status(self, option: str, display: str, value: str) -> bool:
        # Construct the command payload for "status"
        command = {
            "cmd": "status",
            "option": option,
            "value": value
        }
        payload = {
            "to": {"name": display},
            "data": command
        }
        headers = {
            "Authorization": f"Bearer {self.token}"
        }
        async with httpx.AsyncClient() as client:
            response = await client.post(self.api_url, headers=headers, json=payload)
            if response.status_code != 200:
                print(f"SmartScreen status failed: {response.text}")
                return False
            result = response.json()
            if result.get("ErrCode", -1) != 0:
                print(f"SmartScreen status error: {result.get('ErrMsg', 'Unknown error')}")
                return False
        return True

    async def dj(self, option: str, display: str, value: str) -> bool:
        # Construct the command payload for "dj"
        command = {
            "cmd": "dj",
            "option": option,
            "value": value
        }
        payload = {
            "to": {"name": display},
            "data": command
        }
        headers = {
            "Authorization": f"Bearer {self.token}"
        }
        async with httpx.AsyncClient() as client:
            response = await client.post(self.api_url, headers=headers, json=payload)
            if response.status_code != 200:
                print(f"SmartScreen dj failed: {response.text}")
                return False
            result = response.json()
            if result.get("ErrCode", -1) != 0:
                print(f"SmartScreen dj error: {result.get('ErrMsg', 'Unknown error')}")
                return False
        return True

# Create an instance of our SmartScreen client.
smartscreen_client = SmartScreenClient()

@app.head("/mcp")
async def mcp_head():
    return Response(status_code=200)

@app.get("/search-youtube", summary="Search YouTube Video", response_description="URL of the most relevant YouTube video")
async def get_youtube_video_link(
    query: str = Query(..., description="The search query for YouTube.")
):
    """
    Accepts a search query and returns the YouTube link 
    for the most relevant video found.
    """
    if not YOUTUBE_API_KEY:
         raise HTTPException(status_code=500, detail="API key configuration error. Please check server logs.")

    video_id = await search_youtube_video(query)

    if video_id:
        # Construct the standard YouTube URL
        video_url = f"https://www.youtube.com/watch?v={video_id}"
        return {"video_url": video_url}
    else:
        # Handles cases where search failed or no results were found
        raise HTTPException(status_code=404, detail=f"No relevant YouTube video found for query: '{query}'")

@app.get("/test", tags=["test"])
async def test_endpoint():
    """
    Test endpoint to check if the API is running.
    """
    return {"message": "My name is MCP Vincent, and I am here to help you."}

# API endpoint for drop command
@app.post("/drop")
async def drop_item(request: DropRequest):
    """
    Drop an item onto the SmartScreen at the specified coordinates.
    MCP Tool Exposure: This endpoint is exposed via MCP as the 'drop' tool.
    What it does:
Displays an image or video on the screen.
Why/When to use:
Use it whenever you want to show media content—like a video, slideshow, or static image—on a particular screen area.
Key options:
• type: "url" (URLs for images/videos)
• src: one or more URLs of the media files
• duration: how long to display/play (0 = play to end)
• frame: on which screen area (e.g., main, t1, t2)
• animate: animation style (fade, slide, etc.)
• pmode: playback mode (“loop” or “random”)
    """
    success = await smartscreen_client.drop(item=request.item,x=request.x,y=request.y,display=request.display)
    if not success:
        raise HTTPException(status_code=500, detail="Failed to drop item")
    return {"message": "Item dropped successfully", "item": request.item, "x": request.x, "y": request.y}

# API endpoint for notify command
@app.post("/notify")
async def notify_user(request: NotifyRequest):
    """
    Send a notification message to the SmartScreen.
    MCP Tool Exposure: This endpoint is exposed via MCP as the 'notify' tool.
    What it does:
Shows a notification banner containing simple text.
Why/When to use:
If you want to display a short message or alert without interrupting the main screen content.
Key options:
• msg: notification text
• duration: how long to display
• color, size: appearance settings for text
    """
    success = await smartscreen_client.notify(message=request.message,priority=request.priority,display=request.display)
    if not success:
        raise HTTPException(status_code=500, detail="Failed to send notification")
    return {"message": "Notification sent successfully", "notification": request.message}

# New API endpoints for additional SmartScreen commands
@app.post("/toast")
async def toast_message(request: ToastRequest):
    """
    Display a toast message on the SmartScreen.
    MCP Tool Exposure: This endpoint is exposed via MCP as the 'toast' tool.
    What it does:
Shows a popup message with optional heading, icon, and transition effects.
Why/When to use:
Ideal for quick, transient notes (toasts) that appear and then disappear without disturbing the main content.
Key options:
• msg: main text content
• heading: optional title/heading
• icon: can be info, warning, error, or success
• transition: how it appears (plain, fade, slide)
• duration: how long the toast stays visible
    """
    success = await smartscreen_client.toast(message=request.message,heading=request.heading, 
                                          icon=request.icon,transition=request.transition,duration=request.duration,display=request.display)
    if not success:
        raise HTTPException(status_code=500, detail="Failed to show toast message")
    return {"message": "Toast displayed successfully", "toast": request.message}

@app.post("/marquee")
async def marquee_text(request: MarqueeRequest):
    """
    Display scrolling marquee text on the SmartScreen.
    MCP Tool Exposure: This endpoint is exposed via MCP as the 'marquee' tool.
    What it does:
Displays scrolling text across the screen.
Why/When to use:
Useful for announcements or continuously running messages (e.g., news ticker, short updates) in a marquee format.
Key options:
• msg: marquee text
• duration: how long to keep scrolling
• color, size: text styling
• bgcolor: background color
    """
    success = await smartscreen_client.marquee(message=request.message,duration=request.duration, 
                                           color=request.color,size=request.size,bgcolor=request.bgcolor,display=request.display)
    if not success:
        raise HTTPException(status_code=500, detail="Failed to show marquee text")
    return {"message": "Marquee displayed successfully", "marquee": request.message}

@app.post("/text")
async def display_text(request: TextRequest):
    """
    Display text on the SmartScreen.
    MCP Tool Exposure: This endpoint is exposed via MCP as the 'text' tool.
    What it does:
Shows a static block of text with customizable appearance and alignment.
Why/When to use:
Use to render text that remains on the screen for a set duration, possibly in a specific frame or layout region.
Key options:
• msg: text to display
• duration: display length (seconds)
• color, size, bgcolor: styling
• align: left, center, or right
• animate: how text appears (fade, slide, etc.)
    """
    success = await smartscreen_client.text(message=request.message,duration=request.duration, 
                                        color=request.color,size=request.size,bgcolor=request.bgcolor,
                                        align=request.align,frame=request.frame,animate=request.animate,
                                        aniduration=request.aniduration,display=request.display)
    if not success:
        raise HTTPException(status_code=500, detail="Failed to show text")
    return {"message": "Text displayed successfully", "text": request.message}

@app.post("/app")
async def launch_app(request: AppRequest):
    """
    Launch a web application on the SmartScreen.
    MCP Tool Exposure: This endpoint is exposed via MCP as the 'app' tool.
    What it does:
Loads and displays a webpage within a specified frame.
Why/When to use:
If you need to embed a browser view or show dynamic web content on the SmartScreen.
Key options:
• url: the webpage’s URL
• duration: how long to display (0 = indefinite)
• frame: target screen region
    """
    success = await smartscreen_client.app(url=request.url,duration=request.duration,frame=request.frame,display=request.display)
    if not success:
        raise HTTPException(status_code=500, detail="Failed to launch app")
    return {"message": "App launched successfully", "url": request.url}

@app.post("/touch")
async def touch_command(request: TouchRequest):
    """
    Send a touch command to control the SmartScreen.
    MCP Tool Exposure: This endpoint is exposed via MCP as the 'touch' tool.
    What it does:
Sends control commands like play/pause, next/previous, volume up/down, or console commands.
Why/When to use:
Ideal for interactive control of media playback or SmartScreen settings without reloading the main content.
Key options:
• option: specifies the control action (playnext, mute, console, etc.)
• value: additional parameter if needed (e.g., console sub-command)
    """
    success = await smartscreen_client.touch(option=request.option,value=request.value,display=request.display)
    if not success:
        raise HTTPException(status_code=500, detail="Failed to execute touch command")
    return {"message": "Touch command executed successfully", "option": request.option, "value": request.value}

@app.post("/status")
async def status_command(request: StatusRequest):
    """
    Get or set status information on the SmartScreen.
    MCP Tool Exposure: This endpoint is exposed via MCP as the 'status' tool.
    What it does:
Requests status information from the SmartScreen, such as version, frame usage, or device info.
Why/When to use:
If you need to check what’s currently playing, or retrieve system details (e.g., debugging, system health checks).
Key options:
• option: type of status (frame, ver, devinfo)
• value: additional context (like which frame)
    """
    success = await smartscreen_client.status(option=request.option,value=request.value,display=request.display)
    if not success:
        raise HTTPException(status_code=500, detail="Failed to execute status command")
    return {"message": "Status command executed successfully", "option": request.option, "value": request.value}

@app.post("/dj")
async def dj_command(request: DjRequest):
    """
    Execute a DJ command on the SmartScreen.
    MCP Tool Exposure: This endpoint is exposed via MCP as the 'dj' tool.
    What it does:
Manages higher-level features like scheduling commands, changing startup behavior, toggling kiosk mode, or updating the logo.
Why/When to use:
When you want to schedule commands (to run at startup or a specific time), change the logo, or restart the SmartScreen service.
Key options:
• logo: sets a new logo URL
• kiosk: toggles kiosk mode on/off
• startcmd/schcmd: schedule commands for startup or specific time triggers
• restart: restarts the SmartScreen.
    """
    success = await smartscreen_client.dj(option=request.option, value=request.value,display=request.display)
    if not success:
        raise HTTPException(status_code=500, detail="Failed to execute DJ command")
    return {"message": "DJ command executed successfully", "option": request.option, "value": request.value}

mcp = FastApiMCP(
    app,
    name="ultra-mcp-ss"
)
mcp.mount()  # mounts at '/mcp' by default


if __name__ == "__main__":
    import uvicorn
    uvicorn.run("main:app", host="0.0.0.0", port=8000, reload=True)