diff --git a/src/py/mat3ra/api_client/endpoints/jobs.py b/src/py/mat3ra/api_client/endpoints/jobs.py
index 0001695..c751858 100644
--- a/src/py/mat3ra/api_client/endpoints/jobs.py
+++ b/src/py/mat3ra/api_client/endpoints/jobs.py
@@ -27,6 +27,26 @@ def __init__(self, host, port, account_id, auth_token, version=DEFAULT_API_VERSI
         super(JobEndpoints, self).__init__(host, port, account_id, auth_token, version, secure, **kwargs)
         self.name = "jobs"
 
+    def create(self, config, owner_id=None):
+        """
+        Creates a new job.
+
+        The server takes the workflow as a full document (the job keeps a snapshot of it), so a workflow given only
+        by its `_id` is fetched and embedded first.
+
+        Args:
+            config (dict): job config.
+            owner_id (str): owner ID. Job is created under user's default account if not specified.
+
+        Returns:
+             dict: new job.
+        """
+        workflow = config.get("workflow")
+        if isinstance(workflow, dict) and "_id" in workflow and "subworkflows" not in workflow:
+            workflow_document = self.request("GET", "/".join(("workflows", workflow["_id"])), headers=self.headers)
+            config = {**config, "workflow": workflow_document}
+        return super(JobEndpoints, self).create(config, owner_id)
+
     def submit(self, id_):
         """
         Submits a given job.
diff --git a/tests/py/unit/test_jobs.py b/tests/py/unit/test_jobs.py
index 1e69c40..64907c0 100644
--- a/tests/py/unit/test_jobs.py
+++ b/tests/py/unit/test_jobs.py
@@ -1,9 +1,11 @@
+import json
 from unittest import mock
 
 from mat3ra.api_client.endpoints.jobs import JobEndpoints
 from tests.py.unit.entity import EntityEndpointsUnitTest
 
 ENDPOINT_NAME = "jobs"
+WORKFLOW = {"_id": "workflowId", "name": "Total Energy", "subworkflows": [{"name": "scf", "units": []}]}
 
 
 class EndpointJobsUnitTest(EntityEndpointsUnitTest):
@@ -31,3 +33,50 @@ def test_create(self, mock_request):
     @mock.patch("requests.sessions.Session.request")
     def test_delete(self, mock_request):
         self.create(mock_request)
+
+    def respond(self, mock_request, *data):
+        responses = [json.dumps({"status": "success", "data": item}) for item in data]
+        mock_request.side_effect = [self.mock_response(response) for response in responses]
+
+    @mock.patch("requests.sessions.Session.request")
+    def test_create_embeds_a_workflow_given_by_id(self, mock_request):
+        self.respond(mock_request, WORKFLOW, {"_id": "job"})
+        config = {"name": "job", "workflow": {"_id": WORKFLOW["_id"]}}
+
+        self.assertEqual(self.endpoints.create(config), {"_id": "job"})
+
+        fetch, create = mock_request.call_args_list
+        self.assertEqual(fetch[1]["method"], "get")
+        self.assertEqual(fetch[1]["url"], f"https://{self.host}:{self.port}/api/{self.version}/workflows/{WORKFLOW['_id']}")
+        self.assertEqual(create[1]["url"], f"{self.base_url}/create")
+        self.assertEqual(json.loads(create[1]["data"]), {"name": "job", "workflow": WORKFLOW})
+        self.assertEqual(config["workflow"], {"_id": WORKFLOW["_id"]})
+
+    @mock.patch("requests.sessions.Session.request")
+    def test_create_leaves_a_full_workflow_alone(self, mock_request):
+        self.respond(mock_request, {"_id": "job"})
+
+        self.endpoints.create({"name": "job", "workflow": WORKFLOW})
+
+        self.assertEqual(mock_request.call_count, 1)
+        self.assertEqual(json.loads(mock_request.call_args[1]["data"])["workflow"], WORKFLOW)
+
+    @mock.patch("requests.sessions.Session.request")
+    def test_create_without_a_workflow_fetches_nothing(self, mock_request):
+        self.respond(mock_request, {"_id": "job"})
+
+        self.endpoints.create({"name": "job"}, owner_id="owner")
+
+        self.assertEqual(mock_request.call_count, 1)
+        self.assertEqual(json.loads(mock_request.call_args[1]["data"]), {"name": "job", "owner": {"_id": "owner"}})
+
+    @mock.patch("requests.sessions.Session.request")
+    def test_create_by_ids_embeds_the_workflow(self, mock_request):
+        self.respond(mock_request, WORKFLOW, {"_id": "job"})
+
+        jobs = self.endpoints.create_by_ids(
+            [{"_id": "material", "formula": "Si"}], WORKFLOW["_id"], "project", "prefix", owner_id="owner"
+        )
+
+        self.assertEqual(jobs, [{"_id": "job"}])
+        self.assertEqual(json.loads(mock_request.call_args[1]["data"])["workflow"], WORKFLOW)