base_readQuery
Execute SQL queries via SQLAlchemy, bind parameters for prepared SQL, and retrieve fully rendered SQL with literals in metadata. Ideal for querying and analyzing Teradata databases.
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
- Core implementation of the base_readQuery tool handler. Executes SQL query using SQLAlchemy, handles bind parameters, fetches and formats results as JSON, compiles final SQL with literals, and returns structured response with metadata.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:344-363 (registration)Dynamic registration of all handle_* functions (including handle_base_readQuery as 'base_readQuery') as MCP tools using module_loader to discover functions from loaded tool modules based on profile config.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}")