diff --git a/docs/docs.md b/docs/docs.md
index 6d4d7e3d..cba8ea13 100644
--- a/docs/docs.md
+++ b/docs/docs.md
@@ -352,7 +352,7 @@ Return True when the actor is running on the Apify platform, and False otherwise
 Return a dictionary with information parsed from all the APIFY_XXX environment variables.
 
 For a list of all the environment variables,
-see the [Actor documentation]([https://docs.apify.com/actor/run#environment-variables](https://docs.apify.com/actor/run#environment-variables)).
+see the [Actor documentation]([https://docs.apify.com/actors/development/environment-variables](https://docs.apify.com/actors/development/environment-variables)).
 If some variables are not defined or are invalid, the corresponding value in the resulting dictionary will be None.
 
 * **Return type**
diff --git a/pytest.ini b/pytest.ini
index 40880458..1a2cc47c 100644
--- a/pytest.ini
+++ b/pytest.ini
@@ -1,2 +1,3 @@
 [pytest]
 asyncio_mode=auto
+timeout = 1200
diff --git a/setup.py b/setup.py
index fc425344..40cfaaf3 100644
--- a/setup.py
+++ b/setup.py
@@ -77,7 +77,9 @@
             'pre-commit ~= 2.20.0',
             'pytest ~= 7.2.0',
             'pytest-asyncio ~= 0.20.3',
+            'pytest-only ~= 2.0.0',
             'pytest-randomly ~= 3.12.0',
+            'pytest-timeout ~= 2.1.0',
             'pytest-xdist ~= 3.1.0',
             'respx ~= 0.20.1',
             'sphinx ~= 5.3.0',
diff --git a/src/apify/actor.py b/src/apify/actor.py
index 8cc3432c..22b9931b 100644
--- a/src/apify/actor.py
+++ b/src/apify/actor.py
@@ -698,7 +698,7 @@ def get_env(cls) -> Dict:
         """Return a dictionary with information parsed from all the `APIFY_XXX` environment variables.
 
         For a list of all the environment variables,
-        see the [Actor documentation](https://docs.apify.com/actor/run#environment-variables).
+        see the [Actor documentation](https://docs.apify.com/actors/development/environment-variables).
         If some variables are not defined or are invalid, the corresponding value in the resulting dictionary will be None.
         """
         return cls._get_default_instance().get_env()
diff --git a/tests/integration/README.md b/tests/integration/README.md
index e35eecd5..63451f64 100644
--- a/tests/integration/README.md
+++ b/tests/integration/README.md
@@ -30,11 +30,15 @@ This fixture returns a factory function for creating actors on the Apify Platfor
 For the actor source, the fixture takes the files from `tests/integration/actor_source_base`,
 builds the Apify SDK wheel from the current codebase,
 and adds the actor source you passed to the fixture as an argument.
+You have to pass exactly one of the `main_func`, `main_py` and `source_files` arguments.
 
 The created actor will be uploaded to the platform, built there, and after the test finishes, it will be automatically deleted.
 If the actor build fails, it will not be deleted, so that you can check why the build failed.
 
-You can create actors straight from a Python function:
+### Creating test actor straight from a Python function
+
+You can create actors straight from a Python function.
+This is great because you can have the test actor source code checked with the linter.
 
 ```python
 async def test_something(self, make_actor: ActorFactory) -> None:
@@ -50,8 +54,34 @@ async def test_something(self, make_actor: ActorFactory) -> None:
     assert run_result['status'] == 'SUCCEEDED'
 ```
 
-Or you can pass the `src/main.py` file, if you need something more complex
-(e.g. specify more imports or pass some fixed value to the actor source code):
+These actors will have the `src/main.py` file set to the `main` function definition,
+prepended with `import asyncio` and `from apify import Actor`, for your convenience.
+
+You can also pass extra imports directly to the main function:
+
+```python
+async def test_something(self, make_actor: ActorFactory) -> None:
+    async def main():
+        import os
+        from apify.consts import ActorEventType, ApifyEnvVars
+        async with Actor:
+            print('The actor is running with ' + os.getenv(ApifyEnvVars.MEMORY_MBYTES) + 'MB of memory')
+            await Actor.on(ActorEventType.SYSTEM_INFO, lambda event_data: print(event_data))
+
+    actor = await make_actor('something', main_func=main)
+
+    run_result = await actor.call()
+
+    assert run_result is not None
+    assert run_result['status'] == 'SUCCEEDED'
+```
+
+### Creating actor from source files
+
+You can also pass the source files directly if you need something more complex
+(e.g. pass some fixed value to the actor source code or use multiple source files).
+
+To pass the source code of the `src/main.py` file directly, use the `main_py` argument to `make_actor`:
 
 ```python
 async def test_something(self, make_actor: ActorFactory) -> None:
@@ -76,7 +106,8 @@ async def test_something(self, make_actor: ActorFactory) -> None:
 
 ```
 
-Or you can pass multiple source files, if you need something really complex:
+Or you can pass multiple source files with the `source_files` argument,
+if you need something really complex:
 
 ```python
 async def test_something(self, make_actor: ActorFactory) -> None:
@@ -104,5 +135,3 @@ async def test_something(self, make_actor: ActorFactory) -> None:
     assert actor_run is not None
     assert actor_run['status'] == 'SUCCEEDED'
 ```
-
-You have to pass exactly one of the `main_func`, `main_py` and `source_files` arguments.
diff --git a/tests/integration/test_actor_api_helpers.py b/tests/integration/test_actor_api_helpers.py
new file mode 100644
index 00000000..a6d2fd5a
--- /dev/null
+++ b/tests/integration/test_actor_api_helpers.py
@@ -0,0 +1,392 @@
+import asyncio
+import json
+
+from apify import Actor
+from apify._utils import _crypto_random_object_id
+from apify_client import ApifyClientAsync
+
+from ._utils import generate_unique_resource_name
+from .conftest import ActorFactory
+
+
+class TestActorIsAtHome:
+    async def test_actor_is_at_home(self, make_actor: ActorFactory) -> None:
+        async def main() -> None:
+            async with Actor:
+                assert Actor.is_at_home() is True
+
+        actor = await make_actor('is-at-home', main_func=main)
+
+        run_result = await actor.call()
+
+        assert run_result is not None
+        assert run_result['status'] == 'SUCCEEDED'
+
+
+class TestActorGetEnv:
+    async def test_actor_get_env(self, make_actor: ActorFactory) -> None:
+        async def main() -> None:
+            async with Actor:
+                env_dict = Actor.get_env()
+
+                assert env_dict.get('is_at_home') is True
+                assert env_dict.get('token') is not None
+                assert env_dict.get('actor_events_ws_url') is not None
+                assert env_dict.get('input_key') is not None
+
+                assert len(env_dict.get('actor_id', '')) == 17
+                assert len(env_dict.get('actor_run_id', '')) == 17
+                assert len(env_dict.get('user_id', '')) == 17
+                assert len(env_dict.get('default_dataset_id', '')) == 17
+                assert len(env_dict.get('default_key_value_store_id', '')) == 17
+                assert len(env_dict.get('default_request_queue_id', '')) == 17
+
+        actor = await make_actor('get-env', main_func=main)
+
+        run_result = await actor.call()
+
+        assert run_result is not None
+        assert run_result['status'] == 'SUCCEEDED'
+
+
+class TestActorNewClient:
+    async def test_actor_new_client(self, make_actor: ActorFactory) -> None:
+        async def main() -> None:
+            import os
+
+            from apify.consts import ApifyEnvVars
+
+            async with Actor:
+                new_client = Actor.new_client()
+                assert new_client is not Actor.apify_client
+
+                default_key_value_store_id = os.getenv(ApifyEnvVars.DEFAULT_KEY_VALUE_STORE_ID)
+                assert default_key_value_store_id is not None
+                kv_store_client = new_client.key_value_store(default_key_value_store_id)
+                await kv_store_client.set_record('OUTPUT', 'TESTING-OUTPUT')
+
+        actor = await make_actor('new-client', main_func=main)
+
+        run_result = await actor.call()
+
+        assert run_result is not None
+        assert run_result['status'] == 'SUCCEEDED'
+
+        output_record = await actor.last_run().key_value_store().get_record('OUTPUT')
+        assert output_record is not None
+        assert output_record['value'] == 'TESTING-OUTPUT'
+
+
+class TestActorSetStatusMessage:
+    async def test_actor_set_status_message(self, make_actor: ActorFactory) -> None:
+        async def main() -> None:
+            async with Actor:
+                await Actor.set_status_message('testing-status-message')
+
+        actor = await make_actor('set-status-message', main_func=main)
+
+        run_result = await actor.call()
+
+        assert run_result is not None
+        assert run_result['status'] == 'SUCCEEDED'
+        assert run_result['statusMessage'] == 'testing-status-message'
+
+
+class TestActorStart:
+    async def test_actor_start(self, make_actor: ActorFactory) -> None:
+        async def main_inner() -> None:
+            async with Actor:
+                await asyncio.sleep(5)
+                input = await Actor.get_input() or {}
+                test_value = input.get('test_value')
+                await Actor.set_value('OUTPUT', f'{test_value}_XXX_{test_value}')
+
+        async def main_outer() -> None:
+            async with Actor:
+                input = await Actor.get_input() or {}
+                inner_actor_id = input.get('inner_actor_id')
+                test_value = input.get('test_value')
+
+                assert inner_actor_id is not None
+
+                await Actor.start(inner_actor_id, run_input={'test_value': test_value})
+
+                inner_run_status = await Actor.apify_client.actor(inner_actor_id).last_run().get()
+                assert inner_run_status is not None
+                assert inner_run_status.get('status') in ['READY', 'RUNNING']
+
+        inner_actor = await make_actor('start-inner', main_func=main_inner)
+        outer_actor = await make_actor('start-outer', main_func=main_outer)
+
+        inner_actor_id = (await inner_actor.get() or {})['id']
+        test_value = _crypto_random_object_id()
+
+        outer_run_result = await outer_actor.call(run_input={'test_value': test_value, 'inner_actor_id': inner_actor_id})
+
+        assert outer_run_result is not None
+        assert outer_run_result['status'] == 'SUCCEEDED'
+
+        await inner_actor.last_run().wait_for_finish()
+
+        inner_output_record = await inner_actor.last_run().key_value_store().get_record('OUTPUT')
+        assert inner_output_record is not None
+        assert inner_output_record['value'] == f'{test_value}_XXX_{test_value}'
+
+
+class TestActorCall:
+    async def test_actor_call(self, make_actor: ActorFactory) -> None:
+        async def main_inner() -> None:
+            async with Actor:
+                await asyncio.sleep(5)
+                input = await Actor.get_input() or {}
+                test_value = input.get('test_value')
+                await Actor.set_value('OUTPUT', f'{test_value}_XXX_{test_value}')
+
+        async def main_outer() -> None:
+            async with Actor:
+                input = await Actor.get_input() or {}
+                inner_actor_id = input.get('inner_actor_id')
+                test_value = input.get('test_value')
+
+                assert inner_actor_id is not None
+
+                await Actor.call(inner_actor_id, run_input={'test_value': test_value})
+
+                inner_run_status = await Actor.apify_client.actor(inner_actor_id).last_run().get()
+                assert inner_run_status is not None
+                assert inner_run_status.get('status') == 'SUCCEEDED'
+
+        inner_actor = await make_actor('call-inner', main_func=main_inner)
+        outer_actor = await make_actor('call-outer', main_func=main_outer)
+
+        inner_actor_id = (await inner_actor.get() or {})['id']
+        test_value = _crypto_random_object_id()
+
+        outer_run_result = await outer_actor.call(run_input={'test_value': test_value, 'inner_actor_id': inner_actor_id})
+
+        assert outer_run_result is not None
+        assert outer_run_result['status'] == 'SUCCEEDED'
+
+        await inner_actor.last_run().wait_for_finish()
+
+        inner_output_record = await inner_actor.last_run().key_value_store().get_record('OUTPUT')
+        assert inner_output_record is not None
+        assert inner_output_record['value'] == f'{test_value}_XXX_{test_value}'
+
+
+class TestActorCallTask:
+    async def test_actor_call_task(self, make_actor: ActorFactory, apify_client_async: ApifyClientAsync) -> None:
+        async def main_inner() -> None:
+            async with Actor:
+                await asyncio.sleep(5)
+                input = await Actor.get_input() or {}
+                test_value = input.get('test_value')
+                await Actor.set_value('OUTPUT', f'{test_value}_XXX_{test_value}')
+
+        async def main_outer() -> None:
+            async with Actor:
+                input = await Actor.get_input() or {}
+                inner_task_id = input.get('inner_task_id')
+
+                assert inner_task_id is not None
+
+                await Actor.call_task(inner_task_id)
+
+                inner_run_status = await Actor.apify_client.task(inner_task_id).last_run().get()
+                assert inner_run_status is not None
+                assert inner_run_status.get('status') == 'SUCCEEDED'
+
+        inner_actor = await make_actor('call-task-inner', main_func=main_inner)
+        outer_actor = await make_actor('call-task-outer', main_func=main_outer)
+
+        inner_actor_id = (await inner_actor.get() or {})['id']
+        test_value = _crypto_random_object_id()
+
+        task = await apify_client_async.tasks().create(
+            actor_id=inner_actor_id,
+            name=generate_unique_resource_name('actor-call-task'),
+            task_input={'test_value': test_value},
+        )
+
+        outer_run_result = await outer_actor.call(run_input={'test_value': test_value, 'inner_task_id': task['id']})
+
+        assert outer_run_result is not None
+        assert outer_run_result['status'] == 'SUCCEEDED'
+
+        await inner_actor.last_run().wait_for_finish()
+
+        inner_output_record = await inner_actor.last_run().key_value_store().get_record('OUTPUT')
+        assert inner_output_record is not None
+        assert inner_output_record['value'] == f'{test_value}_XXX_{test_value}'
+
+        await apify_client_async.task(task['id']).delete()
+
+
+class TestActorAbort:
+    async def test_actor_abort(self, make_actor: ActorFactory) -> None:
+        async def main_inner() -> None:
+            async with Actor:
+                await asyncio.sleep(60)
+                # This should not be set, the actor should be aborted by now
+                await Actor.set_value('OUTPUT', 'dummy')
+
+        async def main_outer() -> None:
+            async with Actor:
+                input = await Actor.get_input() or {}
+                inner_run_id = input.get('inner_run_id')
+
+                assert inner_run_id is not None
+
+                await Actor.abort(inner_run_id)
+
+        inner_actor = await make_actor('abort-inner', main_func=main_inner)
+        outer_actor = await make_actor('abort-outer', main_func=main_outer)
+
+        inner_run_id = (await inner_actor.start())['id']
+
+        outer_run_result = await outer_actor.call(run_input={'inner_run_id': inner_run_id})
+
+        assert outer_run_result is not None
+        assert outer_run_result['status'] == 'SUCCEEDED'
+
+        await inner_actor.last_run().wait_for_finish()
+        inner_actor_last_run = await inner_actor.last_run().get()
+        assert inner_actor_last_run is not None
+        assert inner_actor_last_run['status'] == 'ABORTED'
+
+        inner_output_record = await inner_actor.last_run().key_value_store().get_record('OUTPUT')
+        assert inner_output_record is None
+
+
+class TestActorMetamorph:
+    async def test_actor_metamorph(self, make_actor: ActorFactory) -> None:
+        async def main_inner() -> None:
+            import os
+
+            from apify.consts import ApifyEnvVars
+
+            async with Actor:
+                assert os.getenv(ApifyEnvVars.INPUT_KEY) is not None
+                assert os.getenv(ApifyEnvVars.INPUT_KEY) != 'INPUT'
+                input = await Actor.get_input() or {}
+
+                test_value = input.get('test_value', '')
+                assert test_value.endswith('_BEFORE_METAMORPH')
+
+                output = test_value.replace('_BEFORE_METAMORPH', '_AFTER_METAMORPH')
+                await Actor.set_value('OUTPUT', output)
+
+        async def main_outer() -> None:
+            async with Actor:
+                input = await Actor.get_input() or {}
+                inner_actor_id = input.get('inner_actor_id')
+                test_value = input.get('test_value')
+                new_test_value = f'{test_value}_BEFORE_METAMORPH'
+
+                assert inner_actor_id is not None
+
+                await Actor.metamorph(inner_actor_id, run_input={'test_value': new_test_value})
+
+                # This should not be called
+                await Actor.set_value('RECORD_AFTER_METAMORPH_CALL', 'dummy')
+                raise AssertionError('The actor should have been metamorphed by now')
+
+        inner_actor = await make_actor('metamorph-inner', main_func=main_inner)
+        outer_actor = await make_actor('metamorph-outer', main_func=main_outer)
+
+        inner_actor_id = (await inner_actor.get() or {})['id']
+        test_value = _crypto_random_object_id()
+
+        outer_run_result = await outer_actor.call(run_input={'test_value': test_value, 'inner_actor_id': inner_actor_id})
+
+        assert outer_run_result is not None
+        assert outer_run_result['status'] == 'SUCCEEDED'
+
+        outer_run_key_value_store = outer_actor.last_run().key_value_store()
+
+        outer_output_record = await outer_run_key_value_store.get_record('OUTPUT')
+        assert outer_output_record is not None
+        assert outer_output_record['value'] == f'{test_value}_AFTER_METAMORPH'
+
+        assert await outer_run_key_value_store.get_record('RECORD_AFTER_METAMORPH_CALL') is None
+
+        # After metamorph, the run still belongs to the original actor, so the inner one should have no runs
+        assert await inner_actor.last_run().get() is None
+
+
+class TestActorAddWebhook:
+    async def test_actor_add_webhook(self, make_actor: ActorFactory) -> None:
+        async def main_server() -> None:
+            import os
+            from http.server import BaseHTTPRequestHandler, HTTPServer
+
+            from apify.consts import ApifyEnvVars
+
+            webhook_body = ''
+
+            async with Actor:
+                class WebhookHandler(BaseHTTPRequestHandler):
+                    def do_GET(self) -> None:  # noqa: N802
+                        self.send_response(200)
+                        self.end_headers()
+                        self.wfile.write(bytes('Hello, world!', encoding='utf-8'))
+
+                    def do_POST(self) -> None:  # noqa: N802
+                        nonlocal webhook_body
+                        content_length = self.headers.get('content-length')
+                        length = int(content_length) if content_length else 0
+
+                        webhook_body = self.rfile.read(length).decode('utf-8')
+
+                        self.send_response(200)
+                        self.end_headers()
+                        self.wfile.write(bytes('Hello, world!', encoding='utf-8'))
+
+                container_port = int(os.getenv(ApifyEnvVars.CONTAINER_PORT, ''))
+                with HTTPServer(('', container_port), WebhookHandler) as server:
+                    await Actor.set_value('INITIALIZED', True)
+                    while not webhook_body:
+                        server.handle_request()
+
+                await Actor.set_value('WEBHOOK_BODY', webhook_body)
+
+        async def main_client() -> None:
+            from apify_client.consts import WebhookEventType
+            async with Actor:
+                input = await Actor.get_input() or {}
+                server_actor_container_url = str(input.get('server_actor_container_url'))
+
+                await Actor.add_webhook(
+                    event_types=[WebhookEventType.ACTOR_RUN_SUCCEEDED],
+                    request_url=server_actor_container_url,
+                )
+
+        server_actor, client_actor = await asyncio.gather(
+            make_actor('add-webhook-server', main_func=main_server),
+            make_actor('add-webhook-client', main_func=main_client),
+        )
+
+        server_actor_run = await server_actor.start()
+        server_actor_container_url = server_actor_run['containerUrl']
+
+        # Give the server actor some time to start running
+        server_actor_initialized = await server_actor.last_run().key_value_store().get_record('INITIALIZED')
+        while not server_actor_initialized:
+            server_actor_initialized = await server_actor.last_run().key_value_store().get_record('INITIALIZED')
+            await asyncio.sleep(1)
+
+        client_actor_run_result = await client_actor.call(run_input={'server_actor_container_url': server_actor_container_url})
+        assert client_actor_run_result is not None
+        assert client_actor_run_result['status'] == 'SUCCEEDED'
+
+        server_actor_run_result = await server_actor.last_run().wait_for_finish()
+        assert server_actor_run_result is not None
+        assert server_actor_run_result['status'] == 'SUCCEEDED'
+
+        webhook_body_record = await server_actor.last_run().key_value_store().get_record('WEBHOOK_BODY')
+        assert webhook_body_record is not None
+        assert webhook_body_record['value'] != ''
+        parsed_webhook_body = json.loads(webhook_body_record['value'])
+
+        assert parsed_webhook_body['eventData']['actorId'] == client_actor_run_result['actId']
+        assert parsed_webhook_body['eventData']['actorRunId'] == client_actor_run_result['id']
diff --git a/tests/integration/test_fixtures.py b/tests/integration/test_fixtures.py
index a6bd9df0..dd68bf3a 100644
--- a/tests/integration/test_fixtures.py
+++ b/tests/integration/test_fixtures.py
@@ -10,8 +10,12 @@
 class TestMakeActorFixture:
     async def test_main_func(self, make_actor: ActorFactory) -> None:
         async def main() -> None:
+            import os
+
+            from apify.consts import ApifyEnvVars
+
             async with Actor:
-                await Actor.set_value('OUTPUT', 'TESTING_123')
+                await Actor.set_value('OUTPUT', os.getenv(ApifyEnvVars.ACTOR_ID))
 
         actor = await make_actor('make-actor-main-func', main_func=main)
 
@@ -22,7 +26,7 @@ async def main() -> None:
 
         output_record = await actor.last_run().key_value_store().get_record('OUTPUT')
         assert output_record is not None
-        assert output_record['value'] == 'TESTING_123'
+        assert run_result['actId'] == output_record['value']
 
     async def test_main_py(self, make_actor: ActorFactory) -> None:
         expected_output = f'ACTOR_OUTPUT_{_crypto_random_object_id(5)}'