Teradata MCP Server

Instructions

Implementation Reference

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

  The core handler function implementing the base_readQuery tool. It executes the provided SQL on a SQLAlchemy Connection, handles bind parameters, fetches and formats results as JSON, compiles the final rendered SQL, builds metadata including columns and row count, 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)

  Automatic registration of all handle_* functions (including handle_base_readQuery as 'base_readQuery') as MCP tools using the module_loader. The tool schema is inferred from the function signature after wrapping to remove internal parameters like 'conn' and 'tool_name'.

  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: