get_recent_audit_logs

Instructions

Implementation Reference

• doris_mcp_server/utils/schema_extractor.py:1665-1692 (handler)

  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")

• doris_mcp_server/utils/schema_extractor.py:1415-1438 (helper)

  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
              })

• doris_mcp_server/tools/tools_manager.py:1535-1542 (handler)

  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)

• doris_mcp_server/tools/tools_manager.py:927-944 (schema)

  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},
                      },
                  },
              ),