# server.py
from mcp.server.stdio import stdio_server
from mcp.server import Server
from mcp.types import Tool, TextContent
from .music import AmazonMusicController
app = Server("amazon-music-mcp")
music = AmazonMusicController()
async def ensure_connected():
"""Ensure browser connection is established"""
if music.page is None:
await music.attach_to_chrome()
@app.list_tools()
async def list_tools() -> list[Tool]:
return [
Tool(
name="play",
description="Play current track",
inputSchema={"type": "object", "properties": {}}
),
Tool(
name="pause",
description="Pause current track",
inputSchema={"type": "object", "properties": {}}
),
Tool(
name="next_track",
description="Skip to next track",
inputSchema={"type": "object", "properties": {}}
),
Tool(
name="search_and_play",
description="Search Amazon Music and play first result",
inputSchema={
"type": "object",
"properties": {
"query": {"type": "string", "description": "Search query"}
},
"required": ["query"]
}
),
]
@app.call_tool()
async def call_tool(name: str, arguments: dict) -> list[TextContent]:
await ensure_connected()
if name == "play":
await music.play()
return [TextContent(type="text", text="Playing track")]
elif name == "pause":
await music.pause()
return [TextContent(type="text", text="Paused track")]
elif name == "next_track":
await music.next()
return [TextContent(type="text", text="Skipped to next track")]
elif name == "search_and_play":
query = arguments["query"]
await music.search_and_play(query)
return [TextContent(type="text", text=f"Searching and playing: {query}")]
else:
raise ValueError(f"Unknown tool: {name}")
async def main():
async with stdio_server() as (read_stream, write_stream):
await app.run(read_stream, write_stream, app.create_initialization_options())
if __name__ == "__main__":
import asyncio
asyncio.run(main())
