sql-query-mcp

Instructions

Implementation Reference

• sql_query_mcp/introspection.py:104-146 (handler)

  The main implementation of the `list_tables` tool logic, which handles connection resolution, auditing, and adapter calls.

  def list_tables(
      self,
      connection_id: str,
      schema: Optional[str] = None,
      database: Optional[str] = None,
  ) -> Dict[str, object]:
      started = time.perf_counter()
      config = None
      try:
          config = self._registry.get_connection_config(connection_id)
          namespace = resolve_namespace(config, schema=schema, database=database)
          with self._registry.connection_from_config(config) as (conn, adapter):
              _apply_statement_timeout(
                  adapter, conn, self._settings.statement_timeout_ms
              )
              tables = adapter.list_tables(conn, namespace.value)
          duration_ms = _elapsed_ms(started)
          self._audit.log(
              tool="list_tables",
              connection_id=connection_id,
              success=True,
              duration_ms=duration_ms,
              row_count=len(tables),
              extra={"engine": config.engine, namespace.field_name: namespace.value},
          )
          return {
              "connection_id": connection_id,
              "engine": config.engine,
              namespace.field_name: namespace.value,
              "tables": tables,
          }
      except Exception as exc:
          duration_ms = _elapsed_ms(started)
          sanitized = sanitize_error_message(str(exc))
          self._audit.log(
              tool="list_tables",
              connection_id=connection_id,
              success=False,
              duration_ms=duration_ms,
              error=sanitized,
              extra=_build_audit_extra(config, schema=schema, database=database),
          )
          raise QueryExecutionError(sanitized) from exc

• sql_query_mcp/app.py:44-52 (registration)

  Registration of the `list_tables` MCP tool in the FastAPI app.

  @mcp.tool()
  def list_tables(
      connection_id: str,
      schema: Optional[str] = None,
      database: Optional[str] = None,
  ) -> dict:
      """List tables and views for a resolved PostgreSQL schema or MySQL database."""
  
      return _run_tool(lambda: metadata.list_tables(connection_id, schema, database))

• sql_query_mcp/adapters/mysql.py:57-58 (handler)

  MySQL implementation for listing tables.

  def list_tables(self, conn: object, database: str):
      with conn.cursor() as cur:

• sql_query_mcp/adapters/postgres.py:53-54 (handler)

  PostgreSQL implementation for listing tables.

  def list_tables(self, conn: object, schema: str):
      with conn.cursor() as cur: