twitchAPI Python Wrapper

User authentication requires the user to explicitly authorize your application via a Twitch login prompt.

1. Setup Redirect URL: In your Twitch Developer Dashboard, add http://localhost:17563 as an OAuth Redirect URL.
2. Use UserAuthenticator: Use the UserAuthenticator class to trigger the browser-based authorization flow.
3. Apply Token: Once the token is obtained, apply it to the Twitch instance using set_user_authentication.

from twitchAPI.twitch import Twitch
from twitchAPI.oauth import UserAuthenticator
from twitchAPI.type import AuthScope

twitch = await Twitch('my_app_id', 'my_app_secret')

target_scope = [AuthScope.BITS_READ]
auth = UserAuthenticator(twitch, target_scope, force_verify=False)

# Opens browser for Twitch verification
token, refresh_token = await auth.authenticate()

# Apply the user authentication to the instance
await twitch.set_user_authentication(token, target_scope, refresh_token)

Version 3 of pytwitchapi is fully asynchronous. This is a breaking change from version 2, which used a mix of sync and async patterns.

Key changes include:

• The Twitch object must now be initialized using await.
• All refresh callbacks (app_auth_refresh_callback, user_auth_refresh_callback) must be async functions.
• All PubSub callbacks must be async functions.
• All EventSub methods (including listen_, unsubscribe_, and stop()) are now asynchronous and must be called with await.
• API responses have switched from raw dictionaries to specialized objects and generators.

from twitchAPI.twitch import Twitch

# V3 Initialization
async def user_refresh(token: str, refresh_token: str):
    print(f'my new user token is: {token}')

async def app_refresh(token: str):
    print(f'my new app token is: {token}')

twitch = await Twitch('my_app_id', 'my_app_secret')
twitch.app_auth_refresh_callback = app_refresh
twitch.user_auth_refresh_callback = user_refresh

EventSub has introduced a new transport mechanism. The original EventSub class is now located in twitchAPI.eventsub.webhook and has been renamed to EventSubWebhook.

Key Changes:

• Class Name: Use EventSubWebhook instead of EventSub.
• Callback Data: Topic callbacks no longer receive plain dictionaries (dict). They now receive specific Event objects (e.g., ChannelFollowEvent). You should update your callback function signatures to expect these objects instead of dictionaries.

# V4 (now)
from twitchAPI.eventsub.webhook import EventSubWebhook
from twitchAPI.object.eventsub import ChannelFollowEvent
import asyncio

EVENTSUB_URL = 'https://url.to.your.webhook.com'

# Note: data is now a ChannelFollowEvent object, not a dict
async def on_follow(data: ChannelFollowEvent):
    print(f'{data.event.user_name} now follows {data.event.broadcaster_user_name}!')


async def eventsub_webhook_example():
    # twitch setup is left out of this example

    eventsub = EventSubWebhook(EVENTSUB_URL, 8080, twitch)
    await eventsub.unsubscribe_all()
    eventsub.start()
    await eventsub.listen_channel_follow_v2(user.id, user.id, on_follow)

    try:
        input('press Enter to shut down...')
    finally:
        await eventsub.stop()
        await twitch.close()
    print('done')


asyncio.run(eventsub_webhook_example())

To implement custom logic for chat commands, extend the BaseCommandMiddleware class. A custom middleware allows you to intercept command execution to decide if a command should run or to perform actions after a command has been executed.

To create a middleware, you must implement two primary methods:

1. can_execute(cmd: ChatCommand) -> bool: This method is called before the command runs. Return True if the command should proceed, or False to block it.
2. was_executed(cmd: ChatCommand): This method is called after the command has been executed. Use this to update internal state or perform post-execution logic.

It is recommended to include an execute_blocked_handler in your __init__ method to maintain compatibility with default middlewares, though this is optional.

from typing import Callable, Optional, Awaitable
from twitchAPI.chat.middleware import BaseCommandMiddleware

class MyOwnCoinFlipMiddleware(BaseCommandMiddleware):

   # Recommended for compatibility with default middlewares
   def __init__(self, execute_blocked_handler: Optional[Callable[[ChatCommand], Awaitable[None]]] = None):
     self.execute_blocked_handler = execute_blocked_handler

   async def can_execute(self, cmd: ChatCommand) -> bool:
      # Return True if the command should execute and False otherwise
      import random
      return random.choice([True, False])

   async def was_executed(self, cmd: ChatCommand):
      # Called whenever a command this Middleware is attached to was executed
      pass

To implement OAuth authentication in a headless environment (where you don't have a direct browser interface, such as a web server), use the UserAuthenticator class. This involves setting up a web server (like Flask) to handle the redirect from Twitch.

Workflow

1. Configure Twitch Dashboard: Add your redirect URL (e.g., http://localhost:5000/login/confirm) as an "OAuth Redirect URL" in the Twitch Developer Console.
2. Initialize Twitch and Authenticator: Create a Twitch instance and pass it to UserAuthenticator along with your desired AuthScope and your redirect url.
3. Redirect to Twitch: Create a route (e.g., /login) that redirects the user to the URL generated by auth.return_auth_url().
4. Handle Callback: Create a callback route (e.g., /login/confirm) to receive the state and code parameters from Twitch.
   • Verify that the state from the request matches auth.state to prevent CSRF.
   • Use await auth.authenticate(user_token=code) to exchange the code for tokens.
   • Use await twitch.set_user_authentication(token, TARGET_SCOPE, refresh) to apply the new authentication to your Twitch instance.

Implementation Example

import asyncio
from twitchAPI.twitch import Twitch
from twitchAPI.oauth import UserAuthenticator
from twitchAPI.type import AuthScope, TwitchAPIException
from flask import Flask, redirect, request

APP_ID = 'my_app_id'
APP_SECRET = 'my_app_secret'
TARGET_SCOPE = [AuthScope.CHAT_EDIT, AuthScope.CHAT_READ]
MY_URL = 'http://localhost:5000/login/confirm'

app = Flask(__name__)
twitch: Twitch
auth: UserAuthenticator

@app.route('/login')
def login():
    return redirect(auth.return_auth_url())

@app.route('/login/confirm')
async def login_confirm():
    state = request.args.get('state')
    if state != auth.state:
        return 'Bad state', 401
    code = request.args.get('code')
    if code is None:
        return 'Missing code', 400
    try:
        token, refresh = await auth.authenticate(user_token=code)
        await twitch.set_user_authentication(token, TARGET_SCOPE, refresh)
    except TwitchAPIException as e:
        return 'Failed to generate auth token', 400
    return 'Sucessfully authenticated!'

async def twitch_setup():
    global twitch, auth
    twitch = await Twitch(APP_ID, APP_SECRET)
    auth = UserAuthenticator(twitch, TARGET_SCOPE, url=MY_URL)

asyncio.run(twitch_setup())