Kaltura MCP Server

Instructions

Implementation Reference

• src/kaltura_mcp/tools/analytics.py:102-195 (handler)

  Primary handler function for get_analytics_timeseries tool. Processes input arguments, determines default metrics, calls the core graph analytics function, reformats the response into a time-series structure optimized for charts with 'series' and 'metadata'.
  async def get_analytics_timeseries( manager: KalturaClientManager, from_date: str, to_date: str, report_type: str = "content", metrics: Optional[List[str]] = None, entry_id: Optional[str] = None, interval: str = "days", dimension: Optional[str] = None, ) -> str: """ Get time-series analytics data optimized for charts and visualizations. This function returns analytics data as time-series with consistent intervals, perfect for creating line charts, area charts, and trend visualizations. USE WHEN: - Creating charts or graphs showing trends over time - Building dashboards with visual analytics - Comparing metrics across different time periods - Showing growth or decline patterns - Visualizing seasonal trends Args: manager: Kaltura client manager from_date: Start date (YYYY-MM-DD) to_date: End date (YYYY-MM-DD) report_type: Type of report (default: "content") metrics: List of metrics to include (e.g., ["plays", "views", "avg_time"]) entry_id: Optional specific entry ID interval: Time interval - "hours", "days", "weeks", "months" (default: "days") dimension: Optional dimension for grouping Returns: JSON with time-series data formatted for visualization: { "series": [ { "metric": "count_plays", "data": [{"date": "2024-01-01", "value": 150}, ...] }, { "metric": "unique_viewers", "data": [{"date": "2024-01-01", "value": 89}, ...] } ], "metadata": {...} } Examples: # Daily play counts for a video get_analytics_timeseries(manager, from_date, to_date, entry_id="1_abc", metrics=["plays"]) # Monthly platform trends get_analytics_timeseries(manager, from_date, to_date, interval="months", report_type="platforms") """ from .analytics_core import get_analytics_graph # If no metrics specified, use common ones based on report type if not metrics: metrics_map = { "content": ["count_plays", "unique_viewers", "avg_time_viewed"], "user_engagement": ["count_plays", "unique_known_users", "avg_completion_rate"], "geographic": ["count_plays", "unique_viewers"], "platforms": ["count_plays", "unique_viewers"], } metrics = metrics_map.get(report_type, ["count_plays", "unique_viewers"]) result = await get_analytics_graph( manager=manager, from_date=from_date, to_date=to_date, report_type=report_type, entry_id=entry_id, interval=interval, dimension=dimension, ) # Reformat to emphasize time-series nature data = json.loads(result) if "graphs" in data: # Rename "graphs" to "series" for clarity data["series"] = data.pop("graphs") # Add interval metadata data["metadata"] = { "interval": interval, "report_type": report_type, "date_range": data.get("dateRange", {}), } return json.dumps(data, indent=2)
• src/kaltura_mcp/tools/analytics_core.py:205-398 (handler)

  Core implementation that performs the actual Kaltura API calls to getGraphs and getTotal for time-series data, parses the results, and structures them into graphs suitable for visualization.
  async def get_analytics_graph( manager: KalturaClientManager, from_date: str, to_date: str, report_type: str = "content", entry_id: Optional[str] = None, user_id: Optional[str] = None, object_ids: Optional[str] = None, interval: str = "days", dimension: Optional[str] = None, filters: Optional[Dict[str, str]] = None, ) -> str: """ Get analytics data in graph format suitable for visualization. This function returns time-series data for creating charts and graphs. Use this when you need data for visualization purposes. Args: manager: Kaltura client manager from_date: Start date (YYYY-MM-DD) to_date: End date (YYYY-MM-DD) report_type: Type of analytics report entry_id: Specific entry ID (if applicable) user_id: Specific user ID (if applicable) object_ids: Comma-separated object IDs interval: Time interval (days, weeks, months) dimension: Additional dimension for grouping filters: Additional filters Returns: JSON with graph data including: - Multiple metrics as separate time series - Each series has metric name and array of date/value pairs - Summary totals for the period Example response: { "graphs": [ { "metric": "count_plays", "data": [ {"date": "2024-01-01", "value": 100}, {"date": "2024-01-02", "value": 150} ] }, { "metric": "sum_time_viewed", "data": [ {"date": "2024-01-01", "value": 3600}, {"date": "2024-01-02", "value": 5400} ] } ], "summary": { "total_plays": 250, "total_time_viewed": 9000 } } """ # Validate inputs if report_type not in REPORT_TYPE_MAP: return json.dumps( { "error": f"Unknown report type: {report_type}", "available_types": list(REPORT_TYPE_MAP.keys()), }, indent=2, ) # Validate entry ID if provided if entry_id and not validate_entry_id(entry_id): return json.dumps({"error": "Invalid entry ID format"}, indent=2) # Check if report type requires specific IDs requires_object_ids = [ "engagement_timeline", "specific_user_engagement", "specific_user_usage", ] if report_type in requires_object_ids and not (entry_id or user_id or object_ids): return json.dumps({"error": f"Report type '{report_type}' requires object IDs"}, indent=2) try: # Try to import KalturaReportType try: from KalturaClient.Plugins.Report import ( KalturaEndUserReportInputFilter, KalturaReportInputFilter, ) except ImportError: # Fallback if imports fail pass # Get client client = manager.get_client() # Build report filter try: # Try to use KalturaEndUserReportInputFilter for user-facing reports user_reports = [ "user_engagement", "specific_user_engagement", "user_top_content", "user_usage", "specific_user_usage", ] if report_type in user_reports: report_filter = KalturaEndUserReportInputFilter() else: report_filter = KalturaReportInputFilter() except Exception: # Fallback to generic object with toParams method class FallbackFilter: def __init__(self): pass def toParams(self): """Return dict representation for API calls.""" return {k: v for k, v in self.__dict__.items() if not k.startswith("_")} report_filter = FallbackFilter() # Set filter properties report_filter.fromDate = int(datetime.strptime(from_date, "%Y-%m-%d").timestamp()) report_filter.toDate = int(datetime.strptime(to_date, "%Y-%m-%d").timestamp()) # Set interval if hasattr(report_filter, "interval"): interval_map = { "days": "days", "weeks": "weeks", "months": "months", "years": "years", } report_filter.interval = interval_map.get(interval, "days") # Add object filters if entry_id: report_filter.entryIdIn = entry_id if user_id: report_filter.userIds = user_id # Set object IDs obj_ids = object_ids if entry_id and not obj_ids: obj_ids = entry_id elif user_id and not obj_ids: obj_ids = user_id else: obj_ids = None # Get report type ID report_type_id = REPORT_TYPE_MAP[report_type] # Call getGraphs API graphs_result = client.report.getGraphs( reportType=report_type_id, reportInputFilter=report_filter, dimension=dimension, objectIds=obj_ids, ) # Also get totals totals_result = client.report.getTotal( reportType=report_type_id, reportInputFilter=report_filter, objectIds=obj_ids, ) # Parse results response = { "reportType": REPORT_TYPE_NAMES.get(report_type, "Analytics Report"), "reportTypeCode": report_type, "reportTypeId": report_type_id, "dateRange": {"from": from_date, "to": to_date, "interval": interval}, "graphs": [], "summary": {}, } # Parse graph data if graphs_result and isinstance(graphs_result, list): for graph in graphs_result: if hasattr(graph, "id") and hasattr(graph, "data"): graph_data = {"metric": graph.id, "data": parse_graph_data(graph.data)} response["graphs"].append(graph_data) # Parse totals if totals_result: response["summary"] = parse_summary_data(totals_result) return json.dumps(response, indent=2)
• src/kaltura_mcp/server.py:146-181 (schema)

  MCP tool schema definition, including detailed input schema with properties, descriptions, enums, and required fields.
  types.Tool( name="get_analytics_timeseries", description="Get time-series analytics data optimized for charts and visualizations. Use this when creating graphs, dashboards, or tracking trends over time.", inputSchema={ "type": "object", "properties": { "from_date": { "type": "string", "description": "Start date in YYYY-MM-DD format (e.g., '2024-01-01')", }, "to_date": { "type": "string", "description": "End date in YYYY-MM-DD format (e.g., '2024-01-31')", }, "report_type": { "type": "string", "description": "Report type (default: 'content')", }, "metrics": { "type": "array", "items": {"type": "string"}, "description": "Metrics to include (e.g., ['plays', 'views'])", }, "entry_id": { "type": "string", "description": "Optional: Track single video's performance over time (e.g., '1_abc123'). Leave empty for platform-wide trends.", }, "interval": { "type": "string", "enum": ["hours", "days", "weeks", "months"], "description": "Time grouping for data points (default: 'days'). Use 'hours' for <7 day ranges, 'days' for monthly, 'weeks' for quarterly, 'months' for yearly views. Affects data granularity.", }, }, "required": ["from_date", "to_date"], }, ),
• src/kaltura_mcp/server.py:493-535 (registration)

  Tool dispatch registration in call_tool handler and inclusion in list_tools, mapping tool name to the handler execution with Kaltura manager.
  @server.call_tool() async def call_tool(name: str, arguments: Dict[str, Any]) -> List[types.TextContent]: """Execute a Kaltura API tool.""" try: if name == "get_media_entry": result = await get_media_entry(kaltura_manager, **arguments) elif name == "list_categories": result = await list_categories(kaltura_manager, **arguments) elif name == "get_analytics": result = await get_analytics(kaltura_manager, **arguments) elif name == "get_analytics_timeseries": result = await get_analytics_timeseries(kaltura_manager, **arguments) elif name == "get_video_retention": result = await get_video_retention(kaltura_manager, **arguments) elif name == "get_realtime_metrics": result = await get_realtime_metrics(kaltura_manager, **arguments) elif name == "get_quality_metrics": result = await get_quality_metrics(kaltura_manager, **arguments) elif name == "get_geographic_breakdown": result = await get_geographic_breakdown(kaltura_manager, **arguments) elif name == "list_analytics_capabilities": result = await list_analytics_capabilities(kaltura_manager, **arguments) elif name == "get_download_url": result = await get_download_url(kaltura_manager, **arguments) elif name == "get_thumbnail_url": result = await get_thumbnail_url(kaltura_manager, **arguments) elif name == "search_entries": result = await search_entries_intelligent(kaltura_manager, **arguments) elif name == "list_caption_assets": result = await list_caption_assets(kaltura_manager, **arguments) elif name == "get_caption_content": result = await get_caption_content(kaltura_manager, **arguments) elif name == "list_attachment_assets": result = await list_attachment_assets(kaltura_manager, **arguments) elif name == "get_attachment_content": result = await get_attachment_content(kaltura_manager, **arguments) else: return [types.TextContent(type="text", text=f"Unknown tool: {name}")] return [types.TextContent(type="text", text=result)] except Exception as e: return [types.TextContent(type="text", text=f"Error executing {name}: {str(e)}")]
• src/kaltura_mcp/tools/analytics_core.py:12-94 (helper)

  Mapping of human-readable report_type strings to Kaltura internal report type IDs, essential for API calls across all analytics functions.
  REPORT_TYPE_MAP = { # Content Performance Reports (1-10, 34, 44) "content": 1, # TOP_CONTENT "content_dropoff": 2, # CONTENT_DROPOFF "content_interactions": 3, # CONTENT_INTERACTIONS "engagement_timeline": 34, # USER_ENGAGEMENT_TIMELINE "content_contributions": 7, # CONTENT_CONTRIBUTIONS "content_report_reasons": 44, # CONTENT_REPORT_REASONS "content_spread": 10, # CONTENT_SPREAD # User Analytics Reports (11-18, 35, 40) "user_engagement": 11, # USER_ENGAGEMENT "specific_user_engagement": 12, # SPECIFIC_USER_ENGAGEMENT "user_top_content": 13, # USER_TOP_CONTENT "user_content_dropoff": 14, # USER_CONTENT_DROPOFF "user_content_interactions": 15, # USER_CONTENT_INTERACTIONS "user_usage": 17, # USER_USAGE "unique_users": 35, # UNIQUE_USERS_PLAY "user_highlights": 40, # USER_HIGHLIGHTS "specific_user_usage": 18, # SPECIFIC_USER_USAGE # Geographic & Demographic Reports (4, 30, 36-37) "geographic": 4, # MAP_OVERLAY "geographic_country": 36, # MAP_OVERLAY_COUNTRY "geographic_region": 37, # MAP_OVERLAY_REGION "geographic_city": 30, # MAP_OVERLAY_CITY # Platform & Technology Reports (21-23, 32-33) "platforms": 21, # PLATFORMS "operating_system": 22, # OPERATING_SYSTEM "browsers": 23, # BROWSERS "operating_system_families": 32, # OPERATING_SYSTEM_FAMILIES "browsers_families": 33, # BROWSERS_FAMILIES # Creator & Contributor Reports (5, 20, 38-39) "contributors": 5, # TOP_CONTRIBUTORS "creators": 20, # TOP_CREATORS "content_creator": 38, # TOP_CONTENT_CREATOR "content_contributors": 39, # TOP_CONTENT_CONTRIBUTORS # Distribution & Syndication Reports (6, 25, 41-42) "syndication": 6, # TOP_SYNDICATION "playback_context": 25, # TOP_PLAYBACK_CONTEXT "sources": 41, # TOP_SOURCES "syndication_usage": 42, # TOP_SYNDICATION_DISTRIBUTION # Usage & Infrastructure Reports (19, 26-27, 60, 64, 201) "partner_usage": 201, # PARTNER_USAGE "var_usage": 19, # VAR_USAGE "vpaas_usage": 26, # VPAAS_USAGE "entry_usage": 27, # ENTRY_USAGE "self_serve_usage": 60, # SELF_SERVE_USAGE "cdn_bandwidth": 64, # CDN_BANDWIDTH_USAGE # Interactive & Advanced Reports (43, 45-50) "percentiles": 43, # PERCENTILES - Video timeline retention analysis "video_timeline": 43, # Alias for PERCENTILES - clearer for LLMs "retention_curve": 43, # Another alias for PERCENTILES "viewer_retention": 43, # PERCENTILES - Per-video retention analysis "drop_off_analysis": 43, # PERCENTILES - Where viewers stop watching "replay_detection": 43, # PERCENTILES - Identify replay hotspots "player_interactions": 45, # PLAYER_RELATED_INTERACTIONS "playback_rate": 46, # PLAYBACK_RATE "interactive_video": 49, # USER_INTERACTIVE_VIDEO "interactive_nodes": 50, # INTERACTIVE_VIDEO_TOP_NODES # Live & Real-time Reports (48, 10001-10006) "live_stats": 48, # LIVE_STATS "realtime_country": 10001, # MAP_OVERLAY_COUNTRY_REALTIME "realtime_users": 10005, # USERS_OVERVIEW_REALTIME "realtime_qos": 10006, # QOS_OVERVIEW_REALTIME # Quality of Experience Reports (30001-30050) "qoe_overview": 30001, # QOE_OVERVIEW "qoe_experience": 30002, # QOE_EXPERIENCE "qoe_engagement": 30014, # QOE_ENGAGEMENT "qoe_stream_quality": 30026, # QOE_STREAM_QUALITY "qoe_error_tracking": 30038, # QOE_ERROR_TRACKING # Business Intelligence & Webcast Reports (40001-40013) "webcast_highlights": 40001, # HIGHLIGHTS_WEBCAST "webcast_engagement": 40011, # ENGAGEMENT_TIMELINE_WEBCAST # Additional Reports "discovery": 51, # DISCOVERY "discovery_realtime": 52, # DISCOVERY_REALTIME "realtime": 53, # REALTIME "peak_usage": 54, # PEAK_USAGE "flavor_params_usage": 55, # FLAVOR_PARAMS_USAGE "content_spread_country": 56, # CONTENT_SPREAD_COUNTRY "top_contributors_country": 57, # TOP_CONTRIBUTORS_COUNTRY "contribution_source": 58, # CONTRIBUTION_SOURCE "vod_performance": 59, # VOD_PERFORMANCE }
• src/kaltura_mcp/tools/__init__.py:4-6 (helper)

  Package-level export making get_analytics_timeseries available for import from the tools module.
  from .analytics import ( get_analytics, get_analytics_timeseries,