The Xtream Codes API standard remains the universal middleware protocol for IPTV players, OTT set-top boxes, and media streaming middleware. Understanding its endpoint lifecycle allows developers to build robust client players, proxy caches, and stream validation services.
🔑 Authentication Handshake (/player_api.php)
All authentication flows originate via a simple GET/POST query to /player_api.php. A valid request carries username and password parameters.
The server returns a structured JSON payload containing two primary dictionaries:
-
user_info: status, max active connections, expiry timestamp, active trial flags. -
server_info: base URL, port, protocol, RTMP port, timezone, server timestamp.
💻 Python Implementation: Async Xtream Client SDK
Below is an asynchronous client SDK built on top of aiohttp:
import aiohttp
import asyncio
from typing import Dict, Any, Optional
class XtreamCodesClient:
def __init__(self, host: str, port: int, username: str, password: str, use_https: bool = False):
protocol = "https" if use_https else "http"
self.base_url = f"{protocol}://{host}:{port}/player_api.php"
self.auth_params = {"username": username, "password": password}
self.session: Optional[aiohttp.ClientSession] = None
async def __aenter__(self):
self.session = aiohttp.ClientSession(headers={"User-Agent": "XtreamTech-SDK/1.0"})
return self
async def __aexit__(self, exc_type, exc_val, exc_tb):
if self.session:
await self.session.close()
async def authenticate(self) -> Dict[str, Any]:
async with self.session.get(self.base_url, params=self.auth_params) as resp:
data = await resp.json(content_type=None)
if data.get("user_info", {}).get("auth") == 1:
return data
raise PermissionError("Xtream authentication failed or expired account.")
async def get_live_categories(self):
params = {**self.auth_params, "action": "get_live_categories"}
async with self.session.get(self.base_url, params=params) as resp:
return await resp.json(content_type=None)
async def get_live_streams(self, category_id: Optional[str] = None):
params = {**self.auth_params, "action": "get_live_streams"}
if category_id:
params["category_id"] = category_id
async with self.session.get(self.base_url, params=params) as resp:
return await resp.json(content_type=None)
⚡ Optimization & Stream Caching Strategies
To reduce backend load during peak streaming hours, implement in-memory caching (e.g. Redis) for category trees and EPG XML dumps. Set an automated TTL of 3,600 seconds on category metadata while refreshing stream health statuses asynchronously.
🌐 Technical Reference & Middleware Documentation
For in-depth middleware architectures, Xtream API code generators, and streaming technology benchmarks, visit the developer hub at XtreamTech Engineering Portal.
Top comments (1)
Your breakdown of the Xtream Codes API—especially the separation of authentication, EPG, and stream endpoints—highlights the classic micro‑service boundaries that make scaling IPTV pipelines easier. In a recent project I wrapped a similar API with an async FastAPI gateway to handle token refresh and rate‑limit back‑pressure, which cut latency by ~30 %. Have you considered adding a lightweight health‑check layer or using OpenAPI specs to auto‑generate clien…