base_tableUsage

Instructions

Implementation Reference

• src/teradata_mcp_server/tools/base/base_tools.py:364-441 (handler)

  Handler function that implements the core logic for the 'base_tableUsage' tool. It executes a SQL query against DBC.DBQLObjTbl and DBC.DBQLogTbl to compute table usage frequency, recency, and ranking within a specified database.

  def handle_base_tableUsage(conn: TeradataConnection, database_name: str | None = None, *args, **kwargs):
      """
      Measure the usage of a table and views by users in a given schema, this is helpful to infer what database objects are most actively used or drive most value via SQLAlchemy, bind parameters if provided (prepared SQL), and return the fully rendered SQL (with literals) in metadata.
  
      Arguments:
        database_name - Database name
  
      Returns:
        ResponseType: formatted response with query results + metadata
      """
  
      logger.debug("Tool: handle_base_tableUsage: Args: database_name:")
      database_name_filter = f"AND objectdatabasename = '{database_name}'" if database_name else ""
  
      table_usage_sql="""
      LOCKING ROW for ACCESS
      sel
      DatabaseName
      ,TableName
      ,Weight as "QueryCount"
      ,100*"Weight" / sum("Weight") over(partition by 1) PercentTotal
      ,case
          when PercentTotal >=10 then 'High'
          when PercentTotal >=5 then 'Medium'
          else 'Low'
      end (char(6)) usage_freq
      ,FirstQueryDaysAgo
      ,LastQueryDaysAgo
  
      from
      (
          SELECT   TRIM(QTU1.TableName)  AS "TableName"
                  , TRIM(QTU1.DatabaseName)  AS "DatabaseName"
                  ,max((current_timestamp - CollectTimeStamp) day(4)) as "FirstQueryDaysAgo"
                  ,min((current_timestamp - CollectTimeStamp) day(4)) as "LastQueryDaysAgo"
                  , COUNT(DISTINCT QTU1.QueryID) as "Weight"
          FROM    (
                              SELECT   objectdatabasename AS DatabaseName
                                  , ObjectTableName AS TableName
                                  , QueryId
                              FROM DBC.DBQLObjTbl /* uncomment for DBC */
                              WHERE Objecttype in ('Tab', 'Viw')
                              {database_name_filter}
                              AND ObjectTableName IS NOT NULL
                              AND ObjectColumnName IS NULL
                              -- AND LogDate BETWEEN '2017-01-01' AND '2017-08-01' /* uncomment for PDCR */
                              --	AND LogDate BETWEEN current_date - 90 AND current_date - 1 /* uncomment for PDCR */
                              GROUP BY 1,2,3
                          ) AS QTU1
          INNER JOIN DBC.DBQLogTbl QU /* uncomment for DBC */
          ON QTU1.QueryID=QU.QueryID
          AND (QU.AMPCPUTime + QU.ParserCPUTime) > 0
  
          GROUP BY 1,2
      ) a
      order by PercentTotal desc
      qualify PercentTotal>0
      ;
  
      """
  
  
      with conn.cursor() as cur:
          rows = cur.execute(table_usage_sql.format(database_name_filter=database_name_filter))
          data = rows_to_json(cur.description, rows.fetchall())
      if len(data):
          info=f'This data contains the list of tables most frequently queried objects in database schema {database_name}'
      else:
          info=f'No tables have recently been queried in the database schema {database_name}.'
      metadata = {
          "tool_name": "handle_base_tableUsage",
          "database": database_name,
          "table_count": len(data),
          "comment": info
      }
      logger.debug(f"Tool: handle_base_tableUsage: metadata: {metadata}")
      return create_response(data, metadata)

• src/teradata_mcp_server/app.py:273-281 (registration)

  Dynamic registration of Python handler functions as MCP tools. Strips 'handle_' prefix from function names to derive the tool name (e.g., 'handle_base_tableUsage' becomes 'base_tableUsage') and wraps with DB connection injection and QueryBand support.

  for name, func in all_functions.items():
      if not (inspect.isfunction(func) and name.startswith("handle_")):
          continue
      tool_name = name[len("handle_"):]
      if not any(re.match(p, tool_name) for p in config.get('tool', [])):
          continue
      wrapped = make_tool_wrapper(func)
      mcp.tool(name=tool_name, description=wrapped.__doc__)(wrapped)
      logger.info(f"Created tool: {tool_name}")