### Install betterKickAPI with pip

Source: https://github.com/benjas333/pykickapi/blob/master/README.md

Install the library using pip.

```bash
pip install betterKickAPI
```

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

### Install betterKickAPI with uv

Source: https://github.com/benjas333/pykickapi/blob/master/README.md

Install the library using uv.

```bash
uv add betterKickAPI
```

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

### Install betterKickAPI

Source: https://context7.com/benjas333/pykickapi/llms.txt

Install the library using pip or uv. Ensure you are using Python 3.9+.

```bash
pip install betterKickAPI
```

```bash
uv add betterKickAPI
```

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

### App Authentication Setup

Source: https://github.com/benjas333/pykickapi/blob/master/README.md

Initialize the Kick API instance using App Authentication. Requires 'APP_ID' and 'APP_SECRET'.

```python
from betterKickAPI.kick import Kick
kick = await Kick('APP_ID', 'APP_SECRET')
```

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

### Set Up Kick EventSub Webhook Server

Source: https://context7.com/benjas333/pykickapi/llms.txt

Use `KickWebhook` to start an internal HTTP server that receives, verifies, and dispatches Kick EventSub webhook events. This requires setting up listeners for specific events and starting the webhook server. Ensure `auto_fetch_public_key` is set to `True` for automatic public key retrieval.

```python
import asyncio
from betterKickAPI.kick import Kick
from betterKickAPI.oauth import UserAuthenticationStorageHelper
from betterKickAPI.eventsub.webhook import KickWebhook, SSLOptions
from betterKickAPI.eventsub.events import ChatMessageEvent, ChannelFollowEvent, LivestreamStatusUpdatedEvent
from betterKickAPI.types import OAuthScope

BROADCASTER_ID = 668  # Your channel's broadcaster_user_id

async def on_chat_message(event: ChatMessageEvent):
    print(f"[{event.broadcaster.username}] {event.sender.username}: {event.content}")
    if event.emotes:
        for emote in event.emotes:
            print(f"  Emote: {emote.emote_id}")

async def on_follow(event: ChannelFollowEvent):
    print(f"New follower: {event.follower.username} followed {event.broadcaster.username}")

async def on_stream_status(event: LivestreamStatusUpdatedEvent):
    state = "LIVE" if event.is_live else "OFFLINE"
    print(f"Stream {state}: '{event.title}'")

async def main():
    kick = await Kick("YOUR_APP_ID", "YOUR_APP_SECRET")

    # Optional: use SSL
    # ssl_opts = SSLOptions(cert_file_name="cert.pem", key_file_name="key.pem")

    webhook = KickWebhook(
        kick,
        auto_fetch_public_key=True,  # fetches key from API automatically
        msg_id_history_max_length=100,
    )

    # Register event listeners (subscribes and maps callbacks)
    await webhook.listen_chat_message_sent(BROADCASTER_ID, on_chat_message)
    await webhook.listen_channel_follow(BROADCASTER_ID, on_follow)
    await webhook.listen_livestream_status_updated(BROADCASTER_ID, on_stream_status)

    # Start the internal webhook server
    webhook.start(port=3330, host_binding="0.0.0.0", endpoint="/callback")
    print("Webhook server running on port 3330")

    try:
        await asyncio.sleep(3600)  # run for 1 hour
    finally:
        await webhook.stop()  # auto-unsubscribes all known subscriptions
        await kick.close()

asyncio.run(main())
```

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

### User Authentication with OAuth Scopes

Source: https://context7.com/benjas333/pykickapi/llms.txt

Authenticate a user and set their authentication token and scopes for the Kick client. This example demonstrates how to define and use various `OAuthScope` permissions.

```python
from betterKickAPI.types import OAuthScope

# Available scopes:
# OAuthScope.USER_READ        — "user:read"       — View user info
# OAuthScope.CHANNEL_READ     — "channel:read"    — View channel info
# OAuthScope.CHANNEL_WRITE    — "channel:write"   — Update stream metadata
# OAuthScope.CHAT_WRITE       — "chat:write"      — Send chat messages
# OAuthScope.STREAMKEY_READ   — "streamkey:read"  — Read stream key/URL
# OAuthScope.EVENTS_SUBSCRIBE — "events:subscribe"— Subscribe to events
# OAuthScope.MODERATION_BAN   — "moderation:ban"  — Execute bans/timeouts
# OAuthScope.KICKS_READ       — "kicks:read"      — View KICKs leaderboards

from betterKickAPI.kick import Kick
from betterKickAPI.oauth import UserAuthenticator

async def main():
    kick = await Kick("YOUR_APP_ID", "YOUR_APP_SECRET")
    scopes = [
        OAuthScope.USER_READ,
        OAuthScope.CHANNEL_READ,
        OAuthScope.CHANNEL_WRITE,
        OAuthScope.CHAT_WRITE,
        OAuthScope.EVENTS_SUBSCRIBE,
        OAuthScope.MODERATION_BAN,
        OAuthScope.KICKS_READ,
    ]
    auth = UserAuthenticator(kick, scopes)
    token, refresh_token = await auth.authenticate()
    await kick.set_user_authentication(token=token, scope=scopes, refresh_token=refresh_token)
    print("Authorized scopes:", [str(s) for s in kick.user_auth_scope])

```

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

### KickWebhook

Source: https://context7.com/benjas333/pykickapi/llms.txt

Starts an internal HTTP server to receive, verify, and dispatch Kick EventSub webhook events to callback functions.

```APIDOC
## `KickWebhook` — EventSub Webhook Client

Starts an internal HTTP server (via `socketify`) in a separate process that receives, signature-verifies, deduplicates, and dispatches Kick EventSub webhook events to async or sync callback functions.

```python
import asyncio
from betterKickAPI.kick import Kick
from betterKickAPI.oauth import UserAuthenticationStorageHelper
from betterKickAPI.eventsub.webhook import KickWebhook, SSLOptions
from betterKickAPI.eventsub.events import ChatMessageEvent, ChannelFollowEvent, LivestreamStatusUpdatedEvent
from betterKickAPI.types import OAuthScope

BROADCASTER_ID = 668  # Your channel's broadcaster_user_id

async def on_chat_message(event: ChatMessageEvent):
    print(f"[{event.broadcaster.username}] {event.sender.username}: {event.content}")
    if event.emotes:
        for emote in event.emotes:
            print(f"  Emote: {emote.emote_id}")

async def on_follow(event: ChannelFollowEvent):
    print(f"New follower: {event.follower.username} followed {event.broadcaster.username}")

async def on_stream_status(event: LivestreamStatusUpdatedEvent):
    state = "LIVE" if event.is_live else "OFFLINE"
    print(f"Stream {state}: '{event.title}'")

async def main():
    kick = await Kick("YOUR_APP_ID", "YOUR_APP_SECRET")

    # Optional: use SSL
    # ssl_opts = SSLOptions(cert_file_name="cert.pem", key_file_name="key.pem")

    webhook = KickWebhook(
        kick,
        auto_fetch_public_key=True,  # fetches key from API automatically
        msg_id_history_max_length=100,
    )

    # Register event listeners (subscribes and maps callbacks)
    await webhook.listen_chat_message_sent(BROADCASTER_ID, on_chat_message)
    await webhook.listen_channel_follow(BROADCASTER_ID, on_follow)
    await webhook.listen_livestream_status_updated(BROADCASTER_ID, on_stream_status)

    # Start the internal webhook server
    webhook.start(port=3330, host_binding="0.0.0.0", endpoint="/callback")
    print("Webhook server running on port 3330")

    try:
        await asyncio.sleep(3600)  # run for 1 hour
    finally:
        await webhook.stop()  # auto-unsubscribes all known subscriptions
        await kick.close()

asyncio.run(main())
```
```

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

### Basic Kick API Call Example

Source: https://github.com/benjas333/pykickapi/blob/master/README.md

Initialize the Kick API instance and retrieve a user ID. Requires 'APP_ID' and 'APP_SECRET'. Ensure asyncio is used for asynchronous operations.

```python
from betterKickAPI.kick import Kick
from betterKickAPI.helper import first
import asyncio

async def kick_example():
        # Initialize the kick instance, this will by default also create an app authentication for you
        kick = await Kick('APP_ID', 'APP_SECRET')

        users = await kick.get_users(slug='your_kick_user')
        # print the ID of your user
        print(user[0].broadcaster_user_id)

# run this example
asyncio.run(kick_example())
```

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

### User Authentication Setup

Source: https://github.com/benjas333/pykickapi/blob/master/README.md

Set up User Authentication for the Kick API. This requires user consent via a redirect URI configured in Kick developer settings. The `UserAuthenticator` handles the OAuth flow.

```python
from betterKickAPI.kick import Kick
from betterKickAPI.oauth import UserAuthenticator
from betterKickAPI.type import OAuthScope

kick = await Kick('APP_ID', 'APP_SECRET')

target_scope = [OAuthScope.CHANNEL_READ]
auth = UserAuthenticator(kick, target_scope, force_verify=False)
# this will open your default browser and prompt you with the kick auth website
token, refresh_token = await auth.authenticate()

await kick.set_user_authentication(token, target_scope, refresh_token)
```

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

### Listen to All Supported Events with EventSubBase

Source: https://context7.com/benjas333/pykickapi/llms.txt

Subscribe to various Kick events using the `listen_*` methods. Each method requires a broadcaster ID and an async callback function. Ensure the `KickWebhook` is started and stopped appropriately. The `Kick` client should also be closed.

```python
import asyncio
from betterKickAPI.kick import Kick
from betterKickAPI.eventsub.webhook import KickWebhook
from betterKickAPI.eventsub.events import (
    ChannelFollowEvent,
    ChannelSubscriptionNewEvent,
    ChannelSubscriptionRenewalEvent,
    ChannelSubscriptionGiftsEvent,
    LivestreamStatusUpdatedEvent,
    LivestreamMetadataUpdatedEvent,
    ModerationBannedEvent,
    KicksGiftedEvent,
)

BROADCASTER_ID = 668

async def main():
    kick = await Kick("YOUR_APP_ID", "YOUR_APP_SECRET")
    webhook = KickWebhook(kick)

    sub_ids = []

    sub_ids.append(await webhook.listen_channel_follow(
        BROADCASTER_ID,
        lambda e: print(f"Follow: {e.follower.username}"),
    ))
    sub_ids.append(await webhook.listen_channel_subscription_new(
        BROADCASTER_ID,
        lambda e: print(f"New sub: {e.subscriber.username} for {e.duration}mo"),
    ))
    sub_ids.append(await webhook.listen_channel_subscription_renewal(
        BROADCASTER_ID,
        lambda e: print(f"Resub: {e.subscriber.username}"),
    ))
    sub_ids.append(await webhook.listen_channel_subscription_gifts(
        BROADCASTER_ID,
        lambda e: print(f"Gift subs from {e.gifter.username if not e.gifter.is_anonymous else 'Anonymous'} to {len(e.giftees)} users"),
    ))
    sub_ids.append(await webhook.listen_livestream_status_updated(
        BROADCASTER_ID,
        lambda e: print(f"Stream {'started' if e.is_live else 'ended'}: {e.title}"),
    ))
    sub_ids.append(await webhook.listen_livestream_metadata_updated(
        BROADCASTER_ID,
        lambda e: print(f"Metadata update: title='{e.metadata.title}' lang={e.metadata.language}"),
    ))
    sub_ids.append(await webhook.listen_moderation_banned(
        BROADCASTER_ID,
        lambda e: print(f"Banned: {e.banned_user.username} by {e.moderator.username} — {e.metadata.reason}"),
    ))
    sub_ids.append(await webhook.listen_kicks_gifted(
        BROADCASTER_ID,
        lambda e: print(f"KICKs gifted: {e.sender.username} gave {e.gift.amount} {e.gift.name}"),
    ))

    webhook.start(port=3330)
    print(f"Listening for {len(sub_ids)} events...")
    await asyncio.sleep(3600)
    await webhook.stop()
    await kick.close()

asyncio.run(main())

```

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

### Kick.get_category

Source: https://context7.com/benjas333/pykickapi/llms.txt

Get a Single Category by ID.

```APIDOC
## `Kick.get_category` — Get a Single Category by ID

Retrieves a single category by its unique identifier.

### Parameters
#### Path Parameters
- **category_id** (int) - Required - The ID of the category to retrieve.

### Returns
- A `Category` object if found.
- Raises `KickResourceNotFound` if the category does not exist.
```

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

### Get KICKs Gifting Leaderboard with KickAPI

Source: https://context7.com/benjas333/pykickapi/llms.txt

Retrieve weekly, monthly, and all-time KICKs gifting leaderboards using `get_kicks_leaderboard`. Requires `OAuthScope.KICKS_READ`.

```python
import asyncio
from betterKickAPI.kick import Kick
from betterKickAPI.types import OAuthScope

async def main():
    kick = await Kick("YOUR_APP_ID", "YOUR_APP_SECRET")
    # Set user auth with OAuthScope.KICKS_READ

    leaderboard = await kick.get_kicks_leaderboard(top=5)

    print("=== Weekly Top 5 ===")
    for entry in leaderboard.week:
        print(f"#{entry.rank} {entry.username}: {entry.gifted_amount} KICKs")

    print("=== Monthly Top 5 ===")
    for entry in leaderboard.month:
        print(f"#{entry.rank} {entry.username}: {entry.gifted_amount} KICKs")

    print("=== All-Time Top 5 ===")
    for entry in leaderboard.lifetime:
        print(f"#{entry.rank} {entry.username}: {entry.gifted_amount} KICKs")

    await kick.close()

asyncio.run(main())
```

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

### Get Channel Information with Kick.get_channels

Source: https://context7.com/benjas333/pykickapi/llms.txt

Retrieve channel information by broadcaster user ID(s) or slug(s). Accepts up to 50 entries per request. No user authentication is needed for lookup by slug.

```python
import asyncio
from betterKickAPI.kick import Kick

async def main():
    kick = await Kick("YOUR_APP_ID", "YOUR_APP_SECRET")

    # Lookup by slug (no user auth needed)
    channels = await kick.get_channels(slug=["xqc", "trainwreckstv"])
    for ch in channels:
        print(f"{ch.slug} (ID {ch.broadcaster_user_id}): {ch.stream_title}")
        if ch.stream and ch.stream.is_live:
            print(f"  LIVE — {ch.stream.viewer_count} viewers")
        if ch.category:
            print(f"  Category: {ch.category.name}")
    # xqc (ID 668): Just Chatting
    #   LIVE — 42000 viewers
    #   Category: Just Chatting

    await kick.close()

asyncio.run(main())
```

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

### Get User Information with Kick.get_users

Source: https://context7.com/benjas333/pykickapi/llms.txt

Retrieve user information by user ID(s). If no arguments are provided and user authentication is active, it returns the currently authenticated user. Requires OAuthScope.USER_READ for full details.

```python
import asyncio
from betterKickAPI.kick import Kick
from betterKickAPI.types import OAuthScope

async def main():
    kick = await Kick("YOUR_APP_ID", "YOUR_APP_SECRET")
    # Assume user auth already set with OAuthScope.USER_READ

    # Get current authenticated user
    current_users = await kick.get_users()
    # Returns: list[api.User] — each has .name, .user_id, .email, .profile_picture

    # Get specific users by ID
    users = await kick.get_users(user_id=[12345, 67890])
    for u in users:
        print(f"User {u.user_id}: {u.name} ({u.email})")
    # User 12345: streamer_one (streamer@example.com)
    # User 67890: streamer_two (other@example.com)

    await kick.close()

asyncio.run(main())
```

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

### Async Generator Utilities: first and limit

Source: https://context7.com/benjas333/pykickapi/llms.txt

Utilize `first` to get a single item from an async generator and `limit` to retrieve a specified number of items. These helpers are useful for paginated API endpoints.

```python
import asyncio
from betterKickAPI.kick import Kick
from betterKickAPI.helper import first, limit

async def main():
    kick = await Kick("YOUR_APP_ID", "YOUR_APP_SECRET")

    # Get only the very first category matching "chess"
    cat = await first(kick.get_categories("chess"))
    print("First result:", cat.name if cat else "None found")
    # First result: Chess

    # Get at most 3 results
    top3 = [cat async for cat in limit(kick.get_categories("fps"), 3)]
    print("Top 3 FPS categories:", [c.name for c in top3])
    # Top 3 FPS categories: ['Call of Duty', 'Valorant', 'CS2']

    await kick.close()

asyncio.run(main())

```

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

### Get Single Category by ID

Source: https://context7.com/benjas333/pykickapi/llms.txt

Use `kick.get_category` to fetch a specific category using its ID. Handles `KickResourceNotFound` exception if the category does not exist.

```python
import asyncio
from betterKickAPI.kick import Kick
from betterKickAPI.types import KickResourceNotFound

async def main():
    kick = await Kick("YOUR_APP_ID", "YOUR_APP_SECRET")

    try:
        category = await kick.get_category(7)
        print(f"{category.name} — thumbnail: {category.thumbnail}")
        # Minecraft — thumbnail: https://cdn.kick.com/...
    except KickResourceNotFound:
        print("Category not found")

    await kick.close()

asyncio.run(main())
```

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

### Get Livestream Statistics

Source: https://context7.com/benjas333/pykickapi/llms.txt

Fetch aggregate livestream statistics, such as the total live stream count, using `kick.get_livestream_stats`. This endpoint requires app or user authentication.

```python
import asyncio
from betterKickAPI.kick import Kick

async def main():
    kick = await Kick("YOUR_APP_ID", "YOUR_APP_SECRET")
    # Either app or user auth

    stats = await kick.get_livestream_stats()
    print("Total live streams on Kick:", stats.total_count)
    # Total live streams on Kick: 3271

    await kick.close()

asyncio.run(main())
```

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

### Get Webhook Signature Verification Key with KickAPI

Source: https://context7.com/benjas333/pykickapi/llms.txt

Fetches the RSA public key used for verifying webhook signatures with `get_public_key`. This key is essential for ensuring the authenticity of incoming webhook payloads.

```python
import asyncio
from betterKickAPI.kick import Kick

async def main():
    kick = await Kick("YOUR_APP_ID", "YOUR_APP_SECRET")

    public_key_pem = await kick.get_public_key()
    print(public_key_pem[:64] + "...")
    # -----BEGIN PUBLIC KEY-----
    # MFkwEwYHKoZIzj0CAQYIKoZIzj0DAQcDQgAE...

    await kick.close()

asyncio.run(main())
```

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

### Get Active Livestreams

Source: https://context7.com/benjas333/pykickapi/llms.txt

Retrieve active livestreams using `kick.get_livestreams`. Supports filtering by category, language, sort order, and broadcaster IDs. Requires app or user authentication.

```python
import asyncio
from betterKickAPI.kick import Kick

async def main():
    kick = await Kick("YOUR_APP_ID", "YOUR_APP_SECRET")

    # Top 10 streams by viewer count in a specific category
    streams = await kick.get_livestreams(
        category_id=7,
        sort="viewer_count",
        limit=10,
    )
    for s in streams:
        print(f"{s.slug}: {s.viewer_count} viewers — '{s.stream_title}'")
    # xqc: 42000 viewers — 'gambling time'

    # Streams for specific broadcaster IDs
    my_streams = await kick.get_livestreams(broadcaster_user_id=[668, 1234])
    print("Live broadcasts:", len(my_streams))

    await kick.close()

asyncio.run(main())
```

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

### Get Webhook Signature Verification Key

Source: https://context7.com/benjas333/pykickapi/llms.txt

Fetches the RSA public key used to verify ECDSA signatures on incoming webhook payloads.

```APIDOC
## `Kick.get_public_key` — Get Webhook Signature Verification Key

Fetches the RSA public key used to verify ECDSA signatures on incoming webhook payloads.

### Method
GET

### Endpoint
`/oauth/public_key`

### Parameters
None

### Request Example
```python
await kick.get_public_key()
```

### Response
#### Success Response (200)
- **public_key_pem** (string) - The public key in PEM format.

### Response Example
```text
-----BEGIN PUBLIC KEY-----
MFkwEwYHKoZIzj0CAQYIKoZIzj0DAQcDQgAE...
-----END PUBLIC KEY-----
```
```

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

### EventSubBase Listen Methods

Source: https://context7.com/benjas333/pykickapi/llms.txt

Demonstrates how to subscribe to various events for a given broadcaster using `EventSubBase`'s `listen_*` methods. Each method returns a subscription ID.

```APIDOC
## `EventSubBase` Listen Methods — All Supported Events

All `listen_*` methods subscribe to an event for a given broadcaster and register a typed async callback. Each returns the `subscription_id` string.

```python
import asyncio
from betterKickAPI.kick import Kick
from betterKickAPI.eventsub.webhook import KickWebhook
from betterKickAPI.eventsub.events import (
    ChannelFollowEvent,
    ChannelSubscriptionNewEvent,
    ChannelSubscriptionRenewalEvent,
    ChannelSubscriptionGiftsEvent,
    LivestreamStatusUpdatedEvent,
    LivestreamMetadataUpdatedEvent,
    ModerationBannedEvent,
    KicksGiftedEvent,
)

BROADCASTER_ID = 668

async def main():
    kick = await Kick("YOUR_APP_ID", "YOUR_APP_SECRET")
    webhook = KickWebhook(kick)

    sub_ids = []

    sub_ids.append(await webhook.listen_channel_follow(
        BROADCASTER_ID,
        lambda e: print(f"Follow: {e.follower.username}"),
    ))
    sub_ids.append(await webhook.listen_channel_subscription_new(
        BROADCASTER_ID,
        lambda e: print(f"New sub: {e.subscriber.username} for {e.duration}mo"),
    ))
    sub_ids.append(await webhook.listen_channel_subscription_renewal(
        BROADCASTER_ID,
        lambda e: print(f"Resub: {e.subscriber.username}"),
    ))
    sub_ids.append(await webhook.listen_channel_subscription_gifts(
        BROADCASTER_ID,
        lambda e: print(f"Gift subs from {e.gifter.username if not e.gifter.is_anonymous else 'Anonymous'} to {len(e.giftees)} users"),
    ))
    sub_ids.append(await webhook.listen_livestream_status_updated(
        BROADCASTER_ID,
        lambda e: print(f"Stream {'started' if e.is_live else 'ended'}: {e.title}"),
    ))
    sub_ids.append(await webhook.listen_livestream_metadata_updated(
        BROADCASTER_ID,
        lambda e: print(f"Metadata update: title='{e.metadata.title}' lang={e.metadata.language}"),
    ))
    sub_ids.append(await webhook.listen_moderation_banned(
        BROADCASTER_ID,
        lambda e: print(f"Banned: {e.banned_user.username} by {e.moderator.username} — {e.metadata.reason}"),
    ))
    sub_ids.append(await webhook.listen_kicks_gifted(
        BROADCASTER_ID,
        lambda e: print(f"KICKs gifted: {e.sender.username} gave {e.gift.amount} {e.gift.name}"),
    ))

    webhook.start(port=3330)
    print(f"Listening for {len(sub_ids)} events...")
    await asyncio.sleep(3600)
    await webhook.stop()
    await kick.close()

asyncio.run(main())
```
```

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

### Kick Client Initialization

Source: https://context7.com/benjas333/pykickapi/llms.txt

Initialize the Kick client and automatically authenticate with app credentials. You can also disable auto-refresh and register callbacks for token refreshes.

```APIDOC
## Kick Client Initialization

### Description
The `Kick` class is the central client. Awaiting it triggers automatic app authentication. All API calls are methods on this object.

### Method
```python
await Kick(app_id: str, app_secret: str)
```

### Parameters
#### Path Parameters
None

#### Query Parameters
None

#### Request Body
None

### Request Example
```python
import asyncio
from betterKickAPI.kick import Kick

async def main():
    # Auto-fetches an app token using client_credentials grant
    kick = await Kick("YOUR_APP_ID", "YOUR_APP_SECRET")

    # Disable auto-refresh if you manage tokens manually
    kick.auto_refresh_auth = False

    # Register refresh callbacks
    async def on_app_token_refresh(new_token: str):
        print(f"New app token: {new_token[:8]}...")

    kick.app_auth_refresh_callback = on_app_token_refresh

    print("App token:", kick.app_auth_token[:8] + "...")
    await kick.close()

asyncio.run(main())
```

### Response
#### Success Response (200)
None (Initialization is asynchronous and returns the client object)

#### Response Example
None
```

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

### helper.first and helper.limit - Async Generator Utilities

Source: https://context7.com/benjas333/pykickapi/llms.txt

Provides utility functions `first` and `limit` for consuming paginating asynchronous generators, such as those returned by `get_categories`.

```APIDOC
## `helper.first` and `helper.limit` — Async Generator Utilities

Convenience helpers for consuming the paginating async generators returned by endpoints like `get_categories`.

```python
import asyncio
from betterKickAPI.kick import Kick
from betterKickAPI.helper import first, limit

async def main():
    kick = await Kick("YOUR_APP_ID", "YOUR_APP_SECRET")

    # Get only the very first category matching "chess"
    cat = await first(kick.get_categories("chess"))
    print("First result:", cat.name if cat else "None found")
    # First result: Chess

    # Get at most 3 results
    top3 = [cat async for cat in limit(kick.get_categories("fps"), 3)]
    print("Top 3 FPS categories:", [c.name for c in top3])
    # Top 3 FPS categories: ['Call of Duty', 'Valorant', 'CS2']

    await kick.close()

asyncio.run(main())
```
```

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

### Automate User Token Storage with UserAuthenticationStorageHelper

Source: https://context7.com/benjas333/pykickapi/llms.txt

Use UserAuthenticationStorageHelper to manage user tokens persistently. It handles loading from a JSON file, initiating the browser flow on first run, and refreshing tokens if expired.

```python
import asyncio
from pathlib import PurePath
from betterKickAPI.kick import Kick
from betterKickAPI.oauth import UserAuthenticationStorageHelper
from betterKickAPI.types import OAuthScope

async def main():
    kick = await Kick("YOUR_APP_ID", "YOUR_APP_SECRET")
    scopes = [OAuthScope.USER_READ, OAuthScope.CHANNEL_READ, OAuthScope.CHAT_WRITE]

    helper = UserAuthenticationStorageHelper(
        kick,
        scopes,
        storage_path=PurePath("my_token.json"),  # defaults to user_token.json
    )
    # Loads from disk or runs the full OAuth flow; refreshes if expired
    await helper.bind()

    users = await kick.get_users()
    print("Logged-in user:", users[0].name if users else "N/A")
    await kick.close()

asyncio.run(main())
```

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

### Initialize Kick API Client

Source: https://context7.com/benjas333/pykickapi/llms.txt

Initialize the Kick client and automatically authenticate with app credentials. You can optionally disable auto-refresh and register callbacks for token refresh events.

```python
import asyncio
from betterKickAPI.kick import Kick
from betterKickAPI.types import KickAPIException, UnauthorizedException

async def main():
    # Auto-fetches an app token using client_credentials grant
    kick = await Kick("YOUR_APP_ID", "YOUR_APP_SECRET")

    # Disable auto-refresh if you manage tokens manually
    kick.auto_refresh_auth = False

    # Register refresh callbacks
    async def on_app_token_refresh(new_token: str):
        print(f"New app token: {new_token[:8]}...")

    kick.app_auth_refresh_callback = on_app_token_refresh

    print("App token:", kick.app_auth_token[:8] + "...")
    await kick.close()

asyncio.run(main())
```

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

### User OAuth Flow with UserAuthenticator

Source: https://context7.com/benjas333/pykickapi/llms.txt

Initiate the user OAuth flow to obtain access and refresh tokens. The redirect URI must be registered in your Kick developer settings. This method opens the system browser.

```python
import asyncio
from betterKickAPI.kick import Kick
from betterKickAPI.oauth import UserAuthenticator
from betterKickAPI.types import OAuthScope

async def main():
    kick = await Kick("YOUR_APP_ID", "YOUR_APP_SECRET")

    scopes = [OAuthScope.USER_READ, OAuthScope.CHANNEL_READ, OAuthScope.CHAT_WRITE]
    auth = UserAuthenticator(kick, scopes)

    # Opens browser; blocks until user grants access
    token, refresh_token = await auth.authenticate()
    print("Access token:", token[:8] + "...")
    print("Refresh token:", refresh_token[:8] + "...")

    # Activate user auth on the client
    await kick.set_user_authentication(
        token=token,
        scope=scopes,
        refresh_token=refresh_token,
    )
    print("Has user auth:", kick._has_user_auth)  # True
    await kick.close()

asyncio.run(main())
```

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

### Handle Kick API Exceptions

Source: https://context7.com/benjas333/pykickapi/llms.txt

Demonstrates granular error handling for various Kick API exceptions. Import specific exceptions for targeted error management.

```python
import asyncio
from betterKickAPI.kick import Kick
from betterKickAPI.types import (
    KickAPIException,
    KickAuthorizationException,
    UnauthorizedException,
    MissingScopeException,
    InvalidTokenException,
    InvalidRefreshTokenException,
    KickBackendException,
    KickResourceNotFound,
    MissingAppSecretException,
    EventSubSubscriptionError,
)

async def main():
    kick = await Kick("YOUR_APP_ID", "YOUR_APP_SECRET")
    try:
        category = await kick.get_category(999999999)
    except KickResourceNotFound:
        print("Category does not exist")
    except UnauthorizedException:
        print("Not authenticated or token expired")
    except MissingScopeException as e:
        print("Missing OAuth scope:", e)
    except KickBackendException as e:
        print("Kick server error:", e)
    except KickAPIException as e:
        print("Generic API error:", e)

    await kick.close()

asyncio.run(main())
```

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

### OAuthScope - Available Permission Scopes

Source: https://context7.com/benjas333/pykickapi/llms.txt

Lists all available OAuth scopes for user authentication, with descriptions for each scope.

```APIDOC
## `OAuthScope` — Available Permission Scopes

All OAuth scopes available for user authentication.

```python
from betterKickAPI.types import OAuthScope

# Available scopes:
# OAuthScope.USER_READ        — "user:read"       — View user info
# OAuthScope.CHANNEL_READ     — "channel:read"    — View channel info
# OAuthScope.CHANNEL_WRITE    — "channel:write"   — Update stream metadata
# OAuthScope.CHAT_WRITE       — "chat:write"      — Send chat messages
# OAuthScope.STREAMKEY_READ   — "streamkey:read"  — Read stream key/URL
# OAuthScope.EVENTS_SUBSCRIBE — "events:subscribe"— Subscribe to events
# OAuthScope.MODERATION_BAN   — "moderation:ban"  — Execute bans/timeouts
# OAuthScope.KICKS_READ       — "kicks:read"      — View KICKs leaderboards

from betterKickAPI.kick import Kick
from betterKickAPI.oauth import UserAuthenticator

async def main():
    kick = await Kick("YOUR_APP_ID", "YOUR_APP_SECRET")
    scopes = [
        OAuthScope.USER_READ,
        OAuthScope.CHANNEL_READ,
        OAuthScope.CHANNEL_WRITE,
        OAuthScope.CHAT_WRITE,
        OAuthScope.EVENTS_SUBSCRIBE,
        OAuthScope.MODERATION_BAN,
        OAuthScope.KICKS_READ,
    ]
    auth = UserAuthenticator(kick, scopes)
    token, refresh_token = await auth.authenticate()
    await kick.set_user_authentication(token=token, scope=scopes, refresh_token=refresh_token)
    print("Authorized scopes:", [str(s) for s in kick.user_auth_scope])
```
```

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

### Integrate Kick Webhook with External Server

Source: https://context7.com/benjas333/pykickapi/llms.txt

Use `KickWebhook.handle_incoming` to process raw webhook POST requests if you are managing your own web server. This method handles signature verification and event dispatching. Ensure the `Kick` and `KickWebhook` instances are initialized, and event listeners are registered before handling incoming requests.

```python
import asyncio
from fastapi import FastAPI, Request, Response
from betterKickAPI.kick import Kick
from betterKickAPI.eventsub.webhook import KickWebhook
from betterKickAPI.eventsub.events import ChatMessageEvent
from betterKickAPI.types import OAuthScope

app = FastAPI()
kick: Kick = None
webhook: KickWebhook = None

@app.on_event("startup")
async def startup():
    global kick, webhook
    kick = await Kick("YOUR_APP_ID", "YOUR_APP_SECRET")
    webhook = KickWebhook(kick, auto_fetch_public_key=True)

    async def on_message(event: ChatMessageEvent):
        print(f"{event.sender.username}: {event.content}")

    await webhook.listen_chat_message_sent(668, on_message)
    # NOTE: Do NOT call webhook.start() — you're handling routing yourself

@app.post("/kick/webhook")
async def webhook_endpoint(request: Request):
    body = await request.body()
    headers = dict(request.headers)

    result = await webhook.handle_incoming(data=body, headers=headers)
    return Response(content=result.text, status_code=result.status)
```

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

### KickWebhook.handle_incoming

Source: https://context7.com/benjas333/pykickapi/llms.txt

Handles raw webhook POST requests for external web server integrations, allowing custom routing and processing.

```APIDOC
## `KickWebhook.handle_incoming` — External Webhook Integration

If you manage your own web server (e.g., FastAPI, aiohttp), use `handle_incoming` directly instead of the built-in server to handle the raw webhook POST request.

```python
import asyncio
from fastapi import FastAPI, Request, Response
from betterKickAPI.kick import Kick
from betterKickAPI.eventsub.webhook import KickWebhook
from betterKickAPI.eventsub.events import ChatMessageEvent
from betterKickAPI.types import OAuthScope

app = FastAPI()
kick: Kick = None
webhook: KickWebhook = None

@app.on_event("startup")
async def startup():
    global kick, webhook
    kick = await Kick("YOUR_APP_ID", "YOUR_APP_SECRET")
    webhook = KickWebhook(kick, auto_fetch_public_key=True)

    async def on_message(event: ChatMessageEvent):
        print(f"{event.sender.username}: {event.content}")

    await webhook.listen_chat_message_sent(668, on_message)
    # NOTE: Do NOT call webhook.start() — you're handling routing yourself

@app.post("/kick/webhook")
async def webhook_endpoint(request: Request):
    body = await request.body()
    headers = dict(request.headers)

    result = await webhook.handle_incoming(data=body, headers=headers)
    return Response(content=result.text, status_code=result.status)
```
```

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

### UserAuthenticationStorageHelper

Source: https://context7.com/benjas333/pykickapi/llms.txt

Automates loading, saving, and refreshing user tokens from a JSON file. On first run it triggers the browser flow; on subsequent runs it reuses the stored token.

```APIDOC
## `UserAuthenticationStorageHelper` — Persistent Token Storage

Automates loading, saving, and refreshing user tokens from a JSON file. On first run it triggers the browser flow; on subsequent runs it reuses the stored token.

### Usage
```python
import asyncio
from pathlib import PurePath
from betterKickAPI.kick import Kick
from betterKickAPI.oauth import UserAuthenticationStorageHelper
from betterKickAPI.types import OAuthScope

async def main():
    kick = await Kick("YOUR_APP_ID", "YOUR_APP_SECRET")
    scopes = [OAuthScope.USER_READ, OAuthScope.CHANNEL_READ, OAuthScope.CHAT_WRITE]

    helper = UserAuthenticationStorageHelper(
        kick,
        scopes,
        storage_path=PurePath("my_token.json"),  # defaults to user_token.json
    )
    # Loads from disk or runs the full OAuth flow; refreshes if expired
    await helper.bind()

    users = await kick.get_users()
    print("Logged-in user:", users[0].name if users else "N/A")
    await kick.close()

asyncio.run(main())
```
```

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

### Kick.get_users

Source: https://context7.com/benjas333/pykickapi/llms.txt

Retrieves user information by user ID(s). With no arguments and user auth active, returns the currently authenticated user.

```APIDOC
## `Kick.get_users` — Get User Information

Retrieves user information by user ID(s). With no arguments and user auth active, returns the currently authenticated user.

### Parameters
- **user_id** (list[int]) - Optional - A list of user IDs to retrieve information for.

### Returns
- `list[api.User]` - A list of User objects, each containing `.name`, `.user_id`, `.email`, and `.profile_picture`.

### Usage
```python
import asyncio
from betterKickAPI.kick import Kick
from betterKickAPI.types import OAuthScope

async def main():
    kick = await Kick("YOUR_APP_ID", "YOUR_APP_SECRET")
    # Assume user auth already set with OAuthScope.USER_READ

    # Get current authenticated user
    current_users = await kick.get_users()
    # Returns: list[api.User] — each has .name, .user_id, .email, .profile_picture

    # Get specific users by ID
    users = await kick.get_users(user_id=[12345, 67890])
    for u in users:
        print(f"User {u.user_id}: {u.name} ({u.email})")
    # User 12345: streamer_one (streamer@example.com)
    # User 67890: streamer_two (other@example.com)

    await kick.close()

asyncio.run(main())
```
```

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

### Search Categories with Pagination

Source: https://context7.com/benjas333/pykickapi/llms.txt

Use `kick.get_categories` to search for categories. It returns an async generator that auto-paginates through all results. Helper functions `first` and `limit` can be used to control the number of results consumed.

```python
import asyncio
from betterKickAPI.kick import Kick
from betterKickAPI.helper import first, limit

async def main():
    kick = await Kick("YOUR_APP_ID", "YOUR_APP_SECRET")

    # Get only the first result
    category = await first(kick.get_categories("Minecraft"))
    if category:
        print(f"Category: {category.name} (ID {category.id})")
        # Category: Minecraft (ID 7)

    # Consume up to 5 results
    async for cat in limit(kick.get_categories("call of duty"), 5):
        print(cat.id, cat.name)

    # Consume all results (auto-paginated)
    results = [cat async for cat in kick.get_categories("valorant")]
    print(f"Found {len(results)} categories")

    await kick.close()

asyncio.run(main())
```

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

### Manage EventSub Subscriptions with KickAPI

Source: https://context7.com/benjas333/pykickapi/llms.txt

Use `get_events_subscriptions` to list active subscriptions, `post_events_subscriptions` to subscribe to events, and `delete_events_subscriptions` to remove them. The `method` parameter can be set to "webhook".

```python
import asyncio
from betterKickAPI.kick import Kick
from betterKickAPI.types import OAuthScope, EventSubEvents

async def main():
    kick = await Kick("YOUR_APP_ID", "YOUR_APP_SECRET")

    # List all active subscriptions
    subs = await kick.get_events_subscriptions()
    for sub in subs:
        print(f"{sub.id}: {sub.event} (broadcaster {sub.broadcaster_user_id})")

    # Subscribe to chat messages and follows for broadcaster 668
    results = await kick.post_events_subscriptions(
        events=[EventSubEvents.CHAT_MESSAGE, EventSubEvents.CHANNEL_FOLLOW],
        broadcaster_user_id=668,
        method="webhook",
        force_app_auth=True,
    )
    for r in results:
        if r.error:
            print(f"Failed to subscribe to {r.name}: {r.error}")
        else:
            print(f"Subscribed to {r.name} — ID: {r.subscription_id}")

    # Delete subscriptions by ID (max 50 at a time)
    sub_ids = [r.subscription_id for r in results if r.subscription_id]
    if sub_ids:
        deleted = await kick.delete_events_subscriptions(sub_ids, force_app_auth=True)
        print("Deleted:", deleted)  # True if HTTP 204

    await kick.close()

asyncio.run(main())
```

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

### UserAuthenticator.authenticate - Headless Environments

Source: https://context7.com/benjas333/pykickapi/llms.txt

For headless or server environments, disable the automatic browser opening and provide a callback function to handle the authorization URL. This allows users to authenticate from a different device.

```APIDOC
## UserAuthenticator.authenticate — Headless / Server Environments

### Description
For headless servers, disable browser opening and receive the auth URL via a callback or the log.

### Method
```python
await UserAuthenticator(kick_client: Kick, scopes: list[OAuthScope]).authenticate(use_browser: bool = True, auth_url_callback: callable = None)
```

### Parameters
#### Path Parameters
None

#### Query Parameters
None

#### Request Body
None

### Request Example
```python
import asyncio
from betterKickAPI.kick import Kick
from betterKickAPI.oauth import UserAuthenticator
from betterKickAPI.types import OAuthScope

async def main():
    kick = await Kick("YOUR_APP_ID", "YOUR_APP_SECRET")
    scopes = [OAuthScope.CHANNEL_READ]
    auth = UserAuthenticator(kick, scopes)

    async def send_url_to_user(url: str):
        # e.g. send via Discord bot, email, etc.
        print(f"Visit this URL to authenticate:\n{url}")

    token, refresh_token = await auth.authenticate(
        use_browser=False,
        auth_url_callback=send_url_to_user,
    )
    print("Authenticated:", token[:8] + "...")
    await kick.close()

asyncio.run(main())
```

### Response
#### Success Response (200)
- **token** (str) - The user's access token.
- **refresh_token** (str) - The user's refresh token.

#### Response Example
```json
{
  "token": "your_access_token_here",
  "refresh_token": "your_refresh_token_here"
}
```
```

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

### Manage EventSub Subscriptions

Source: https://context7.com/benjas333/pykickapi/llms.txt

Low-level access to the Kick EventSub subscription management endpoints. Typically managed automatically by `KickWebhook`.

```APIDOC
## `Kick.get_events_subscriptions` / `Kick.post_events_subscriptions` / `Kick.delete_events_subscriptions` — Manage EventSub Subscriptions

Low-level access to the Kick EventSub subscription management endpoints. Typically managed automatically by `KickWebhook`.

### Method
GET (list), POST (subscribe), DELETE (unsubscribe)

### Endpoint
`/eventsub/subscriptions`

### Parameters
#### Path Parameters
None

#### Query Parameters (for GET)
None

#### Request Body (for POST)
- **events** (array of strings) - Required - A list of EventSub event types to subscribe to (e.g., `"chat.message"`, `"channel.follow"`).
- **broadcaster_user_id** (integer) - Required - The ID of the broadcaster for whom to subscribe to events.
- **method** (string) - Required - The delivery method for events (e.g., `"webhook"`).
- **force_app_auth** (boolean) - Optional - Forces the use of application authentication.

#### Request Body (for DELETE)
- **ids** (array of strings) - Required - A list of subscription IDs to delete.
- **force_app_auth** (boolean) - Optional - Forces the use of application authentication.

### Request Example (List Subscriptions)
```python
await kick.get_events_subscriptions()
```

### Request Example (Subscribe)
```python
await kick.post_events_subscriptions(
    events=[EventSubEvents.CHAT_MESSAGE, EventSubEvents.CHANNEL_FOLLOW],
    broadcaster_user_id=668,
    method="webhook",
    force_app_auth=True,
)
```

### Request Example (Delete Subscriptions)
```python
await kick.delete_events_subscriptions(sub_ids, force_app_auth=True)
```

### Response
#### Success Response (GET - 200)
- **id** (string) - The subscription ID.
- **event** (string) - The type of event subscribed to.
- **broadcaster_user_id** (integer) - The ID of the broadcaster.

#### Success Response (POST - 200)
- **name** (string) - The name of the event.
- **subscription_id** (string) - The ID of the newly created subscription.
- **error** (string) - An error message if subscription failed.

#### Success Response (DELETE - 204)
Returns an empty response with a 204 status code if successful.

### Response Example (GET)
```json
[
  {
    "id": "sub-123",
    "event": "chat.message",
    "broadcaster_user_id": 668
  }
]
```

### Response Example (POST)
```json
[
  {
    "name": "chat.message",
    "subscription_id": "sub-456",
    "error": null
  }
]
```
```

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

### Kick.get_categories

Source: https://context7.com/benjas333/pykickapi/llms.txt

Returns an async generator over Category objects matching the search query, auto-paginating through all result pages.

```APIDOC
## `Kick.get_categories` — Search Categories (Paginating Generator)

Returns an async generator over `Category` objects matching the search query, auto-paginating through all result pages.

### Parameters
#### Query Parameters
- **query** (string) - Required - The search query for categories.

### Returns
- An async generator yielding `Category` objects.
```

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

### Kick.get_channels

Source: https://context7.com/benjas333/pykickapi/llms.txt

Retrieves channel information by broadcaster user ID(s) or slug(s). Accepts up to 50 entries per request.

```APIDOC
## `Kick.get_channels` — Get Channel Information

Retrieves channel information by broadcaster user ID(s) or slug(s). Accepts up to 50 entries per request.

### Parameters
- **user_id** (list[int]) - Optional - A list of broadcaster user IDs.
- **slug** (list[str]) - Optional - A list of channel slugs.

### Returns
- `list[api.Channel]` - A list of Channel objects, each containing details like `.slug`, `.stream_title`, `.stream` (with `.is_live` and `.viewer_count`), and `.category`.

### Usage
```python
import asyncio
from betterKickAPI.kick import Kick

async def main():
    kick = await Kick("YOUR_APP_ID", "YOUR_APP_SECRET")

    # Lookup by slug (no user auth needed)
    channels = await kick.get_channels(slug=["xqc", "trainwreckstv"])
    for ch in channels:
        print(f"{ch.slug} (ID {ch.broadcaster_user_id}): {ch.stream_title}")
        if ch.stream and ch.stream.is_live:
            print(f"  LIVE — {ch.stream.viewer_count} viewers")
        if ch.category:
            print(f"  Category: {ch.category.name}")
    # xqc (ID 668): Just Chatting
    #   LIVE — 42000 viewers
    #   Category: Just Chatting

    await kick.close()

asyncio.run(main())
```
```