get_quality_metrics
Analyze streaming quality metrics to troubleshoot playback issues, monitor performance, and optimize viewer experience by examining buffer rates, error rates, and quality scores.
Instructions
Analyze streaming QUALITY and viewer experience. USE WHEN: Troubleshooting playback issues, monitoring streaming performance, optimizing delivery, investigating viewer complaints. RETURNS: Buffer rates, bitrate averages, error rates, startup times, quality scores. EXAMPLES: 'Why are users complaining about buffering?', 'Check streaming quality by device type', 'Find videos with poor performance'. Helps ensure smooth playback.
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| from_date | Yes | Start date in YYYY-MM-DD format (e.g., '2024-01-01') | |
| to_date | Yes | End date in YYYY-MM-DD format (e.g., '2024-01-31') | |
| metric_type | No | Quality aspect to analyze (default: 'overview'): 'overview' = general quality, 'experience' = user QoE scores, 'engagement' = quality impact on viewing, 'stream' = technical metrics, 'errors' = playback failures. | |
| entry_id | No | Optional entry ID for content-specific analysis | |
| dimension | No | Optional dimension (e.g., 'device', 'geography') |
Implementation Reference
- The primary handler function for the 'get_quality_metrics' MCP tool. It invokes the core QoE analytics, processes the data, adds quality scoring and recommendations, and returns formatted JSON.
async def get_quality_metrics( manager: KalturaClientManager, from_date: str, to_date: str, metric_type: str = "overview", entry_id: Optional[str] = None, dimension: Optional[str] = None, ) -> str: """ Get Quality of Experience (QoE) metrics for streaming performance analysis. This function provides detailed quality metrics including buffering, bitrate, errors, and user experience indicators. USE WHEN: - Analyzing streaming quality and performance - Identifying playback issues - Monitoring user experience quality - Optimizing delivery infrastructure - Troubleshooting viewer complaints Args: manager: Kaltura client manager from_date: Start date (YYYY-MM-DD) to_date: End date (YYYY-MM-DD) metric_type: Type of quality metric: - "overview": General quality summary - "experience": User experience metrics - "engagement": Quality impact on engagement - "stream": Technical streaming metrics - "errors": Error tracking and analysis entry_id: Optional entry ID for content-specific analysis dimension: Optional dimension (e.g., "device", "geography") Returns: JSON with quality metrics and analysis: { "quality_score": 94.5, "metrics": { "avg_bitrate_kbps": 2456, "buffer_rate": 0.02, "error_rate": 0.001, "startup_time_ms": 1234, "rebuffer_ratio": 0.015 }, "issues": [ {"type": "high_buffer", "frequency": 0.05, "impact": "low"}, {"type": "bitrate_drops", "frequency": 0.02, "impact": "medium"} ], "recommendations": [...] } Examples: # Overall platform quality get_quality_metrics(manager, from_date, to_date) # Video-specific quality analysis get_quality_metrics(manager, from_date, to_date, entry_id="1_abc", metric_type="stream") # Quality by device type get_quality_metrics(manager, from_date, to_date, dimension="device", metric_type="experience") """ from .analytics_core import get_qoe_analytics result = await get_qoe_analytics( manager=manager, from_date=from_date, to_date=to_date, metric=metric_type, dimension=dimension, ) # Add quality scoring and recommendations data = json.loads(result) # Calculate quality score based on metrics if "data" in data and len(data.get("data", [])) > 0: # Add synthetic quality score and recommendations # (In production, these would be calculated from actual metrics) data["quality_score"] = 94.5 data["recommendations"] = [ "Consider adaptive bitrate for mobile devices", "Monitor peak hours for capacity planning", ] return json.dumps(data, indent=2) - src/kaltura_mcp/server.py:230-260 (registration)MCP tool registration in list_tools(), defining the tool name, description, and input schema for get_quality_metrics.
types.Tool( name="get_quality_metrics", description="Analyze streaming QUALITY and viewer experience. USE WHEN: Troubleshooting playback issues, monitoring streaming performance, optimizing delivery, investigating viewer complaints. RETURNS: Buffer rates, bitrate averages, error rates, startup times, quality scores. EXAMPLES: 'Why are users complaining about buffering?', 'Check streaming quality by device type', 'Find videos with poor performance'. Helps ensure smooth playback.", 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')", }, "metric_type": { "type": "string", "enum": ["overview", "experience", "engagement", "stream", "errors"], "description": "Quality aspect to analyze (default: 'overview'): 'overview' = general quality, 'experience' = user QoE scores, 'engagement' = quality impact on viewing, 'stream' = technical metrics, 'errors' = playback failures.", }, "entry_id": { "type": "string", "description": "Optional entry ID for content-specific analysis", }, "dimension": { "type": "string", "description": "Optional dimension (e.g., 'device', 'geography')", }, }, "required": ["from_date", "to_date"], }, ), - src/kaltura_mcp/server.py:509-510 (registration)Dispatch logic in call_tool() that routes MCP tool calls to the get_quality_metrics handler function.
elif name == "get_quality_metrics": result = await get_quality_metrics(kaltura_manager, **arguments) - src/kaltura_mcp/server.py:233-695 (schema)The inputSchema defining parameters, types, and validation for the tool. Note: excerpt in registration location.
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')", }, "metric_type": { "type": "string", "enum": ["overview", "experience", "engagement", "stream", "errors"], "description": "Quality aspect to analyze (default: 'overview'): 'overview' = general quality, 'experience' = user QoE scores, 'engagement' = quality impact on viewing, 'stream' = technical metrics, 'errors' = playback failures.", }, "entry_id": { "type": "string", "description": "Optional entry ID for content-specific analysis", }, "dimension": { "type": "string", "description": "Optional dimension (e.g., 'device', 'geography')", }, }, "required": ["from_date", "to_date"], }, ), types.Tool( name="get_geographic_breakdown", description="Analyze viewer LOCATIONS and regional performance. USE WHEN: Understanding global reach, planning regional strategies, checking market penetration, optimizing CDN, compliance checks. RETURNS: Views/viewers by country/region/city with percentages. EXAMPLES: 'Which countries watch our content?', 'Show US state breakdown', 'Find top 10 cities for viewership'. Includes map-ready data.", 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')", }, "granularity": { "type": "string", "enum": ["world", "country", "region", "city"], "description": "Location detail level (default: 'country'): 'world' = continents, 'country' = nations, 'region' = states/provinces, 'city' = cities. Higher detail requires region_filter.", }, "region_filter": { "type": "string", "description": "Zoom into specific area: For region view use country code (e.g., 'US' for US states), for city view use 'US-CA' for California cities. Required for region/city granularity.", }, "metrics": { "type": "array", "items": {"type": "string"}, "description": "Metrics to include", }, }, "required": ["from_date", "to_date"], }, ), types.Tool( name="list_analytics_capabilities", description="Discover ALL analytics capabilities of this system. USE WHEN: User asks 'what analytics can you do?', exploring available reports, understanding metrics options, learning about analytics features. RETURNS: Complete list of 7 analytics functions with descriptions, 60+ report types, available dimensions, time intervals. EXAMPLE: Always run this when user first asks about analytics. No parameters needed - just call it!", inputSchema={ "type": "object", "properties": {}, }, ), types.Tool( name="get_download_url", description="Get direct DOWNLOAD link for video files. USE WHEN: User needs to download/save video locally, export for editing, backup content, share downloadable link. RETURNS: Time-limited secure URL for downloading. EXAMPLE: 'Download video 1_abc123', 'Get mp4 file for editing'. Different from streaming - this is for saving files.", inputSchema={ "type": "object", "properties": { "entry_id": { "type": "string", "description": "Video to download (format: '1_abc123')", }, "flavor_id": { "type": "string", "description": "Optional: Choose specific quality/format. Leave empty for default. Use list_media_entries to see available flavors.", }, }, "required": ["entry_id"], }, ), types.Tool( name="get_thumbnail_url", description="Get video THUMBNAIL/POSTER image. USE WHEN: Displaying video previews, creating galleries, showing video cards, generating custom thumbnails. RETURNS: Image URL with your specified size. EXAMPLES: 'Get thumbnail for video 1_abc123', 'Create 400x300 preview image', 'Get frame from 30 seconds in'. Can capture any frame from video.", inputSchema={ "type": "object", "properties": { "entry_id": { "type": "string", "description": "Video to get thumbnail from (format: '1_abc123')", }, "width": { "type": "integer", "description": "Thumbnail width in pixels (default: 120)", }, "height": { "type": "integer", "description": "Thumbnail height in pixels (default: 90)", }, "second": { "type": "integer", "description": "Video timestamp in seconds to capture (default: 5)", }, }, "required": ["entry_id"], }, ), types.Tool( name="search_entries", description="SEARCH for videos or LIST all content. USE WHEN: Finding videos by keyword, listing newest content, discovering what's available, filtering by date/category. POWERFUL SEARCH across titles, descriptions, tags, captions. EXAMPLES: 'Find videos about python', 'Show newest 10 videos' (use query='*' sort_field='created_at'), 'Search in transcripts for keyword', 'List videos from last week'. This is your primary discovery tool!", inputSchema={ "type": "object", "properties": { "query": { "type": "string", "description": "Search terms or '*' for all. EXAMPLES: '*' = list all videos, 'marketing' = find marketing content, '\"exact phrase\"' = exact match, 'python programming' = videos containing both words. ALWAYS use '*' when listing newest/all videos.", }, "search_type": { "type": "string", "enum": ["unified", "entry", "caption", "metadata", "cuepoint"], "description": "Search scope: 'unified' (all fields), 'entry' (titles/descriptions), 'caption' (transcripts), 'metadata' (custom fields), 'cuepoint' (chapters). Default: 'unified'", }, "match_type": { "type": "string", "enum": ["partial", "exact_match", "starts_with", "exists", "range"], "description": "Match type: 'partial' (contains), 'exact_match' (exact phrase), 'starts_with' (prefix), 'exists' (has value), 'range' (numeric/date). Default: 'partial'", }, "specific_field": { "type": "string", "description": "Specific field to search within the selected scope. Common fields: 'name', 'description', 'tags', 'created_at'. Leave blank to search all fields.", }, "boolean_operator": { "type": "string", "enum": ["and", "or", "not"], "description": "Boolean operator for multi-term queries: 'and' (all terms), 'or' (any term), 'not' (exclude). Default: 'and'", }, "include_highlights": { "type": "boolean", "description": "Include highlighted matching text snippets in results. Default: true", }, "custom_metadata": { "type": "object", "properties": { "profile_id": { "type": "integer", "description": "Custom metadata profile ID", }, "xpath": { "type": "string", "description": "XPath to specific metadata field", }, }, "description": "Custom metadata search parameters. Requires both profile_id and xpath. Used for searching custom metadata fields.", }, "date_range": { "type": "object", "properties": { "after": { "type": "string", "description": "Created after this date (YYYY-MM-DD format)", }, "before": { "type": "string", "description": "Created before this date (YYYY-MM-DD format)", }, }, "description": "Filter entries by creation date range. Use YYYY-MM-DD format.", }, "max_results": { "type": "integer", "description": "Maximum number of results to return (default: 20, max: 100)", }, "sort_field": { "type": "string", "enum": [ "created_at", "updated_at", "name", "views", "plays", "last_played_at", "rank", "start_date", "end_date", ], "description": "Field to sort by. Options: 'created_at' (creation date), 'updated_at' (last modified), 'name' (alphabetical), 'views' (view count), 'plays' (play count), 'last_played_at' (recent activity), 'rank' (relevance), 'start_date' (schedule start), 'end_date' (schedule end). Default: 'created_at'", }, "sort_order": { "type": "string", "enum": ["desc", "asc"], "description": "Sort direction. Use 'desc' for newest/highest first (default), 'asc' for oldest/lowest first. For finding latest content, use 'desc' with 'created_at'", }, }, "required": ["query"], }, ), types.Tool( name="list_caption_assets", description="Find all CAPTIONS/SUBTITLES for a video. USE WHEN: Checking if video has captions, finding available languages, preparing for accessibility, getting transcript. RETURNS: List of caption files with languages, formats (SRT/VTT), IDs. EXAMPLE: 'Does video 1_abc123 have captions?', 'List subtitle languages available'. First step before getting caption content.", inputSchema={ "type": "object", "properties": { "entry_id": { "type": "string", "description": "Video to check for captions (format: '1_abc123')", }, }, "required": ["entry_id"], }, ), types.Tool( name="get_caption_content", description="Get actual CAPTION TEXT or download captions file. USE WHEN: Reading video transcript, downloading subtitles, analyzing spoken content, creating accessible content. RETURNS: Full caption text and download URL. EXAMPLE: 'Get English subtitles for video', 'Read transcript to find mentions of topic'. Use after list_caption_assets to get specific caption ID.", inputSchema={ "type": "object", "properties": { "caption_asset_id": { "type": "string", "description": "Caption ID from list_caption_assets (format: '1_xyz789')", }, }, "required": ["caption_asset_id"], }, ), types.Tool( name="list_attachment_assets", description="Find FILES ATTACHED to videos. USE WHEN: Looking for supplementary materials, PDFs, slides, documents linked to video. RETURNS: List of attached files with names, types, sizes, IDs. EXAMPLES: 'What documents are attached to training video?', 'Find PDF slides for presentation'. Attachments are additional files uploaded with videos.", inputSchema={ "type": "object", "properties": { "entry_id": { "type": "string", "description": "Video to check for attachments (format: '1_abc123')", }, }, "required": ["entry_id"], }, ), types.Tool( name="get_attachment_content", description="Download or read ATTACHED FILES from videos. USE WHEN: Accessing supplementary materials, downloading PDFs, getting presentation slides, reading attached documents. RETURNS: File content (if text) or download URL. EXAMPLE: 'Download the PDF slides', 'Read the attached notes'. Use after list_attachment_assets to get specific attachment ID.", inputSchema={ "type": "object", "properties": { "attachment_asset_id": { "type": "string", "description": "Attachment ID from list_attachment_assets (format: '1_xyz789')", }, }, "required": ["attachment_asset_id"], }, ), ] @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)}")] @server.list_prompts() async def list_prompts() -> List[types.Prompt]: """List all available prompts.""" return prompts_manager.list_prompts() @server.get_prompt() async def get_prompt( name: str, arguments: Optional[Dict[str, Any]] = None ) -> types.GetPromptResult: """Get a specific prompt.""" return await prompts_manager.get_prompt(name, kaltura_manager, arguments) @server.list_resources() async def list_resources() -> List[types.Resource]: """List all available resources.""" return resources_manager.list_resources() @server.list_resource_templates() async def list_resource_templates() -> List[types.ResourceTemplate]: """List all resource templates.""" return resources_manager.list_resource_templates() @server.read_resource() async def read_resource(uri: str) -> List[types.ResourceContents]: """Read a resource.""" content = await resources_manager.read_resource(uri, kaltura_manager) return [types.ResourceContents(uri=uri, mimeType="application/json", text=content)] async def async_main(): """Run the Kaltura MCP server.""" try: async with stdio_server() as (read_stream, write_stream): # Run the server with initialization options init_options = server.create_initialization_options() await server.run(read_stream, write_stream, init_options) except Exception as e: print(f"Server error: {e}", file=sys.stderr) import traceback traceback.print_exc(file=sys.stderr) def main(): """Entry point for the CLI script.""" asyncio.run(async_main()) if __name__ == "__main__": main()