Kaltura MCP Server

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