diff --git a/alembic/versions/502e79c55d2d_add_environment_csp_info.py b/alembic/versions/502e79c55d2d_add_environment_csp_info.py
new file mode 100644
index 00000000..3804e3a9
--- /dev/null
+++ b/alembic/versions/502e79c55d2d_add_environment_csp_info.py
@@ -0,0 +1,30 @@
+"""add Environment csp info
+
+Revision ID: 502e79c55d2d
+Revises: 4a3122ffe898
+Create Date: 2019-09-05 13:54:23.840512
+
+"""
+from alembic import op
+import sqlalchemy as sa
+from sqlalchemy.dialects import postgresql
+
+# revision identifiers, used by Alembic.
+revision = '502e79c55d2d'
+down_revision = '4a3122ffe898' # pragma: allowlist secret
+branch_labels = None
+depends_on = None
+
+
+def upgrade():
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.add_column('environments', sa.Column('baseline_info', postgresql.JSONB(astext_type=sa.Text()), nullable=True))
+    op.add_column('environments', sa.Column('root_user_info', postgresql.JSONB(astext_type=sa.Text()), nullable=True))
+    # ### end Alembic commands ###
+
+
+def downgrade():
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.drop_column('environments', 'root_user_info')
+    op.drop_column('environments', 'baseline_info')
+    # ### end Alembic commands ###
diff --git a/atst/jobs.py b/atst/jobs.py
index 2d4440b1..ceea19db 100644
--- a/atst/jobs.py
+++ b/atst/jobs.py
@@ -1,8 +1,11 @@
 from flask import current_app as app
 
-from atst.queue import celery
 from atst.database import db
+from atst.queue import celery
 from atst.models import EnvironmentJobFailure, EnvironmentRoleJobFailure
+from atst.domain.csp.cloud import CloudProviderInterface, GeneralCSPException
+from atst.domain.environments import Environments
+from atst.domain.users import Users
 
 
 class RecordEnvironmentFailure(celery.Task):
@@ -38,3 +41,79 @@ def send_notification_mail(recipients, subject, body):
         )
     )
     app.mailer.send(recipients, subject, body)
+
+
+def do_create_environment(
+    csp: CloudProviderInterface, environment_id=None, atat_user_id=None
+):
+    environment = Environments.get(environment_id)
+
+    if environment.cloud_id is not None:
+        # TODO: Return value for this?
+        return
+
+    user = Users.get(atat_user_id)
+
+    # we'll need to do some checking in this job for cases where it's retrying
+    # when a failure occured after some successful steps
+    # (e.g. if environment.cloud_id is not None, then we can skip first step)
+
+    # credentials either from a given user or pulled from config?
+    # if using global creds, do we need to log what user authorized action?
+    atat_root_creds = csp.root_creds()
+
+    # user is needed because baseline root account in the environment will
+    # be assigned to the requesting user, open question how to handle duplicate
+    # email addresses across new environments
+    csp_environment_id = csp.create_environment(atat_root_creds, user, environment)
+    environment.cloud_id = csp_environment_id
+    db.session.add(environment)
+    db.session.commit()
+
+
+def do_create_atat_admin_user(csp: CloudProviderInterface, environment_id=None):
+    environment = Environments.get(environment_id)
+    atat_root_creds = csp.root_creds()
+
+    atat_remote_root_user = csp.create_atat_admin_user(
+        atat_root_creds, environment.cloud_id
+    )
+    environment.root_user_info = atat_remote_root_user
+    db.session.add(environment)
+    db.session.commit()
+
+
+def do_create_environment_baseline(csp: CloudProviderInterface, environment_id=None):
+    environment = Environments.get(environment_id)
+
+    # ASAP switch to use remote root user for provisioning
+    atat_remote_root_creds = environment.root_user_info["credentials"]
+
+    baseline_info = csp.create_environment_baseline(
+        atat_remote_root_creds, environment.cloud_id
+    )
+    environment.baseline_info = baseline_info
+    db.session.add(environment)
+    db.session.commit()
+
+
+def do_work(fn, task, csp, **kwargs):
+    try:
+        fn(csp, **kwargs)
+    except GeneralCSPException as e:
+        raise task.retry(exc=e)
+
+
+@celery.task(bind=True)
+def create_environment(self, environment_id=None, atat_user_id=None):
+    do_work(do_create_environment, self, app.csp.cloud, **kwargs)
+
+
+@celery.task(bind=True)
+def create_atat_admin_user(self, environment_id=None):
+    do_work(do_create_atat_admin_user, self, app.csp.cloud, **kwargs)
+
+
+@celery.task(bind=True)
+def create_environment_baseline(self, environment_id=None):
+    do_work(do_create_environment_baseline, self, app.csp.cloud, **kwargs)
diff --git a/atst/models/environment.py b/atst/models/environment.py
index 1c85484a..d85550c6 100644
--- a/atst/models/environment.py
+++ b/atst/models/environment.py
@@ -1,5 +1,6 @@
 from sqlalchemy import Column, ForeignKey, String
 from sqlalchemy.orm import relationship
+from sqlalchemy.dialects.postgresql import JSONB
 
 from atst.models import Base
 from atst.models.types import Id
@@ -18,6 +19,8 @@ class Environment(
     application = relationship("Application")
 
     cloud_id = Column(String)
+    root_user_info = Column(JSONB)
+    baseline_info = Column(JSONB)
 
     job_failures = relationship("EnvironmentJobFailure")
 
diff --git a/tests/test_jobs.py b/tests/test_jobs.py
index 4ad5220c..e1ef988f 100644
--- a/tests/test_jobs.py
+++ b/tests/test_jobs.py
@@ -1,8 +1,16 @@
 import pytest
 
 from atst.jobs import RecordEnvironmentFailure, RecordEnvironmentRoleFailure
+from tests.factories import EnvironmentFactory, EnvironmentRoleFactory, UserFactory
+from uuid import uuid4
+from unittest.mock import Mock
 
-from tests.factories import EnvironmentFactory, EnvironmentRoleFactory
+from atst.jobs import (
+    do_create_environment,
+    do_create_atat_admin_user,
+    do_create_environment_baseline,
+)
+from atst.models import Environment
 
 
 def test_environment_job_failure(celery_app, celery_worker):
@@ -39,3 +47,55 @@ def test_environment_role_job_failure(celery_app, celery_worker):
     assert role.job_failures
     job_failure = role.job_failures[0]
     assert job_failure.task == task
+from tests.factories import EnvironmentFactory, UserFactory
+from atst.domain.csp.cloud import MockCloudProvider
+
+
+@pytest.fixture(autouse=True, scope="function")
+def csp():
+    return Mock(wraps=MockCloudProvider({}, with_delay=False, with_failure=False))
+
+
+def test_create_environment_job(session, csp):
+    user = UserFactory.create()
+    environment = EnvironmentFactory.create()
+    do_create_environment(csp, environment.id, user.id)
+
+    environment_id = environment.id
+    del environment
+
+    updated_environment = session.query(Environment).get(environment_id)
+
+    assert updated_environment.cloud_id
+
+
+def test_create_environment_job_is_idempotent(csp, session):
+    user = UserFactory.create()
+    environment = EnvironmentFactory.create(cloud_id=uuid4().hex)
+    do_create_environment(csp, environment.id, user.id)
+
+    csp.create_environment.assert_not_called()
+
+
+def test_create_atat_admin_user(csp, session):
+    environment = EnvironmentFactory.create(cloud_id="something")
+    do_create_atat_admin_user(csp, environment.id)
+
+    environment_id = environment.id
+    del environment
+    updated_environment = session.query(Environment).get(environment_id)
+
+    assert updated_environment.root_user_info
+
+
+def test_create_environment_baseline(csp, session):
+    environment = EnvironmentFactory.create(
+        root_user_info={"credentials": csp.root_creds()}
+    )
+    do_create_environment_baseline(csp, environment.id)
+
+    environment_id = environment.id
+    del environment
+    updated_environment = session.query(Environment).get(environment_id)
+
+    assert updated_environment.baseline_info