get_realtime_metrics

Monitor live streaming analytics updated every 30 seconds to track active viewers, plays per minute, and bandwidth usage for real-time event monitoring and issue detection.

Instructions

Get LIVE analytics updating every 30 seconds. USE WHEN: Monitoring live events/streams, building real-time dashboards, tracking immediate campaign impact, detecting issues as they happen. RETURNS: Current active viewers, plays per minute, bandwidth usage. EXAMPLES: 'How many people watching right now?', 'Monitor live event performance', 'Track viral video in real-time'. Different from historical analytics - this is NOW.

Input Schema

TableJSON Schema

Name	Required	Description	Default
`report_type`	No	What to monitor (default: 'viewers'): 'viewers' = active viewer count, 'geographic' = viewer locations, 'quality' = streaming performance/buffering.
`entry_id`	No	Optional entry ID for content-specific metrics

Implementation Reference

src/kaltura_mcp/tools/analytics.py:533-605 (handler)

The primary handler function for the get_realtime_metrics tool. Maps user-friendly report types to Kaltura API types, calls the core analytics function, adds a current timestamp, and returns formatted JSON response.

async def get_realtime_metrics(
    manager: KalturaClientManager,
    report_type: str = "viewers",
    entry_id: Optional[str] = None,
) -> str:
    """
    Get real-time analytics data updated every ~30 seconds.

    This function provides live metrics for monitoring current activity,
    perfect for dashboards, live events, and immediate feedback.

    USE WHEN:
    - Monitoring live events or broadcasts
    - Creating real-time dashboards
    - Tracking immediate impact of campaigns
    - Monitoring current platform activity
    - Detecting issues as they happen

    Args:
        manager: Kaltura client manager
        report_type: Type of real-time data:
            - "viewers": Current viewer count and activity
            - "geographic": Live viewer distribution by location
            - "quality": Real-time streaming quality metrics
        entry_id: Optional entry ID for content-specific metrics

    Returns:
        JSON with current metrics and recent trends:
        {
            "timestamp": "2024-01-15T14:30:00Z",
            "current": {
                "active_viewers": 1234,
                "plays_per_minute": 45,
                "bandwidth_mbps": 890
            },
            "trend": {
                "viewers_change": "+12%",
                "peak_viewers": 1456,
                "trend_direction": "increasing"
            },
            "by_content": [...]  // If no entry_id specified
        }

    Examples:
        # Monitor platform-wide activity
        get_realtime_metrics(manager)

        # Track specific live event
        get_realtime_metrics(manager, entry_id="1_live123")

        # Check streaming quality
        get_realtime_metrics(manager, report_type="quality")
    """
    # Map friendly names to report types
    report_map = {
        "viewers": "realtime_users",
        "geographic": "realtime_country",
        "quality": "realtime_qos",
    }

    from .analytics_core import get_realtime_analytics

    result = await get_realtime_analytics(
        manager=manager,
        report_type=report_map.get(report_type, "realtime_users"),
        entry_id=entry_id,
    )

    # Add timestamp and formatting
    data = json.loads(result)
    data["timestamp"] = datetime.now(timezone.utc).isoformat().replace("+00:00", "Z")

    return json.dumps(data, indent=2)

src/kaltura_mcp/server.py:212-229 (schema)

Defines the input schema, description, and parameters for the get_realtime_metrics tool in the MCP server's tool list.

types.Tool(
    name="get_realtime_metrics",
    description="Get LIVE analytics updating every 30 seconds. USE WHEN: Monitoring live events/streams, building real-time dashboards, tracking immediate campaign impact, detecting issues as they happen. RETURNS: Current active viewers, plays per minute, bandwidth usage. EXAMPLES: 'How many people watching right now?', 'Monitor live event performance', 'Track viral video in real-time'. Different from historical analytics - this is NOW.",
    inputSchema={
        "type": "object",
        "properties": {
            "report_type": {
                "type": "string",
                "enum": ["viewers", "geographic", "quality"],
                "description": "What to monitor (default: 'viewers'): 'viewers' = active viewer count, 'geographic' = viewer locations, 'quality' = streaming performance/buffering.",
            },
            "entry_id": {
                "type": "string",
                "description": "Optional entry ID for content-specific metrics",
            },
        },
    },
),

src/kaltura_mcp/server.py:507-508 (registration)
Registers and dispatches calls to the get_realtime_metrics handler in the MCP server's call_tool function.
```
elif name == "get_realtime_metrics":
    result = await get_realtime_metrics(kaltura_manager, **arguments)
```

src/kaltura_mcp/tools/__init__.py:4-12 (registration)

Imports the get_realtime_metrics handler from analytics.py for use in the tools module.

from .analytics import (
    get_analytics,
    get_analytics_timeseries,
    get_geographic_breakdown,
    get_quality_metrics,
    get_realtime_metrics,
    get_video_retention,
    list_analytics_capabilities,
)

Kaltura MCP Server