OddSockets Python SDK

Official Python SDK for OddSockets real-time messaging platform.

Features

• AsyncIO Support: Full async/await support with asyncio
• Sync Support: Traditional synchronous API available
• Type Hints: Complete type annotations for better IDE support
• Cost Effective: No per-message pricing — a monthly message allowance
• Framework Ready: Django, Flask, FastAPI integrations available

Installation

pip install oddsockets
# or
poetry add oddsockets

🏃‍♂️ Quick Start

Basic Usage (Async)

import asyncio
from oddsockets import OddSockets

async def main():
    client = OddSockets({
        'api_key': 'ak_live_1234567890abcdef',
        'user_id': 'server-user-123'
    })

    channel = client.channel('my-channel')
    
    # Subscribe to messages
    async def on_message(message):
        print(f'Received: {message}')
    
    await channel.subscribe(on_message)
    
    # Publish a message
    await channel.publish('Hello from Python!')
    
    # Keep the connection alive
    await asyncio.sleep(10)

if __name__ == '__main__':
    asyncio.run(main())

Basic Usage (Sync)

from oddsockets.sync import OddSockets

client = OddSockets(
    api_key='ak_live_1234567890abcdef',
    manager_url='https://connect.oddsockets.tyga.network'
)

channel = client.channel('my-channel')

# Subscribe to messages
def on_message(message):
    print(f'Received: {message}')

channel.subscribe(on_message)

# Publish a message
channel.publish('Hello from Python!')

# Keep the connection alive
import time
time.sleep(10)

Type Hints

from typing import Dict, Any
from oddsockets import OddSockets, Channel
from oddsockets.types import Message

client: OddSockets = OddSockets({
    'api_key': 'ak_live_1234567890abcdef'
})

channel: Channel = client.channel('typed-channel')

async def typed_handler(message: Message) -> None:
    data: Dict[str, Any] = message.data
    print(f'Typed message: {data}')

Documentation

• Python SDK docs - API reference and setup guide

Examples

• Basic Usage - Simple messaging
• Enhanced Features - Presence, history and typed events

Configuration

Client Options

from oddsockets import OddSockets

client = OddSockets({
    'api_key': 'your-api-key',          # Your OddSockets API key (or use token_provider)
    'user_id': 'user-id',               # Optional: User identifier
    'options': {                        # Optional: connection options
        'auto_connect': True,           #   Auto-connect on creation
        'reconnect_attempts': 5,        #   Max reconnection attempts
        'heartbeat_interval': 30.0      #   Heartbeat interval (seconds)
    }
})

Provide either api_key or token_provider.

Token auth for game clients (token_provider)

Game clients should never ship a raw API key. Instead, exchange the player's own signed JWT for a short-lived, scoped OddSockets token at the OddSockets token front door, and hand the SDK a token_provider callback that returns it. The SDK presents the token on the connection handshake and silently refreshes it — both shortly before it expires and on every reconnect — so the connection never lapses.

import aiohttp

async def token_provider():
    # Called for every (re)connect and by the pre-expiry refresh timer.
    # Return the token string, or a dict {token, expires_at/expiresAt, exp}.
    async with aiohttp.ClientSession() as http:
        async with http.post(
            'https://connect.oddsockets.tyga.network/v1/token',
            headers={'Authorization': f'Bearer {my_player_jwt}'},  # the player's own game JWT
            json={'channels': ['match:9', 'lobby']}                # optional least-privilege scoping
        ) as res:
            return await res.json()  # {token, expiresAt, ...}

client = OddSockets({
    'user_id': 'player_42',
    'token_provider': token_provider,
    'token_refresh_lead_ms': 120000,   # Optional: refresh this long before expiry (default 2 min)
})

client.on('token_refreshed', lambda info: None)  # Optional: observe silent refreshes.

Channel Options

# Subscribe with options
await channel.subscribe(
    callback,
    enable_presence=True,             # Enable presence tracking
    retain_history=True,              # Retain message history
    filter_expression='user.premium == True'  # Message filter
)

# Publish with options
await channel.publish(
    message,
    ttl=3600,                         # Time to live (seconds)
    metadata={'priority': 'high'},    # Additional metadata
    store_in_history=True             # Store in message history
)

Enhanced Features

Beyond core pub/sub, OddSockets ships a Slack-like enhanced surface — reactions, typing indicators, threads, read receipts, presence/status, notifications, DMs, channel management, message editing and search. It lives on client.enhanced. The pattern is always the same:

1. Send an action with an await client.enhanced.* coroutine (snake_case).
2. Receive the paired broadcast with client.on('<event>', handler).

import asyncio
from oddsockets import OddSockets

async def main():
    client = OddSockets({'api_key': 'ak_live_1234567890abcdef', 'user_id': 'alice'})
    channel = client.channel('room-42')
    await channel.subscribe()

    # Receive-path: broadcasts from other users on the channel
    client.on('user_typing', lambda e: print(f"{e['userId']} is typing"))
    client.on('reaction_added', lambda e: print(f"{e['userId']} reacted {e['emoji']}"))
    client.on('thread_reply', lambda e: print('New reply:', e))

    # Send-path: enhanced actions over the live socket
    await client.enhanced.start_typing('alice', 'room-42')
    await client.enhanced.add_reaction(
        message_id='msg-1',
        channel='room-42',
        emoji=':thumbsup:',
        user_id='alice',
        user_name='Alice'
    )
    await client.enhanced.thread_reply(
        channel='room-42',
        parent_message_id='msg-1',
        message='Replying in the thread',
        user_id='alice',
        user_name='Alice'
    )

    await asyncio.sleep(2)
    await client.disconnect()

asyncio.run(main())

Each area exposes coroutine methods on client.enhanced; the worker broadcasts the paired events which you handle with client.on(...). Query methods (get_*, search_*) await and return the worker response.

Area	Requests (await client.enhanced.*)	Broadcast events (client.on)
Typing	start_typing, stop_typing	user_typing, user_stopped_typing
Reactions	add_reaction, remove_reaction, get_reactions	reaction_added, reaction_removed
Threads	thread_reply, get_thread, subscribe_thread, follow_thread, mark_thread_read	thread_reply, thread_subscribed, thread_followed, thread_read_updated
Read receipts	mark_read, mark_all_read, get_unread_counts	user_read, unread_count_updated, all_marked_read
Messages	edit_message, delete_message, pin_message, unpin_message, get_pinned_messages, search_messages	message_edited, message_deleted, message_pinned, message_unpinned
Presence & status	set_status, set_custom_status, set_dnd, get_user_presence	user_status_changed, custom_status_updated, dnd_status_changed
Channels	create_channel, update_channel, archive_channel, invite_to_channel, join_channel, leave_channel	channel_created, channel_updated, user_invited, user_joined_channel, user_left_channel
DMs	create_dm, send_dm, get_dm_conversations	dm_created, dm_received
Notifications	subscribe_notifications, get_notifications, mark_notification_read, clear_notifications	notification, notification_read, notifications_cleared
File uploads	start_file_upload, upload_progress, upload_complete	file_upload_completed, file_upload_progress, file_upload_failed

For any worker event not wrapped above, subscribe with the raw client.on('<event>', handler) API — all enhanced broadcasts are forwarded onto the client surface.

🐍 Python Support

• Python 3.8+
• AsyncIO support
• Type hints included
• Both sync and async APIs

🧪 Testing

# Run tests
pytest

# Run tests with coverage
pytest --cov=oddsockets

# Run async tests
pytest -m asyncio

# Run integration tests
pytest tests/integration/

🔨 Building

# Install development dependencies
pip install -e ".[dev]"

# Run linting
flake8 src/
black src/
mypy src/

# Build package
python -m build

# Install locally
pip install -e .

📈 Performance

• Automatic failover - the cluster reroutes you if a node goes away, and the SDK reconnects and resubscribes on your behalf
• No per-message pricing - you buy a monthly message allowance, not individual sends
• 32 KB maximum message size - split anything larger, or publish a reference to it
• Non-blocking - built on asyncio, so publishes and deliveries never block your event loop

Uptime SLA is per plan (Pro 99.9%, Scale 99.95%, Enterprise 99.999%) — see pricing for the current commitment.

🔐 Security

• End-to-end encryption available
• API key authentication with fine-grained permissions
• Rate limiting and abuse protection
• GDPR compliant data handling

Framework Integrations

Django

# settings.py
ODDSOCKETS = {
    'API_KEY': 'ak_live_1234567890abcdef',
    'MANAGER_URL': 'https://connect.oddsockets.tyga.network'
}

# views.py
from django.http import JsonResponse
from oddsockets.django import get_client

async def send_message(request):
    client = get_client()
    channel = client.channel('notifications')
    await channel.publish({
        'user_id': request.user.id,
        'message': 'Hello from Django!'
    })
    return JsonResponse({'status': 'sent'})

FastAPI

from fastapi import FastAPI
from oddsockets import OddSockets

app = FastAPI()
client = OddSockets(api_key='ak_live_1234567890abcdef')

@app.post("/send-message")
async def send_message(message: str):
    channel = client.channel('api-messages')
    await channel.publish({'text': message})
    return {"status": "sent"}

@app.on_event("startup")
async def startup():
    await client.connect()

@app.on_event("shutdown")
async def shutdown():
    await client.disconnect()

Flask

from flask import Flask, request, jsonify
from oddsockets.sync import OddSockets

app = Flask(__name__)
client = OddSockets(api_key='ak_live_1234567890abcdef')

@app.route('/send-message', methods=['POST'])
def send_message():
    message = request.json.get('message')
    channel = client.channel('flask-messages')
    channel.publish({'text': message})
    return jsonify({'status': 'sent'})

🌍 Other SDKs

OddSockets is available in multiple languages:

• JavaScript SDK - Browser + Node.js, TypeScript ready
• Go SDK - High-performance, goroutines and channels
• Java SDK - Enterprise-ready, Spring Boot integration
• C# SDK - .NET Core/Framework, Azure integrations
• Swift SDK - iOS native, Combine framework
• Kotlin SDK - Android native, coroutines support

📞 Support

• GitHub Issues - Bug reports and feature requests
• Documentation - Complete documentation
• Community Discord - Community support
• Email: support@oddsockets.com

📄 License

Get an API Key

AI agents can sign up with a verified email in two steps — no dashboard, no human required.

Step 1: Request a verification code

curl -X POST https://oddsockets.com/api/agent-signup \
  -H "Content-Type: application/json" \
  -d '{"email": "you@example.com", "agentName": "my-agent", "platform": "python"}'

Step 2: Verify the 6-digit code from your email and get your API key

curl -X POST https://oddsockets.com/api/agent-signup/verify \
  -H "Content-Type: application/json" \
  -d '{"email": "you@example.com", "code": "123456", "agentName": "my-agent"}'

Plans

No free tier — every plan starts with a 7-day free trial.

	Starter	Pro	Scale	Enterprise
Price	$29/mo	$99/mo	$299/mo	Contact sales
Messages/mo	5M	25M	100M	Unlimited
Peak connections	200	1,000	5,000	Unlimited
MAU	Unlimited	Unlimited	Unlimited	Unlimited
Extra messages	$2.50/M	$1.60/M	$1.00/M	Included

Current pricing: oddsockets.com/#pricing.

All limits are enforced in real time. When a limit is reached, the SDK receives a RATE_LIMIT_EXCEEDED error with a retryAfter value.

Get Accredited

Prove you can build and operate real-time features on OddSockets — channels, presence, pub/sub, delivery guarantees and production liveops — on the stack itself. Three tiers (TCU / TCA / TCP), certified through tyga.games and delivered on ClassaaS.

Get accredited on tyga.games →

Support

• Documentation
• Issue Tracker
• Email Support

License

MIT License - Copyright (c) 2026 Joe Wee, Tyga.Cloud Ltd. See LICENSE for details.