base_readQuery
Execute SQL queries on Teradata databases using SQLAlchemy, returning results with rendered SQL metadata for analysis and management.
Instructions
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
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| sql | No |
Implementation Reference
- 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: