get_realtime_metrics

Instructions

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,
  )