tessl/pypi-pypresence

Discord RPC client written in Python for integrating applications with Discord's Rich Presence system

—

Pending

Overview

Eval results

Files

Discord RPC Client

Name: tessl/pypi-pypresence
Author: tessl

Full-featured Discord RPC client with support for authentication, guild/channel operations, voice settings, event subscriptions, and advanced Discord integration. The Client and AioClient classes provide comprehensive access to Discord's RPC API beyond simple Rich Presence.

Capabilities

Synchronous RPC Client

The Client class provides full Discord RPC functionality with synchronous method calls.

class Client(BaseClient):
    def __init__(self, client_id: str, **kwargs):
        """
        Initialize synchronous RPC client.
        
        Parameters:
        - client_id (str): Discord application client ID
        - loop: Optional event loop
        - handler: Optional error handler function
        - pipe (int): Optional specific pipe number (0-9)
        - connection_timeout (int): Connection timeout in seconds (default: 30)
        - response_timeout (int): Response timeout in seconds (default: 10)
        """

def start(self):
    """Start the client and perform handshake with Discord."""

def read(self):
    """
    Read output from Discord RPC.
    
    Returns:
    dict: RPC response data
    """

def close(self):
    """Close the connection to Discord."""

Authentication & Authorization

def authorize(self, client_id: str, scopes: list):
    """
    Request authorization for the application.
    
    Parameters:
    - client_id (str): Discord application client ID
    - scopes (list): List of OAuth2 scopes to request
    
    Returns:
    dict: Authorization response with code
    """

def authenticate(self, token: str):
    """
    Authenticate with Discord using an access token.
    
    Parameters:
    - token (str): OAuth2 access token
    
    Returns:
    dict: Authentication response with user info
    """

Activity Management

def set_activity(self, pid: int = None, state: str = None, details: str = None,
                 start: int = None, end: int = None,
                 large_image: str = None, large_text: str = None,
                 small_image: str = None, small_text: str = None,
                 party_id: str = None, party_size: list = None,
                 join: str = None, spectate: str = None,
                 match: str = None, buttons: list = None,
                 instance: bool = True):
    """
    Set Rich Presence activity with full RPC client features.
    
    Parameters:
    - All parameters same as Presence.update()
    
    Returns:
    dict: Discord RPC response
    """

def clear_activity(self, pid: int = None):
    """
    Clear current activity.
    
    Parameters:
    - pid (int): Process ID (default: current process)
    
    Returns:
    dict: Discord RPC response
    """

def send_activity_join_invite(self, user_id: str):
    """
    Send activity join invite to a user.
    
    Parameters:
    - user_id (str): Target user ID
    
    Returns:
    dict: Discord RPC response
    """

def close_activity_request(self, user_id: str):
    """
    Close activity request from a user.
    
    Parameters:
    - user_id (str): User ID to close request from
    
    Returns:
    dict: Discord RPC response
    """

Guild & Channel Operations

def get_guilds(self):
    """
    Get user's guilds (servers).
    
    Returns:
    dict: List of guild objects
    """

def get_guild(self, guild_id: str):
    """
    Get specific guild information.
    
    Parameters:
    - guild_id (str): Guild (server) ID
    
    Returns:
    dict: Guild object with details
    """

def get_channels(self, guild_id: str):
    """
    Get channels in a guild.
    
    Parameters:
    - guild_id (str): Guild (server) ID
    
    Returns:
    dict: List of channel objects
    """

def get_channel(self, channel_id: str):
    """
    Get specific channel information.
    
    Parameters:
    - channel_id (str): Channel ID
    
    Returns:
    dict: Channel object with details
    """

def select_voice_channel(self, channel_id: str):
    """
    Select a voice channel for the user.
    
    Parameters:
    - channel_id (str): Voice channel ID
    
    Returns:
    dict: Discord RPC response
    """

def get_selected_voice_channel(self):
    """
    Get currently selected voice channel.
    
    Returns:
    dict: Voice channel object or null
    """

def select_text_channel(self, channel_id: str):
    """
    Select a text channel for the user.
    
    Parameters:
    - channel_id (str): Text channel ID
    
    Returns:
    dict: Discord RPC response
    """

Voice Settings

def get_voice_settings(self):
    """
    Get current voice settings.
    
    Returns:
    dict: Voice settings object
    """

def set_voice_settings(self, _input: dict = None, output: dict = None,
                       mode: dict = None, automatic_gain_control: bool = None,
                       echo_cancellation: bool = None, noise_suppression: bool = None,
                       qos: bool = None, silence_warning: bool = None,
                       deaf: bool = None, mute: bool = None):
    """
    Set voice settings.
    
    Parameters:
    - _input (dict): Input device settings
    - output (dict): Output device settings
    - mode (dict): Voice mode settings
    - automatic_gain_control (bool): AGC enabled
    - echo_cancellation (bool): Echo cancellation enabled
    - noise_suppression (bool): Noise suppression enabled
    - qos (bool): Quality of service enabled
    - silence_warning (bool): Silence warning enabled
    - deaf (bool): Deafened state
    - mute (bool): Muted state
    
    Returns:
    dict: Discord RPC response
    """

def set_user_voice_settings(self, user_id: str, pan_left: float = None,
                            pan_right: float = None, volume: int = None,
                            mute: bool = None):
    """
    Set voice settings for a specific user.
    
    Parameters:
    - user_id (str): Target user ID
    - pan_left (float): Left audio pan (-1.0 to 1.0)
    - pan_right (float): Right audio pan (-1.0 to 1.0)
    - volume (int): Volume level (0 to 200)
    - mute (bool): User mute state
    
    Returns:
    dict: Discord RPC response
    """

Event Management

def register_event(self, event: str, func: callable, args: dict = None):
    """
    Register an event handler function.
    
    Parameters:
    - event (str): Event name to listen for
    - func (callable): Function to call when event occurs (must accept 1 parameter)
    - args (dict): Additional arguments for subscription
    
    Raises:
    - ArgumentError: Function doesn't accept exactly 1 argument
    - NotImplementedError: Function is a coroutine (use AioClient)
    """

def unregister_event(self, event: str, args: dict = None):
    """
    Unregister an event handler.
    
    Parameters:
    - event (str): Event name to stop listening for
    - args (dict): Additional arguments for unsubscription
    
    Raises:
    - EventNotFound: Event was not registered
    """

def subscribe(self, event: str, args: dict = None):
    """
    Subscribe to Discord events.
    
    Parameters:
    - event (str): Event name
    - args (dict): Event-specific arguments
    
    Returns:
    dict: Discord RPC response
    """

def unsubscribe(self, event: str, args: dict = None):
    """
    Unsubscribe from Discord events.
    
    Parameters:
    - event (str): Event name
    - args (dict): Event-specific arguments
    
    Returns:
    dict: Discord RPC response
    """

Utility Functions

def capture_shortcut(self, action: str):
    """
    Capture keyboard shortcut.
    
    Parameters:
    - action (str): Action to capture shortcut for
    
    Returns:
    dict: Discord RPC response
    """

Asynchronous RPC Client

The AioClient class provides the same functionality as Client but with async/await support.

class AioClient(BaseClient):
    def __init__(self, client_id: str, **kwargs):
        """Initialize asynchronous RPC client with same parameters as Client."""

All methods from Client are available with async/await:

async def start(self): ...
async def authorize(self, client_id: str, scopes: list): ...
async def authenticate(self, token: str): ...
async def set_activity(self, **kwargs): ...
async def get_guilds(self): ...
async def get_voice_settings(self): ...
async def register_event(self, event: str, func: callable, args: dict = None): ...
async def unregister_event(self, event: str, args: dict = None): ...
# ... all other methods as async

Note: Event handler functions for AioClient must be coroutines (async functions).

Usage Examples

Basic RPC Client

from pypresence import Client

# Initialize and start
client = Client("your_client_id")
client.start()

# Set activity
response = client.set_activity(
    state="In Competitive Match",
    details="Ranked Game - Map: Dust II",
    large_image="game_icon",
    party_size=[2, 5]
)

# Get user's guilds
guilds = client.get_guilds()
print(f"User is in {len(guilds['data']['guilds'])} servers")

client.close()

Authentication Flow

from pypresence import Client

client = Client("your_client_id")
client.start()

# Request authorization
auth_response = client.authorize("your_client_id", ["rpc", "identify"])
print(f"Visit: {auth_response['data']['code']}")

# After user authorizes, authenticate with token
token = "your_oauth_token"  # Get this from OAuth flow
user_info = client.authenticate(token)
print(f"Authenticated as: {user_info['data']['user']['username']}")

client.close()

Event Handling

from pypresence import Client
import time

def on_activity_join(data):
    print(f"Someone wants to join: {data}")
    # Handle join request

def on_activity_spectate(data):
    print(f"Someone wants to spectate: {data}")
    # Handle spectate request

client = Client("your_client_id")
client.start()

# Register event handlers
client.register_event("ACTIVITY_JOIN", on_activity_join)
client.register_event("ACTIVITY_SPECTATE", on_activity_spectate)

# Set activity with join/spectate secrets
client.set_activity(
    state="In Game",
    details="Accepting challenges",
    join="join_secret_123",
    spectate="spectate_secret_456"
)

# Keep alive to receive events
try:
    while True:
        time.sleep(1)
except KeyboardInterrupt:
    client.close()

Async RPC Client

import asyncio
from pypresence import AioClient

async def on_ready(data):
    print(f"Discord ready: {data}")

async def main():
    client = AioClient("your_client_id")
    await client.start()
    
    # Register async event handler
    await client.register_event("READY", on_ready)
    
    # Get voice settings
    voice_settings = await client.get_voice_settings()
    print(f"Current volume: {voice_settings['data']['volume']}")
    
    # Update voice settings
    await client.set_voice_settings(volume=75, mute=False)
    
    # Set activity
    await client.set_activity(
        state="Streaming",
        details="Live coding session",
        large_image="streaming_icon"
    )
    
    # Keep alive
    await asyncio.sleep(60)
    client.close()

asyncio.run(main())

Voice Channel Management

from pypresence import Client

client = Client("your_client_id")
client.start()

# Authenticate first (required for channel operations)
client.authenticate("your_token")

# Get user's guilds and find channels
guilds = client.get_guilds()
if guilds['data']['guilds']:
    guild_id = guilds['data']['guilds'][0]['id']
    channels = client.get_channels(guild_id)
    
    # Find voice channel
    voice_channels = [c for c in channels['data']['channels'] if c['type'] == 2]
    if voice_channels:
        voice_channel_id = voice_channels[0]['id']
        
        # Select voice channel
        client.select_voice_channel(voice_channel_id)
        print(f"Selected voice channel: {voice_channels[0]['name']}")

client.close()

Install with Tessl CLI