diff --git a/CHANGELOG.md b/CHANGELOG.md index 84594b3..47a60fb 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,18 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +### Added +- **Steam Cloud Integration** (`cloud_saves.py`) + - `list_cloud_files` - List cloud save files for a specific game + - Shows file names, sizes, and modification timestamps + - Works with 'me'/'my' Steam ID shortcuts + - `get_cloud_quota` - Get cloud storage usage and limits + - Shows total quota, used space, available space + - Visual usage bar representation + +### Changed +- Tool count increased from 36 to 38 + ## [v0.9.0] - 2025-12-12 ### Added diff --git a/README.md b/README.md index 163b30e..ccdd090 100644 --- a/README.md +++ b/README.md @@ -18,7 +18,7 @@ Once set up, you can ask Claude things like: - "Show my pending trade offers" - "What's the current price for an AK-47 Redline?" -The server includes 36 tools covering player profiles, game libraries, achievements, stats, reviews, wishlists, news, community guides, trading/market data, family sharing, and Steam Workshop. +The server includes 38 tools covering player profiles, game libraries, achievements, stats, reviews, wishlists, news, community guides, trading/market data, family sharing, Steam Workshop, and cloud saves. --- @@ -179,7 +179,7 @@ No registration needed - just drop in the file and restart. --- -## Available Tools (36 total) +## Available Tools (38 total) ### Player Profiles (ISteamUser) - 6 tools @@ -269,6 +269,13 @@ No registration needed - just drop in the file and restart. | `get_workshop_item_details` | Get full details on a Workshop item (description, subscribers, dependencies) | | `get_workshop_collection` | Get items from a Workshop collection | +### Cloud Saves (ISteamRemoteStorage) - 2 tools + +| Tool | What it does | +|------|--------------| +| `list_cloud_files` | List cloud save files for a game (names, sizes, timestamps) | +| `get_cloud_quota` | Get cloud storage usage and limits | + --- ## Configuration Reference diff --git a/src/steam_mcp/endpoints/cloud_saves.py b/src/steam_mcp/endpoints/cloud_saves.py new file mode 100644 index 0000000..57e56bd --- /dev/null +++ b/src/steam_mcp/endpoints/cloud_saves.py @@ -0,0 +1,184 @@ +"""ISteamRemoteStorage and ICloudService API endpoints. + +This module provides MCP tools for Steam Cloud save management, +including listing cloud files and checking storage quotas. + +Reference: https://partner.steamgames.com/doc/webapi/ISteamRemoteStorage +""" + +from steam_mcp.endpoints.base import BaseEndpoint, endpoint + + +class ISteamRemoteStorage(BaseEndpoint): + """Steam Cloud save management endpoints.""" + + @endpoint( + name="list_cloud_files", + description=( + "List cloud save files for a specific game. " + "Returns file names, sizes, and timestamps. " + "Note: Requires user authentication context." + ), + params={ + "steam_id": { + "type": "string", + "description": ( + "Steam ID in any format. Use 'me' or 'my' to query your own profile " + "(requires STEAM_USER_ID to be configured)." + ), + "required": True, + }, + "app_id": { + "type": "integer", + "description": "Game app ID to list cloud files for", + "required": True, + }, + }, + ) + async def list_cloud_files(self, steam_id: str, app_id: int) -> str: + """List cloud save files for a game.""" + normalized_id = await self._resolve_steam_id(steam_id) + if normalized_id.startswith("Error"): + return normalized_id + + try: + result = await self.client.get( + "ISteamRemoteStorage", + "EnumerateUserFiles", + version=1, + params={ + "steamid": normalized_id, + "appid": app_id, + }, + ) + except Exception as e: + error_str = str(e).lower() + if "401" in error_str or "403" in error_str or "unauthorized" in error_str: + return ( + f"Could not access cloud files for Steam ID {normalized_id}.\n" + "This may require owner authentication or the profile is private." + ) + return f"Error fetching cloud files: {e}" + + response = result.get("response", {}) + files = response.get("files", []) + total_count = response.get("totalcount", len(files)) + + if not files: + return ( + f"No cloud files found for app {app_id} (Steam ID: {normalized_id}).\n" + "This game may not use Steam Cloud, or no saves exist yet." + ) + + # Calculate total size + total_bytes = sum(f.get("file_size", 0) for f in files) + + output = [ + f"Cloud Files for App {app_id}", + f"Steam ID: {normalized_id}", + f"Total Files: {total_count}", + f"Total Size: {self._format_bytes(total_bytes)}", + "", + ] + + for f in files: + filename = f.get("filename", "Unknown") + size = f.get("file_size", 0) + timestamp = f.get("timestamp", 0) + + # Format timestamp + if timestamp: + from datetime import datetime + dt = datetime.fromtimestamp(timestamp) + time_str = dt.strftime("%Y-%m-%d %H:%M:%S") + else: + time_str = "Unknown" + + output.append( + f" {filename}\n" + f" Size: {self._format_bytes(size)} | Modified: {time_str}" + ) + + return "\n".join(output) + + @endpoint( + name="get_cloud_quota", + description=( + "Get Steam Cloud storage usage and limits for a user. " + "Shows total quota, used space, and available space." + ), + params={ + "steam_id": { + "type": "string", + "description": ( + "Steam ID in any format. Use 'me' or 'my' to query your own profile " + "(requires STEAM_USER_ID to be configured)." + ), + "required": True, + }, + }, + ) + async def get_cloud_quota(self, steam_id: str) -> str: + """Get cloud storage quota for a user.""" + normalized_id = await self._resolve_steam_id(steam_id) + if normalized_id.startswith("Error"): + return normalized_id + + try: + result = await self.client.get( + "ICloudService", + "GetUploadServerInfo", + version=1, + params={"steamid": normalized_id}, + ) + except Exception as e: + error_str = str(e).lower() + if "401" in error_str or "403" in error_str or "unauthorized" in error_str: + return ( + f"Could not access cloud quota for Steam ID {normalized_id}.\n" + "This may require owner authentication or the profile is private." + ) + return f"Error fetching cloud quota: {e}" + + response = result.get("response", {}) + + # Extract quota information + total_bytes = response.get("quota_bytes", 0) + used_bytes = response.get("used_bytes", 0) + + if total_bytes == 0 and used_bytes == 0: + return ( + f"No cloud quota information available for Steam ID {normalized_id}.\n" + "This may indicate restricted access or no cloud usage." + ) + + available_bytes = total_bytes - used_bytes + usage_percent = (used_bytes / total_bytes * 100) if total_bytes > 0 else 0 + + output = [ + f"Steam Cloud Storage for {normalized_id}", + "", + f"Total Quota: {self._format_bytes(total_bytes)}", + f"Used Space: {self._format_bytes(used_bytes)} ({usage_percent:.1f}%)", + f"Available: {self._format_bytes(available_bytes)}", + ] + + # Add usage bar visualization + bar_length = 20 + filled = int(bar_length * usage_percent / 100) + bar = "[" + "=" * filled + "-" * (bar_length - filled) + "]" + output.append(f"\n{bar} {usage_percent:.1f}% used") + + return "\n".join(output) + + @staticmethod + def _format_bytes(size: int) -> str: + """Format bytes into human-readable string.""" + if size < 1024: + return f"{size} B" + elif size < 1024 * 1024: + return f"{size / 1024:.1f} KB" + elif size < 1024 * 1024 * 1024: + return f"{size / (1024 * 1024):.1f} MB" + else: + return f"{size / (1024 * 1024 * 1024):.2f} GB" diff --git a/tests/test_cloud_saves.py b/tests/test_cloud_saves.py new file mode 100644 index 0000000..940d048 --- /dev/null +++ b/tests/test_cloud_saves.py @@ -0,0 +1,286 @@ +"""Tests for ISteamRemoteStorage endpoint - cloud saves management.""" + +import pytest +from unittest.mock import AsyncMock, MagicMock, patch + +from steam_mcp.endpoints.cloud_saves import ISteamRemoteStorage +from steam_mcp.endpoints.base import SteamIDError + + +@pytest.fixture +def mock_client(): + """Create mock Steam client.""" + client = MagicMock() + client.owner_steam_id = None + client.get = AsyncMock() + return client + + +@pytest.fixture +def cloud_service(mock_client): + """Create ISteamRemoteStorage instance with mock client.""" + return ISteamRemoteStorage(mock_client) + + +class TestListCloudFiles: + """Tests for list_cloud_files endpoint.""" + + @pytest.mark.asyncio + async def test_returns_cloud_files_list(self, cloud_service, mock_client): + """Should return formatted list of cloud files.""" + mock_client.get.return_value = { + "response": { + "totalcount": 2, + "files": [ + { + "filename": "save_slot_1.sav", + "file_size": 1024, + "timestamp": 1704067200, # 2024-01-01 00:00:00 UTC + }, + { + "filename": "config.cfg", + "file_size": 256, + "timestamp": 1704153600, + }, + ], + } + } + + with patch( + "steam_mcp.endpoints.base.normalize_steam_id", + new_callable=AsyncMock, + return_value="76561198000000001", + ): + result = await cloud_service.list_cloud_files( + steam_id="76561198000000001", app_id=440 + ) + + assert "Cloud Files for App 440" in result + assert "Total Files: 2" in result + assert "save_slot_1.sav" in result + assert "config.cfg" in result + assert "1.0 KB" in result + assert "256 B" in result + + @pytest.mark.asyncio + async def test_no_cloud_files_returns_message(self, cloud_service, mock_client): + """Should return message when no cloud files exist.""" + mock_client.get.return_value = {"response": {"files": [], "totalcount": 0}} + + with patch( + "steam_mcp.endpoints.base.normalize_steam_id", + new_callable=AsyncMock, + return_value="76561198000000001", + ): + result = await cloud_service.list_cloud_files( + steam_id="76561198000000001", app_id=440 + ) + + assert "No cloud files found" in result + assert "app 440" in result + + @pytest.mark.asyncio + async def test_handles_private_profile(self, cloud_service, mock_client): + """Should handle 401/403 errors gracefully.""" + mock_client.get.side_effect = Exception("401 Unauthorized") + + with patch( + "steam_mcp.endpoints.base.normalize_steam_id", + new_callable=AsyncMock, + return_value="76561198000000001", + ): + result = await cloud_service.list_cloud_files( + steam_id="76561198000000001", app_id=440 + ) + + assert "Could not access" in result + assert "private" in result.lower() + + @pytest.mark.asyncio + async def test_invalid_steam_id_returns_error(self, cloud_service): + """Invalid Steam ID should return error.""" + with patch( + "steam_mcp.endpoints.base.normalize_steam_id", + new_callable=AsyncMock, + side_effect=SteamIDError("Invalid Steam ID"), + ): + result = await cloud_service.list_cloud_files( + steam_id="invalid_id", app_id=440 + ) + + assert "Error" in result + + @pytest.mark.asyncio + async def test_me_shortcut_without_config_returns_error( + self, cloud_service, mock_client + ): + """Using 'me' without STEAM_USER_ID configured should return error.""" + mock_client.owner_steam_id = None + + result = await cloud_service.list_cloud_files(steam_id="me", app_id=440) + + assert "Error" in result + assert "STEAM_USER_ID" in result + + @pytest.mark.asyncio + async def test_me_shortcut_with_config_works(self, cloud_service, mock_client): + """Using 'me' with STEAM_USER_ID configured should work.""" + mock_client.owner_steam_id = "76561198000000001" + mock_client.get.return_value = { + "response": { + "files": [{"filename": "test.sav", "file_size": 100, "timestamp": 0}], + "totalcount": 1, + } + } + + result = await cloud_service.list_cloud_files(steam_id="my", app_id=440) + + assert "Cloud Files" in result + assert "test.sav" in result + + @pytest.mark.asyncio + async def test_calculates_total_size(self, cloud_service, mock_client): + """Should calculate and display total size of all files.""" + mock_client.get.return_value = { + "response": { + "files": [ + {"filename": "a.sav", "file_size": 1024 * 1024, "timestamp": 0}, + {"filename": "b.sav", "file_size": 1024 * 1024, "timestamp": 0}, + ], + "totalcount": 2, + } + } + + with patch( + "steam_mcp.endpoints.base.normalize_steam_id", + new_callable=AsyncMock, + return_value="76561198000000001", + ): + result = await cloud_service.list_cloud_files( + steam_id="76561198000000001", app_id=440 + ) + + assert "Total Size: 2.0 MB" in result + + +class TestGetCloudQuota: + """Tests for get_cloud_quota endpoint.""" + + @pytest.mark.asyncio + async def test_returns_quota_info(self, cloud_service, mock_client): + """Should return formatted quota information.""" + mock_client.get.return_value = { + "response": { + "quota_bytes": 1024 * 1024 * 100, # 100 MB + "used_bytes": 1024 * 1024 * 25, # 25 MB + } + } + + with patch( + "steam_mcp.endpoints.base.normalize_steam_id", + new_callable=AsyncMock, + return_value="76561198000000001", + ): + result = await cloud_service.get_cloud_quota(steam_id="76561198000000001") + + assert "Steam Cloud Storage" in result + assert "Total Quota: 100.0 MB" in result + assert "Used Space: 25.0 MB" in result + assert "25.0%" in result + assert "Available: 75.0 MB" in result + + @pytest.mark.asyncio + async def test_no_quota_returns_message(self, cloud_service, mock_client): + """Should return message when no quota info available.""" + mock_client.get.return_value = {"response": {}} + + with patch( + "steam_mcp.endpoints.base.normalize_steam_id", + new_callable=AsyncMock, + return_value="76561198000000001", + ): + result = await cloud_service.get_cloud_quota(steam_id="76561198000000001") + + assert "No cloud quota information" in result + + @pytest.mark.asyncio + async def test_handles_private_profile(self, cloud_service, mock_client): + """Should handle 401/403 errors gracefully.""" + mock_client.get.side_effect = Exception("403 Forbidden") + + with patch( + "steam_mcp.endpoints.base.normalize_steam_id", + new_callable=AsyncMock, + return_value="76561198000000001", + ): + result = await cloud_service.get_cloud_quota(steam_id="76561198000000001") + + assert "Could not access" in result + + @pytest.mark.asyncio + async def test_invalid_steam_id_returns_error(self, cloud_service): + """Invalid Steam ID should return error.""" + with patch( + "steam_mcp.endpoints.base.normalize_steam_id", + new_callable=AsyncMock, + side_effect=SteamIDError("Invalid Steam ID"), + ): + result = await cloud_service.get_cloud_quota(steam_id="invalid_id") + + assert "Error" in result + + @pytest.mark.asyncio + async def test_me_shortcut_without_config_returns_error( + self, cloud_service, mock_client + ): + """Using 'me' without STEAM_USER_ID configured should return error.""" + mock_client.owner_steam_id = None + + result = await cloud_service.get_cloud_quota(steam_id="me") + + assert "Error" in result + assert "STEAM_USER_ID" in result + + @pytest.mark.asyncio + async def test_usage_bar_visualization(self, cloud_service, mock_client): + """Should display usage bar visualization.""" + mock_client.get.return_value = { + "response": { + "quota_bytes": 1000, + "used_bytes": 500, # 50% + } + } + + with patch( + "steam_mcp.endpoints.base.normalize_steam_id", + new_callable=AsyncMock, + return_value="76561198000000001", + ): + result = await cloud_service.get_cloud_quota(steam_id="76561198000000001") + + assert "[" in result + assert "]" in result + assert "=" in result + + +class TestFormatBytes: + """Tests for _format_bytes helper method.""" + + def test_formats_bytes(self, cloud_service): + """Should format bytes correctly.""" + assert cloud_service._format_bytes(512) == "512 B" + + def test_formats_kilobytes(self, cloud_service): + """Should format kilobytes correctly.""" + assert cloud_service._format_bytes(1024) == "1.0 KB" + assert cloud_service._format_bytes(2048) == "2.0 KB" + + def test_formats_megabytes(self, cloud_service): + """Should format megabytes correctly.""" + assert cloud_service._format_bytes(1024 * 1024) == "1.0 MB" + assert cloud_service._format_bytes(5 * 1024 * 1024) == "5.0 MB" + + def test_formats_gigabytes(self, cloud_service): + """Should format gigabytes correctly.""" + assert cloud_service._format_bytes(1024 * 1024 * 1024) == "1.00 GB" + assert cloud_service._format_bytes(2 * 1024 * 1024 * 1024) == "2.00 GB"