janus-api 2.0.3

Async Python toolkit for the Janus WebRTC Gateway, with typed protocol bindings, plugin abstractions, and production-ready ASGI service tooling.

janus-api

Async Python bindings and service tooling for the Janus WebRTC Gateway.

This repository now provides two things in one package:

1. A core async Janus client (sessions, transport, typed requests/responses, plugin abstraction).
2. A production-oriented ASGI service layer (manager/readiness APIs, admin monitoring, logs streaming, optional Kafka + Timescale integrations).

Package Metadata

• Name: janus-api
• Version: 2.0.0
• Python: >=3.14
• Author: Leydotpy <[e-mail]>

Core Capabilities

• Async Janus session lifecycle over WebSocket (create, keepalive, destroy)
• Typed protocol models with Pydantic
• Plugin runtime + registry with cached plugin handles
• Built-in plugin implementations:
  • publisher / subscriber (VideoRoom)
  • audiobridge
  • streaming
  • textroom
  • sip
  • peer_to_peer (videocall)
• Event handling through callback and ReactiveX patterns
• ASGI app factory with lifespan orchestration
• Optional leader/follower coordination using Redis for multi-worker deployments
• Optional admin monitor + metrics persistence (Timescale/Postgres)
• Optional Janus EventHandler ingestion to Kafka
• REST and WebSocket endpoints for logs and operational status

Installation

# uv (recommended)
uv add janus-api

# pip
pip install janus-api

For local development in this repo:

uv sync

Quick Start (Client Runtime)

import asyncio

from janus_api.conf import Janus
from janus_api.servers import JanusSessionManager
from janus_api.lib import Plugin


async def main():
    manager = JanusSessionManager()
    Janus.set_manager(manager)
    await manager.start()

    try:
        session = Janus.get_session()
        publisher = await Plugin(
            identifier="publisher",
            session=session,
            room="1234",
            username="alice",
        ).attach()

        # Example operation
        await publisher.join()

        await publisher.detach()
    finally:
        await manager.stop()


asyncio.run(main())

Quick Start (ASGI App)

from janus_api.servers import create_asgi_app

app = create_asgi_app(
    debug=False,
    routes=[],
    mount_rest_api=True,
)

Run with:

uvicorn my_module:app --host 0.0.0.0 --port 8000

By default, internal Janus REST apps are mounted under /janus when MOUNT_REST_API=1.

VideoRoom Example

from janus_api.lib import Plugin
from janus_api.models.videoroom import SubscriberStreams

# Publisher
publisher = await Plugin(
    identifier="publisher",
    session=session,
    room="1234",
    username="alice",
).attach()

resp = await publisher.join_and_configure(
    sdp="<offer-sdp>",
    sdp_type="offer",
    audio=True,
    video=True,
)

# Subscriber
subscriber = await Plugin(
    identifier="subscriber",
    session=session,
    room="1234",
    username="bob",
).attach()

await subscriber.join(streams=[SubscriberStreams(feed="123456")])
await subscriber.watch(sdp="<answer-sdp>", sdp_type="answer")

REST Surface (when mounted)

Base prefix: /janus

• /manager:
  • GET / health details
  • GET /ready readiness check
• /events:
  • POST /janus-events (HTTP Basic auth, forwards events to Kafka)
• /logs:
  • GET / paginated logs
  • GET /levels
  • GET /{idx}
  • WS /ws/logs
• /admin (if enabled):
  • Janus admin/monitor APIs and realtime websocket monitor

Configuration

Settings are loaded from janus_api.conf.settings and can be overridden by env vars.

Important variables:

• JANUS_SESSION_URL (default: ws://localhost:8188/janus)
• LEADER_MODE (true/false, default enabled in local settings)
• REDIS_URL (leader election backend)
• JANUS_SESSION_LEADER_PROXY_HOST
• JANUS_SESSION_LEADER_PROXY_PORT
• MOUNT_REST_API
• MOUNT_LOGGING_APP
• JANUS_ENABLE_ADMIN
• JANUS_ADMIN_URL
• JANUS_ADMIN_WS_URL
• JANUS_ADMIN_SECRET
• JANUS_ADMIN_API_KEY
• TIMESCALE_DSN
• JANUS_ENABLE_EVENTS
• KAFKA_BOOTSTRAP
• KAFKA_EVENT_HANDLER_TOPIC
• EVENT_HANDLER_USER
• EVENT_HANDLER_PASS

Runtime Dependencies

From pyproject.toml:

• websockets, aiohttp, httpx
• pydantic, pyee, reactivex
• asgiref, jinja2
• redis, aiokafka
• asyncpg, psycopg[binary,pool]
• python-decouple

Project Layout

src/janus_api/
  api/rest/        # FastAPI apps (manager, events, logs, admin)
  conf/            # Global settings + runtime Janus accessor
  contrib/admin/   # Janus admin monitor + storage integrations
  core/            # Managers, logging, exceptions, utilities
  lib/             # Plugin runtime, loader, registry, event bus
  models/          # Typed Janus request/response payloads
  servers/         # Session manager, proxy, ASGI app/lifespan
  session/         # Session abstractions and websocket implementation
  transport/       # Transport layer

Known Gaps

• Some plugin classes are scaffolds or partial implementations (p2p, parts of textroom/sip).
• Public API exports at package root are intentionally minimal; most imports are from submodules.
• Current test assets include exploratory scripts; a formal automated test matrix is still limited.

Roadmap / TODO (Future Builds)

• Stabilize and document a single public import surface at janus_api root.
• Complete unimplemented plugin methods (especially P2P/TextRoom/SIP).
• Add HTTP transport mode alongside WebSocket transport.
• Provide first-class pytest suite with CI (unit, integration, failure-mode tests).
• Add typed examples for FastAPI, Django, and worker-style deployments.
• Harden leader election/proxy behavior with stronger distributed guarantees.
• Add OpenTelemetry tracing and richer metrics endpoints.
• Publish versioned API docs (MkDocs or Sphinx) with generated model references.
• Remove accidental runtime artifacts from source tree (__pycache__, logs) during packaging.

Contributing

Contributions are welcome. Prioritize:

• clear reproduction steps for bug reports
• tests for behavior changes
• backward-compatibility notes for API changes

License

MIT