Teradata MCP Server

Instructions

Implementation Reference

• src/teradata_mcp_server/tools/base/base_tools.py:287-359 (handler)

  The core handler function that executes the tool logic. It runs a SQL query against DBQL tables to find other tables commonly queried alongside the given database_name.obj_name, formats the results, and returns a standardized response.
  def handle_base_tableAffinity(conn: TeradataConnection, database_name: str, obj_name: str, *args, **kwargs): """ Get tables commonly used together by database users, this is helpful to infer relationships between tables via SQLAlchemy, bind parameters if provided (prepared SQL), and return the fully rendered SQL (with literals) in metadata. Arguments: database_name - Database name object_name - table or view name Returns: ResponseType: formatted response with query results + metadata """ logger.debug(f"Tool: handle_base_tableAffinity: Args: database_name: {database_name}, obj_name: {obj_name}") table_affiity_sql=""" LOCKING ROW for ACCESS SELECT TRIM(QTU2.DatabaseName) AS "DatabaseName" , TRIM(QTU2.TableName) AS "TableName" , COUNT(DISTINCT QTU1.QueryID) AS "QueryCount" , (current_timestamp - min(QTU2.CollectTimeStamp)) day(4) as "FirstQueryDaysAgo" , (current_timestamp - max(QTU2.CollectTimeStamp)) day(4) as "LastQueryDaysAgo" FROM ( SELECT objectdatabasename AS DatabaseName , ObjectTableName AS TableName , QueryId FROM DBC.DBQLObjTbl /* for DBC */ WHERE Objecttype in ('Tab', 'Viw') AND ObjectTableName = '{table_name}' AND objectdatabasename = '{database_name}' 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 ( SELECT objectdatabasename AS DatabaseName , ObjectTableName AS TableName , QueryId , CollectTimeStamp FROM DBC.DBQLObjTbl /* for DBC */ WHERE Objecttype in ('Tab', 'Viw') AND ObjectTableName IS NOT NULL AND ObjectColumnName IS NULL GROUP BY 1,2,3, 4 ) AS QTU2 ON QTU1.QueryID=QTU2.QueryID INNER JOIN DBC.DBQLogTbl QU /* uncomment for DBC */ -- INNER JOIN DBC.DBQLogTbl QU /* uncomment for PDCR */ ON QTU1.QueryID=QU.QueryID WHERE (TRIM(QTU2.TableName) <> TRIM(QTU1.TableName) OR TRIM(QTU2.DatabaseName) <> TRIM(QTU1.DatabaseName)) AND (QU.AMPCPUTime + QU.ParserCPUTime) > 0 GROUP BY 1,2 ORDER BY 3 DESC, 5 DESC -- having "QueryCount">10 ; """ with conn.cursor() as cur: rows = cur.execute(table_affiity_sql.format(table_name=obj_name, database_name=database_name)) data = rows_to_json(cur.description, rows.fetchall()) if len(data): affinity_info=f'This data contains the list of tables most commonly queried alongside object {database_name}.{obj_name}' else: affinity_info=f'Object {database_name}.{obj_name} is not often queried with any other table or queried at all, try other ways to infer its relationships.' metadata = { "tool_name": "handle_base_tableAffinity", "database": database_name, "object": obj_name, "table_count": len(data), "comment": affinity_info } logger.debug(f"Tool: handle_base_tableAffinity: metadata: {metadata}") return create_response(data, metadata)
• src/teradata_mcp_server/app.py:343-363 (registration)

  Registers all functions starting with 'handle_' as MCP tools by stripping the 'handle_' prefix to derive the tool name (e.g., handle_base_tableAffinity becomes base_tableAffinity). Wraps the handler for MCP compatibility, handling DB connections and QueryBand.
  # Register code tools via module loader module_loader = td.initialize_module_loader(config) if module_loader: 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 # Skip template tools (used for developer reference only) if tool_name.startswith("tmpl_"): logger.debug(f"Skipping template tool: {tool_name}") continue # Skip BAR tools if BAR functionality is disabled if tool_name.startswith("bar_") and not enableBAR: logger.info(f"Skipping BAR tool: {tool_name} (BAR functionality disabled)") continue wrapped = make_tool_wrapper(func) mcp.tool(name=tool_name, description=wrapped.__doc__)(wrapped) logger.info(f"Created tool: {tool_name}")