get_table_sample
Fetch a small sample from database tables to discover schema structure and understand data organization for query planning.
Instructions
Fetch a small sample from a table for schema discovery.
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| connection_id | Yes | ||
| table_name | Yes | ||
| schema | No | ||
| database | No | ||
| limit | No |
Implementation Reference
- sql_query_mcp/executor.py:150-182 (handler)The actual implementation of the get_table_sample tool, which builds the query, executes it, and formats the results.
def get_table_sample( self, connection_id: str, table_name: str, schema: Optional[str] = None, database: Optional[str] = None, limit: Optional[int] = None, ) -> Dict[str, object]: row_limit = clamp_limit( limit, self._settings.default_limit, self._settings.max_limit ) started = time.perf_counter() config = None try: config = self._registry.get_connection_config(connection_id) namespace = resolve_namespace(config, schema=schema, database=database) adapter = self._registry.get_adapter(config) query = adapter.build_sample_query( namespace.value, table_name, row_limit + 1 ) with self._registry.connection_from_config(config) as (conn, adapter): _apply_statement_timeout( adapter, conn, self._settings.statement_timeout_ms ) with conn.cursor() as cur: cur.execute(query) columns = adapter.column_names(cur.description) rows = cur.fetchall() truncated = len(rows) > row_limit trimmed_rows = rows[:row_limit] duration_ms = _elapsed_ms(started) self._audit.log( - sql_query_mcp/app.py:78-87 (registration)The wrapper function in the MCP app that invokes the executor's get_table_sample method.
def get_table_sample( connection_id: str, table_name: str, schema: Optional[str] = None, database: Optional[str] = None, limit: Optional[int] = None, ) -> dict: """Fetch a small sample from a table for schema discovery.""" return _run_tool(lambda: executor.get_table_sample(connection_id, table_name, schema, database, limit))