sql-query-mcp

Instructions

Implementation Reference

• sql_query_mcp/executor.py:34-96 (handler)

  The implementation of the run_select tool, which handles connection retrieval, SQL validation, limit enforcement, execution, auditing, and result formatting.

  def run_select(
      self, connection_id: str, sql_text: str, limit: Optional[int] = None
  ) -> Dict[str, object]:
      started = time.perf_counter()
      config = None
      row_limit = clamp_limit(
          limit, self._settings.default_limit, self._settings.max_limit
      )
  
      try:
          config = self._registry.get_connection_config(connection_id)
          cleaned_sql = validate_select_sql(sql_text, config.engine)
          limited_sql, _ = build_limited_query(cleaned_sql, row_limit)
          sql_summary = summarize_sql(cleaned_sql)
          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(limited_sql)
                  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(
                  tool="run_select",
                  connection_id=connection_id,
                  success=True,
                  duration_ms=duration_ms,
                  row_count=len(trimmed_rows),
                  sql_summary=sql_summary,
                  extra={
                      "engine": config.engine,
                      "limit": row_limit,
                      "truncated": truncated,
                  },
              )
              return {
                  "connection_id": connection_id,
                  "engine": config.engine,
                  "columns": columns,
                  "rows": trimmed_rows,
                  "row_count": len(trimmed_rows),
                  "truncated": truncated,
                  "duration_ms": duration_ms,
                  "applied_limit": row_limit,
              }
      except Exception as exc:
          duration_ms = _elapsed_ms(started)
          sql_summary = summarize_sql(sql_text)
          sanitized = sanitize_error_message(str(exc))
          self._audit.log(
              tool="run_select",
              connection_id=connection_id,
              success=False,
              duration_ms=duration_ms,
              sql_summary=sql_summary,
              error=sanitized,
              extra=_build_audit_extra(config, limit=row_limit),
          )
          raise QueryExecutionError(sanitized) from exc