hope-voice-api/agent/media_mcp.py

"""Media MCP client — lets Hope control the standalone Hope Music service.

Tools call the separate ``hope-media-music`` LAN service over HTTP. That
service owns the library, YouTube Music search/download jobs, stream proxy,
and per-room player state. Its standalone browser page is the actual audio
player: it reconciles against the service and reports real playback.

Environment (set by the voice agent when spawning this MCP):
- MUSIC_BASE_URL  e.g. http://192.168.86.2:8094
- MUSIC_ROOM      stable player scope shared with the Hope Music page

The voice agent polls the same service state to keep music and conversation
mutually exclusive; there is no music server or music player in the voice
container anymore.
"""
from __future__ import annotations

import json
import os

import httpx
from mcp.server.fastmcp import FastMCP

MUSIC_BASE_URL = os.environ.get(
    "MUSIC_BASE_URL", "http://192.168.86.2:8094"
).rstrip("/")
MUSIC_ROOM = os.environ.get("MUSIC_ROOM", "voice-room")

mcp = FastMCP("media")


def _media_get(path: str, params: dict | None = None) -> dict:
    resp = httpx.get(f"{MUSIC_BASE_URL}{path}", params=params, timeout=10)
    resp.raise_for_status()
    return resp.json()


def _media_post(path: str, body: dict) -> dict:
    resp = httpx.post(f"{MUSIC_BASE_URL}{path}", json=body, timeout=30)
    if resp.status_code >= 400:
        try:
            detail = resp.json().get("error", resp.text)
        except Exception:  # noqa: BLE001
            detail = resp.text
        return {"error": f"{resp.status_code}: {detail}"}
    return resp.json()


def _control(action: str, **extra) -> str:
    data = _media_post("/media/api/control", {"room": MUSIC_ROOM, "action": action, **extra})
    if "error" in data:
        return f"Could not {action} the music: {data['error']}"
    track = data.get("track") or {}
    desired = data.get("desired", "")
    name = track.get("title") or "nothing"
    return f"OK — {action} applied. Loaded track: {name} (player state: {desired})."


def _fmt_duration(seconds) -> str:
    try:
        seconds = int(seconds or 0)
    except (TypeError, ValueError):
        return ""
    if seconds <= 0:
        return ""
    minutes, secs = divmod(seconds, 60)
    hours, minutes = divmod(minutes, 60)
    return f"{hours}:{minutes:02d}:{secs:02d}" if hours else f"{minutes}:{secs:02d}"


@mcp.tool()
def search_music(query: str, limit: int = 5) -> str:
    """Search for songs: already-downloaded library matches come first, then
    YouTube Music results. Call this before playing anything the user
    describes by name, artist or lyrics. Results carry video_id, title,
    channel, duration and in_library flags. ALWAYS prefer a result with
    in_library=true — it is already saved and plays instantly. Pick the
    result that clearly matches what the user asked for; if two or more could
    be right, use offer_song_choices instead of guessing."""
    data = _media_post("/media/api/search", {"query": query, "limit": max(1, min(limit, 10))})
    if "error" in data:
        return f"Search failed: {data['error']}"
    results = data.get("results", [])
    if not results:
        return f"No results for '{query}'."
    return json.dumps(results, ensure_ascii=False)


@mcp.tool()
def play_song(video_id: str, title: str = "", channel: str = "") -> str:
    """Play a song in the user's standalone Hope Music player. Pass the video_id from
    search_music — for an in_library result this plays the saved copy with no
    download; a new song downloads and starts playing automatically while it
    arrives. After calling this, tell the user what is playing in one short
    sentence and stop talking — the player takes over and the conversation
    pauses until they return."""
    data = _media_post("/media/api/play", {
        "room": MUSIC_ROOM, "video_id": video_id, "title": title, "channel": channel,
    })
    if "error" in data:
        return f"Could not start playback: {data['error']}"
    track = data.get("track") or {}
    status = track.get("status")
    name = track.get("title") or title or video_id
    if status == "downloading":
        return (
            f"'{name}' is downloading and starts playing within seconds — "
            "audio streams in while the download finishes. Say it's on the "
            "way, then hand over to the player."
        )
    return f"Playing '{name}' now. Hand over to the player with a short sign-off."


@mcp.tool()
def offer_song_choices(choices: list[dict]) -> str:
    """Offer the user 2-4 song options when a search is ambiguous. Pass a list
    like [{"video_id": "...", "title": "...", "channel": "...", "duration": 213}].
    The choices appear as clickable cards in Hope Music; the user can
    also just say which one they want. Briefly name the options aloud, then
    wait for their pick."""
    if not choices:
        return "No choices provided."
    data = _media_post("/media/api/choices", {"room": MUSIC_ROOM, "choices": choices})
    if "error" in data:
        return f"Could not present choices: {data['error']}"
    names = [
        f"{c.get('title')}" + (f" by {c['channel']}" if c.get("channel") else "")
        for c in choices
    ]
    return "Choices shown in the player: " + "; ".join(names) + ". Ask the user to pick one."


@mcp.tool()
def pause_music() -> str:
    """Pause the currently playing music. The user stays in the conversation
    afterwards."""
    return _control("pause")


@mcp.tool()
def resume_music() -> str:
    """Resume the loaded music. Warn the user you'll stop listening while it
    plays — they'll need to come back to you to talk again."""
    return _control("resume")


@mcp.tool()
def stop_music() -> str:
    """Stop playback completely and clear the loaded track."""
    return _control("stop")


@mcp.tool()
def skip_song() -> str:
    """Skip to the next track in the library (most recently added first).
    If the library has nothing else, playback stops."""
    return _control("skip")


@mcp.tool()
def seek_music(delta_seconds: int) -> str:
    """Skip forward or backward in the current track. Positive delta_seconds
    jumps ahead, negative goes back (e.g. -30 rewinds 30 seconds, 20 skips
    20 seconds forward)."""
    try:
        reported = (_media_get("/media/api/state", {"room": MUSIC_ROOM}).get("reported") or {})
        position = float(reported.get("position") or 0.0)
    except Exception:  # noqa: BLE001
        position = 0.0
    target = max(0.0, position + float(delta_seconds))
    return _control("seek", position=target)


@mcp.tool()
def now_playing() -> str:
    """Return what is currently loaded/playing in the user's web player and
    its playback position. Use when the user asks what's playing or what this
    song is."""
    try:
        st = _media_get("/media/api/state", {"room": MUSIC_ROOM})
    except Exception as exc:  # noqa: BLE001
        return f"Media server unreachable: {exc}"
    track = st.get("track") or {}
    if not track:
        return "Nothing is loaded in the player right now."
    reported = st.get("reported") or {}
    state = reported.get("state") or st.get("desired") or "unknown"
    position = _fmt_duration(reported.get("position"))
    duration = _fmt_duration(track.get("duration"))
    bits = [f"'{track.get('title')}'"]
    if track.get("channel"):
        bits.append(f"by {track['channel']}")
    if state == "playing" and position:
        bits.append(f"at {position}" + (f" of {duration}" if duration else ""))
    else:
        bits.append(f"(playback state: {state})")
    return "Currently loaded: " + " ".join(bits) + "."


@mcp.tool()
def list_library() -> str:
    """List the permanent music library: downloaded YouTube Music tracks and
    added live streams, ready to play instantly. Use when the user asks what
    music is available or wants something from their collection."""
    data = _media_get("/media/api/library")
    tracks = data.get("tracks", [])
    if not tracks:
        return "The library is empty — search for something with search_music first."
    lines = []
    for t in tracks[:25]:
        status = "" if t.get("status") == "ready" else f" ({t.get('status')})"
        dur = _fmt_duration(t.get("duration"))
        dur = f" [{dur}]" if dur else ""
        channel = f" — {t['channel']}" if t.get("channel") else ""
        lines.append(f"{t['title']}{channel}{dur}{status}")
    return f"{len(tracks)} tracks in the library:\n" + "\n".join(lines)


@mcp.tool()
def add_stream(name: str, url: str) -> str:
    """Add a live audio stream URL (internet radio station, direct mp3/aac
    stream) to the permanent library. It then shows up in the web player's
    library and can be played by name. Pass a short friendly name."""
    if not url.lower().startswith(("http://", "https://")):
        return "That does not look like an http(s) stream URL."
    data = _media_post("/media/api/streams", {"name": name, "url": url})
    if "error" in data:
        return f"Could not add the stream: {data['error']}"
    return f"Added '{name}' to the library as a live stream. You can play it on request."


if __name__ == "__main__":
    mcp.run(transport="stdio")