TIDAL MCP Server

Instructions

Implementation Reference

• src/tidal_mcp/server.py:157-227 (handler)

  Implementation of the 'login' tool handler. Uses @mcp.tool decorator for automatic registration. Performs OAuth login via browser, persists session to file, checks authentication status.
  @mcp.tool( annotations={ "title": "Authenticate with TIDAL", "readOnlyHint": False, "openWorldHint": True, "idempotentHint": False, } ) async def login() -> AuthResult: """ Authenticate with TIDAL using OAuth browser flow. Opens browser automatically for secure login. Session is persisted for future use. """ if await ensure_authenticated(): return AuthResult( status="success", message="Already authenticated with TIDAL", authenticated=True, ) try: # Start OAuth device code flow login_obj, future = await anyio.to_thread.run_sync(session.login_oauth) auth_url = login_obj.verification_uri_complete if not auth_url.startswith("http"): auth_url = "https://" + auth_url # Try to open browser automatically try: await anyio.to_thread.run_sync(webbrowser.open, auth_url) except Exception: pass # Wait for user to complete authentication (polls TIDAL's server) # This blocks until auth completes or the link expires (~5 minutes) await anyio.to_thread.run_sync(future.result) # Check if login succeeded if await anyio.to_thread.run_sync(session.check_login): # Save session for future use session_data = { "token_type": {"data": session.token_type or "Bearer"}, "session_id": {"data": session.session_id or ""}, "access_token": {"data": session.access_token}, "refresh_token": {"data": session.refresh_token}, "is_pkce": {"data": session.is_pkce}, } async with await anyio.open_file(SESSION_FILE, "w") as f: await f.write(json.dumps(session_data)) return AuthResult( status="success", message="Successfully authenticated with TIDAL", authenticated=True, ) else: raise ToolError("Authentication failed - please try again") except ToolError: raise except Exception as e: error_msg = str(e) if "too long" in error_msg.lower() or "timeout" in error_msg.lower(): raise ToolError( f"Authentication timed out. Please authenticate using the helper script:\n\n" f" cd /home/ubuntu/code/personal/tidal/tidal-mcp && uv run python authenticate.py\n\n" f"After authenticating, all MCP tools will work automatically." ) raise ToolError(f"Authentication error: {error_msg}")
• src/tidal_mcp/models.py:123-129 (schema)

  Pydantic BaseModel defining the structured output schema for the login tool response.
  class AuthResult(BaseModel): """Result of authentication attempt.""" status: str = Field(description="Operation status (success/error)") message: str = Field(description="Status message") authenticated: bool = Field(description="Whether authentication was successful")
• src/tidal_mcp/server.py:120-151 (helper)

  Helper function to check and load persisted authentication session. Used by login and all other authenticated tools.
  async def ensure_authenticated() -> bool: """ Check if user is authenticated with TIDAL. Automatically loads persisted session if available. """ if await anyio.Path(SESSION_FILE).exists(): try: async with await anyio.open_file(SESSION_FILE, "r") as f: content = await f.read() data = json.loads(content) # Load OAuth session result = await anyio.to_thread.run_sync( session.load_oauth_session, data["token_type"]["data"], data["access_token"]["data"], data["refresh_token"]["data"], None, # expiry time ) if result: is_valid = await anyio.to_thread.run_sync(session.check_login) if not is_valid: await anyio.Path(SESSION_FILE).unlink() return is_valid return False except Exception: await anyio.Path(SESSION_FILE).unlink() return False return await anyio.to_thread.run_sync(session.check_login)
• src/tidal_mcp/server.py:157-164 (registration)

  @mcp.tool decorator registers the login function as an MCP tool with custom annotations.
  @mcp.tool( annotations={ "title": "Authenticate with TIDAL", "readOnlyHint": False, "openWorldHint": True, "idempotentHint": False, } )