Skip to content

feat: add pagination support to truncated endpoints #30

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
CodeKeanu wants to merge 1 commit into
base: main
Choose a base branch
from
Open

feat: add pagination support to truncated endpoints #30

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
35 changes: 25 additions & 10 deletions src/steam_mcp/endpoints/player_service.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -58,6 +58,13 @@ class IPlayerService(BaseEndpoint):
"default": 25,
"minimum": 0,
},
"offset": {
"type": "integer",
"description": "Number of games to skip (for pagination). Default is 0.",
"required": False,
"default": 0,
"minimum": 0,
},
},
)
async def get_owned_games(
Expand All @@ -66,6 +73,7 @@ async def get_owned_games(
include_free_games: bool = True,
sort_by: str = "playtime",
limit: int = 25,
offset: int = 0,
format: str = "text",
) -> str:
"""Get owned games for a Steam user."""
Expand Down Expand Up @@ -112,15 +120,19 @@ async def get_owned_games(
total_minutes = sum(g.get("playtime_forever", 0) for g in games)
total_hours = total_minutes / 60

# Determine display limit (0 = show all)
# Apply pagination
display_limit = limit if limit > 0 else len(games)
paginated_games = games[offset:offset + display_limit]

if format == "json":
data = {
"steam_id": normalized_id,
"total_games": game_count,
"total_playtime_hours": round(total_hours, 1),
"sort_by": sort_by,
"offset": offset,
"limit": display_limit,
"returned_count": len(paginated_games),
"games": [
{
"app_id": g.get("appid"),
Expand All @@ -130,12 +142,12 @@ async def get_owned_games(
"playtime_2weeks_minutes": g.get("playtime_2weeks", 0),
"last_played": g.get("rtime_last_played"),
}
for g in games[:display_limit]
for g in paginated_games
],
}
if len(games) > display_limit:
data["truncated"] = True
data["remaining_games"] = len(games) - display_limit
remaining = len(games) - offset - len(paginated_games)
if remaining > 0:
data["remaining_games"] = remaining
return json.dumps(data, indent=2)

# Text format
Expand All @@ -146,13 +158,15 @@ async def get_owned_games(
"",
]

if display_limit < len(games):
output.append(f"Top {display_limit} games (sorted by {sort_by}):")
if offset > 0:
output.append(f"Showing from offset {offset} (sorted by {sort_by}):")
elif len(paginated_games) < len(games):
output.append(f"Top {len(paginated_games)} games (sorted by {sort_by}):")
else:
output.append(f"All games (sorted by {sort_by}):")
output.append("")

for game in games[:display_limit]:
for game in paginated_games:
name = game.get("name", f"App {game.get('appid', 'Unknown')}")
appid = game.get("appid", "?")
playtime_mins = game.get("playtime_forever", 0)
Expand All @@ -178,8 +192,9 @@ async def get_owned_games(

output.append(f" [{appid}] {name}: {playtime_str}{recent_str}")

if len(games) > display_limit:
output.append(f"\n ... and {len(games) - display_limit} more games")
remaining = len(games) - offset - len(paginated_games)
if remaining > 0:
output.append(f"\n ... and {remaining} more games (use offset to paginate)")

return "\n".join(output)

Expand Down
44 changes: 34 additions & 10 deletions src/steam_mcp/endpoints/steam_user.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -14,9 +14,6 @@
from steam_mcp.endpoints.base import BaseEndpoint, endpoint
from steam_mcp.utils.steam_id import normalize_steam_id, SteamIDError

# Maximum friends to display in output to avoid overwhelming responses
MAX_FRIENDS_DISPLAY = 50


class ISteamUser(BaseEndpoint):
"""ISteamUser API endpoints for player identity and profile data."""
Expand Down Expand Up @@ -232,9 +229,25 @@ async def resolve_vanity_url(self, vanity_name: str) -> str:
),
"required": True,
},
"limit": {
"type": "integer",
"description": "Maximum number of friends to return. Use 0 for all friends. Default is 50.",
"required": False,
"default": 50,
"minimum": 0,
},
"offset": {
"type": "integer",
"description": "Number of friends to skip (for pagination). Default is 0.",
"required": False,
"default": 0,
"minimum": 0,
},
},
)
async def get_friend_list(self, steam_id: str, format: str = "text") -> str:
async def get_friend_list(
self, steam_id: str, limit: int = 50, offset: int = 0, 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"):
Expand Down Expand Up @@ -272,31 +285,42 @@ async def get_friend_list(self, steam_id: str, format: str = "text") -> str:
return json.dumps({"error": error_msg})
return error_msg

# Apply pagination
total_friends = len(friends_list)
paginated_list = friends_list[offset:] if limit == 0 else friends_list[offset:offset + limit]

if format == "json":
data = {
"steam_id": normalized_id,
"total_friends": len(friends_list),
"total_friends": total_friends,
"offset": offset,
"limit": limit if limit > 0 else total_friends,
"returned_count": len(paginated_list),
"friends": [
{
"steam_id": friend["steamid"],
"friend_since": friend.get("friend_since"),
}
for friend in friends_list
for friend in paginated_list
],
}
return json.dumps(data, indent=2)

# Text format
output = [f"Friend list for {normalized_id} ({len(friends_list)} friends):\n"]
output = [f"Friend list for {normalized_id} ({total_friends} total friends):\n"]

if offset > 0:
output.append(f"Showing from offset {offset}:\n")

for friend in friends_list[:MAX_FRIENDS_DISPLAY]:
for friend in paginated_list:
friend_since = friend.get("friend_since", "Unknown")
output.append(
f" - {friend['steamid']} (friends since: {friend_since})"
)

if len(friends_list) > MAX_FRIENDS_DISPLAY:
output.append(f"\n ... and {len(friends_list) - MAX_FRIENDS_DISPLAY} more friends")
remaining = total_friends - offset - len(paginated_list)
if remaining > 0:
output.append(f"\n ... and {remaining} more friends (use offset to paginate)")

return "\n".join(output)

Expand Down
34 changes: 31 additions & 3 deletions src/steam_mcp/endpoints/steam_wishlist.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -152,12 +152,28 @@ def _format_price(self, price_info: dict[str, Any] | None, is_free: bool) -> str
"required": False,
"default": "us",
},
"limit": {
"type": "integer",
"description": "Maximum number of wishlist items to return. Use 0 for all. Default is 50.",
"required": False,
"default": 50,
"minimum": 0,
},
"offset": {
"type": "integer",
"description": "Number of items to skip (for pagination). Default is 0.",
"required": False,
"default": 0,
"minimum": 0,
},
},
)
async def get_wishlist(
self,
steam_id: str,
country_code: str = "us",
limit: int = 50,
offset: int = 0,
) -> str:
"""Get a user's Steam wishlist with current pricing."""
# Normalize Steam ID
Expand Down Expand Up @@ -192,13 +208,21 @@ async def get_wishlist(

items.sort(key=lambda x: (x[0], x[1].lower()))

# Apply pagination
total_items = len(items)
display_limit = limit if limit > 0 else total_items
paginated_items = items[offset:offset + display_limit]

output = [
f"Steam Wishlist ({len(items)} games)",
f"Steam Wishlist ({total_items} total games)",
f"Prices shown for region: {country_code.upper()}",
"",
]

for priority, name, info in items:
if offset > 0:
output.append(f"Showing from offset {offset}")
output.append("")

for priority, name, info in paginated_items:
app_id = info["app_id"]
details = info["details"]
is_free = details.get("is_free", False)
Expand All @@ -224,6 +248,10 @@ async def get_wishlist(
line += f" (Priority: {priority_str})"
output.append(line)

remaining = total_items - offset - len(paginated_items)
if remaining > 0:
output.append(f"\n ... and {remaining} more items (use offset to paginate)")

return "\n".join(output)

@endpoint(
Expand Down
77 changes: 55 additions & 22 deletions src/steam_mcp/endpoints/user_stats.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -14,10 +14,6 @@
from steam_mcp.utils.steam_id import normalize_steam_id, SteamIDError


# Maximum achievements to display in detailed output
MAX_ACHIEVEMENTS_DISPLAY = 30


class ISteamUserStats(BaseEndpoint):
"""ISteamUserStats API endpoints for achievements and game statistics."""

Expand Down Expand Up @@ -48,13 +44,29 @@ class ISteamUserStats(BaseEndpoint):
"required": False,
"default": "english",
},
"limit": {
"type": "integer",
"description": "Maximum number of achievements to return per section (unlocked/locked). Use 0 for all. Default is 30.",
"required": False,
"default": 30,
"minimum": 0,
},
"offset": {
"type": "integer",
"description": "Number of achievements to skip in each section (for pagination). Default is 0.",
"required": False,
"default": 0,
"minimum": 0,
},
},
)
async def get_player_achievements(
self,
steam_id: str,
app_id: int,
language: str = "english",
limit: int = 30,
offset: int = 0,
format: str = "text",
) -> str:
"""Get player achievements for a specific game."""
Expand Down Expand Up @@ -118,12 +130,24 @@ async def get_player_achievements(
return json.dumps({"error": msg})
return msg

# Count unlocked
# Count unlocked and locked
unlocked = [a for a in achievements if a.get("achieved", 0) == 1]
locked = [a for a in achievements if a.get("achieved", 0) == 0]

# Sort unlocked by unlock time (most recent first)
unlocked_sorted = sorted(
unlocked,
key=lambda a: a.get("unlocktime", 0),
reverse=True
)

completion_pct = (len(unlocked) / len(achievements)) * 100 if achievements else 0

# Apply pagination to each section
display_limit = limit if limit > 0 else len(achievements)
unlocked_paginated = unlocked_sorted[offset:offset + display_limit]
locked_paginated = locked[offset:offset + display_limit]

if format == "json":
data = {
"steam_id": normalized_id,
Expand All @@ -133,15 +157,24 @@ async def get_player_achievements(
"unlocked_count": len(unlocked),
"locked_count": len(locked),
"completion_percent": round(completion_pct, 1),
"achievements": [
"offset": offset,
"limit": display_limit,
"unlocked": [
{
"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
for a in unlocked_paginated
],
"locked": [
{
"api_name": a.get("apiname", ""),
"name": a.get("name", a.get("apiname", "Unknown")),
"description": a.get("description", ""),
}
for a in locked_paginated
],
}
return json.dumps(data, indent=2)
Expand All @@ -154,15 +187,13 @@ async def get_player_achievements(
"",
]

if offset > 0:
output.append(f"Showing from offset {offset}:\n")

# Show unlocked achievements first (most recent first by unlock time)
if unlocked:
unlocked_sorted = sorted(
unlocked,
key=lambda a: a.get("unlocktime", 0),
reverse=True
)
output.append(f"Unlocked ({len(unlocked)}):")
for ach in unlocked_sorted[:MAX_ACHIEVEMENTS_DISPLAY]:
output.append(f"Unlocked ({len(unlocked)} total):")
for ach in unlocked_paginated:
name = ach.get("name", ach.get("apiname", "Unknown"))
desc = ach.get("description", "")
unlock_time = ach.get("unlocktime", 0)
Expand All @@ -177,14 +208,15 @@ async def get_player_achievements(
else:
output.append(f" ✓ {name} [{unlock_str}]")

if len(unlocked) > MAX_ACHIEVEMENTS_DISPLAY:
output.append(f" ... and {len(unlocked) - MAX_ACHIEVEMENTS_DISPLAY} more unlocked")
remaining_unlocked = len(unlocked) - offset - len(unlocked_paginated)
if remaining_unlocked > 0:
output.append(f" ... and {remaining_unlocked} more unlocked (use offset to paginate)")
output.append("")

# Show a few locked achievements
# Show locked achievements
if locked:
output.append(f"Locked ({len(locked)}):")
for ach in locked[:10]:
output.append(f"Locked ({len(locked)} total):")
for ach in locked_paginated:
name = ach.get("name", ach.get("apiname", "Unknown"))
desc = ach.get("description", "")

Expand All @@ -193,8 +225,9 @@ async def get_player_achievements(
else:
output.append(f" ○ {name}")

if len(locked) > 10:
output.append(f" ... and {len(locked) - 10} more locked")
remaining_locked = len(locked) - offset - len(locked_paginated)
if remaining_locked > 0:
output.append(f" ... and {remaining_locked} more locked (use offset to paginate)")

return "\n".join(output)

Expand Down
2 changes: 1 addition & 1 deletion tests/test_steam_wishlist.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -72,7 +72,7 @@ async def test_get_wishlist_success(self, wishlist_endpoint, mock_client):
steam_id="76561198000000001"
)

assert "Steam Wishlist (2 games)" in result
assert "Steam Wishlist (2 total games)" in result
assert "Counter-Strike 2" in result
assert "Dota 2" in result

Expand Down