Official Python SDK for Stream

Check out our:

• ⭐ Chat API
• 📱 Video API
• 🔔 Activity Feeds

Features

• Video call creation and management
• Chat session creation and management
• Token generation for user authentication

Installation

To install the Stream Client Library, run the following command:

pip install getstream

# or if like us, you fell in love with uv
uv add getstream

If you want to build audio or video AI integrations, make sure to check Vision-Agents:

pip install getstream[webrtc]

# or using uv
uv add 'getstream[webrtc]'

Migrating from stream-chat?

If you are currently using stream-chat, we have a detailed migration guide with side-by-side code examples for common Chat use cases. See the Migration Guide.

Usage

To get started, you need to import the Stream class from the library and create a new instance with your API key and secret:

from getstream import Stream

client = Stream(api_key="your_api_key", api_secret="your_api_secret")

Users and Authentication

from getstream.models import UserRequest

# sync two users using the update_users method, both users will get insert or updated
client.upsert_users(
    UserRequest(
        id="tommaso-id", name="tommaso", role="admin", custom={"country": "NL"}
    ),
    UserRequest(
        id="thierry-id", name="thierry", role="admin", custom={"country": "US"}
    ),
)

# Create a JWT token for the user to connect client-side (e.g. browser/mobile app)
token = client.create_token("tommaso-id")

Video API - Calls

To create a video call, use the client.video.call method:

import uuid
from getstream.models import (
    CallRequest,
    MemberRequest,
)

call = client.video.call("default", uuid.uuid4())
call.get_or_create(
    data=CallRequest(
        created_by_id="tommaso-id",
        members=[
            MemberRequest(user_id="thierry-id"),
            MemberRequest(user_id="tommaso-id"),
        ],
    ),
)

Getting Response Data

Many calls return a StreamResponse object, with the specific dataclass for the method call nested inside. You can access this via:

response: StreamResponse[StartClosedCaptionsResponse] = call.start_closed_captions()
response.data  # Gives the StartClosedCaptionsResponse model

Reusing the client and connection pooling

Create one Stream (or AsyncStream) client when your app starts and reuse it everywhere, for example as a module-level object or on your web framework's app state. The client keeps a pool of keep-alive HTTP connections, so most requests skip the TCP and TLS handshake. Creating a new client per request throws that pool away and adds latency to every call. The sub-clients (client.video, client.chat, client.moderation, client.feeds) all share the same pool. Call client.close() (or await client.aclose()) when your app shuts down.

By default the pool allows up to 100 concurrent connections, closes connections that have been idle for 55 seconds, and uses a 10 second connect timeout and a 30 second request timeout. When every connection is busy, further requests wait for one to free up. If your app sends more concurrent requests than that, raise the limit:

client = Stream(
    api_key="your_api_key",
    api_secret="your_api_secret",
    max_conns_per_host=200,  # max concurrent connections, also kept alive for reuse
    idle_timeout=55.0,       # seconds an idle connection stays open
    connect_timeout=10.0,    # seconds for the TCP + TLS handshake
    request_timeout=30.0,    # seconds per request
)

The same settings can be set with the STREAM_MAX_CONNS_PER_HOST, STREAM_IDLE_TIMEOUT, STREAM_CONNECT_TIMEOUT and STREAM_REQUEST_TIMEOUT environment variables. For full control, pass your own httpx.Client (or httpx.AsyncClient) as http_client=; the settings above are then ignored and your client's configuration is used as-is.

Logging

The SDK emits structured log events (client.initialized, http.request.sent, http.response.received, http.request.failed) through the stdlib logging module. By default nothing is printed: pass a logging.Logger to see them.

import logging

logging.basicConfig(level=logging.DEBUG)
client = Stream(api_key="key", api_secret="secret", logger=logging.getLogger("myapp.stream"))

Each event carries structured fields via the standard extra={} mechanism (for example http.response.status_code, duration_ms, stream.endpoint_name). Query and body values for known secret keys (api_key, api_secret, token, password) are always redacted. Request/response bodies are omitted by default; pass log_bodies=True to include them (still redacted, and this emits one WARNING at construction since bodies can contain other sensitive data).

Retries

By default the client makes exactly one attempt per request and surfaces errors unchanged. Pass a RetryConfig to opt in to auto-retry:

from getstream import Stream, RetryConfig

client = Stream(api_key=..., api_secret=..., retry=RetryConfig(enabled=True, max_attempts=3, max_backoff=30.0))

Only idempotent GET/HEAD requests are retried, and only on HTTP 429 (unless the backend marked it unrecoverable) or a transport-level failure (timeout, connection reset, DNS, TLS). A 429's Retry-After header is honored (clamped to max_backoff); otherwise the delay uses full jitter over an exponential backoff. A retried failure logs http.request.failed at DEBUG; a final, non-retried failure logs it at ERROR (or not at all for a final 429, since that's already covered by http.response.received).

App configuration

# Video: update settings for a call type

# Chat: update settings for a channel type

Chat API - Channels

To work with chat sessions, use the client.chat object and implement the desired chat methods in the Chat class:

chat_instance = client.chat

# TODO: implement and call chat-related methods with chat_instance

Development

We use uv to manage dependencies and run tests.

🚀 Quick Start

Prerequisites:

• Python 3.10+ (recommended: 3.12.2)
• uv package manager

Setup:

# 1. Clone and enter the repository
git clone https://github.com/GetStream/stream-py.git
cd stream-py

# 2. Create virtual environment and install everything
uv venv --python 3.12.2
uv sync --all-extras --dev

# 3. Set up pre-commit hooks
pre-commit install

# 4. Create environment file for API credentials
cp .env.example .env
# Edit .env with your Stream API credentials

🧪 Testing

Run all tests:

uv run pytest                          # Everything
uv run pytest -v                       # Verbose output
uv run pytest -x                       # Stop on first failure

Run specific test suites:

# Main package tests
uv run pytest tests/
uv run pytest tests/test_video.py      # Specific test file

Test with coverage:

uv run pytest --cov=getstream --cov-report=html

Test configuration:

• Configuration: pytest.ini
• Fixtures: tests/fixtures.py
• Test assets: tests/assets/ (keep files < 256KB)

Testing best practices:

• Write tests as simple Python functions with assert statements
• Use fixtures from tests/fixtures.py for common setup
• Place test assets in tests/assets/ directory
• Avoid mocks unless specifically required
• Always run tests from the project root directory

CI considerations:

import pytest

@pytest.mark.skip_in_ci
def test_something():
    # This test will be skipped in GitHub Actions
    ...

🎯 Common Tasks

Install new dependency:

# Main package
uv add "new-package>=1.0.0"

# Plugin-specific
cd getstream/plugins/stt/my-plugin/
uv add "plugin-specific-dep>=2.0.0"

# WebRTC-related (add to webrtc extra)
# Edit pyproject.toml [project.optional-dependencies] webrtc section

Run linting and formatting:

uv run ruff check getstream/ tests/        # Check for issues
uv run ruff format getstream/ tests/       # Format code
uv run pre-commit run --all-files          # Run all hooks

Generate code:

./generate_webrtc.sh                       # Regenerate WebRTC bindings

Note: regenerating code requires access to internal code available only to Stream developers

🐛 Troubleshooting

Test failures:

# Run with verbose output
uv run pytest -v -s

# Run specific test
uv run pytest tests/test_video.py::test_specific_function -v

Releases

Releases are driven by release-please.

• Merge PRs to main with conventional-commit titles. The repo is squash-only with squash_merge_commit_title: PR_TITLE, so the PR title becomes the commit subject and decides the next version: feat: is a minor, fix: and perf: are a patch, feat!: or <type>(scope)!: is a major. Other types (chore, ci, docs, test, refactor) ship nothing. Both settings are load-bearing: release-please reads commit messages, never PR titles, and a merge commit's subject is not conventional.
• release-please keeps a Release PR open with the version bump in pyproject.toml, uv.lock and CHANGELOG.md. It is opened by github-actions[bot], so approve it and run its held checks like any other PR.
• Merging the Release PR creates the tag and the GitHub Release on that merge commit and publishes to PyPI via Trusted Publishing (OIDC), with no further test run: the Release PR adds only the version bump and changelog to an already-tested main. A hotfix release from N.x runs the unit lane first, since its commits were pushed without a PR. A tag and a GitHub Release cannot be withdrawn, a failed publish can be retried. If the commit waiting to be tagged is not this run's commit, the workflow stands down instead of tagging it, and says so in the run summary.

While a merged Release PR is waiting to be tagged, no new Release PR is opened or refreshed, so that release-please has a release commit to stop its walk at. If the release job failed on that merge commit, use "Re-run failed jobs" on its Release run. If the commit is genuinely broken, remove the autorelease: pending label from the merged Release PR by hand, then release forward; nothing clears that state automatically.

To retry a publish that failed after the release was tagged, use "Re-run failed jobs" on that workflow run. Once GitHub has retired the run, dispatch Release from main with publish_tag set to the tag (for example v6.1.1), which builds and publishes that tag without touching release-please.

To force a specific version, type Release-As: X.Y.Z in the commit message box of the squash dialog when merging a PR; the PR description is not copied there. To hotfix while main carries unreleased work, branch N.x from the last tag, cherry-pick the fix, and merge the Release PR that release-please opens against that branch.

last-release-sha in release-please-config.json is temporary. v6.1.0 sits on a bump commit the previous workflow created off-branch and never pushed, so release-please cannot reach it by walking main and would otherwise treat the whole history as unreleased. Delete the key once a release-please-created release exists on main; the walk stops at that release commit before it reaches the pin.

License

This project is licensed under the MIT License.

Contributing

Contributions are welcome! Please read the contributing guidelines to get started.