search_logs
Search application logs using filters for service, namespace, text, level, and time range to investigate issues without writing complex queries.
Instructions
Search logs with simple filters. Easier than writing raw LogQL queries.
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| service | No | Service/job name to filter logs | |
| namespace | No | Namespace to filter logs | |
| search_text | No | Text to search for in log messages | |
| level | No | Log level filter: 'error', 'warn', 'info', 'debug' | |
| start | No | Start time (relative like '1h' or absolute). Default: 1h ago | |
| end | No | End time. Default: now | |
| limit | No | Maximum number of log entries. Default: 100 |
Implementation Reference
- src/tools/loki_tools.py:58-125 (handler)Implementation of the search_logs tool handler in src/tools/loki_tools.py.
async def search_logs( client: LokiClient, service: Optional[str] = None, namespace: Optional[str] = None, search_text: Optional[str] = None, level: Optional[str] = None, start: Optional[str] = None, end: Optional[str] = None, limit: int = 100 ) -> Dict[str, Any]: """ Search logs with simple filters. Args: client: Loki client service: Service/job name filter namespace: Namespace filter search_text: Text to search for in logs level: Log level filter (error, warn, info, debug) start: Start time end: End time limit: Maximum number of results Returns: Filtered log entries """ try: # Build LogQL query labels = [] if service: labels.append(f'job="{service}"') if namespace: labels.append(f'namespace="{namespace}"') query = "{" + ",".join(labels) + "}" if labels else "{job=~\".+\"}" # Add search text filter if search_text: query += f' |= "{search_text}"' # Add level filter if level: query += f' | json | level="{level}"' # Parse time range start_dt, end_dt = parse_time_range(start, end) start_ns = to_loki_time(start_dt) end_ns = to_loki_time(end_dt) result = await client.query_range(query, start_ns, end_ns, limit) return { "success": True, "query": query, "filters": { "service": service, "namespace": namespace, "search_text": search_text, "level": level }, "result": result } except Exception as e: logger.error(f"Error searching logs: {e}") return { "success": False, "error": str(e) }