From 22c201a01c3a06e2eecd43393eb81e264564139a Mon Sep 17 00:00:00 2001 From: CodeKeanu <85296798+CodeKeanu@users.noreply.github.com> Date: Fri, 12 Dec 2025 00:05:46 +0000 Subject: [PATCH 1/2] feat: add optional JSON output format for all tools (#14) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Add `supports_json` parameter to the @endpoint decorator that auto-adds a `format` parameter to endpoints. When format="json" is passed, endpoints return structured JSON instead of text. Changes: - Enhanced @endpoint decorator with supports_json=True option - Auto-adds format parameter with enum ["text", "json"] - Implemented JSON output for 5 key endpoints: - get_player_summary: Player profile data - get_friend_list: Friends with timestamps - get_owned_games: Games with playtime stats - get_player_achievements: Achievements with unlock times - get_app_details: App metadata, pricing, platforms - Text output remains unchanged (backward compatible) - JSON output uses consistent snake_case field names 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude Opus 4.5 --- src/steam_mcp/endpoints/base.py | 21 +++++- src/steam_mcp/endpoints/player_service.py | 34 ++++++++- src/steam_mcp/endpoints/steam_apps.py | 68 ++++++++++++++++-- src/steam_mcp/endpoints/steam_user.py | 87 +++++++++++++++++++++-- src/steam_mcp/endpoints/user_stats.py | 60 +++++++++++++--- 5 files changed, 244 insertions(+), 26 deletions(-) diff --git a/src/steam_mcp/endpoints/base.py b/src/steam_mcp/endpoints/base.py index a592856..c0b53e9 100644 --- a/src/steam_mcp/endpoints/base.py +++ b/src/steam_mcp/endpoints/base.py @@ -58,6 +58,7 @@ class EndpointTool: input_schema: dict[str, Any] handler: Callable[..., Coroutine[Any, Any, str]] endpoint_class: type["BaseEndpoint"] + supports_json: bool = False @dataclass @@ -206,6 +207,7 @@ def endpoint( name: str, description: str, params: dict[str, dict[str, Any] | EndpointParam] | None = None, + supports_json: bool = False, ) -> Callable[[F], F]: """ Decorator to register a method as an MCP tool endpoint. @@ -219,6 +221,8 @@ def endpoint( description: Human-readable description of what the tool does params: Dictionary of parameter definitions. Each parameter can be a dict with keys: type, description, required, enum, default + supports_json: If True, adds a 'format' parameter that allows switching + between 'text' (default) and 'json' output formats Returns: Decorated function @@ -227,6 +231,7 @@ def endpoint( @endpoint( name="get_player_summary", description="Get Steam player profile", + supports_json=True, params={ "steam_id": { "type": "string", @@ -241,10 +246,22 @@ def endpoint( }, }, ) - async def get_player_summary(self, steam_id: str, include_avatar: bool = True) -> str: + async def get_player_summary(self, steam_id: str, include_avatar: bool = True, format: str = "text") -> str: ... """ params = params or {} + + # If supports_json, auto-add the format parameter + if supports_json: + params = dict(params) # Make a copy to avoid mutating the original + params["format"] = { + "type": "string", + "description": "Output format: 'text' for human-readable output, 'json' for structured JSON", + "enum": ["text", "json"], + "default": "text", + "required": False, + } + input_schema = _build_input_schema(params) def decorator(func: F) -> F: @@ -255,6 +272,7 @@ def decorator(func: F) -> F: "description": description, "input_schema": input_schema, "params": params, + "supports_json": supports_json, } return func @@ -289,6 +307,7 @@ def __new__( input_schema=meta["input_schema"], handler=attr_value, endpoint_class=cls, # type: ignore[arg-type] + supports_json=meta.get("supports_json", False), ) EndpointRegistry.register_tool(tool) diff --git a/src/steam_mcp/endpoints/player_service.py b/src/steam_mcp/endpoints/player_service.py index 7c504cd..21e6753 100644 --- a/src/steam_mcp/endpoints/player_service.py +++ b/src/steam_mcp/endpoints/player_service.py @@ -7,6 +7,7 @@ """ import asyncio +import json from typing import Any from steam_mcp.endpoints.base import BaseEndpoint, endpoint @@ -27,6 +28,7 @@ class IPlayerService(BaseEndpoint): "Returns game names, App IDs, and total playtime. " "Note: Only works for public profiles unless querying your own profile." ), + supports_json=True, params={ "steam_id": { "type": "string", @@ -64,11 +66,14 @@ async def get_owned_games( include_free_games: bool = True, sort_by: str = "playtime", limit: int = 25, + format: str = "text", ) -> str: """Get owned games for a Steam user.""" # Handle "me" / "my" shortcut normalized_id = await self._resolve_steam_id(steam_id) if normalized_id.startswith("Error"): + if format == "json": + return json.dumps({"error": normalized_id}) return normalized_id result = await self.client.get( @@ -87,10 +92,13 @@ async def get_owned_games( game_count = response.get("game_count", len(games)) if not games: - return ( + error_msg = ( f"No games found for Steam ID {normalized_id}.\n" "This may indicate a private profile or an account with no games." ) + if format == "json": + return json.dumps({"error": error_msg, "steam_id": normalized_id, "games": []}) + return error_msg # Sort games if sort_by == "playtime": @@ -107,6 +115,30 @@ async def get_owned_games( # Determine display limit (0 = show all) display_limit = limit if limit > 0 else len(games) + if format == "json": + data = { + "steam_id": normalized_id, + "total_games": game_count, + "total_playtime_hours": round(total_hours, 1), + "sort_by": sort_by, + "games": [ + { + "app_id": g.get("appid"), + "name": g.get("name", f"App {g.get('appid', 'Unknown')}"), + "playtime_minutes": g.get("playtime_forever", 0), + "playtime_hours": round(g.get("playtime_forever", 0) / 60, 1), + "playtime_2weeks_minutes": g.get("playtime_2weeks", 0), + "last_played": g.get("rtime_last_played"), + } + for g in games[:display_limit] + ], + } + if len(games) > display_limit: + data["truncated"] = True + data["remaining_games"] = len(games) - display_limit + return json.dumps(data, indent=2) + + # Text format output = [ f"Game Library for {normalized_id}", f"Total Games: {game_count}", diff --git a/src/steam_mcp/endpoints/steam_apps.py b/src/steam_mcp/endpoints/steam_apps.py index 3c9ce06..dccd1b7 100644 --- a/src/steam_mcp/endpoints/steam_apps.py +++ b/src/steam_mcp/endpoints/steam_apps.py @@ -9,6 +9,7 @@ """ import asyncio +import json import re from dataclasses import dataclass from datetime import datetime @@ -190,6 +191,7 @@ async def check_app_up_to_date( "Get detailed information about a Steam app including description, " "price, genres, release date, and more. Uses the Steam Store API." ), + supports_json=True, params={ "app_id": { "type": "integer", @@ -208,6 +210,7 @@ async def get_app_details( self, app_id: int, country_code: str = "us", + format: str = "text", ) -> str: """Get detailed app information from the Store API.""" try: @@ -220,24 +223,33 @@ async def get_app_details( }, ) except Exception as e: - return f"Error fetching app details: {e}" + err = f"Error fetching app details: {e}" + if format == "json": + return json.dumps({"error": err}) + return err app_data = result.get(str(app_id), {}) if not app_data.get("success", False): - return f"App ID {app_id} not found or unavailable in region '{country_code}'." + err = f"App ID {app_id} not found or unavailable in region '{country_code}'." + if format == "json": + return json.dumps({"error": err}) + return err data = app_data.get("data", {}) if not data: - return f"No data available for App ID {app_id}." + err = f"No data available for App ID {app_id}." + if format == "json": + return json.dumps({"error": err}) + return err name = data.get("name", "Unknown") app_type = data.get("type", "unknown") is_free = data.get("is_free", False) short_desc = data.get("short_description", "") - developers = ", ".join(data.get("developers", ["Unknown"])) - publishers = ", ".join(data.get("publishers", ["Unknown"])) + developers = data.get("developers", []) + publishers = data.get("publishers", []) # Release date release_info = data.get("release_date", {}) @@ -279,6 +291,48 @@ async def get_app_details( # Metacritic metacritic = data.get("metacritic", {}) + + if format == "json": + json_data: dict[str, Any] = { + "app_id": app_id, + "name": name, + "type": app_type, + "is_free": is_free, + "short_description": short_desc, + "developers": developers, + "publishers": publishers, + "release_date": release_date, + "coming_soon": release_info.get("coming_soon", False), + "platforms": { + "windows": platforms.get("windows", False), + "mac": platforms.get("mac", False), + "linux": platforms.get("linux", False), + }, + "genres": genres, + "categories": categories, + "store_url": f"https://store.steampowered.com/app/{app_id}", + } + # Add price info + if is_free: + json_data["price"] = {"is_free": True} + elif price_info: + json_data["price"] = { + "is_free": False, + "currency": price_info.get("currency", ""), + "initial": price_info.get("initial", 0), + "final": price_info.get("final", 0), + "discount_percent": price_info.get("discount_percent", 0), + "final_formatted": price_info.get("final_formatted", ""), + } + # Add metacritic if available + if metacritic: + json_data["metacritic"] = { + "score": metacritic.get("score"), + "url": metacritic.get("url", ""), + } + return json.dumps(json_data, indent=2) + + # Text format metacritic_str = "" if metacritic: score = metacritic.get("score", "N/A") @@ -287,8 +341,8 @@ async def get_app_details( output = [ f"{name}", f"App ID: {app_id} | Type: {app_type.title()}", - f"Developer: {developers}", - f"Publisher: {publishers}", + f"Developer: {', '.join(developers) if developers else 'Unknown'}", + f"Publisher: {', '.join(publishers) if publishers else 'Unknown'}", f"Release Date: {release_date}", f"Price: {price_str}", f"Platforms: {platforms_str}", diff --git a/src/steam_mcp/endpoints/steam_user.py b/src/steam_mcp/endpoints/steam_user.py index 038de06..27269ba 100644 --- a/src/steam_mcp/endpoints/steam_user.py +++ b/src/steam_mcp/endpoints/steam_user.py @@ -6,6 +6,7 @@ Reference: https://partner.steamgames.com/doc/webapi/ISteamUser """ +import json from datetime import datetime from typing import Any @@ -90,6 +91,7 @@ async def get_my_steam_id(self) -> str: "online status, and profile visibility. Accepts any Steam ID format " "(SteamID64, vanity URL, profile URL, etc.)." ), + supports_json=True, params={ "steam_id": { "type": "string", @@ -103,18 +105,27 @@ async def get_my_steam_id(self) -> str: }, }, ) - async def get_player_summary(self, steam_id: str) -> str: + async def get_player_summary(self, steam_id: str, format: str = "text") -> str: """Get player summary for a Steam user.""" normalized_id = await self._resolve_steam_id(steam_id) if normalized_id.startswith("Error"): + if format == "json": + return json.dumps({"error": normalized_id}) return normalized_id players = await self.client.get_player_summaries([normalized_id]) if not players: - return f"Player not found for Steam ID: {steam_id}" + error_msg = f"Player not found for Steam ID: {steam_id}" + if format == "json": + return json.dumps({"error": error_msg}) + return error_msg player = players[0] + data = self._build_player_data(player) + + if format == "json": + return json.dumps(data, indent=2) return self._format_player_summary(player) @endpoint( @@ -212,6 +223,7 @@ async def resolve_vanity_url(self, vanity_name: str) -> str: "Get the friend list for a Steam user. Note: This only works for " "users with public profiles. Returns friend Steam IDs and relationship info." ), + supports_json=True, params={ "steam_id": { "type": "string", @@ -222,10 +234,12 @@ async def resolve_vanity_url(self, vanity_name: str) -> str: }, }, ) - async def get_friend_list(self, steam_id: str) -> str: + async def get_friend_list(self, steam_id: str, format: str = "text") -> str: """Get friend list for a Steam user.""" normalized_id = await self._resolve_steam_id(steam_id) if normalized_id.startswith("Error"): + if format == "json": + return json.dumps({"error": normalized_id}) return normalized_id try: @@ -238,20 +252,41 @@ async def get_friend_list(self, steam_id: str) -> str: except SteamAPIError as e: # 401 Unauthorized typically means private profile if e.status_code == 401: - return ( + error_msg = ( f"Cannot access friend list for Steam ID {normalized_id}. " "This profile's friend list is private." ) + if format == "json": + return json.dumps({"error": error_msg}) + return error_msg raise friends_list = result.get("friendslist", {}).get("friends", []) if not friends_list: - return ( + error_msg = ( f"No friends found for Steam ID {normalized_id}. " "This may indicate a private profile or an account with no friends." ) - + if format == "json": + return json.dumps({"error": error_msg, "steam_id": normalized_id, "friends": []}) + return error_msg + + if format == "json": + data = { + "steam_id": normalized_id, + "total_friends": len(friends_list), + "friends": [ + { + "steam_id": friend["steamid"], + "friend_since": friend.get("friend_since"), + } + for friend in friends_list + ], + } + return json.dumps(data, indent=2) + + # Text format output = [f"Friend list for {normalized_id} ({len(friends_list)} friends):\n"] for friend in friends_list[:MAX_FRIENDS_DISPLAY]: @@ -346,6 +381,46 @@ async def get_player_bans(self, steam_ids: list[str]) -> str: return "\n".join(output) + def _build_player_data(self, player: dict[str, Any]) -> dict[str, Any]: + """Build structured player data for JSON output.""" + visibility_map = {1: "private", 2: "friends_only", 3: "public"} + status_map = { + 0: "offline", + 1: "online", + 2: "busy", + 3: "away", + 4: "snooze", + 5: "looking_to_trade", + 6: "looking_to_play", + } + + data: dict[str, Any] = { + "steam_id": player.get("steamid", "Unknown"), + "persona_name": player.get("personaname", "Unknown"), + "profile_url": player.get("profileurl", ""), + "visibility": visibility_map.get( + player.get("communityvisibilitystate", 1), "unknown" + ), + "status": status_map.get(player.get("personastate", 0), "unknown"), + "avatar_url": player.get("avatarfull", ""), + } + + # Additional fields only available for public profiles + if player.get("communityvisibilitystate") == 3: + if player.get("realname"): + data["real_name"] = player["realname"] + if player.get("loccountrycode"): + data["country"] = player["loccountrycode"] + if player.get("gameextrainfo"): + data["currently_playing"] = { + "name": player["gameextrainfo"], + "app_id": player.get("gameid", ""), + } + if player.get("timecreated"): + data["account_created"] = player["timecreated"] + + return data + def _format_player_summary(self, player: dict[str, Any]) -> str: """Format a player summary for display.""" steam_id = player.get("steamid", "Unknown") diff --git a/src/steam_mcp/endpoints/user_stats.py b/src/steam_mcp/endpoints/user_stats.py index 83a3adb..c55b6de 100644 --- a/src/steam_mcp/endpoints/user_stats.py +++ b/src/steam_mcp/endpoints/user_stats.py @@ -6,6 +6,8 @@ Reference: https://partner.steamgames.com/doc/webapi/ISteamUserStats """ +import json +from datetime import datetime from typing import Any from steam_mcp.endpoints.base import BaseEndpoint, endpoint @@ -26,6 +28,7 @@ class ISteamUserStats(BaseEndpoint): "Shows which achievements are unlocked and when. " "Note: Requires the player's game details to be public." ), + supports_json=True, params={ "steam_id": { "type": "string", @@ -52,10 +55,13 @@ async def get_player_achievements( steam_id: str, app_id: int, language: str = "english", + format: str = "text", ) -> str: """Get player achievements for a specific game.""" normalized_id = await self._resolve_steam_id(steam_id) if normalized_id.startswith("Error"): + if format == "json": + return json.dumps({"error": normalized_id}) return normalized_id try: @@ -72,35 +78,45 @@ async def get_player_achievements( except Exception as e: error_msg = str(e).lower() if "profile is not public" in error_msg or "private" in error_msg: - return ( - f"Cannot access achievements for Steam ID {normalized_id}.\n" + err = ( + f"Cannot access achievements for Steam ID {normalized_id}. " "This player's game details are set to private." ) + if format == "json": + return json.dumps({"error": err}) + return err raise playerstats = result.get("playerstats", {}) if not playerstats.get("success", False): - return ( - f"Could not retrieve achievements for App ID {app_id}.\n" + err = ( + f"Could not retrieve achievements for App ID {app_id}. " "The game may not have achievements or the profile is private." ) + if format == "json": + return json.dumps({"error": err}) + return err # Verify the Steam ID matches (detect misrouted responses) response_steamid = playerstats.get("steamID") if response_steamid and response_steamid != normalized_id: - return ( - f"Error: Steam API returned data for wrong player.\n" - f"Requested: {normalized_id}\n" - f"Received: {response_steamid}\n" - "This may be a Steam API issue. Please try again." + err = ( + f"Error: Steam API returned data for wrong player. " + f"Requested: {normalized_id}, Received: {response_steamid}" ) + if format == "json": + return json.dumps({"error": err}) + return err game_name = playerstats.get("gameName", f"App {app_id}") achievements = playerstats.get("achievements", []) if not achievements: - return f"No achievements found for {game_name}." + msg = f"No achievements found for {game_name}." + if format == "json": + return json.dumps({"error": msg, "steam_id": normalized_id, "app_id": app_id, "achievements": []}) + return msg # Count unlocked unlocked = [a for a in achievements if a.get("achieved", 0) == 1] @@ -108,6 +124,29 @@ async def get_player_achievements( completion_pct = (len(unlocked) / len(achievements)) * 100 if achievements else 0 + if format == "json": + data = { + "steam_id": normalized_id, + "app_id": app_id, + "game_name": game_name, + "total_achievements": len(achievements), + "unlocked_count": len(unlocked), + "locked_count": len(locked), + "completion_percent": round(completion_pct, 1), + "achievements": [ + { + "api_name": a.get("apiname", ""), + "name": a.get("name", a.get("apiname", "Unknown")), + "description": a.get("description", ""), + "achieved": a.get("achieved", 0) == 1, + "unlock_time": a.get("unlocktime") if a.get("unlocktime") else None, + } + for a in achievements + ], + } + return json.dumps(data, indent=2) + + # Text format output = [ f"Achievements for {game_name}", f"Player: {normalized_id}", @@ -129,7 +168,6 @@ async def get_player_achievements( unlock_time = ach.get("unlocktime", 0) if unlock_time: - from datetime import datetime unlock_str = datetime.fromtimestamp(unlock_time).strftime("%Y-%m-%d") else: unlock_str = "Unknown date" From 730d6c0172585d0d76a17c2c75d74717b03a1f8d Mon Sep 17 00:00:00 2001 From: CodeKeanu <85296798+CodeKeanu@users.noreply.github.com> Date: Fri, 12 Dec 2025 09:00:20 +0000 Subject: [PATCH 2/2] fix: standardize JSON error response format MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit All JSON error responses now consistently return {"error": msg} without extra fields like empty arrays or IDs. 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude Opus 4.5 --- src/steam_mcp/endpoints/player_service.py | 2 +- src/steam_mcp/endpoints/steam_user.py | 2 +- src/steam_mcp/endpoints/user_stats.py | 2 +- 3 files changed, 3 insertions(+), 3 deletions(-) diff --git a/src/steam_mcp/endpoints/player_service.py b/src/steam_mcp/endpoints/player_service.py index 21e6753..47e0b47 100644 --- a/src/steam_mcp/endpoints/player_service.py +++ b/src/steam_mcp/endpoints/player_service.py @@ -97,7 +97,7 @@ async def get_owned_games( "This may indicate a private profile or an account with no games." ) if format == "json": - return json.dumps({"error": error_msg, "steam_id": normalized_id, "games": []}) + return json.dumps({"error": error_msg}) return error_msg # Sort games diff --git a/src/steam_mcp/endpoints/steam_user.py b/src/steam_mcp/endpoints/steam_user.py index 27269ba..459785c 100644 --- a/src/steam_mcp/endpoints/steam_user.py +++ b/src/steam_mcp/endpoints/steam_user.py @@ -269,7 +269,7 @@ async def get_friend_list(self, steam_id: str, format: str = "text") -> str: "This may indicate a private profile or an account with no friends." ) if format == "json": - return json.dumps({"error": error_msg, "steam_id": normalized_id, "friends": []}) + return json.dumps({"error": error_msg}) return error_msg if format == "json": diff --git a/src/steam_mcp/endpoints/user_stats.py b/src/steam_mcp/endpoints/user_stats.py index c55b6de..a19d618 100644 --- a/src/steam_mcp/endpoints/user_stats.py +++ b/src/steam_mcp/endpoints/user_stats.py @@ -115,7 +115,7 @@ async def get_player_achievements( if not achievements: msg = f"No achievements found for {game_name}." if format == "json": - return json.dumps({"error": msg, "steam_id": normalized_id, "app_id": app_id, "achievements": []}) + return json.dumps({"error": msg}) return msg # Count unlocked