search_by_metadata
Search for content in Alfresco by metadata fields like creator, content type, or custom terms to find specific documents efficiently.
Instructions
Search for content in Alfresco by metadata fields.
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| term | No | ||
| creator | No | ||
| content_type | No | ||
| max_results | No |
Output Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| result | Yes |
Implementation Reference
- alfresco_mcp_server/fastmcp_server.py:69-79 (registration)Registration of the 'search_by_metadata' tool using FastMCP @mcp.tool decorator. This defines the input schema via type hints and docstring, and serves as the entry point handler delegating to the implementation.
@mcp.tool async def search_by_metadata( term: str = "", creator: str = "", content_type: str = "", max_results: int = 25, ctx: Context = None ) -> str: """Search for content in Alfresco by metadata fields.""" return await search_by_metadata_impl(term, creator, content_type, max_results, ctx) - Core handler implementation executing the tool logic: parameter handling, Alfresco connection via ensure_connection, query building with search_utils.build_query, search execution, result parsing, and formatted output.
async def search_by_metadata_impl( term: str = "", creator: str = "", content_type: str = "", max_results: int = 25, ctx: Optional[Context] = None ) -> str: """Search for content in Alfresco by metadata fields. Args: term: Search term to include (primary search parameter) creator: Creator username to filter by content_type: Content type to filter by (e.g., "cm:content", "cm:folder") max_results: Maximum number of results to return (default: 25) ctx: MCP context for progress reporting Returns: Formatted search results with metadata """ # Parameter validation and extraction try: # Extract parameters with fallback handling if hasattr(creator, 'value'): actual_creator = str(creator.value) else: actual_creator = str(creator) if hasattr(content_type, 'value'): actual_content_type = str(content_type.value) else: actual_content_type = str(content_type) if hasattr(term, 'value'): actual_term = str(term.value) else: actual_term = str(term) if hasattr(max_results, 'value'): actual_max_results = int(max_results.value) else: actual_max_results = int(max_results) # Clean and normalize for display (preserve Unicode characters) safe_creator_display = str(actual_creator) safe_content_type_display = str(actual_content_type) safe_term_display = str(actual_term) except Exception as e: logger.error(f"Parameter extraction error: {e}") return f"ERROR: Parameter error: {str(e)}" if ctx: await ctx.info(safe_format_output(f"Searching by metadata in Alfresco...")) await ctx.report_progress(0.0) try: # Get all clients that ensure_connection() already created master_client = await ensure_connection() # Import search_utils from python_alfresco_api.utils import search_utils # Access the search client that was already created search_client = master_client.search logger.info(f"Searching Alfresco by metadata - creator: '{safe_creator_display}', type: '{safe_content_type_display}', term: '{safe_term_display}'") if ctx: await ctx.report_progress(0.3) # Build query using search_utils.build_query() utility search_query = search_utils.build_query( term=actual_term if actual_term.strip() else None, content_type=actual_content_type if actual_content_type.strip() else None, creator=actual_creator if actual_creator.strip() else None ) # If no parameters provided, search everything if not search_query or search_query.strip() == "": search_query = "*" # Execute search using existing search client if ctx: await ctx.report_progress(0.5) try: # Use correct working pattern: search_utils.simple_search with existing search_client search_results = search_utils.simple_search(search_client, search_query, max_items=actual_max_results) if not search_results or not hasattr(search_results, 'list_'): return safe_format_output(f"ERROR: Search failed - invalid response from Alfresco") except Exception as e: logger.error(f"Search failed: {e}") return safe_format_output(f"ERROR: Search failed: {str(e)}") # Process results using correct pattern entries = [] if hasattr(search_results, 'list_') and search_results.list_ and hasattr(search_results.list_, 'entries'): entries = search_results.list_.entries if search_results.list_ else [] if ctx: await ctx.report_progress(1.0) # Process final results if entries: logger.info(f"Found {len(entries)} search results") result_text = f"Found {len(entries)} item(s) matching the metadata criteria:\n\n" for i, entry in enumerate(entries, 1): # Debug: Log the entry structure logger.debug(f"Entry {i} type: {type(entry)}, content: {entry}") # Handle different possible entry structures node = None if isinstance(entry, dict): if 'entry' in entry: node = entry['entry'] elif 'name' in entry: # Direct node structure node = entry else: logger.warning(f"Unknown entry structure: {entry}") continue elif hasattr(entry, 'entry'): # ResultSetRowEntry object node = entry.entry else: logger.warning(f"Entry is not a dict or ResultSetRowEntry: {type(entry)}") continue if node: # Handle both dict and ResultNode objects if isinstance(node, dict): name = str(node.get('name', 'Unknown')) node_id = str(node.get('id', 'Unknown')) node_type_actual = str(node.get('nodeType', 'Unknown')) created_at = str(node.get('createdAt', 'Unknown')) else: # ResultNode object - access attributes directly name = str(getattr(node, 'name', 'Unknown')) node_id = str(getattr(node, 'id', 'Unknown')) node_type_actual = str(getattr(node, 'node_type', 'Unknown')) created_at = str(getattr(node, 'created_at', 'Unknown')) # Clean JSON-friendly formatting (no markdown syntax) # Apply safe formatting to individual fields to prevent emoji encoding issues safe_name = safe_format_output(name) safe_node_id = safe_format_output(node_id) safe_node_type = safe_format_output(node_type_actual) safe_created_at = safe_format_output(created_at) result_text += f"{i}. {safe_name}\n" result_text += f" - ID: {safe_node_id}\n" result_text += f" - Type: {safe_node_type}\n" result_text += f" - Created: {safe_created_at}\n\n" return safe_format_output(result_text) else: # Simple "0" for zero results as requested return "0" except Exception as e: # Preserve Unicode characters in error messages error_msg = f"ERROR: Metadata search failed: {str(e)}" if ctx: await ctx.error(safe_format_output(error_msg)) return safe_format_output(error_msg) if ctx: await ctx.info(safe_format_output("Metadata search completed!"))