OBS Integration¶

StarStreamer provides comprehensive integration with OBS Studio through WebSocket connections, enabling real-time scene control, source management, stream automation, and advanced features directly from Python code.

Overview¶

The OBS integration provides complete control over OBS Studio with 110+ implemented methods covering:

Scene Management - Switch scenes, manage scene collections, create/remove scenes
Source Control - Manage sources, filters, and scene items with full transform support
Stream & Recording - Control streaming, recording with pause/resume, chapters, and file splitting
Virtual Camera & Replay Buffer - Full control over virtual camera and replay buffer features
Studio Mode - Preview/program scene management and transitions
Advanced Features - Screenshot capture, batch operations, media control, and filter settings
Audio Management - Input controls, audio tracks, monitoring, and balance adjustment
Profile & Collections - Scene collection and profile switching

Installation¶

OBS integration is an optional plugin. Install with:

# Install with OBS support
pip install starstreamer[obs]

# Or using uv
uv pip install starstreamer[obs]

Configuration¶

YAML Configuration¶

Configure OBS WebSocket connection in your config.yaml file:

# OBS WebSocket Configuration
obs:
  host: "localhost"
  port: 4455
  password: "your_obs_password"
  timeout: 60                    # Request timeout in seconds
  event_subscriptions: "minimal" # Event subscription level (none, minimal, all, or custom)
  custom_subscriptions: 0        # Custom subscription bitmask if using custom mode

Or use environment variable substitution:

obs:
  host: "${OBS_HOST:-localhost}"
  port: "${OBS_PORT:-4455}"
  password: "${OBS_PASSWORD}"
  timeout: "${OBS_TIMEOUT:-60}"
  event_subscriptions: "${OBS_EVENT_SUBSCRIPTIONS:-minimal}"
  custom_subscriptions: "${OBS_CUSTOM_SUBSCRIPTIONS:-0}"

OBS Studio Setup¶

Enable WebSocket Server (OBS Studio 28.0+):
Go to Tools → WebSocket Server Settings
Check "Enable WebSocket server"
Set Server Port to 4455 (default)
Set a secure Server Password
Click OK
For Older OBS Versions (pre-28.0):
Install the obs-websocket plugin
Configure the plugin with the same settings
Network Configuration:
For local OBS: Use localhost or 127.0.0.1
For remote OBS: Use the computer's IP address
Ensure firewall allows connections on the WebSocket port

Test Connection¶

# Test OBS WebSocket connection
uv run python -c "
import asyncio
from starstreamer.plugins.obs import OBSClient

async def test():
    obs = OBSClient.get_instance()
    try:
        await obs.connect()
        version = await obs.get_version()
        print(f'Connected to OBS {version[\"obs_version\"]}')
        scenes = await obs.get_scene_list()
        print(f'Available scenes: {scenes}')
        await obs.disconnect()
    except Exception as e:
        print(f'Connection failed: {e}')

asyncio.run(test())
"

Quick Start¶

Basic Scene Control¶

from starstreamer import on_event
from starstreamer.triggers import trigger, CommandTrigger
from starstreamer.plugins.obs import OBSClient
from starstreamer.plugins.twitch import TwitchClient

@on_event("twitch.chat.message")
@trigger(CommandTrigger("!scene"))
async def switch_scene(event: Event, twitch: TwitchClient, obs: OBSClient) -> None:
    """Switch to a specific OBS scene"""
    message = event.data.get("message", "")
    parts = message.split(maxsplit=1)

    if len(parts) < 2:
        current = await obs.get_current_scene()
        await twitch.send_message(f"Current scene: {current}")
        return

    scene_name = parts[1]

    try:
        await obs.set_current_scene(scene_name)
        await twitch.send_message(f"✅ Switched to scene: {scene_name}")
    except Exception as e:
        await twitch.send_message(f"❌ Failed to switch scene: {e}")

Available Chat Commands¶

StarStreamer includes 40+ pre-built OBS chat commands:

Scene Control¶

!scene [name] - Switch scene or show current
!scenes - List all available scenes

Stream Control¶

!startstream - Start streaming
!stopstream - Stop streaming
!streamstatus - Show stream status

Recording Control¶

!startrecord - Start recording
!stoprecord - Stop recording
!pauserecord - Pause recording
!resumerecord - Resume recording
!togglerecordpause - Toggle recording pause

Virtual Camera¶

!startvcam - Start virtual camera
!stopvcam - Stop virtual camera
!togglevcam - Toggle virtual camera
!vcamstatus - Show virtual camera status

Replay Buffer¶

!startreplay - Start replay buffer
!stopreplay - Stop replay buffer
!togglereplay - Toggle replay buffer
!savereplay - Save replay buffer
!replaystatus - Show replay buffer status

Studio Mode¶

!studiomode [on/off] - Enable/disable studio mode
!preview [scene] - Set preview scene
!transition - Transition preview to program

Transitions¶

!transitions - List available transitions
!settransition [name] - Set transition type
!triggertransition - Trigger scene transition

Source Management¶

!mute [input] - Toggle input mute
!sources - List all sources
!sourceactive [name] - Check if source is active
!filters [source] - List filters on source
!togglefilter [source] [filter] - Toggle filter

Output Control¶

!outputs - List all outputs
!startoutput [name] - Start specific output
!stopoutput [name] - Stop specific output

Scene Collections¶

!collections - List scene collections
!setcollection [name] - Switch scene collection

System Information¶

!obsinfo - Show OBS version info
!obsstats - Show OBS performance stats

Batch Operations¶

!batchtoggle - Demo batch command (toggles multiple features)

OBS Client API¶

Connection Management¶

The OBS client uses a singleton pattern for consistent connection management:

from starstreamer.plugins.obs import OBSClient

# Get singleton instance (recommended)
obs = OBSClient.get_instance()

# Connect to OBS
await obs.connect()

# Check connection status
if obs.connected:
    print("Connected to OBS!")

# Disconnect
await obs.disconnect()

# Context manager support
async with OBSClient.get_instance() as obs:
    # OBS operations here
    await obs.set_current_scene("Gaming")
    # Automatically disconnects on exit

Scene Management¶

# Get current scene
current_scene = await obs.get_current_scene()

# Switch to a specific scene
await obs.set_current_scene("Gaming Scene")

# Get list of all scenes
scenes = await obs.get_scene_list()

# Create a new scene
await obs.create_scene("New Scene")

# Remove a scene
await obs.remove_scene("Old Scene")

# Scene items
items = await obs.get_scene_item_list("Gaming Scene")
await obs.set_scene_item_enabled("Scene", item_id=5, enabled=True)
await obs.create_scene_item("Scene", "Source Name")
await obs.duplicate_scene_item("Scene", item_id=5, destination_scene="Other Scene")

Source and Input Management¶

# List all sources
sources = await obs.get_source_list()

# Check if source is active
is_active = await obs.get_source_active("Webcam")

# Input management
inputs = await obs.get_input_list()
input_kinds = await obs.get_input_kind_list()
special = await obs.get_special_inputs()  # Desktop audio, mic, etc.

# Create/remove inputs
await obs.create_input("Scene", "Browser Source", "browser_source", {"url": "https://example.com"})
await obs.remove_input("Old Input")

# Audio control
await obs.toggle_input_mute("Microphone")
await obs.set_input_mute("Microphone", muted=True)
is_muted = await obs.get_input_mute("Microphone")
await obs.set_input_volume("Microphone", volume_db=-10.0)

# Audio configuration
await obs.set_input_audio_balance("Microphone", balance=0.5)  # 0.0=left, 1.0=right
await obs.set_input_audio_sync_offset("Microphone", offset=50)  # milliseconds
await obs.set_input_audio_monitor_type("Microphone", "monitor_and_output")
tracks = await obs.get_input_audio_tracks("Microphone")
await obs.set_input_audio_tracks("Microphone", {"1": True, "2": False})

Filter Management¶

# List filters on a source
filters = await obs.get_source_filter_list("Webcam")

# Enable/disable filters
await obs.set_source_filter_enabled("Webcam", "Color Correction", enabled=True)
is_enabled = await obs.get_source_filter_enabled("Webcam", "Color Correction")

# Create/remove filters
await obs.create_source_filter("Webcam", "My Filter", "color_filter", {"brightness": 0.5})
await obs.remove_source_filter("Webcam", "Old Filter")

# Get/set filter settings (Move It plugin support!)
settings = await obs.get_source_filter_settings("Source", "Move It Filter")
await obs.set_source_filter_settings("Source", "Move It Filter", {
    "transform_x": 100,
    "transform_y": 50,
    "duration": 1000
})

# Get default settings for a filter type
defaults = await obs.get_source_filter_default_settings("color_filter")

Stream and Recording Control¶

# Stream control
await obs.start_stream()
await obs.stop_stream()
status = await obs.get_stream_status()
# Returns: {"active": true, "state": "OBS_WEBSOCKET_OUTPUT_STARTED", 
#           "reconnecting": false, "timecode": "00:05:30", ...}

# Recording control
await obs.start_recording()
await obs.stop_recording()
await obs.pause_recording()
await obs.resume_recording()
await obs.toggle_recording_pause()

# Recording management
recording_dir = await obs.get_record_directory()
await obs.set_record_directory("/path/to/recordings")
await obs.split_record_file()  # Split into new file
await obs.create_record_chapter("Chapter Name")

status = await obs.get_recording_status()
# Returns: {"active": true, "state": "OBS_WEBSOCKET_OUTPUT_STARTED",
#           "timecode": "00:10:45", "duration": 645000, "bytes": 2048000}

# Stream settings
service = await obs.get_stream_service_settings()
await obs.set_stream_service_settings("rtmp_common", {"server": "rtmp://...", "key": "..."})
await obs.send_stream_caption("Live caption text")

Virtual Camera & Replay Buffer¶

# Virtual camera
await obs.start_virtual_cam()
await obs.stop_virtual_cam()
is_active = await obs.toggle_virtual_cam()  # Returns new state
status = await obs.get_virtual_cam_status()

# Replay buffer
await obs.start_replay_buffer()
await obs.stop_replay_buffer()
is_active = await obs.toggle_replay_buffer()  # Returns new state
await obs.save_replay_buffer()
last_replay = await obs.get_last_replay_buffer_replay()  # File path
status = await obs.get_replay_buffer_status()

Studio Mode & Transitions¶

# Studio mode control
is_enabled = await obs.get_studio_mode_enabled()
await obs.set_studio_mode_enabled(True)

# Preview scene (studio mode)
preview = await obs.get_current_preview_scene()
await obs.set_current_preview_scene("Next Scene")
await obs.trigger_studio_mode_transition()

# Transitions
transitions = await obs.get_scene_transition_list()
# Returns: [{"name": "Fade", "fixed": false, "configurable": true}, ...]

current = await obs.get_current_scene_transition()
# Returns: {"name": "Fade", "duration": 300, "fixed": false, ...}

await obs.set_current_scene_transition("Cut")
await obs.trigger_scene_transition()

# Transition cursor (for manual transitions)
cursor = await obs.get_scene_transition_cursor()  # 0.0 to 1.0
await obs.set_scene_transition_cursor(0.5)

Screenshot Functionality¶

# Save screenshot to file
file_path = await obs.save_screenshot("png", "/path/to/screenshot.png")
file_path = await obs.save_screenshot("jpg", width=1920, height=1080)

# Get screenshot as base64
base64_data = await obs.get_screenshot("png")

# Preview screenshot (studio mode only)
preview_path = await obs.save_preview_screenshot("png", "/path/to/preview.png")
preview_data = await obs.get_preview_screenshot("png")

# Source screenshot
source_data = await obs.get_source_screenshot("Webcam", "png")
source_path = await obs.save_source_screenshot("Webcam", "png", "/path/to/source.png")

Scene Item Transforms¶

# Get transform properties
transform = await obs.get_scene_item_transform("Scene", item_id=5)
# Returns: {"position": {"x": 100, "y": 200}, "rotation": 0, 
#           "scale": {"x": 1.0, "y": 1.0}, "crop": {...}, ...}

# Set transform properties
await obs.set_scene_item_transform("Scene", item_id=5, {
    "position": {"x": 150, "y": 250},
    "scale": {"x": 1.5, "y": 1.5},
    "rotation": 45.0
})

# Lock/unlock items
is_locked = await obs.get_scene_item_locked("Scene", item_id=5)
await obs.set_scene_item_locked("Scene", item_id=5, locked=True)

# Reorder items
index = await obs.get_scene_item_index("Scene", item_id=5)
await obs.set_scene_item_index("Scene", item_id=5, index=0)  # Move to front

Media Control¶

# Get media input status
status = await obs.get_media_input_status("Media Source")
# Returns: {"state": "playing", "duration": 120000, "cursor": 30000}

# Control media playback
await obs.set_media_input_cursor("Media Source", cursor=60000)  # Seek to 1 minute
await obs.offset_media_input_cursor("Media Source", offset=5000)  # Skip 5 seconds
await obs.trigger_media_input_action("Media Source", "play")
await obs.trigger_media_input_action("Media Source", "pause")
await obs.trigger_media_input_action("Media Source", "restart")
await obs.trigger_media_input_action("Media Source", "stop")
await obs.trigger_media_input_action("Media Source", "next")
await obs.trigger_media_input_action("Media Source", "previous")

Output Management¶

# List all outputs
outputs = await obs.get_output_list()
# Returns: [{"name": "adv_stream", "kind": "rtmp_output", "active": true}, ...]

# Control outputs
await obs.start_output("NDI Output")
await obs.stop_output("NDI Output")
status = await obs.get_output_status("NDI Output")

# Output settings
settings = await obs.get_output_settings("Virtual Output")
await obs.set_output_settings("Virtual Output", {"resolution": "1920x1080"})

Scene Collections & Profiles¶

# Scene collections
collections = await obs.get_scene_collection_list()
current_collection = await obs.get_current_scene_collection()
await obs.set_current_scene_collection("Stream Layout")
await obs.create_scene_collection("New Collection")

# Profiles
profiles = await obs.get_profile_list()
current_profile = await obs.get_current_profile()
await obs.set_current_profile("High Quality")
await obs.create_profile("Custom Profile")
await obs.remove_profile("Old Profile")

Batch Operations¶

Execute multiple OBS operations atomically:

# Create batch request
batch_requests = [
    {"requestType": "SetCurrentProgramScene", "requestData": {"sceneName": "Starting"}},
    {"requestType": "Sleep", "requestData": {"sleepMillis": 2000}},
    {"requestType": "SetInputMute", "requestData": {"inputName": "Mic", "inputMuted": false}},
    {"requestType": "StartStream", "requestData": {}},
    {"requestType": "StartRecord", "requestData": {}},
]

# Execute batch atomically
results = await obs.batch_request(batch_requests)

# Check results
for i, result in enumerate(results):
    if result.get("requestStatus", {}).get("result", False):
        print(f"Request {i} succeeded")
    else:
        print(f"Request {i} failed: {result.get('requestStatus', {}).get('comment')}")

# Sleep utility for delays
await obs.sleep(1000)  # Sleep 1 second

System Information¶

# Get OBS version and capabilities
version = await obs.get_version()
# Returns: {"obs_version": "30.0.0", "obs_web_socket_version": "5.4.0",
#           "rpc_version": 1, "available_requests": [...], 
#           "supported_image_formats": ["png", "jpg", "bmp"]}

# Get performance statistics
stats = await obs.get_stats()
# Returns: {"cpu_usage": 25.5, "memory_usage": 1024.0, "active_fps": 29.97,
#           "render_skipped_frames": 5, "output_total_frames": 9995, ...}

Advanced Integration Examples¶

Scene-Based Automation¶

from starstreamer import on_event
from starstreamer.triggers import trigger, CommandTrigger, ModOnlyTrigger

# Auto-configure scene based on game
@on_event("twitch.chat.message")
@trigger(CommandTrigger("!game"))
@trigger(ModOnlyTrigger())
async def setup_game_scene(event: Event, twitch: TwitchClient, obs: OBSClient) -> None:
    """Configure OBS for specific game"""
    message = event.data.get("message", "")
    game = message.split(maxsplit=1)[1] if len(message.split()) > 1 else ""

    game_configs = {
        "valorant": {
            "scene": "FPS Gaming",
            "filters": {"Webcam": {"Chroma Key": True}},
            "audio": {"Game Audio": -5.0, "Microphone": -10.0}
        },
        "minecraft": {
            "scene": "Casual Gaming",
            "filters": {"Webcam": {"Chroma Key": False}},
            "audio": {"Game Audio": -8.0, "Microphone": -12.0}
        }
    }

    if game.lower() not in game_configs:
        await twitch.send_message(f"No config for {game}")
        return

    config = game_configs[game.lower()]

    # Switch scene
    await obs.set_current_scene(config["scene"])

    # Configure filters
    for source, filters in config["filters"].items():
        for filter_name, enabled in filters.items():
            await obs.set_source_filter_enabled(source, filter_name, enabled)

    # Set audio levels
    for input_name, volume in config["audio"].items():
        await obs.set_input_volume(input_name, volume)

    await twitch.send_message(f"✅ OBS configured for {game}")

Dynamic Filter Control (Move It Plugin)¶

@on_event("twitch.chat.message")
@trigger(CommandTrigger("!animate"))
async def animate_source(event: Event, twitch: TwitchClient, obs: OBSClient) -> None:
    """Animate source using Move It plugin"""
    message = event.data.get("message", "")
    parts = message.split()

    if len(parts) < 3:
        await twitch.send_message("Usage: !animate <source> <preset>")
        return

    source_name = parts[1]
    preset = parts[2]

    animation_presets = {
        "bounce": {"transform_y": -100, "duration": 500, "easing": "bounce"},
        "slide": {"transform_x": 1920, "duration": 1000, "easing": "ease_out"},
        "spin": {"rotation": 360, "duration": 2000, "easing": "linear"}
    }

    if preset not in animation_presets:
        await twitch.send_message(f"Unknown preset. Available: {', '.join(animation_presets.keys())}")
        return

    settings = animation_presets[preset]
    await obs.set_source_filter_settings(source_name, "Move It", settings)
    await twitch.send_message(f"✨ Animating {source_name} with {preset}")

Stream Startup Sequence¶

@on_event("twitch.chat.message")
@trigger(CommandTrigger("!goLive"))
@trigger(ModOnlyTrigger())
async def automated_stream_start(event: Event, twitch: TwitchClient, obs: OBSClient, logger: logging.Logger) -> None:
    """Automated stream startup sequence"""
    try:
        # Batch all startup operations
        startup_batch = [
            # Set starting scene
            {"requestType": "SetCurrentProgramScene", "requestData": {"sceneName": "Starting Soon"}},
            {"requestType": "Sleep", "requestData": {"sleepMillis": 1000}},

            # Unmute audio sources
            {"requestType": "SetInputMute", "requestData": {"inputName": "Microphone", "inputMuted": False}},
            {"requestType": "SetInputMute", "requestData": {"inputName": "Desktop Audio", "inputMuted": False}},

            # Start outputs
            {"requestType": "StartVirtualCam", "requestData": {}},
            {"requestType": "StartReplayBuffer", "requestData": {}},
            {"requestType": "StartRecord", "requestData": {}},
            {"requestType": "Sleep", "requestData": {"sleepMillis": 2000}},
            {"requestType": "StartStream", "requestData": {}},
        ]

        await twitch.send_message("🚀 Initiating stream startup sequence...")
        results = await obs.batch_request(startup_batch)

        # Check results
        success_count = sum(1 for r in results if r.get("requestStatus", {}).get("result", False))

        if success_count == len(results):
            await twitch.send_message("✅ Stream is LIVE! All systems operational!")
            logger.info("Stream started successfully")
        else:
            await twitch.send_message(f"⚠️ Stream started with {len(results) - success_count} warnings")
            logger.warning(f"Stream startup had {len(results) - success_count} failed operations")

    except Exception as e:
        await twitch.send_message(f"❌ Stream startup failed: {e}")
        logger.error(f"Failed to start stream: {e}")

Screenshot on Demand¶

@on_event("twitch.chat.message")
@trigger(CommandTrigger("!clip"))
async def save_moment(event: Event, twitch: TwitchClient, obs: OBSClient) -> None:
    """Save current moment as screenshot"""
    username = event.data.get("user", {}).get("username", "Unknown")

    try:
        # Save screenshot with timestamp
        from datetime import datetime
        timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
        filename = f"clip_{username}_{timestamp}.png"
        filepath = f"/screenshots/{filename}"

        saved_path = await obs.save_screenshot("png", filepath)
        await twitch.send_message(f"📸 Moment captured! Saved as {filename}")

        # Also save replay buffer if active
        status = await obs.get_replay_buffer_status()
        if status["active"]:
            await obs.save_replay_buffer()
            await twitch.send_message("🎬 Video clip also saved!")

    except Exception as e:
        await twitch.send_message(f"❌ Failed to capture moment: {e}")

Error Handling¶

Connection Resilience¶

from starstreamer.plugins.obs import OBSClient
import asyncio

class ResilientOBSManager:
    def __init__(self):
        self.obs = OBSClient.get_instance()
        self.reconnect_attempts = 0
        self.max_reconnects = 5

    async def ensure_connected(self) -> bool:
        """Ensure OBS is connected with automatic reconnection"""
        if self.obs.connected:
            return True

        for attempt in range(self.max_reconnects):
            try:
                await self.obs.connect()
                self.reconnect_attempts = 0
                return True
            except Exception as e:
                self.reconnect_attempts += 1
                wait_time = 2 ** attempt  # Exponential backoff
                await asyncio.sleep(wait_time)

        return False

    async def safe_operation(self, operation, *args, **kwargs):
        """Execute OBS operation with automatic reconnection"""
        try:
            return await operation(*args, **kwargs)
        except Exception as e:
            # Try to reconnect and retry once
            if await self.ensure_connected():
                return await operation(*args, **kwargs)
            raise

Graceful Degradation¶

@on_event("twitch.chat.message")
@trigger(CommandTrigger("!status"))
async def stream_status_resilient(event: Event, twitch: TwitchClient, obs: OBSClient) -> None:
    """Stream status that handles OBS disconnection gracefully"""
    status_parts = []

    # Always available: basic info
    status_parts.append("Stream Status:")

    # Try to get OBS info
    try:
        if obs.connected:
            stream = await obs.get_stream_status()
            recording = await obs.get_recording_status()
            scene = await obs.get_current_scene()

            status_parts.append(f"Scene: {scene}")
            status_parts.append(f"Live: {'✅' if stream['active'] else '❌'}")
            status_parts.append(f"Recording: {'✅' if recording['active'] else '❌'}")
        else:
            status_parts.append("(OBS disconnected)")
    except Exception as e:
        status_parts.append("(OBS unavailable)")

    await twitch.send_message(" | ".join(status_parts))

Testing¶

Mock OBS Client¶

import pytest
from unittest.mock import AsyncMock, MagicMock
from starstreamer.plugins.obs import OBSClient

@pytest.fixture
def mock_obs():
    """Create a mock OBS client for testing"""
    obs = AsyncMock(spec=OBSClient)
    obs.connected = True
    obs.get_current_scene.return_value = "Test Scene"
    obs.get_scene_list.return_value = ["Test Scene", "Gaming", "BRB"]
    obs.get_stream_status.return_value = {"active": True, "timecode": "00:05:30"}
    obs.get_version.return_value = {"obs_version": "30.0.0"}
    return obs

@pytest.mark.asyncio
async def test_scene_switch_command(mock_obs):
    """Test scene switching command"""
    from modules.obs.actions.obs_commands import scene_command

    # Mock event and services
    event = MagicMock()
    event.data = {"message": "!scene Gaming"}
    twitch = AsyncMock()

    # Execute command
    await scene_command(event, mock_obs, twitch)

    # Verify OBS interaction
    mock_obs.set_current_scene.assert_called_once_with("Gaming")
    twitch.send_message.assert_called_once()

Performance Optimization¶

Caching Frequently Accessed Data¶

from functools import lru_cache
import time

class CachedOBSClient:
    def __init__(self, obs: OBSClient):
        self.obs = obs
        self._cache = {}
        self._cache_ttl = 30  # seconds

    async def get_scene_list_cached(self):
        """Get scene list with caching"""
        cache_key = "scene_list"
        now = time.time()

        if cache_key in self._cache:
            data, timestamp = self._cache[cache_key]
            if now - timestamp < self._cache_ttl:
                return data

        # Cache miss or expired
        data = await self.obs.get_scene_list()
        self._cache[cache_key] = (data, now)
        return data

Batch Operations for Efficiency¶

async def configure_stream_layout(obs: OBSClient, layout: str) -> None:
    """Configure complete stream layout in one batch"""
    layouts = {
        "gaming": [
            {"requestType": "SetCurrentProgramScene", "requestData": {"sceneName": "Gaming"}},
            {"requestType": "SetInputVolume", "requestData": {"inputName": "Game Audio", "inputVolumeDb": -5}},
            {"requestType": "SetInputVolume", "requestData": {"inputName": "Microphone", "inputVolumeDb": -10}},
            {"requestType": "SetSourceFilterEnabled", "requestData": {
                "sourceName": "Webcam", "filterName": "Chroma Key", "filterEnabled": True
            }},
        ],
        "chatting": [
            {"requestType": "SetCurrentProgramScene", "requestData": {"sceneName": "Just Chatting"}},
            {"requestType": "SetInputVolume", "requestData": {"inputName": "Microphone", "inputVolumeDb": -8}},
            {"requestType": "SetSourceFilterEnabled", "requestData": {
                "sourceName": "Webcam", "filterName": "Chroma Key", "filterEnabled": False
            }},
        ]
    }

    if layout in layouts:
        await obs.batch_request(layouts[layout])

Best Practices¶

Security¶

Validate Input: Always validate scene and source names from user input
Use Permissions: Restrict sensitive commands to moderators or broadcasters
Sanitize Settings: When accepting filter settings from users, validate the values
Password Protection: Always use strong passwords for OBS WebSocket

Reliability¶

Connection Checks: Always verify connection before operations
Error Handling: Implement proper error handling for all OBS operations
Graceful Degradation: Design features to work even when OBS is unavailable
Reconnection Logic: Implement automatic reconnection with exponential backoff

Performance¶

Batch Operations: Use batch requests for multiple related operations
Cache Data: Cache frequently accessed data like scene lists
Async Operations: Leverage async/await for non-blocking operations
Event Subscriptions: Only subscribe to events you actually use

Troubleshooting¶

Connection Issues¶

# Debug connection problems
import logging
logging.basicConfig(level=logging.DEBUG)

async def debug_obs_connection():
    obs = OBSClient.get_instance()
    try:
        print(f"Attempting connection to {obs.config.host}:{obs.config.port}")
        await obs.connect()
        print("✅ Connected successfully!")

        version = await obs.get_version()
        print(f"OBS Version: {version['obs_version']}")
        print(f"WebSocket Version: {version['obs_web_socket_version']}")

    except Exception as e:
        print(f"❌ Connection failed: {e}")
        print("\nTroubleshooting checklist:")
        print("1. Is OBS running?")
        print("2. Is WebSocket Server enabled in OBS?")
        print("3. Is the password correct?")
        print("4. Is the port (4455) correct?")
        print("5. Check firewall settings")

Common Error Messages¶

"OBS client is not connected": Call await obs.connect() first
"obsws-python package is required": Install with pip install obsws-python
"Invalid OBS configuration": Check host/port settings in config.yaml
"Authentication failed": Verify OBS WebSocket password

OBS Integration¶

Overview¶

Installation¶

Configuration¶

YAML Configuration¶

OBS Studio Setup¶

Test Connection¶

Quick Start¶

Basic Scene Control¶

Available Chat Commands¶

Scene Control¶

Stream Control¶

Recording Control¶

Virtual Camera¶

Replay Buffer¶

Studio Mode¶

Transitions¶

Source Management¶

Output Control¶

Scene Collections¶

System Information¶

Batch Operations¶

OBS Client API¶

Connection Management¶

Scene Management¶

Source and Input Management¶

Filter Management¶

Stream and Recording Control¶

Virtual Camera & Replay Buffer¶

Studio Mode & Transitions¶

Screenshot Functionality¶

Scene Item Transforms¶

Media Control¶

Output Management¶

Scene Collections & Profiles¶

Batch Operations¶

System Information¶

Advanced Integration Examples¶

Scene-Based Automation¶

Dynamic Filter Control (Move It Plugin)¶

Stream Startup Sequence¶

Screenshot on Demand¶

Error Handling¶

Connection Resilience¶

Graceful Degradation¶

Testing¶

Mock OBS Client¶

Performance Optimization¶

Caching Frequently Accessed Data¶

Batch Operations for Efficiency¶

Best Practices¶

Security¶

Reliability¶

Performance¶

Troubleshooting¶

Connection Issues¶

Common Error Messages¶

See Also¶