> ## Documentation Index
> Fetch the complete documentation index at: https://docs.pipecat.ai/llms.txt
> Use this file to discover all available pages before exploring further.

# ServiceSwitcher

> Dynamically switch between different service instances at runtime using configurable strategies

## Overview

`ServiceSwitcher` is a specialized parallel pipeline that enables dynamic switching between multiple service instances at runtime. This is useful when you need to change between different STT providers, TTS providers, or other frame processors based on user preferences, costs, performance requirements, or other runtime conditions.

The switcher uses a strategy pattern to determine which service is active. Two built-in strategies are provided: manual switching for explicit control, and automatic failover for handling service errors.

## How It Works

`ServiceSwitcher` wraps multiple services in a parallel pipeline where only the active service processes frames. Each service is "sandwiched" between two filters that check if it's the currently active service before allowing frames to pass through. When you switch services, the filters update to redirect frame flow to the newly active service.

## Constructor

```python theme={null}
from pipecat.pipeline.service_switcher import ServiceSwitcher

# Uses ServiceSwitcherStrategyManual by default
switcher = ServiceSwitcher(services=[stt_service1, stt_service2])

# Or explicitly specify a strategy
from pipecat.pipeline.service_switcher import ServiceSwitcherStrategyFailover
switcher = ServiceSwitcher(
    services=[stt_service1, stt_service2],
    strategy_type=ServiceSwitcherStrategyFailover
)
```

<ParamField path="services" type="List[FrameProcessor]" required>
  List of service instances to switch between. Can be any frame processors (STT,
  TTS, or custom processors).
</ParamField>

<ParamField path="strategy_type" type="Type[ServiceSwitcherStrategy]" default="ServiceSwitcherStrategyManual">
  The strategy class to use for switching logic. Pass the class itself, not an
  instance. Defaults to `ServiceSwitcherStrategyManual`.
</ParamField>

## Switching Strategies

### ServiceSwitcherStrategyManual

The manual strategy allows explicit control over which service is active by pushing `ManuallySwitchServiceFrame` frames into the pipeline.

**Initial State**: The first service in the list is active by default.

**Switching**: Push a `ManuallySwitchServiceFrame` with the desired service instance.

### ServiceSwitcherStrategyFailover

The failover strategy automatically switches to the next available service when the active service reports a non-fatal error. This enables automatic recovery from service failures without manual intervention.

**Initial State**: The first service in the list is active by default.

**Automatic Failover**: When the active service pushes a non-fatal `ErrorFrame`, the strategy automatically switches to the next service in the list (wrapping around to the first service if needed).

**Recovery**: The failed service remains in the list and can be switched back to manually or via application logic in the `on_service_switched` event handler. This allows implementing custom recovery policies.

```python theme={null}
from pipecat.pipeline.service_switcher import ServiceSwitcher, ServiceSwitcherStrategyFailover

switcher = ServiceSwitcher(
    services=[primary_stt, backup_stt],
    strategy_type=ServiceSwitcherStrategyFailover
)

@switcher.strategy.event_handler("on_service_switched")
async def on_switched(strategy, service):
    # Application decides when/how to recover the failed service
    print(f"Switched to: {service.name}")
```

### Custom Strategies

You can create your own switching strategy by subclassing `ServiceSwitcherStrategy` and implementing the `handle_frame` and/or `handle_error` methods.

* **`handle_frame(frame, direction)`**: Called for control frames (like `ManuallySwitchServiceFrame`). Should return the newly active service if a switch occurred, or `None` otherwise.
* **`handle_error(error)`**: Called when the active service reports a non-fatal error. Override this to implement custom error-handling logic. Should return the newly active service if a switch occurred, or `None` otherwise.

Additionally, if you want to maintain either manual switching or automatic failover as an option when writing a custom strategy, your new strategy should inherit from `ServiceSwitcherStrategyManual` or `ServiceSwitcherStrategyFailover`, respectively.

## Usage Examples

### Switching Between TTS Services

```python theme={null}
from pipecat.frames.frames import ManuallySwitchServiceFrame
from pipecat.pipeline.service_switcher import ServiceSwitcher
from pipecat.services.elevenlabs.tts import ElevenLabsTTSService
from pipecat.services.cartesia.tts import CartesiaTTSService

# Create TTS services
elevenlabs = ElevenLabsTTSService(api_key=os.getenv("ELEVENLABS_API_KEY"), voice_id=os.getenv("ELEVENLABS_VOICE_ID"))
cartesia = CartesiaTTSService(api_key=os.getenv("CARTESIA_API_KEY"), voice_id=os.getenv("CARTESIA_VOICE_ID"))

# Create switcher with both services (uses ServiceSwitcherStrategyManual by default)
tts_switcher = ServiceSwitcher(services=[elevenlabs, cartesia])

# Use in pipeline
pipeline = Pipeline([
    transport.input(),
    stt,
    context_aggregator.user(),
    llm,
    tts_switcher,
    transport.output(),
    context_aggregator.assistant()
])

# Later, switch to Cartesia
await task.queue_frame(ManuallySwitchServiceFrame(service=cartesia))
```

## Event Handlers

| Event                 | Description                 |
| --------------------- | --------------------------- |
| `on_service_switched` | Active service was switched |

```python theme={null}
@switcher.event_handler("on_service_switched")
async def on_service_switched(switcher, service):
    print(f"Switched to: {service.name}")
```

**Parameters:**

| Parameter  | Type                      | Description              |
| ---------- | ------------------------- | ------------------------ |
| `switcher` | `ServiceSwitcherStrategy` | The switcher instance    |
| `service`  | `FrameProcessor`          | The newly active service |