Kaltura MCP Server

Instructions

Implementation Reference

• src/kaltura_mcp/tools/analytics.py:608-695 (handler)

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