query_loki
Execute LogQL queries to search and filter logs in Loki for troubleshooting and root cause analysis.
Instructions
Execute raw LogQL query against Loki. Use this to search and filter logs.
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| query | Yes | LogQL query string (e.g., '{namespace="prod-services"}', '{job="api-server"} |= "error"') | |
| start | No | Start time. Can be: relative like '1h', '30m', or absolute. Default: 1h ago | |
| end | No | End time. Can be: 'now', relative, or absolute. Default: now | |
| limit | No | Maximum number of log entries to return. Default: 100 | |
| direction | No | Query direction: 'forward' (oldest first) or 'backward' (newest first). Default: backward |
Implementation Reference
- src/tools/loki_tools.py:11-55 (handler)The implementation of the `query_loki` tool handler, which executes a raw LogQL query using the provided Loki client.
async def query_loki( client: LokiClient, query: str, start: Optional[str] = None, end: Optional[str] = None, limit: int = 100, direction: str = "backward" ) -> Dict[str, Any]: """ Execute raw LogQL query. Args: client: Loki client query: LogQL query string start: Start time (relative like '1h' or absolute) end: End time (relative like 'now' or absolute) limit: Maximum number of log entries direction: Query direction (forward or backward) Returns: Query results with log entries """ try: # 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, direction) return { "success": True, "query": query, "start": start or "auto", "end": end or "now", "limit": limit, "result": result } except Exception as e: logger.error(f"Error executing Loki query: {e}") return { "success": False, "error": str(e), "query": query }