Create Your Own Plugin

This guide will take you through what you need to know to create your own plugin to connect the Stream Python AI SDK to a third-party AI provider.

Currently, we support TTS, STT, STS and VAD plugins but we’ll be adding more functionality and features.

The audience is developers familiar with the Stream Video/Audio SDK in Python who want to add support for an additional provider.

Plugin Categories

Each plugin implements one of the abstract base classes in getstream/plugins/common:

All base classes ship with:

• WebRTC integration: Your plugin sends and receives audio frames to and from a Stream video call, with the webrtc code handled for you.
• Async event emitter: Plugins use PyEE to emit events which can be handled by event listeners in your application.

Implementing the abstract methods in the base class is all that is required for a minimal integration.

System Architecture & Lifecycle

An example workflow could look like this:

1. You instantiate the plugin client and add it to your app.
2. You listen for an event (e.g. audio received), which fires and triggers your plugin.
3. Your plugin calls the third-party API.
4. Results are dispatched via an event or directly into the call, e.g. for an STT plugin a transcript event is fired.

For an STT plugin, the workflow looks like:

Call → WebRTC Track → STT.process_audio()
                               ↓
                        calls provider
                               ↓
                        Provider transcript  →  emit("transcript")
                                                        ↓
                                                Your application consumes the event

Required Overrides

The base class for each type of plugin has different abstract methods which must be implemented in your plugin code, e.g. for an STT plugin, you must override the STT.process_audio and STT.close methods.

from getstream.plugins.common import STT
from getstream.video.rtc.track_util import PcmData

class MySTT(STT):

    async def process_audio(self, pcm: PcmData, user):
        """Send PCM data to provider and await transcripts."""

    async def close(self):
        """Gracefully shut down connections."""

Quickstart Template

Below is the skeleton for a new STT plugin named AcmeSTT. Replace placeholders with real provider logic. For other plugin types, replace with their equivalent objects.

First, fork the stream-py GitHub repo and create a new branch from the webrtc branch to add your plugin.

Create these files in the getstream/plugins/acme folder:

getstream/plugins/acme/
├── pyproject.toml
├── README.md
├── stt/
│   ├── __init__.py
│   └── stt.py
└── tests/
    └── test_stt.py

pyproject.toml

[build-system]
requires = ["hatchling"]
build-backend = "hatchling.build"

[project]
name = "getstream-plugin-acme"
version = "0.1.0"
description = "Acme STT integration for GetStream"
readme = "README.md"
requires-python = ">=3.10"
dependencies = [
    "getstream[webrtc]>=2.3.0a4",
    "getstream-plugins-common>=0.1.1",
    "acmestt-sdk~=1.2",    # provider SDK
]

[tool.hatch.build.targets.wheel]
packages = ["."]

[tool.uv.sources]
getstream = { workspace = true }
getstream-plugins-common = { workspace = true }

[dependency-groups]
dev = [
    "pytest>=8.4.1",
    "pytest-asyncio>=1.0.0",
]

acme/stt/__init__.py

Leave this blank!

acme/stt/stt.py

from __future__ import annotations
import os
from getstream.plugins.common import STT
from getstream.video.rtc.track_util import PcmData

class AcmeSTT(STT):
    """Real-time STT via fictional provider Acme."""

    # This will be called by the base class `STT.process_audio` method
    async def _process_audio_impl(self, pcm: PcmData, user):
        """Send PCM data to provider and await transcripts."""

    async def close(self):
        """Gracefully shut down connections."""

tests/test_stt.py

Fill out your test suite to cover all the plugin’s functions and features.

Run the suite with:

# Install dependencies and create virtual environment
uv sync

# Run tests
uv run pytest -v

README.md

Model your readme on other plugins - explain how to use the plugin, what’s possible with it, what needs to be set, the events it emits, etc.

Emitting Custom Events

The base classes inherit from a PyEE class: pyee.asyncio.AsyncIOEventEmitter. This means emitting custom events can be done like this:

# Inside the AcmeSTT class
self.emit("custom_event", data)

Don’t forget to document them!

Packaging & Distribution

First, install dev dependencies and make sure your tests all pass.

Then, run quality checks:

uv sync
uv run pre-commit run --all-files

Once everything looks good, open a PR to the Stream repo. Once merged, we will publish the plugin under the getstream-plugin-<provider> namespace on PyPI.

Contribution Checklist

1. Fork stream-py, create a feature branch
2. Add your plugin by implementing one of the base classes
3. Ensure tests are all passing
4. Ensure new code passes the pre-commit hooks
5. List runtime and test dependencies in pyproject.toml
6. Add documentation (README.md in the plugin folder)

Support

Need help? Open an issue or discussion on the stream-py GitHub repo or email support@getstream.io. We look forward to seeing what you build with us!