search_log_all_records
Search all log records with filtering by scope and content patterns, providing context before and after matches for comprehensive analysis.
Instructions
Search for all log records, optionally filtering by scope and content patterns, with context.
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| scope | No | default | |
| context_before | No | ||
| context_after | No | ||
| log_dirs_override | No | ||
| log_content_patterns_override | No |
Output Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| result | Yes |
Implementation Reference
- The handler function decorated with @mcp.tool() that executes the tool logic: parses input parameters, initializes AnalysisEngine, builds filter criteria, searches logs, and returns results or raises MCP error.
@mcp.tool() async def search_log_all_records( scope: str = "default", context_before: int = 2, context_after: int = 2, log_dirs_override: str = "", log_content_patterns_override: str = "", ) -> list[dict[str, Any]]: """Search for all log records, optionally filtering by scope and content patterns, with context.""" # Forcing re-initialization of analysis_engine for debugging module caching. # Pass project_root_for_config=None to allow AnalysisEngine to determine it. current_analysis_engine = AnalysisEngine(logger_instance=logger, project_root_for_config=None) print( f"DEBUG_MCP_TOOL_SEARCH_ALL: Entered search_log_all_records with log_dirs_override='{log_dirs_override}'", file=sys.stderr, flush=True, ) logger.info( "MCP search_log_all_records called with scope='%s', context=%sB/%sA, " "log_dirs_override='%s', log_content_patterns_override='%s'", scope, context_before, context_after, log_dirs_override, log_content_patterns_override, ) log_dirs_list = log_dirs_override.split(",") if log_dirs_override else None log_content_patterns_list = log_content_patterns_override.split(",") if log_content_patterns_override else None filter_criteria = build_filter_criteria( scope=scope, context_before=context_before, context_after=context_after, log_dirs_override=log_dirs_list, log_content_patterns_override=log_content_patterns_list, ) try: results = await asyncio.to_thread(current_analysis_engine.search_logs, filter_criteria) logger.info("search_log_all_records returning %s records.", len(results)) return results except Exception as e: # pylint: disable=broad-exception-caught logger.error("Error in search_log_all_records: %s", e, exc_info=True) custom_message = f"Failed to search all logs: {e!s}" raise McpError(ErrorData(code=-32603, message=custom_message)) from e - Pydantic model defining the input schema for the search_log_all_records tool, inheriting from BaseSearchInput.
class SearchLogAllInput(BaseSearchInput): """Input for search_log_all_records.""" - Base Pydantic model providing common input fields for search tools, used by SearchLogAllInput.
class BaseSearchInput(BaseModel): """Base model for common search parameters.""" scope: str = Field(default="default", description="Logging scope to search within (from .env scopes or default).") context_before: int = Field(default=2, description="Number of lines before a match.", ge=0) context_after: int = Field(default=2, description="Number of lines after a match.", ge=0) log_dirs_override: str = Field( default="", description="Comma-separated list of log directories, files, or glob patterns (overrides .env for file locations).", ) log_content_patterns_override: str = Field( default="", description="Comma-separated list of REGEX patterns for log messages (overrides .env content filters).", ) - src/log_analyzer_mcp/log_analyzer_mcp_server.py:587-587 (registration)The @mcp.tool() decorator registers the search_log_all_records function as an MCP tool.
@mcp.tool()