search_entries
Find videos and media content by searching across titles, descriptions, tags, and transcripts. Filter results by date, category, or custom metadata to discover relevant content in Kaltura's media library.
Instructions
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!
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| query | Yes | 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 | No | Search scope: 'unified' (all fields), 'entry' (titles/descriptions), 'caption' (transcripts), 'metadata' (custom fields), 'cuepoint' (chapters). Default: 'unified' | |
| match_type | No | Match type: 'partial' (contains), 'exact_match' (exact phrase), 'starts_with' (prefix), 'exists' (has value), 'range' (numeric/date). Default: 'partial' | |
| specific_field | No | Specific field to search within the selected scope. Common fields: 'name', 'description', 'tags', 'created_at'. Leave blank to search all fields. | |
| boolean_operator | No | Boolean operator for multi-term queries: 'and' (all terms), 'or' (any term), 'not' (exclude). Default: 'and' | |
| include_highlights | No | Include highlighted matching text snippets in results. Default: true | |
| custom_metadata | No | Custom metadata search parameters. Requires both profile_id and xpath. Used for searching custom metadata fields. | |
| date_range | No | Filter entries by creation date range. Use YYYY-MM-DD format. | |
| max_results | No | Maximum number of results to return (default: 20, max: 100) | |
| sort_field | No | 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 | No | Sort direction. Use 'desc' for newest/highest first (default), 'asc' for oldest/lowest first. For finding latest content, use 'desc' with 'created_at' |
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) - 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",