EventSub Client#

EventSub lets you listen for events that happen on Twitch.

The EventSub client runs in its own thread, calling the given callback function whenever an event happens.

Look at the Twitch EventSub reference to find the topics you are interested in.

Requirements#

Note

Please note that Your Endpoint URL has to be HTTPS, has to run on Port 443 and requires a valid, non self signed certificate This most likely means, that you need a reverse proxy like nginx. You can also hand in a valid ssl context to be used in the constructor.

In the case that you don’t hand in a valid ssl context to the constructor, you can specify any port you want in the constructor and handle the bridge between this program and your public URL on port 443 via reverse proxy.

You can check on whether or not your webhook is publicly reachable by navigating to the URL set in callback_url. You should get a 200 response with the text pyTwitchAPI eventsub.

Listening to topics#

After you started your EventSub client, you can use the listen_ prefixed functions to listen to the topics you are interested in.

The function you hand in as callback will be called whenever that event happens with the event data as a parameter.

Code Example#from twitchAPI.twitch import Twitch
from twitchAPI.helper import first
from twitchAPI.eventsub import EventSub
import asyncio


TARGET_USERNAME = 'target_username_here'
EVENTSUB_URL = 'https://url.to.your.webhook.com'
APP_ID = 'your_app_id'
APP_SECRET = 'your_app_secret'


async def on_follow(data: dict):
    # our event happend, lets do things with the data we got!
    print(data)


async def eventsub_example():
    # create the api instance and get the ID of the target user
    twitch = await Twitch(APP_ID, APP_SECRET)
    user = await first(twitch.get_users(logins=TARGET_USERNAME))

    # basic setup, will run on port 8080 and a reverse proxy takes care of the https and certificate
    event_sub = EventSub(EVENTSUB_URL, APP_ID, 8080, twitch)

    # unsubscribe from all old events that might still be there
    # this will ensure we have a clean slate
    await event_sub.unsubscribe_all()
    # start the eventsub client
    event_sub.start()
    # subscribing to the desired eventsub hook for our user
    # the given function will be called every time this event is triggered
    await event_sub.listen_channel_follow(user.id, on_follow)

    # eventsub will run in its own process
    # so lets just wait for user input before shutting it all down again
    try:
        input('press Enter to shut down...')
    finally:
        # stopping both eventsub as well as gracefully closing the connection to the API
        await event_sub.stop()
        await twitch.close()
    print('done')


# lets run our example
asyncio.run(eventsub_example())

Class Documentation#

class twitchAPI.eventsub.EventSub#

Bases: object

EventSub integration for the Twitch Helix API.

__init__(callback_url, api_client_id, port, twitch, ssl_context=None)#

Parameters:

callback_url (str) – The full URL of the webhook.
api_client_id (str) – The id of your API client
port (int) – the port on which this webhook should run
twitch (Twitch) – a app authenticated instance of Twitch
ssl_context (Optional[SSLContext]) – optional ssl context to be used
Default: None

callback_url: str#: The full URL of the webhook.

logger: Logger#: The logger used for EventSub related log messages

secret: str#: A random secret string. Set this for added security.
Default: A random 20 character long string

wait_for_subscription_confirm: bool#: Set this to false if you don’t want to wait for a subscription confirm.
Default: True

wait_for_subscription_confirm_timeout: int#: Max time in seconds to wait for a subscription confirmation. Only used if wait_for_subscription_confirm is set to True.
Default: 30

unsubscribe_on_stop: bool#: Unsubscribe all currently active Webhooks on calling stop()
Default: True

start()#

Starts the EventSub client

Return type:: None
Raises:: RuntimeError – if EventSub is already running

async stop()#

Stops the EventSub client

This also unsubscribes from all known subscriptions if unsubscribe_on_stop is True

Return type:: None

async unsubscribe_all()#: Unsubscribe from all subscriptions

async unsubscribe_all_known()#: Unsubscribe from all subscriptions known to this client.

async unsubscribe_topic(topic_id)#

Unsubscribe from a specific topic.

Return type:: bool

async listen_channel_update(broadcaster_user_id, callback)#

A broadcaster updates their channel properties e.g., category, title, mature flag, broadcast, or language.

For more information see here: https://dev.twitch.tv/docs/eventsub/eventsub-subscription-types#channelupdate

Parameters:

broadcaster_user_id (str) – the id of the user you want to listen to
callback (Callable[[dict], Awaitable[None]]) – function for callback

Raises:

EventSubSubscriptionConflict – if a conflict was found with this subscription (e.g. already subscribed to this exact topic)
EventSubSubscriptionTimeout – if wait_for_subscription_confirm is true and the subscription was not fully confirmed in time
EventSubSubscriptionError – if the subscription failed (see error message for details)
TwitchBackendException – if the subscription failed due to a twitch backend error

Return type: