sql-query-mcp

Instructions

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))