get_recent_audit_logs
Retrieve audit log records for a specified recent period, defaulting to the last 7 days and up to 100 records. Customize days and limit parameters for targeted log analysis.
Instructions
[Function Description]: Get audit log records for a recent period.
[Parameter Content]:
days (integer) [Optional] - Number of recent days of logs to retrieve, default is 7
limit (integer) [Optional] - Maximum number of records to return, default is 100
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| days | No | ||
| limit | No |
Implementation Reference
- Primary handler for the get_recent_audit_logs tool. Retrieves audit logs using async helper, converts to JSON, serializes datetimes, and formats MCP response.async def get_recent_audit_logs_for_mcp(self, days: int = 7, limit: int = 100) -> Dict[str, Any]: """Get recent audit log records - MCP interface""" logger.info(f"Getting audit logs: Days: {days}, Limit: {limit}") try: logs_df = await self.get_recent_audit_logs_async(days=days, limit=limit) # Convert DataFrame to JSON format if hasattr(logs_df, 'to_dict'): try: logs_data = logs_df.to_dict('records') except Exception as e: logger.warning(f"DataFrame.to_dict failed, trying manual conversion: {e}") # Manually convert DataFrame to records format logs_data = [] if not logs_df.empty: for _, row in logs_df.iterrows(): logs_data.append(dict(row)) # Serialize datetime objects logs_data = self._serialize_datetime_objects(logs_data) else: logs_data = self._serialize_datetime_objects(logs_df) return self._format_response(success=True, result=logs_data) except Exception as e: logger.error(f"Failed to get audit logs: {str(e)}", exc_info=True) return self._format_response(success=False, error=str(e), message="Error occurred while getting audit logs")
- Helper function that executes the SQL query to fetch recent successful audit logs from __internal_schema.audit_log table, filtering non-trivial statements.async def get_recent_audit_logs_async(self, days: int = 7, limit: int = 100): """Async version: get recent audit logs and return a pandas DataFrame.""" try: start_date = (datetime.now() - timedelta(days=days)).strftime('%Y-%m-%d') query = f""" SELECT client_ip, user, db, time, stmt_id, stmt, state, error_code FROM `__internal_schema`.`audit_log` WHERE `time` >= '{start_date}' AND state = 'EOF' AND error_code = 0 AND `stmt` NOT LIKE 'SHOW%' AND `stmt` NOT LIKE 'DESC%' AND `stmt` NOT LIKE 'EXPLAIN%' AND `stmt` NOT LIKE 'SELECT 1%' ORDER BY time DESC LIMIT {limit} """ rows = await self._execute_query_async(query) import pandas as pd return pd.DataFrame(rows or []) except Exception as e: logger.error(f"Error getting audit logs asynchronously: {str(e)}") import pandas as pd return pd.DataFrame()
- doris_mcp_server/tools/tools_manager.py:247-265 (registration)MCP tool registration decorator defining the tool name, description, parameters (schema), and wrapper function that delegates to internal call_tool.@mcp.tool( "get_recent_audit_logs", description="""[Function Description]: Get audit log records for a recent period. [Parameter Content]: - days (integer) [Optional] - Number of recent days of logs to retrieve, default is 7 - limit (integer) [Optional] - Maximum number of records to return, default is 100 """, ) async def get_recent_audit_logs_tool( days: int = 7, limit: int = 100 ) -> str: """Get audit logs""" return await self.call_tool("get_recent_audit_logs", { "days": days, "limit": limit })
- Tool manager routing handler that extracts parameters and delegates to MetadataExtractor.get_recent_audit_logs_for_mcp.async def _get_recent_audit_logs_tool(self, arguments: Dict[str, Any]) -> Dict[str, Any]: """Get audit logs tool routing""" days = arguments.get("days", 7) limit = arguments.get("limit", 100) # Delegate to metadata extractor for processing return await self.metadata_extractor.get_recent_audit_logs_for_mcp(days, limit)
- Explicit input schema definition for the tool in list_tools method (for stdio mode), matching the decorator parameters.Tool( name="get_recent_audit_logs", description="""[Function Description]: Get audit log records for a recent period. [Parameter Content]: - days (integer) [Optional] - Number of recent days of logs to retrieve, default is 7 - limit (integer) [Optional] - Maximum number of records to return, default is 100 """, inputSchema={ "type": "object", "properties": { "days": {"type": "integer", "description": "Number of recent days", "default": 7}, "limit": {"type": "integer", "description": "Maximum number of records", "default": 100}, }, }, ),