### Example Plugin Directory Structure

Source: https://github.com/gogcom/galaxy-integrations-python-api/blob/master/README.md

This is an example of how your plugin files and installed dependencies should be structured for deployment.

```bash
installed
└── my_integration
    ├── galaxy
    │   └── api
    ├── requests
    │   └── ...
    ├── plugin.py
    └── manifest.json
```

--------------------------------

### Control Platform Client Lifecycle

Source: https://context7.com/gogcom/galaxy-integrations-python-api/llms.txt

Implement `launch_platform_client` and `shutdown_platform_client` to allow GOG GALAXY to start and gracefully stop the third-party platform client. This example uses `subprocess.Popen` to launch and `psutil` to terminate the client process.

```python
import subprocess, psutil

async def launch_platform_client(self) -> None:
    subprocess.Popen(["C:/Program Files/Steam/Steam.exe", "-silent"])

async def shutdown_platform_client(self) -> None:
    for proc in psutil.process_iter(["name"]):
        if proc.info["name"] == "steam.exe":
            proc.terminate()
            break
```

--------------------------------

### Install Dependencies with Pip

Source: https://github.com/gogcom/galaxy-integrations-python-api/blob/master/README.md

Use this command to install third-party packages for your plugin. Ensure you specify the target directory and Python version.

```bash
pip install DEP --target DIR --implementation cp --python-version 313
```

--------------------------------

### Manage Game Launch, Install, and Uninstall

Source: https://context7.com/gogcom/galaxy-integrations-python-api/llms.txt

Implement `launch_game`, `install_game`, and `uninstall_game` to handle client requests for game management. These functions typically use `subprocess.Popen` to execute platform-specific commands.

```python
import subprocess

async def launch_game(self, game_id: str) -> None:
    subprocess.Popen(["steam", f"steam://run/{game_id}"])

async def install_game(self, game_id: str) -> None:
    subprocess.Popen(["steam", f"steam://install/{game_id}"])

async def uninstall_game(self, game_id: str) -> None:
    subprocess.Popen(["steam", f"steam://uninstall/{game_id}"])
```

--------------------------------

### Entry-point Helper for Plugin Creation

Source: https://context7.com/gogcom/galaxy-integrations-python-api/llms.txt

Top-level function that parses command-line arguments (token + port), connects the plugin to GOG GALAXY over TCP, and runs the asyncio event loop until shutdown. This example demonstrates authentication and retrieving owned games.

```python
import sys
from galaxy.api.plugin import Plugin, create_and_run_plugin
from galaxy.api.consts import Platform
from galaxy.api.types import Authentication, Game, LicenseInfo, LicenseType

class MyPlugin(Plugin):
    def __init__(self, reader, writer, token):
        super().__init__(Platform.Epic, "0.1", reader, writer, token)

    async def authenticate(self, stored_credentials=None):
        return Authentication("user_123", "GamerTag")

    async def get_owned_games(self):
        return [Game("fortnite", "Fortnite", None, LicenseInfo(LicenseType.FreeToPlay))]

# invoked as: python plugin.py <token> <port>
if __name__ == "__main__":
    create_and_run_plugin(MyPlugin, sys.argv)
```

--------------------------------

### Plugin.launch_game, Plugin.install_game, Plugin.uninstall_game

Source: https://context7.com/gogcom/galaxy-integrations-python-api/llms.txt

Provides functionality to launch, install, or uninstall a specified game. Implementing all three methods enables automatic exposure of `LaunchGame`, `InstallGame`, and `UninstallGame` features.

```APIDOC
## `Plugin.launch_game` / `install_game` / `uninstall_game` — Game management

Called by the client to launch, install, or uninstall the specified game. Implement all three to expose `LaunchGame`, `InstallGame`, and `UninstallGame` features automatically.

```python
import subprocess

async def launch_game(self, game_id: str) -> None:
    subprocess.Popen(["steam", f"steam://run/{game_id}"])

async def install_game(self, game_id: str) -> None:
    subprocess.Popen(["steam", f"steam://install/{game_id}"])

async def uninstall_game(self, game_id: str) -> None:
    subprocess.Popen(["steam", f"steam://uninstall/{game_id}"])
```
```

--------------------------------

### Plugin Lifecycle Hooks

Source: https://context7.com/gogcom/galaxy-integrations-python-api/llms.txt

Implement `handshake_complete` for post-connection setup and `tick` for regular background polling. Ensure `asyncio` is imported for asynchronous tasks.

```python
import asyncio

def handshake_complete(self):
    # Restore persisted state
    self._owned_ids = set(self.persistent_cache.get("owned_ids", []))
    self._friend_ids = set(self.persistent_cache.get("friend_ids", []))

def tick(self):
    # Spin up polling tasks if not already running
    if not hasattr(self, "_library_task") or self._library_task.done():
        self._library_task = asyncio.create_task(self._check_for_library_changes())
    if not hasattr(self, "_status_task") or self._status_task.done():
        self._status_task = asyncio.create_task(self._poll_game_statuses())

async def shutdown(self):
    await self._http_client.close()
```

--------------------------------

### Get Installed Game Local Size

Source: https://context7.com/gogcom/galaxy-integrations-python-api/llms.txt

Implement `get_local_size` to retrieve the installed size of a game in bytes. This function should prefer reading from platform-specific manifests over scanning files for accuracy and performance.

```python
from typing import Any, Optional

async def get_local_size(self, game_id: str, context: Any) -> Optional[int]:
    install_path = self._install_paths.get(game_id)
    if not install_path:
        return None
    # Read from platform-specific manifest, not file scan
    manifest = await self._read_manifest(install_path)
    return manifest.get("SizeOnDisk")  # bytes
```

--------------------------------

### Detect Installed Games

Source: https://context7.com/gogcom/galaxy-integrations-python-api/llms.txt

Implement `get_local_games` to report games installed on the user's machine, including their current state (Installed, Running, or both). Uses `_scanner` to find installed games and `_process_watcher` to check if they are running.

```python
from galaxy.api.types import LocalGame
from galaxy.api.consts import LocalGameState

async def get_local_games(self):
    local = []
    for app_id, info in self._scanner.scan_installed().items():
        state = LocalGameState.Installed
        if self._process_watcher.is_running(app_id):
            state |= LocalGameState.Running
        local.append(LocalGame(str(app_id), state))
    return local
```

--------------------------------

### Plugin.get_local_games

Source: https://context7.com/gogcom/galaxy-integrations-python-api/llms.txt

Detects and returns a list of games installed on the user's machine, including their current local game state (Installed, Running, or both).

```APIDOC
## `Plugin.get_local_games` — Detect installed games

Return a list of games currently installed on the user's machine, each with a `LocalGameState` (Installed / Running / both).

```python
from galaxy.api.types import LocalGame
from galaxy.api.consts import LocalGameState

async def get_local_games(self):
    local = []
    for app_id, info in self._scanner.scan_installed().items():
        state = LocalGameState.Installed
        if self._process_watcher.is_running(app_id):
            state |= LocalGameState.Running
        local.append(LocalGame(str(app_id), state))
    return local
```
```

--------------------------------

### Plugin.get_local_size

Source: https://context7.com/gogcom/galaxy-integrations-python-api/llms.txt

Retrieves the installed size of a game in bytes. It is recommended to use registry or manifest lookups rather than scanning files.

```APIDOC
## Plugin.get_local_size

### Description
Returns the installed size of a game in bytes. Prefers registry/manifest lookups over file scanning.

### Method
`get_local_size`

### Parameters
- **game_id** (str) - The unique identifier of the game.
- **context** (Any) - Contextual information for the operation.

### Response
#### Success Response (Optional[int])
- Returns the size of the installed game in bytes, or `None` if the install path is not found.

### Request Example
```python
size = await plugin.get_local_size("game_123", context)
if size is not None:
    print(f"Installed size: {size} bytes")
```
```

--------------------------------

### Get Subscriptions and Subscription Games

Source: https://context7.com/gogcom/galaxy-integrations-python-api/llms.txt

Implement `get_subscriptions` to retrieve available subscription tiers and `get_subscription_games` as an async generator for paginated subscription game data. Ensure proper handling of subscription details and game information.

```python
from galaxy.api.types import Subscription, SubscriptionGame
from galaxy.api.consts import SubscriptionDiscovery
from typing import Any, AsyncGenerator, List

async def get_subscriptions(self) -> List[Subscription]:
    subs = await self._backend.get_active_subscriptions()
    return [
        Subscription(
            subscription_name=s["name"],
            owned=s["active"],
            end_time=s.get("expiry_timestamp"),
            subscription_discovery=SubscriptionDiscovery.AUTOMATIC | SubscriptionDiscovery.USER_ENABLED
        )
        for s in subs
    ]
```

```python
async def get_subscription_games(
    self, subscription_name: str, context: Any
) -> AsyncGenerator[List[SubscriptionGame], None]:
    page = 1
    while True:
        games_page = await self._backend.get_subscription_catalog(subscription_name, page)
        if not games_page:
            yield None
            return
        yield [
            SubscriptionGame(
                game_title=g["title"],
                game_id=g["id"],
                start_time=g.get("added_ts"),
                end_time=g.get("leaving_ts")
            )
            for g in games_page
        ]
        page += 1
```

--------------------------------

### Integration Manifest JSON

Source: https://github.com/gogcom/galaxy-integrations-python-api/blob/master/README.md

The obligatory manifest file for a GOG GALAXY integration. Ensure the `guid` is a custom UUID, `version` matches the `Plugin` constructor, and `script` points to the entry point module.

```json
{
    "name": "Example plugin",
    "platform": "test",
    "guid": "xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx",
    "version": "0.1",
    "description": "Example plugin",
    "author": "Name",
    "email": "author@email.com",
    "url": "https://github.com/user/galaxy-plugin-example",
    "script": "plugin.py"
}

```

--------------------------------

### create_and_run_plugin - Entry-point helper

Source: https://context7.com/gogcom/galaxy-integrations-python-api/llms.txt

This top-level function handles command-line argument parsing, establishes a TCP connection between the plugin and GOG GALAXY, and manages the asyncio event loop until the integration is shut down.

```APIDOC
### `create_and_run_plugin` — Entry-point helper

Top-level function that parses command-line arguments (token + port), connects the plugin to GOG GALAXY over TCP, and runs the asyncio event loop until shutdown.

```python
import sys
from galaxy.api.plugin import Plugin, create_and_run_plugin
from galaxy.api.consts import Platform
from galaxy.api.types import Authentication, Game, LicenseInfo, LicenseType

class MyPlugin(Plugin):
    def __init__(self, reader, writer, token):
        super().__init__(Platform.Epic, "0.1", reader, writer, token)

    async def authenticate(self, stored_credentials=None):
        return Authentication("user_123", "GamerTag")

    async def get_owned_games(self):
        return [Game("fortnite", "Fortnite", None, LicenseInfo(LicenseType.FreeToPlay))]

# invoked as: python plugin.py <token> <port>
if __name__ == "__main__":
    create_and_run_plugin(MyPlugin, sys.argv)
```
```

--------------------------------

### prepare_game_library_settings_context

Source: https://context7.com/gogcom/galaxy-integrations-python-api/llms.txt

Prepares the context for game library settings import by fetching settings for a list of game IDs in a single batch request.

```APIDOC
## prepare_game_library_settings_context

### Description
Fetches library settings for a list of game IDs to be used as context for `get_game_library_settings`.

### Method
Asynchronous function

### Parameters
- **game_ids** (List[str]) - Required - A list of game IDs.

### Response
- **Any** - Context object containing library settings, typically a dictionary keyed by game ID.
```

--------------------------------

### Library Settings Import with Batched Context

Source: https://context7.com/gogcom/galaxy-integrations-python-api/llms.txt

Retrieves per-game visibility and tag metadata using `prepare_game_library_settings_context` for batched requests. `get_game_library_settings` then constructs `GameLibrarySettings` objects from the context.

```python
from galaxy.api.types import GameLibrarySettings
from typing import Any

async def prepare_game_library_settings_context(self, game_ids):
    return await self._backend.get_library_settings(game_ids)

async def get_game_library_settings(self, game_id: str, context: Any) -> GameLibrarySettings:
    settings = context.get(game_id, {})
    return GameLibrarySettings(
        game_id=game_id,
        tags=settings.get("tags", []),
        hidden=settings.get("hidden", False)
    )
```

--------------------------------

### Initialize Steam Plugin with Plugin Base Class

Source: https://context7.com/gogcom/galaxy-integrations-python-api/llms.txt

Subclass `Plugin` and call `super().__init__` with the target platform and version. All supported features are auto-detected from overridden methods; nothing extra needs to be registered.

```python
import sys
from galaxy.api.plugin import Plugin, create_and_run_plugin
from galaxy.api.consts import Platform
from galaxy.api.types import Authentication, Game, LicenseInfo, LicenseType

class SteamPlugin(Plugin):
    def __init__(self, reader, writer, token):
        super().__init__(
            Platform.Steam,   # platform this integration covers
            "1.0.0",          # integration version
            reader, writer, token
        )
        self._games_cache = []

def main():
    create_and_run_plugin(SteamPlugin, sys.argv)

if __name__ == "__main__":
    main()
```

--------------------------------

### Basic Plugin Implementation in Python

Source: https://github.com/gogcom/galaxy-integrations-python-api/blob/master/README.md

A minimal GOG GALAXY integration plugin. Requires overriding `authenticate` and `get_owned_games`. Ensure the platform is chosen from `galaxy.api.consts.Platform` and other methods are implemented as needed.

```python
import sys
from galaxy.api.plugin import Plugin, create_and_run_plugin
from galaxy.api.consts import Platform
from galaxy.api.types import Authentication, Game, LicenseInfo, LicenseType


class PluginExample(Plugin):
    def __init__(self, reader, writer, token):
        super().__init__(
            Platform.Test,  # choose platform from available list
            "0.1",  # version
            reader,
            writer,
            token
        )

    # implement methods

    # required
    async def authenticate(self, stored_credentials=None):
        return Authentication('test_user_id', 'Test User Name')

    # required
    async def get_owned_games(self):
        return [
            Game('test', 'The Test', None, LicenseInfo(LicenseType.SinglePurchase))
        ]


def main():
    create_and_run_plugin(PluginExample, sys.argv)

# run plugin event loop
if __name__ == "__main__":
    main()

```

--------------------------------

### Plugin.launch_platform_client, Plugin.shutdown_platform_client

Source: https://context7.com/gogcom/galaxy-integrations-python-api/llms.txt

Enables GOG GALAXY to initiate and gracefully terminate the third-party platform client.

```APIDOC
## `Plugin.launch_platform_client` / `shutdown_platform_client` — Client lifecycle

Allow GOG GALAXY to start and gracefully stop the third-party platform client itself.

```python
import subprocess, psutil

async def launch_platform_client(self) -> None:
    subprocess.Popen(["C:/Program Files/Steam/Steam.exe", "-silent"])

async def shutdown_platform_client(self) -> None:
    for proc in psutil.process_iter(["name"]):
        if proc.info["name"] == "steam.exe":
            proc.terminate()
            break
```
```

--------------------------------

### Plugin.get_os_compatibility

Source: https://context7.com/gogcom/galaxy-integrations-python-api/llms.txt

Retrieves the operating system compatibility flags for a given game ID. It fetches game information to determine compatibility.

```APIDOC
## Plugin.get_os_compatibility

### Description
Determines and returns the OS compatibility flags for a given game.

### Method
Asynchronous function

### Parameters
- **game_id** (str) - Required - The ID of the game.
- **context** (Any) - Required - Pre-fetched context containing game information.

### Response
- **Optional[OSCompatibility]** - An OSCompatibility bitflag indicating supported operating systems, or None if game info is not found.
```

--------------------------------

### prepare_game_times_context

Source: https://context7.com/gogcom/galaxy-integrations-python-api/llms.txt

Prepares the context for game time import by fetching all game time data in a single batch request.

```APIDOC
## prepare_game_times_context

### Description
Fetches all game time data to be used as context for `get_game_time`.

### Method
Asynchronous function

### Parameters
None

### Response
- **Any** - Context object containing game time data, typically a dictionary keyed by game ID.
```

--------------------------------

### Game Time Import with Batched Context

Source: https://context7.com/gogcom/galaxy-integrations-python-api/llms.txt

Utilize `prepare_game_times_context` for batched prefetching of game time data. `get_game_time` then extracts playtime and last-played timestamps from the provided context.

```python
from galaxy.api.types import GameTime
from typing import Any

async def prepare_game_times_context(self, game_ids):
    return await self._backend.get_all_game_times()  # {game_id: {minutes, last_play}}

async def get_game_time(self, game_id: str, context: Any) -> GameTime:
    data = context.get(game_id, {})
    return GameTime(
        game_id=game_id,
        time_played=data.get("minutes"),
        last_played_time=data.get("last_play")  # unix timestamp
    )

def game_times_import_complete(self):
    self.push_cache()
```

--------------------------------

### Plugin.authenticate - User authentication (required)

Source: https://context7.com/gogcom/galaxy-integrations-python-api/llms.txt

This method is called by GOG GALAXY during startup to authenticate the user. It should return an `Authentication` object upon successful login or a `NextStep` object to initiate a browser-based login flow. `stored_credentials` can be used to re-authenticate using saved credentials from a previous session.

```APIDOC
### `Plugin.authenticate` — User authentication (required)

Called by GOG GALAXY on startup. Return `Authentication` when login is complete, or `NextStep` to redirect the built-in browser to an OAuth/web-login page. `stored_credentials` carries credentials saved from the previous session.

```python
from galaxy.api.types import Authentication, NextStep, Cookie
from galaxy.api.errors import InvalidCredentials

PARAMS = {
    "window_title": "Login to Steam",
    "window_width": 900,
    "window_height": 700,
    "start_uri": "https://steamcommunity.com/login",
    "end_uri_regex": r"^https://steamcommunity\.com/profiles/.*"
}

async def authenticate(self, stored_credentials=None):
    if stored_credentials:
        try:
            user = await self._backend.login_with_token(stored_credentials["token"])
            return Authentication(user.id, user.name)
        except Exception:
            raise InvalidCredentials()
    # No saved credentials — open browser-based login
    return NextStep(
        "web_session",
        PARAMS,
        cookies=[Cookie("SteamLoginSecure", "", ".steamcommunity.com")]
    )
```
```

--------------------------------

### prepare_user_presence_context

Source: https://context7.com/gogcom/galaxy-integrations-python-api/llms.txt

Prepares the context for user presence import by fetching presence data for a list of user IDs in a single batch request.

```APIDOC
## prepare_user_presence_context

### Description
Fetches presence data for a list of user IDs to be used as context for `get_user_presence`.

### Method
Asynchronous function

### Parameters
- **user_id_list** (List[str]) - Required - A list of user IDs.

### Response
- **Any** - Context object containing presence data, typically a dictionary keyed by user ID.
```

--------------------------------

### Plugin.get_game_library_settings

Source: https://context7.com/gogcom/galaxy-integrations-python-api/llms.txt

Retrieves the visibility and tag metadata for a specific game within the GOG GALAXY library. It uses context prepared by `prepare_game_library_settings_context`.

```APIDOC
## Plugin.get_game_library_settings

### Description
Retrieves library settings (tags and hidden status) for a specific game using pre-fetched context.

### Method
Asynchronous function

### Parameters
- **game_id** (str) - Required - The ID of the game.
- **context** (Any) - Required - Pre-fetched context containing library settings for multiple games.

### Response
- **GameLibrarySettings** - A GameLibrarySettings object containing `tags` and `hidden` status.
```

--------------------------------

### Handle User Authentication with Stored Credentials or Web Login

Source: https://context7.com/gogcom/galaxy-integrations-python-api/llms.txt

Called by GOG GALAXY on startup. This method handles user authentication, either by using stored credentials from a previous session or by initiating a browser-based login flow. It returns `Authentication` on success or `NextStep` to redirect to a login page.

```python
from galaxy.api.types import Authentication, NextStep, Cookie
from galaxy.api.errors import InvalidCredentials

PARAMS = {
    "window_title": "Login to Steam",
    "window_width": 900,
    "window_height": 700,
    "start_uri": "https://steamcommunity.com/login",
    "end_uri_regex": r"^https://steamcommunity\.com/profiles/.*"
}

async def authenticate(self, stored_credentials=None):
    if stored_credentials:
        try:
            user = await self._backend.login_with_token(stored_credentials["token"])
            return Authentication(user.id, user.name)
        except Exception:
            raise InvalidCredentials()
    # No saved credentials — open browser-based login
    return NextStep(
        "web_session",
        PARAMS,
        cookies=[Cookie("SteamLoginSecure", "", ".steamcommunity.com")]
    )
```

--------------------------------

### prepare_achievements_context

Source: https://context7.com/gogcom/galaxy-integrations-python-api/llms.txt

Prepares the context for achievements import by fetching all achievements for a list of game IDs in a single batch request.

```APIDOC
## prepare_achievements_context

### Description
Fetches all achievements for a list of game IDs to be used as context for `get_unlocked_achievements`.

### Method
Asynchronous function

### Parameters
- **game_ids** (List[str]) - Required - A list of game IDs.

### Response
- **Any** - Context object containing achievement data, typically a dictionary keyed by game ID.
```

--------------------------------

### Real-time Game-List Updates

Source: https://context7.com/gogcom/galaxy-integrations-python-api/llms.txt

Implement `add_game`, `remove_game`, and `update_game` to push incremental game-list changes to the client. This avoids the need for a full re-import by handling additions, removals, and updates based on library changes.

```python
from galaxy.api.types import Game, LicenseInfo
from galaxy.api.consts import LicenseType

async def _check_for_library_changes(self):
    current_ids = {g["appid"] for g in await self._backend.fetch_library()}
    cached_ids = set(self._owned_cache.keys())
    for appid in current_ids - cached_ids:
        info = await self._backend.get_game_info(appid)
        new_game = Game(str(appid), info["name"], None, LicenseInfo(LicenseType.SinglePurchase))
        self._owned_cache[appid] = new_game
        self.add_game(new_game)
    for appid in cached_ids - current_ids:
        del self._owned_cache[appid]
        self.remove_game(str(appid))
```

--------------------------------

### Handle Multi-Step Web Authentication

Source: https://context7.com/gogcom/galaxy-integrations-python-api/llms.txt

Implement `pass_login_credentials` to manage multi-stage authentication flows, returning `NextStep` for subsequent authentication stages or `Authentication` upon success. This is called after the built-in browser reaches the `end_uri_regex`.

```python
from galaxy.api.types import Authentication, NextStep

async def pass_login_credentials(self, step, credentials, cookies):
    # credentials["end_uri"] is the URL the browser landed on
    if "two_factor" in credentials.get("end_uri", ""):
        return NextStep("web_session", {
            "window_title": "2FA required",
            "window_width": 500,
            "window_height": 400,
            "start_uri": credentials["end_uri"],
            "end_uri_regex": r"^https://steamcommunity\.com/profiles/.*"
        })
    session_cookie = next(c for c in cookies if c["name"] == "SteamLoginSecure")
    user = await self._backend.exchange_cookie(session_cookie["value"])
    self.store_credentials({"token": user.token})
    return Authentication(user.id, user.name)
```

--------------------------------

### User Presence with Batched Context

Source: https://context7.com/gogcom/galaxy-integrations-python-api/llms.txt

Fetches user presence status (online/away/offline) and in-game information using batched requests via `prepare_user_presence_context`. `get_user_presence` maps backend states to `PresenceState` enum.

```python
from galaxy.api.types import UserPresence
from galaxy.api.consts import PresenceState
from typing import Any

async def prepare_user_presence_context(self, user_id_list):
    return await self._backend.get_presence_bulk(user_id_list)  # {user_id: {...}}

async def get_user_presence(self, user_id: str, context: Any) -> UserPresence:
    data = context.get(user_id, {})
    state_map = {1: PresenceState.Online, 3: PresenceState.Away}
    return UserPresence(
        presence_state=state_map.get(data.get("personastate", 0), PresenceState.Offline),
        game_id=str(data["gameid"]) if data.get("gameid") else None,
        game_title=data.get("gameextrainfo"),
        in_game_status=data.get("gamestateflags"),
        full_status=data.get("personaname")
    )
```

--------------------------------

### Achievements Import with Batched Context

Source: https://context7.com/gogcom/galaxy-integrations-python-api/llms.txt

Use `prepare_achievements_context` to fetch all achievements in a single batch request before iterating through games. `get_unlocked_achievements` then processes this context to return achievement data.

```python
from galaxy.api.types import Achievement
from typing import Any, List

async def prepare_achievements_context(self, game_ids: List[str]) -> Any:
    # Fetch all stats in one request, cache by game id
    return await self._backend.get_all_achievements(game_ids)

async def get_unlocked_achievements(self, game_id: str, context: Any) -> List[Achievement]:
    raw_achs = context.get(game_id, [])
    return [
        Achievement(
            unlock_time=int(a["unlocktime"]),
            achievement_id=a["apiname"],
            achievement_name=a["name"]
        )
        for a in raw_achs
        if a.get("achieved")
    ]

def achievements_import_complete(self):
    self.push_cache()   # persist updated cache
```

--------------------------------

### Import Owned Games

Source: https://context7.com/gogcom/galaxy-integrations-python-api/llms.txt

Implement `get_owned_games` to return a list of all games owned by the authenticated user. Each `Game` object requires a unique `game_id`. Raises `AuthenticationRequired` if the user is not authenticated.

```python
from galaxy.api.types import Game, LicenseInfo, Dlc
from galaxy.api.consts import LicenseType
from galaxy.api.errors import AuthenticationRequired

async def get_owned_games(self):
    if not self._authenticated:
        raise AuthenticationRequired()
    raw = await self._backend.fetch_library()
    return [
        Game(
            game_id=str(item["appid"]),
            game_title=item["name"],
            dlcs=[
                Dlc(str(d["id"]), d["name"], LicenseInfo(LicenseType.SinglePurchase))
                for d in item.get("dlcs", [])
            ],
            license_info=LicenseInfo(LicenseType.SinglePurchase)
        )
        for item in raw["games"]
    ]
```

--------------------------------

### Plugin.get_subscriptions and get_subscription_games

Source: https://context7.com/gogcom/galaxy-integrations-python-api/llms.txt

Manages subscription tiers and retrieves games associated with a subscription. `get_subscriptions` returns available subscription tiers, while `get_subscription_games` is an async generator that yields pages of `SubscriptionGame` objects.

```APIDOC
## Plugin.get_subscriptions

### Description
Returns a list of available subscription tiers.

### Method
`get_subscriptions`

### Parameters
None

### Response
#### Success Response (List[Subscription])
- **subscription_name** (str) - The name of the subscription tier.
- **owned** (bool) - Indicates if the user owns this subscription.
- **end_time** (Optional[int]) - The timestamp when the subscription ends, if applicable.
- **subscription_discovery** (SubscriptionDiscovery) - Flags indicating how the subscription was discovered.

### Request Example
```python
await plugin.get_subscriptions()
```

## Plugin.get_subscription_games

### Description
An async generator that yields pages of `SubscriptionGame` objects for a given subscription.

### Method
`get_subscription_games`

### Parameters
- **subscription_name** (str) - The name of the subscription to retrieve games for.
- **context** (Any) - Contextual information for the operation.

### Response
#### Success Response (AsyncGenerator[List[SubscriptionGame], None])
Yields pages of `SubscriptionGame` objects. Each page is a list of `SubscriptionGame`.
- **game_title** (str) - The title of the game.
- **game_id** (str) - The unique identifier for the game.
- **start_time** (Optional[int]) - The timestamp when the game was added to the subscription, if applicable.
- **end_time** (Optional[int]) - The timestamp when the game will leave the subscription, if applicable.

### Request Example
```python
async for games_page in plugin.get_subscription_games("premium_tier", context):
    if games_page:
        for game in games_page:
            print(game.game_title)
```
```

--------------------------------

### Managed HTTP Client Session

Source: https://context7.com/gogcom/galaxy-integrations-python-api/llms.txt

Use `create_client_session` and `create_tcp_connector` for pre-configured `aiohttp` sessions. `handle_exception` automatically translates common network errors and HTTP status codes into `galaxy.api.errors`.

```python
from galaxy.http import create_client_session, create_tcp_connector, handle_exception
from galaxy.api.errors import AuthenticationRequired, BackendError
import aiohttp

class BackendClient:
    BASE_URL = "https://api.example-platform.com"

    def __init__(self):
        self._session = create_client_session(
            headers={"User-Agent": "MyPlugin/1.0", "Accept": "application/json"},
            connector=create_tcp_connector(limit=10),
        )

    async def close(self):
        await self._session.close()

    async def get_library(self, user_id: str) -> dict:
        with handle_exception():
            resp = await self._session.get(
                f"{self.BASE_URL}/users/{user_id}/games"
            )
            return await resp.json()
        # handle_exception() automatically maps:
        #   asyncio.TimeoutError        -> BackendTimeout
        #   401 HTTP response           -> AuthenticationRequired
        #   403 HTTP response           -> AccessDenied
        #   429 HTTP response           -> TooManyRequests
        #   5xx HTTP responses          -> BackendError
        #   ServerDisconnectedError     -> BackendNotAvailable
```

--------------------------------

### State Management Notifications

Source: https://context7.com/gogcom/galaxy-integrations-python-api/llms.txt

Utilize `store_credentials`, `lost_authentication`, and `push_cache` for state management. `store_credentials` persists tokens after login, `lost_authentication` signals auth loss, and `push_cache` saves arbitrary data between sessions.

```python
# store_credentials: called after successful web-login to persist token
async def pass_login_credentials(self, step, credentials, cookies):
    token = await self._exchange_cookies_for_token(cookies)
    self.store_credentials({"access_token": token["access"], "refresh": token["refresh"]})
    return Authentication(token["user_id"], token["username"])
```

```python
# lost_authentication: called when a refresh token expires mid-session
async def _make_api_request(self, url):
    try:
        return await self._session.get(url)
    except Exception:
        self.lost_authentication()   # client will re-trigger authenticate()
```

```python
# push_cache: save arbitrary data between sessions
def handshake_complete(self):
    self._game_cache = self.persistent_cache.get("games", {})

def tick(self):
    if self._cache_dirty:
        self.persistent_cache["games"] = self._game_cache
        self.push_cache()
        self._cache_dirty = False
```

--------------------------------

### Push Real-time Game State Changes

Source: https://context7.com/gogcom/galaxy-integrations-python-api/llms.txt

Implement `update_local_game_status` to notify GOG GALAXY about changes in a game's local state, such as download completion or starting/stopping play. This is typically called from the `tick` loop, often by a background task like `_poll_running_games`.

```python
from galaxy.api.types import LocalGame
from galaxy.api.consts import LocalGameState
import asyncio

async def _poll_running_games(self):
    while True:
        for game_id, proc in self._tracked_processes.items():
            running = proc.poll() is None
            new_state = (LocalGameState.Installed | LocalGameState.Running) if running \
                        else LocalGameState.Installed
            if new_state != self._last_states.get(game_id):
                self.update_local_game_status(LocalGame(game_id, new_state))
                self._last_states[game_id] = new_state
        await asyncio.sleep(5)

def tick(self):
    if not hasattr(self, "_poll_task") or self._poll_task.done():
        self._poll_task = asyncio.create_task(self._poll_running_games())
```

--------------------------------

### Plugin.get_user_presence

Source: https://context7.com/gogcom/galaxy-integrations-python-api/llms.txt

Retrieves the presence status (online, away, offline) and in-game information for a given user ID. It uses context prepared by `prepare_user_presence_context`.

```APIDOC
## Plugin.get_user_presence

### Description
Retrieves the presence status and in-game information for a specific user using pre-fetched context.

### Method
Asynchronous function

### Parameters
- **user_id** (str) - Required - The ID of the user.
- **context** (Any) - Required - Pre-fetched context containing presence data for multiple users.

### Response
- **UserPresence** - A UserPresence object containing presence state, game ID, and other status details.
```

--------------------------------

### Plugin.get_unlocked_achievements

Source: https://context7.com/gogcom/galaxy-integrations-python-api/llms.txt

Imports unlocked achievements for a given game. It utilizes a context prepared by `prepare_achievements_context` for batched API requests.

```APIDOC
## Plugin.get_unlocked_achievements

### Description
Fetches unlocked achievements for a specific game using pre-fetched context data.

### Method
Asynchronous function

### Parameters
- **game_id** (str) - Required - The ID of the game.
- **context** (Any) - Required - Pre-fetched context containing achievement data for multiple games.

### Response
- **List[Achievement]** - A list of Achievement objects representing the unlocked achievements.
```

--------------------------------

### Plugin.pass_login_credentials

Source: https://context7.com/gogcom/galaxy-integrations-python-api/llms.txt

Handles multi-step web authentication, processing credentials and cookies to authenticate users and manage multi-stage login flows.

```APIDOC
## `Plugin.pass_login_credentials` — Multi-step web authentication

Called after the built-in browser reaches the `end_uri_regex` URL from a previous `NextStep`. Receives the final URL and any extracted cookies. Can return another `NextStep` for multi-stage flows.

```python
from galaxy.api.types import Authentication, NextStep

async def pass_login_credentials(self, step, credentials, cookies):
    # credentials["end_uri"] is the URL the browser landed on
    if "two_factor" in credentials.get("end_uri", ""):
        return NextStep("web_session", {
            "window_title": "2FA required",
            "window_width": 500,
            "window_height": 400,
            "start_uri": credentials["end_uri"],
            "end_uri_regex": r"^https://steamcommunity\.com/profiles/.*"
        })
    session_cookie = next(c for c in cookies if c["name"] == "SteamLoginSecure")
    user = await self._backend.exchange_cookie(session_cookie["value"])
    self.store_credentials({"token": user.token})
    return Authentication(user.id, user.name)
```
```

--------------------------------

### Plugin Base Class - galaxy.api.plugin.Plugin

Source: https://context7.com/gogcom/galaxy-integrations-python-api/llms.txt

The Plugin base class is the foundation for creating GOG Galaxy integrations. Developers subclass this class and override methods to define the integration's features. Integrations are auto-detected based on overridden methods.

```APIDOC
## Plugin Base Class — `galaxy.api.plugin.Plugin`

### `Plugin.__init__` — Initialize a platform integration

Subclass `Plugin` and call `super().__init__` with the target platform and version. All supported features are auto-detected from overridden methods; nothing extra needs to be registered.

```python
import sys
from galaxy.api.plugin import Plugin, create_and_run_plugin
from galaxy.api.consts import Platform
from galaxy.api.types import Authentication, Game, LicenseInfo, LicenseType

class SteamPlugin(Plugin):
    def __init__(self, reader, writer, token):
        super().__init__(
            Platform.Steam,   # platform this integration covers
            "1.0.0",          # integration version
            reader, writer, token
        )
        self._games_cache = []

def main():
    create_and_run_plugin(SteamPlugin, sys.argv)

if __name__ == "__main__":
    main()
```
```

--------------------------------

### OS Compatibility Flags

Source: https://context7.com/gogcom/galaxy-integrations-python-api/llms.txt

Determines the operating systems a game is compatible with. Fetches game info and applies bitflags from `OSCompatibility` based on platform availability (Windows, macOS, Linux).

```python
from galaxy.api.consts import OSCompatibility
from typing import Any, Optional

async def get_os_compatibility(self, game_id: str, context: Any) -> Optional[OSCompatibility]:
    info = await self._backend.get_game_info(game_id)
    if info is None:
        return None
    compat = OSCompatibility.Windows  # always Windows
    if info.get("mac"):
        compat |= OSCompatibility.MacOS
    if info.get("linux"):
        compat |= OSCompatibility.Linux
    return compat
```

--------------------------------

### Plugin.add_game, remove_game, and update_game

Source: https://context7.com/gogcom/galaxy-integrations-python-api/llms.txt

Enables real-time updates to the game list by pushing incremental changes to the client, avoiding the need for a full re-import.

```APIDOC
## Plugin.add_game

### Description
Adds a new game to the client's game list. Used for real-time updates.

### Method
`add_game`

### Parameters
- **game** (Game) - A `Game` object representing the game to add.

### Request Example
```python
# Example usage within _check_for_library_changes
new_game = Game(str(appid), info["name"], None, LicenseInfo(LicenseType.SinglePurchase))
self.add_game(new_game)
```

## Plugin.remove_game

### Description
Removes a game from the client's game list. Used for real-time updates.

### Method
`remove_game`

### Parameters
- **game_id** (str) - The unique identifier of the game to remove.

### Request Example
```python
# Example usage within _check_for_library_changes
self.remove_game(str(appid))
```

## Plugin.update_game

### Description
Updates information for an existing game in the client's game list. Used for real-time updates.

### Method
`update_game`

### Parameters
- **game** (Game) - A `Game` object with updated information for the game.

### Request Example
```python
# This method is declared but not shown in use in the provided source.
# Example placeholder:
# await plugin.update_game(updated_game_object)
```
```

--------------------------------

### Plugin.store_credentials, lost_authentication, and push_cache

Source: https://context7.com/gogcom/galaxy-integrations-python-api/llms.txt

Handles state management notifications, including storing credentials after login, signaling authentication loss, and persisting cache data between sessions.

```APIDOC
## Plugin.store_credentials

### Description
Stores authentication credentials, typically called after a successful web login to persist tokens.

### Method
`store_credentials`

### Parameters
- **credentials** (dict) - A dictionary containing authentication tokens (e.g., `{"access_token": "...", "refresh": "..."}`).

### Request Example
```python
# Example usage within pass_login_credentials
self.store_credentials({"access_token": token["access"], "refresh": token["refresh"]})
```

## Plugin.lost_authentication

### Description
Signals that authentication has been lost, typically called when a refresh token expires mid-session. This prompts the client to re-trigger the authentication process.

### Method
`lost_authentication`

### Parameters
None

### Request Example
```python
# Example usage within _make_api_request error handling
except Exception:
    self.lost_authentication()
```

## Plugin.push_cache

### Description
Persists arbitrary data to a cache that is saved between sessions. This is used to save data that should be available the next time the application runs.

### Method
`push_cache`

### Parameters
None

### Request Example
```python
# Example usage within tick to save game cache
self.persistent_cache["games"] = self._game_cache
self.push_cache()
```
```

--------------------------------

### Plugin.get_game_time

Source: https://context7.com/gogcom/galaxy-integrations-python-api/llms.txt

Retrieves the total playtime and last-played timestamp for a specific game. It uses context prepared by `prepare_game_times_context`.

```APIDOC
## Plugin.get_game_time

### Description
Retrieves the playtime and last played timestamp for a given game using pre-fetched context.

### Method
Asynchronous function

### Parameters
- **game_id** (str) - Required - The ID of the game.
- **context** (Any) - Required - Pre-fetched context containing game time data for multiple games.

### Response
- **GameTime** - A GameTime object containing `time_played` and `last_played_time`.
```

--------------------------------

### Galaxy API Constants Module

Source: https://github.com/gogcom/galaxy-integrations-python-api/blob/master/docs/source/galaxy.api.rst

Documentation for the galaxy.api.consts module, listing its members, undocumented members, and inheritance.

```APIDOC
## Module: galaxy.api.consts

This module contains constants used throughout the Galaxy API.

### Members:
- :py:mod:`galaxy.api.consts`

### Undocumented Members:
- All undocumented members are included.

### Inheritance:
- :py:class:`Feature`
```

--------------------------------

### Lifecycle hooks

Source: https://context7.com/gogcom/galaxy-integrations-python-api/llms.txt

`handshake_complete` fires once after the client connection is established. `tick` is called every ~1 second for non-blocking background polling tasks.

```APIDOC
## `Plugin.tick` + `Plugin.handshake_complete`

### Description
`handshake_complete` fires once after the client connection is established (cache is available). `tick` is called every ~1 second for non-blocking background polling tasks.

### Usage
```python
import asyncio

def handshake_complete(self):
    # Restore persisted state
    self._owned_ids = set(self.persistent_cache.get("owned_ids", []))
    self._friend_ids = set(self.persistent_cache.get("friend_ids", []))

def tick(self):
    # Spin up polling tasks if not already running
    if not hasattr(self, "_library_task") or self._library_task.done():
        self._library_task = asyncio.create_task(self._check_for_library_changes())
    if not hasattr(self, "_status_task") or self._status_task.done():
        self._status_task = asyncio.create_task(self._poll_game_statuses())

async def shutdown(self):
    await self._http_client.close()
```
```

--------------------------------

### Managed HTTP client

Source: https://context7.com/gogcom/galaxy-integrations-python-api/llms.txt

Factory functions that produce pre-configured `aiohttp` sessions with SSL, timeouts, and automatic error translation to `galaxy.api.errors` types.

```APIDOC
## HTTP Utilities — `galaxy.http`

### `create_client_session` / `create_tcp_connector` / `handle_exception`

### Description
Factory functions that produce pre-configured `aiohttp` sessions with SSL, timeouts, and automatic error translation to `galaxy.api.errors` types.

### Usage
```python
from galaxy.http import create_client_session, create_tcp_connector, handle_exception
from galaxy.api.errors import AuthenticationRequired, BackendError
import aiohttp

class BackendClient:
    BASE_URL = "https://api.example-platform.com"

    def __init__(self):
        self._session = create_client_session(
            headers={"User-Agent": "MyPlugin/1.0", "Accept": "application/json"},
            connector=create_tcp_connector(limit=10),
        )

    async def close(self):
        await self._session.close()

    async def get_library(self, user_id: str) -> dict:
        with handle_exception():
            resp = await self._session.get(
                f"{self.BASE_URL}/users/{user_id}/games"
            )
            return await resp.json()
        # handle_exception() automatically maps:
        #   asyncio.TimeoutError        -> BackendTimeout
        #   401 HTTP response           -> AuthenticationRequired
        #   403 HTTP response           -> AccessDenied
        #   429 HTTP response           -> TooManyRequests
        #   5xx HTTP responses          -> BackendError
        #   ServerDisconnectedError     -> BackendNotAvailable
```
```

--------------------------------

### Push Real-time Data Updates

Source: https://context7.com/gogcom/galaxy-integrations-python-api/llms.txt

Use these methods to push single achievement unlocks, updated game time, or user presence changes to the client without a full import. Ensure you have the necessary types imported.

```python
from galaxy.api.types import Achievement, GameTime, UserPresence
from galaxy.api.consts import PresenceState

# Notify about a freshly unlocked achievement
self.unlock_achievement("game_456", Achievement(
    unlock_time=1700000000,
    achievement_id="WIN_FIRST_BATTLE",
    achievement_name="First Blood"
))

# Update game time after a play session ends
self.update_game_time(GameTime(
    game_id="game_456",
    time_played=320,           # total minutes
    last_played_time=1700000000
))

# Update a friend's presence
self.update_user_presence("friend_789", UserPresence(
    presence_state=PresenceState.Online,
    game_id="game_456",
    game_title="Witcher 3",
    in_game_status="In Combat"
))
```

--------------------------------

### Galaxy API Plugin Module

Source: https://github.com/gogcom/galaxy-integrations-python-api/blob/master/docs/source/galaxy.api.rst

Documentation for the galaxy.api.plugin module, listing its members and excluding specific ones.

```APIDOC
## Module: galaxy.api.plugin

This module provides functionalities related to plugins within the Galaxy API.

### Members:
- :py:mod:`galaxy.api.plugin`

### Excluded Members:
- JSONEncoder
- features
```

--------------------------------

### Real-time Friends Updates

Source: https://context7.com/gogcom/galaxy-integrations-python-api/llms.txt

Implement `add_friend`, `remove_friend`, and `update_friend_info` to push incremental changes to the friends list and friend metadata. This ensures the client reflects real-time updates such as avatar or profile URL changes.

```python
from galaxy.api.types import UserInfo

async def _sync_friends(self):
    remote = {f["id"]: f for f in await self._backend.get_friends()}
    cached = self._friends_cache

    for uid, data in remote.items():
        if uid not in cached:
            self.add_friend(UserInfo(uid, data["name"], data.get("avatar")))
        elif cached[uid]["name"] != data["name"]:
            self.update_friend_info(UserInfo(uid, data["name"], data.get("avatar")))
    for uid in list(cached):
        if uid not in remote:
            self.remove_friend(uid)
    self._friends_cache = remote
```

--------------------------------

### Patch async plugin methods with async_raise

Source: https://context7.com/gogcom/galaxy-integrations-python-api/llms.txt

Use `async_raise` to mock an asynchronous exception for a plugin method. This is essential for testing how your integration handles errors and unexpected backend behavior.

```python
import pytest
from unittest.mock import MagicMock, patch
from galaxy.unittest.mock import async_raise
from galaxy.api.errors import BackendNotAvailable

@pytest.fixture
def plugin(reader, writer):
    return MyPlugin(reader, writer, "handshake-token")

@pytest.mark.asyncio
async def test_get_owned_games_backend_down(plugin):
    with patch.object(plugin._backend, "fetch_library",
                      side_effect=async_raise(BackendNotAvailable())):
        with pytest.raises(BackendNotAvailable):
            await plugin.get_owned_games()
```