Kaltura MCP Server

Instructions

Implementation Reference

• src/kaltura_mcp/tools/search.py:376-521 (handler)

  Primary handler for 'search_entries' tool. Intelligently handles parameters, calls esearch_entries, adds rich search context, and formats results for MCP.
  async def search_entries_intelligent( manager: KalturaClientManager, query: str, search_type: str = "unified", match_type: str = "partial", specific_field: Optional[str] = None, boolean_operator: str = "and", include_highlights: bool = True, custom_metadata: Optional[Dict[str, Any]] = None, date_range: Optional[Dict[str, str]] = None, max_results: int = 20, sort_field: str = "created_at", sort_order: str = "desc", ) -> str: """ Advanced search function that exposes the full power of Kaltura's eSearch API. This is the PRIMARY tool for finding and listing media entries. Use this for: - General searches with keywords - Listing all entries (query="*") - Filtering by date ranges, content types, fields - Searching in transcripts, metadata, or specific fields - Complex boolean queries with multiple criteria This function provides access to all eSearch capabilities while maintaining clear parameter naming and comprehensive documentation for LLM usage. """ # Validate and cap max_results max_results = min(max_results, 100) # Handle special case for listing all entries if query == "*": # For wildcard searches, we need to avoid field-specific restrictions # Use unified search with a common term that should match most entries search_type = "unified" # Force unified search to avoid field restrictions match_type = "partial" # Use partial matching # Use common terms that are likely to appear in most entries (video, audio, or metadata) # "e" is the most common letter in English and should match most content query = "e" specific_field = None # Clear any specific field to avoid API restrictions # Extract date range if provided date_after = None date_before = None if date_range: date_after = date_range.get("after") date_before = date_range.get("before") # Extract custom metadata parameters metadata_profile_id = None metadata_xpath = None if custom_metadata: metadata_profile_id = custom_metadata.get("profile_id") metadata_xpath = custom_metadata.get("xpath") # Call the underlying eSearch function with all parameters try: result = await esearch_entries( manager=manager, search_term=query, search_type=search_type, item_type=match_type, field_name=specific_field, add_highlight=include_highlights, operator_type=boolean_operator, metadata_profile_id=metadata_profile_id, metadata_xpath=metadata_xpath, date_range_start=date_after, date_range_end=date_before, limit=max_results, sort_field=sort_field, sort_order=sort_order, ) # Parse the result to add comprehensive search context import json result_data = json.loads(result) # Add detailed search context information if "entries" in result_data: # Determine operation type original_query = query if query else "*" operation_type = "list_all" if original_query == "*" else "search" search_context = { "searchQuery": original_query, "operationType": operation_type, "searchConfiguration": { "scope": search_type, "matchType": match_type, "specificField": specific_field, "booleanOperator": boolean_operator, "highlightsEnabled": include_highlights, }, "filters": { "dateRange": date_range, "customMetadata": custom_metadata if custom_metadata else None, }, "results": { "count": len(result_data["entries"]), "totalMatches": result_data.get("totalCount", 0), "maxRequested": max_results, }, "searchCapabilities": { "availableScopes": [ "unified (all content)", "entry (metadata)", "caption (transcripts)", "metadata (custom fields)", "cuepoint (temporal markers)", ], "availableMatchTypes": [ "partial (contains)", "exact_match (exact phrase)", "starts_with (prefix)", "exists (field has value)", "range (numeric/date)", ], "advancedFeatures": [ "highlighting", "boolean operators", "custom metadata search", "date filtering", "field-specific search", ], }, } result_data["searchContext"] = search_context return json.dumps(result_data, indent=2) except Exception as e: # If eSearch fails, provide detailed error information logger.error(f"eSearch failed: {e}") return handle_kaltura_error( e, "search entries", { "searchQuery": query, "searchType": search_type, "suggestion": "Try using simpler parameters or check your search syntax", }, )
• src/kaltura_mcp/server.py:346-433 (schema)

  Input schema and description for the 'search_entries' tool, defining all parameters like query, search_type, sort_field, etc.
  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"], }, ),
• src/kaltura_mcp/server.py:519-520 (registration)

  Dispatch/registration in call_tool handler mapping 'search_entries' to search_entries_intelligent function.
  elif name == "search_entries": result = await search_entries_intelligent(kaltura_manager, **arguments)
• src/kaltura_mcp/tools/search.py:155-374 (helper)

  Core helper implementing Kaltura eSearch API calls, used by search_entries_intelligent.
  async def esearch_entries( manager: KalturaClientManager, search_term: str, search_type: str = "unified", item_type: str = "partial", field_name: Optional[str] = None, add_highlight: bool = True, operator_type: str = "and", metadata_profile_id: Optional[int] = None, metadata_xpath: Optional[str] = None, date_range_start: Optional[str] = None, date_range_end: Optional[str] = None, limit: int = 20, sort_field: str = "created_at", sort_order: str = "desc", ) -> str: """Enhanced search using Kaltura eSearch API with advanced capabilities.""" client = manager.get_client() try: # Use the ElasticSearch service through the client # Create search items based on search type search_items = [] if search_type == "unified": # Unified search across all entry data unified_item = KalturaESearchUnifiedItem() unified_item.searchTerm = search_term unified_item.itemType = _get_item_type(item_type) unified_item.addHighlight = add_highlight search_items.append(unified_item) elif search_type == "entry": # Search in specific entry fields entry_item = KalturaESearchEntryItem() entry_item.searchTerm = search_term entry_item.itemType = _get_item_type(item_type) entry_item.fieldName = _get_entry_field_name(field_name or "name") entry_item.addHighlight = add_highlight search_items.append(entry_item) elif search_type == "caption": # Search in captions caption_item = KalturaESearchCaptionItem() caption_item.searchTerm = search_term caption_item.itemType = _get_item_type(item_type) caption_item.fieldName = _get_caption_field_name(field_name or "content") caption_item.addHighlight = add_highlight search_items.append(caption_item) elif search_type == "metadata": # Search in custom metadata metadata_item = KalturaESearchEntryMetadataItem() metadata_item.searchTerm = search_term metadata_item.itemType = _get_item_type(item_type) metadata_item.addHighlight = add_highlight if metadata_profile_id: metadata_item.metadataProfileId = metadata_profile_id if metadata_xpath: metadata_item.xpath = metadata_xpath search_items.append(metadata_item) elif search_type == "cuepoint": # Search in cue points cuepoint_item = KalturaESearchCuePointItem() cuepoint_item.searchTerm = search_term cuepoint_item.itemType = _get_item_type(item_type) cuepoint_item.fieldName = _get_cuepoint_field_name(field_name or "text") cuepoint_item.addHighlight = add_highlight search_items.append(cuepoint_item) # Add date range filtering if specified if date_range_start and date_range_end: try: start_timestamp = int(datetime.strptime(date_range_start, "%Y-%m-%d").timestamp()) end_timestamp = int(datetime.strptime(date_range_end, "%Y-%m-%d").timestamp()) range_item = KalturaESearchRange() range_item.greaterThanOrEqual = start_timestamp range_item.lessThanOrEqual = end_timestamp date_item = KalturaESearchEntryItem() date_item.itemType = KalturaESearchItemType.RANGE date_item.fieldName = KalturaESearchEntryFieldName.CREATED_AT date_item.range = range_item search_items.append(date_item) except ValueError: return json.dumps({"error": "Invalid date format. Use YYYY-MM-DD"}) # Create search operator search_operator = KalturaESearchEntryOperator() search_operator.operator = _get_operator_type(operator_type) search_operator.searchItems = search_items # Create search parameters search_params = KalturaESearchEntryParams() search_params.searchOperator = search_operator # Add sorting order_by = KalturaESearchOrderBy() order_item = KalturaESearchEntryOrderByItem() order_item.sortField = _get_sort_field(sort_field) order_item.sortOrder = _get_sort_order(sort_order) order_by.orderItems = [order_item] search_params.orderBy = order_by # Add paging pager = KalturaFilterPager() pager.pageSize = limit # Execute search using the client's elasticsearch service search_results = client.elasticSearch.eSearch.searchEntry(search_params, pager) # Process results entries = [] for result in search_results.objects: entry_data = { "id": result.object.id, "name": result.object.name, "description": result.object.description, "mediaType": result.object.mediaType.value if hasattr(result.object.mediaType, "value") else result.object.mediaType, "createdAt": datetime.fromtimestamp(result.object.createdAt).isoformat() if result.object.createdAt else None, "duration": result.object.duration, "tags": result.object.tags, "thumbnailUrl": result.object.thumbnailUrl, "plays": result.object.plays, "views": result.object.views, } # Add highlights if available if hasattr(result, "highlight") and result.highlight: highlights = [] for highlight in result.highlight: highlight_data = { "fieldName": highlight.fieldName, "hits": [hit.value for hit in highlight.hits] if highlight.hits else [], } highlights.append(highlight_data) entry_data["highlights"] = highlights # Add items data (for captions, metadata, etc.) if hasattr(result, "itemsData") and result.itemsData: items_data = [] for item_data in result.itemsData: item_info = {"totalCount": item_data.totalCount, "items": []} if hasattr(item_data, "items") and item_data.items: for item in item_data.items: item_detail = {} # Handle different item types if hasattr(item, "line"): # Caption item item_detail.update( { "line": item.line, "startsAt": item.startsAt, "endsAt": item.endsAt, "language": item.language, "captionAssetId": item.captionAssetId, } ) elif hasattr(item, "valueText"): # Metadata item item_detail.update( { "xpath": item.xpath, "metadataProfileId": item.metadataProfileId, "metadataFieldId": item.metadataFieldId, "valueText": item.valueText, } ) # Add highlights for this item if hasattr(item, "highlight") and item.highlight: item_highlights = [] for highlight in item.highlight: item_highlight = { "fieldName": highlight.fieldName, "hits": [hit.value for hit in highlight.hits] if highlight.hits else [], } item_highlights.append(item_highlight) item_detail["highlights"] = item_highlights item_info["items"].append(item_detail) items_data.append(item_info) entry_data["itemsData"] = items_data entries.append(entry_data) return json.dumps( { "searchTerm": search_term, "searchType": search_type, "totalCount": search_results.totalCount, "entries": entries, }, indent=2, ) except Exception as e: return json.dumps( { "error": f"eSearch failed: {str(e)}", "searchTerm": search_term, "searchType": search_type, }, indent=2, )
• src/kaltura_mcp/tools/__init__.py:27-62 (registration)

  Exports search_entries and search_entries_intelligent for import in server.py.
  from .search import ( esearch_entries, list_categories, search_entries, search_entries_intelligent, ) from .utils import ( handle_kaltura_error, safe_serialize_kaltura_field, validate_entry_id, ) # Export all tools __all__ = [ # Utilities "handle_kaltura_error", "safe_serialize_kaltura_field", "validate_entry_id", # Media operations "get_download_url", "get_media_entry", "get_thumbnail_url", "list_media_entries", # Analytics operations "get_analytics", "get_analytics_timeseries", "get_video_retention", "get_realtime_metrics", "get_quality_metrics", "get_geographic_breakdown", "list_analytics_capabilities", # Search operations "esearch_entries", "list_categories", "search_entries", "search_entries_intelligent",