Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
21 changes: 20 additions & 1 deletion src/steam_mcp/endpoints/base.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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.
Expand All @@ -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
Expand All @@ -227,6 +231,7 @@ def endpoint(
@endpoint(
name="get_player_summary",
description="Get Steam player profile",
supports_json=True,
params={
"steam_id": {
"type": "string",
Expand All @@ -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:
Expand All @@ -255,6 +272,7 @@ def decorator(func: F) -> F:
"description": description,
"input_schema": input_schema,
"params": params,
"supports_json": supports_json,
}
return func

Expand Down Expand Up @@ -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)

Expand Down
34 changes: 33 additions & 1 deletion src/steam_mcp/endpoints/player_service.py
Original file line number Diff line number Diff line change
Expand Up @@ -7,6 +7,7 @@
"""

import asyncio
import json
from typing import Any

from steam_mcp.endpoints.base import BaseEndpoint, endpoint
Expand All @@ -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",
Expand Down Expand Up @@ -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(
Expand All @@ -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})
return error_msg

# Sort games
if sort_by == "playtime":
Expand All @@ -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}",
Expand Down
68 changes: 61 additions & 7 deletions src/steam_mcp/endpoints/steam_apps.py
Original file line number Diff line number Diff line change
Expand Up @@ -9,6 +9,7 @@
"""

import asyncio
import json
import re
from dataclasses import dataclass
from datetime import datetime
Expand Down Expand Up @@ -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",
Expand All @@ -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:
Expand All @@ -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", {})
Expand Down Expand Up @@ -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")
Expand All @@ -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}",
Expand Down
Loading