Teradata MCP Server

Instructions

Implementation Reference

• src/teradata_mcp_server/tools/dba/dba_tools.py:49-88 (handler)

  The core handler function for the 'dba_userSqlList' tool. It executes SQL queries against Teradata's DBC.QryLog views to retrieve recent SQL statements run by a specific user (or all users if none specified), formats the results, and returns them with metadata.
  def handle_dba_userSqlList(conn: TeradataConnection, user_name: str, no_days: int | None = 7, *args, **kwargs): """ Get a list of SQL run by a user in the last number of days if a user name is provided, otherwise get list of all SQL in the last number of days. Arguments: user_name - user name no_days - number of days Returns: ResponseType: formatted response with query results + metadata """ logger.debug(f"Tool: handle_dba_userSqlList: Args: user_name: {user_name}") with conn.cursor() as cur: if user_name == "": logger.debug("No user name provided, returning all SQL queries.") rows = cur.execute(f"""SELECT t1.QueryID, t1.ProcID, t1.CollectTimeStamp, t1.SqlTextInfo, t2.UserName FROM DBC.QryLogSqlV t1 JOIN DBC.QryLogV t2 ON t1.QueryID = t2.QueryID WHERE t1.CollectTimeStamp >= CURRENT_TIMESTAMP - INTERVAL '{no_days}' DAY ORDER BY t1.CollectTimeStamp DESC;""") else: logger.debug(f"User name provided: {user_name}, returning SQL queries for this user.") rows = cur.execute(f"""SELECT t1.QueryID, t1.ProcID, t1.CollectTimeStamp, t1.SqlTextInfo, t2.UserName FROM DBC.QryLogSqlV t1 JOIN DBC.QryLogV t2 ON t1.QueryID = t2.QueryID WHERE t1.CollectTimeStamp >= CURRENT_TIMESTAMP - INTERVAL '{no_days}' DAY AND t2.UserName = '{user_name}' ORDER BY t1.CollectTimeStamp DESC;""") data = rows_to_json(cur.description, rows.fetchall()) metadata = { "tool_name": "dba_userSqlList", "user_name": user_name, "no_days": no_days, "total_queries": len(data) } logger.debug(f"Tool: handle_dba_userSqlList: metadata: {metadata}") return create_response(data, metadata)
• src/teradata_mcp_server/app.py:273-281 (registration)

  Dynamic registration of handler functions as MCP tools. Functions named 'handle_*' are discovered via module_loader, wrapped, and registered with mcp.tool using the tool name derived from the function name (e.g., handle_dba_userSqlList -> dba_userSqlList).
  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/tools/dba/__init__.py:1-2 (helper)

  Package init file that imports dba_tools (containing the handler) and dba_resources, making the functions available for discovery by the module_loader.
  from .dba_resources import * from .dba_tools import *