Teradata MCP Server

Instructions

Implementation Reference

• src/teradata_mcp_server/tools/base/base_tools.py:14-67 (handler)

  Core handler function that executes the provided SQL query using SQLAlchemy Connection, handles bind parameters, fetches and formats results, compiles final SQL with literals for metadata, and returns a structured response.
  def handle_base_readQuery( conn: Connection, sql: str | None = None, tool_name: str | None = None, *args, **kwargs ): """ Execute a SQL query via SQLAlchemy, bind parameters if provided (prepared SQL), and return the fully rendered SQL (with literals) in metadata. Arguments: sql - SQL text, with optional bind-parameter placeholders Returns: ResponseType: formatted response with query results + metadata """ logger.debug(f"Tool: handle_base_readQuery: Args: sql: {sql}, args={args!r}, kwargs={kwargs!r}") # 1. Build a textual SQL statement stmt = text(sql) # 2. Execute with bind parameters if provided result = conn.execute(stmt, kwargs) if kwargs else conn.execute(stmt) # 3. Fetch rows & column metadata cursor = result.cursor # underlying DB-API cursor raw_rows = cursor.fetchall() or [] data = rows_to_json(cursor.description, raw_rows) columns = [ { "name": col[0], "type": getattr(col[1], "__name__", str(col[1])) } for col in (cursor.description or []) ] # 4. Compile the statement with literal binds for “final SQL” # Fallback to DefaultDialect if conn has no `.dialect` dialect = getattr(conn, "dialect", default.DefaultDialect()) compiled = stmt.compile( dialect=dialect, compile_kwargs={"literal_binds": True} ) final_sql = str(compiled) # 5. Build metadata using the rendered SQL metadata = { "tool_name": tool_name if tool_name else "base_readQuery", "sql": final_sql, "columns": columns, "row_count": len(data), } logger.debug(f"Tool: handle_base_readQuery: metadata: {metadata}") return create_response(data, metadata)
• src/teradata_mcp_server/app.py:270-282 (registration)

  Registers all discovered handle_* functions (including handle_base_readQuery) as MCP tools via the module loader, wrapping them with connection injection and QueryBand support.
  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 wrapped = make_tool_wrapper(func) mcp.tool(name=tool_name, description=wrapped.__doc__)(wrapped) logger.info(f"Created tool: {tool_name}") else:
• src/teradata_mcp_server/tools/__init__.py:24-35 (helper)

  Dynamic attribute getter in tools package __init__.py that provides access to loaded handle_* functions via the module loader when td.handle_base_readQuery is referenced.
  def __getattr__(name): """ Dynamic attribute access for lazy loading of functions. This is called when an attribute is not found in the module. """ if _module_loader: all_functions = _module_loader.get_all_functions() if name in all_functions: return all_functions[name] # If not found, raise AttributeError as usual raise AttributeError(f"module '{__name__}' has no attribute '{name}'")
• src/teradata_mcp_server/tools/__init__.py:13-18 (helper)

  Initializes the ModuleLoader instance used for lazy-loading tool modules including base_tools.py based on profile config.
  def initialize_module_loader(config: dict): """Initialize the module loader with the given profile configuration.""" global _module_loader _module_loader = ModuleLoader() _module_loader.determine_required_modules(config) return _module_loader