"""Media MCP server — lets Hope find and control music by voice. Tools talk to the local media server (web/media_server.py) over HTTP, which owns the library, YouTube Music search/download jobs and per-room player state. The browser is the actual player: it reconciles against the media server and reports playback back, so voice controls and the web panel stay in sync. Per-session environment (set by the agent when spawning this MCP): - MEDIA_SERVER_URL e.g. http://127.0.0.1:8093 - MEDIA_ROOM LiveKit room name this session controls The "stop listening while music plays" rule lives in the agent itself: the UI reports media_state over the data channel and the agent disables its mic while a track is playing. """ from __future__ import annotations import json import os import httpx from mcp.server.fastmcp import FastMCP MEDIA_SERVER_URL = os.environ.get("MEDIA_SERVER_URL", "http://127.0.0.1:8093").rstrip("/") MEDIA_ROOM = os.environ.get("MEDIA_ROOM", "default") mcp = FastMCP("media") def _media_get(path: str, params: dict | None = None) -> dict: resp = httpx.get(f"{MEDIA_SERVER_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"{MEDIA_SERVER_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": MEDIA_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 web 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": MEDIA_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 the music panel; 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": MEDIA_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": MEDIA_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": MEDIA_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")