Teradata MCP Server

Instructions

Implementation Reference

• src/teradata_mcp_server/tools/dba/dba_tools.py:207-342 (handler)

  The handler function that implements the core logic of the 'dba_resusageSummary' tool. It dynamically constructs and executes a complex SQL query against DBC.DBQLogTbl to provide resource usage summaries aggregated by user-specified dimensions (e.g., LogDate, hourOfDay, workloadType) and filters.

  def handle_dba_resusageSummary(conn: TeradataConnection,
                                   dimensions: list[str] | None = None,
                                   user_name: str | None = None,
                                   date:  str | None = None,
                                   dayOfWeek:  str | None = None,
                                   hourOfDay:  str | None = None,
                                   workloadType: str | None = None,
                                   workloadComplexity: str | None = None,
                                   AppId: str | None = None,
                                   *args, **kwargs):
  
      """
      Get the Teradata system usage summary metrics by weekday and hour for each workload type and query complexity bucket.
  
      Arguments:
        dimensions - list of dimensions to aggregate the resource usage summary. All dimensions are: ["LogDate", "hourOfDay", "dayOfWeek", "workloadType", "workloadComplexity", "UserName", "AppId"]
        user_name - user name
        date - Date to analyze, formatted as `YYYY-MM-DD`
        dayOfWeek - day of the week to analyze
        hourOfDay - hour of day to analyze
        workloadType - workload type to analyze, example: 'LOAD', 'ETL/ELT', 'EXPORT', 'QUERY', 'ADMIN', 'OTHER'
        workloadComplexity - workload complexity to analyze, example: 'Ingest & Prep', 'Answers', 'System/Procedural'
        AppId - Application ID to analyze, example: 'TPTLOAD%', 'TPTUPD%', 'FASTLOAD%', 'MULTLOAD%', 'EXECUTOR%', 'JDBC%'
  
      """
      logger.debug(f"Tool: handle_dba_resusageSummary: Args: dimensions: {dimensions}")
  
      comment="Total system resource usage summary."
  
      # If dimensions is not None or empty, filter in the allowed dimensions
      allowed_dimensions = ["LogDate", "hourOfDay", "dayOfWeek", "workloadType", "workloadComplexity","UserName","AppId"]
      unsupported_dimensions = []
      if dimensions is not None:
          unsupported_dimensions = [dim for dim in dimensions if dim not in allowed_dimensions]
          dimensions = [dim for dim in dimensions if dim in allowed_dimensions]
      else:
          dimensions=[]
  
  
      # Update comment string based on dimensions used and supported.
      if dimensions:
          comment+="Metrics aggregated by " + ", ".join(dimensions) + "."
      if unsupported_dimensions:
          comment+="The following dimensions are not supported and will be ignored: " + ", ".join(unsupported_dimensions) + "."
  
      # Dynamically construct the SELECT and GROUP BY clauses based on dimensions
      dim_string = ", ".join(dimensions)
      group_by_clause = ("GROUP BY " if dimensions else "")+dim_string
      dim_string += ("," if dimensions else "")
  
      filter_clause = ""
      filter_clause += f"AND UserName = '{user_name}' " if user_name else ""
      filter_clause += f"AND LogDate = '{date}' " if date else ""
      filter_clause += f"AND dayOfWeek = '{dayOfWeek}' " if dayOfWeek else ""
      filter_clause += f"AND hourOfDay = '{hourOfDay}' " if hourOfDay else ""
      filter_clause += f"AND workloadType = '{workloadType}' " if workloadType else ""
      filter_clause += f"AND workloadComplexity = '{workloadComplexity}' " if workloadComplexity else ""
      filter_clause += f"AND AppID LIKE '{AppId}' " if AppId else ""
  
      query = f"""
      SELECT
          {dim_string}
          COUNT(*) AS "Request Count",
          SUM(AMPCPUTime) AS "Total AMPCPUTime",
          SUM(TotalIOCount) AS "Total IOCount",
          SUM(ReqIOKB) AS "Total ReqIOKB",
          SUM(ReqPhysIO) AS "Total ReqPhysIO",
          SUM(ReqPhysIOKB) AS "Total ReqPhysIOKB",
          SUM(SumLogIO_GB) AS "Total ReqIO GB",
          SUM(SumPhysIO_GB) AS "Total ReqPhysIOGB",
          SUM(TotalServerByteCount) AS "Total Server Byte Count"
      FROM
          (
              SELECT
                  CAST(QryLog.Starttime as DATE) AS LogDate,
                  EXTRACT(HOUR FROM StartTime) AS hourOfDay,
                  CASE QryCal.day_of_week
                      WHEN 1 THEN 'Sunday'
                      WHEN 2 THEN 'Monday'
                      WHEN 3 THEN 'Tuesday'
                      WHEN 4 THEN 'Wednesday'
                      WHEN 5 THEN 'Thursday'
                      WHEN 6 THEN 'Friday'
                      WHEN 7 THEN 'Saturday'
                  END AS dayOfWeek,
                  QryLog.UserName,
                  QryLog.AcctString,
                  QryLog.AppID ,
                  CASE
                      WHEN QryLog.AppID LIKE ANY('TPTLOAD%', 'TPTUPD%', 'FASTLOAD%', 'MULTLOAD%', 'EXECUTOR%', 'JDBCL%') THEN 'LOAD'
                      WHEN QryLog.StatementType IN ('Insert', 'Update', 'Delete', 'Create Table', 'Merge Into')
                          AND QryLog.AppID NOT LIKE ANY('TPTLOAD%', 'TPTUPD%', 'FASTLOAD%', 'MULTLOAD%', 'EXECUTOR%', 'JDBCL%') THEN 'ETL/ELT'
                      WHEN QryLog.StatementType = 'Select' AND (AppID IN ('TPTEXP', 'FASTEXP') OR AppID LIKE 'JDBCE%') THEN 'EXPORT'
                      WHEN QryLog.StatementType = 'Select'
                          AND QryLog.AppID NOT LIKE ANY('TPTLOAD%', 'TPTUPD%', 'FASTLOAD%', 'MULTLOAD%', 'EXECUTOR%', 'JDBCL%') THEN 'QUERY'
                      WHEN QryLog.StatementType IN ('Dump Database', 'Unrecognized type', 'Release Lock', 'Collect Statistics') THEN 'ADMIN'
                      ELSE 'OTHER'
                  END AS workloadType,
                  CASE
                      WHEN StatementType = 'Merge Into' THEN 'Ingest & Prep'
                      WHEN StatementType = 'Select' THEN 'Answers'
                      ELSE 'System/Procedural'
                  END AS workloadComplexity,
                  QryLog.AMPCPUTime,
                  QryLog.TotalIOCount,
                  QryLog.ReqIOKB,
                  QryLog.ReqPhysIO,
                  QryLog.ReqPhysIOKB,
                  QryLog.TotalServerByteCount,
                  (QryLog.ReqIOKB / 1024 / 1024) AS SumLogIO_GB,
                  (QryLog.ReqPhysIOKB / 1024 / 1024) AS SumPhysIO_GB
              FROM
                  DBC.DBQLogTbl QryLog
                  INNER JOIN Sys_Calendar.CALENDAR QryCal
                      ON QryCal.calendar_date = CAST(QryLog.Starttime as DATE)
              WHERE
                  CAST(QryLog.Starttime as DATE) BETWEEN CURRENT_DATE - 30 AND CURRENT_DATE
                  AND StartTime IS NOT NULL
                  {filter_clause}
          ) AS QryDetails
          {group_by_clause}
      """
      logger.debug(f"Tool: handle_dba_resusageSummary: Query: {query}")
      with conn.cursor() as cur:
          logger.debug("Resource usage summary requested.")
          rows = cur.execute(query)
  
          data = rows_to_json(cur.description, rows.fetchall())
          metadata = {
              "tool_name": "dba_resusageSummary",
              "total_rows": len(data) ,
              "comment": comment,
              "rows": len(data)
          }
          logger.debug(f"Tool: handle_dba_resusageSummary: metadata: {metadata}")
          return create_response(data, metadata)

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

  Dynamic registration of Python handler functions as MCP tools. Functions named 'handle_<tool_name>' are automatically discovered from loaded modules, wrapped, and registered with FastMCP using the tool name 'dba_resusageSummary'.

  all_functions = module_loader.get_all_functions()
  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}")

• src/teradata_mcp_server/app.py:228-267 (schema)

  Generates the input schema for MCP tools by inspecting the handler function's signature, removing internal params like 'conn', preserving type annotations and docstring for Pydantic-based validation and client forms.

  def make_tool_wrapper(func):
      """Create an MCP-facing wrapper for a handle_* function.
  
      - Removes internal parameters (conn, tool_name, fs_config) from the MCP
        signature while still injecting them into the underlying handler.
      - Preserves the handler's parameter names and types so MCP clients can
        render friendly forms.
      """
      sig = inspect.signature(func)
      inject_kwargs = {}
      removable = {"conn", "tool_name"}
      if "fs_config" in sig.parameters:
          inject_kwargs["fs_config"] = fs_config
          removable.add("fs_config")
  
      params = [
          p for name, p in sig.parameters.items()
          if name not in removable and p.kind in (inspect.Parameter.POSITIONAL_OR_KEYWORD, inspect.Parameter.KEYWORD_ONLY)
      ]
      new_sig = sig.replace(parameters=params)
  
      # Preserve annotations for Pydantic schema generation
      annotations = {}
      for name, p in sig.parameters.items():
          if name in removable:
              continue
          if p.kind not in (inspect.Parameter.POSITIONAL_OR_KEYWORD, inspect.Parameter.KEYWORD_ONLY):
              continue
          if p.annotation is not inspect._empty:
              annotations[name] = p.annotation
  
      def _exec(*args, **kwargs):
          return execute_db_tool(func, **inject_kwargs, **kwargs)
  
      _exec.__name__ = getattr(func, "__name__", "wrapped_tool")
      _exec.__signature__ = new_sig
      _exec.__doc__ = func.__doc__
      if annotations:
          _exec.__annotations__ = annotations
      return _exec