@@ -533,3 +533,11 @@ def _register_task(
         if task.broker != self:
             raise TaskBrokerMismatchError(broker=task.broker)
         self.local_task_registry[task_name] = task
+
+    async def __aenter__(self) -> None:
+        """Starts the broker as ctx manager."""
+        await self.startup()
+
+    async def __aexit__(self, *args: object, **kwargs: Any) -> None:
+        """Shuts down the broker as ctx manager."""
+        await self.shutdown()
@@ -4,7 +4,7 @@
 import inspect
 import sys
 from collections.abc import Callable
-from concurrent.futures import Executor
+from concurrent.futures import Executor, ProcessPoolExecutor
 from logging import getLogger
 from time import time
 from typing import Any, get_type_hints
@@ -28,6 +28,33 @@
 QUEUE_DONE = b"-1"
 
 
+def _execute_sync_task_in_executor(
+    target: Callable[..., Any],
+    args: tuple[Any, ...],
+    kwargs: dict[str, Any],
+) -> Any:
+    """Execute a sync task.
+
+    This is a wrapper to ensure we pass the target function directly
+    to the executor, avoiding issues with pickling bound methods like ctx.run.
+
+    :param target: function to execute
+    :param args: positional arguments
+    :param kwargs: keyword arguments
+    :return: result of the function call
+    """
+    return target(*args, **kwargs)
+
+
+def _is_process_pool_executor(executor: Executor) -> bool:
+    """Check if the executor is a ProcessPoolExecutor.
+
+    :param executor: executor instance
+    :return: True if it's a ProcessPoolExecutor
+    """
+    return isinstance(executor, ProcessPoolExecutor)
+
+
 class Receiver:
     """Class that uses as a callback handler."""
 
@@ -247,13 +274,28 @@ async def run_task(  # noqa: C901, PLR0912, PLR0915
                 is_coroutine = False
                 # If this is a synchronous function, we
                 # run it in executor and preserve the context.
-                ctx = contextvars.copy_context()
-                func = functools.partial(target, *message.args, **kwargs)
-                target_future = loop.run_in_executor(
-                    self.executor,
-                    ctx.run,
-                    func,
-                )
+                if _is_process_pool_executor(self.executor):
+                    # For ProcessPoolExecutor, we can't use ctx.run because it contains
+                    # a reference to contextvars.Context which cannot be pickled.
+                    # Instead, we call the target function directly in the executor.
+                    # Each worker process starts with its own context, so we don't need
+                    # to preserve the parent context.
+                    target_future = loop.run_in_executor(
+                        self.executor,
+                        _execute_sync_task_in_executor,
+                        target,
+                        message.args,
+                        kwargs,
+                    )
+                else:
+                    # For ThreadPoolExecutor, we can use ctx.run with functools.partial
+                    ctx = contextvars.copy_context()
+                    func = functools.partial(target, *message.args, **kwargs)
+                    target_future = loop.run_in_executor(
+                        self.executor,
+                        ctx.run,
+                        func,
+                    )
             timeout = message.labels.get("timeout")
             if timeout is not None:
                 if not is_coroutine:
 
@@ -1,9 +1,13 @@
 from collections.abc import AsyncGenerator
 from copy import copy
 
+import pytest
+
 from taskiq.abc.broker import AsyncBroker
 from taskiq.decor import AsyncTaskiqDecoratedTask
+from taskiq.events import TaskiqEvents
 from taskiq.message import BrokerMessage
+from taskiq.state import TaskiqState
 
 
 class _TestBroker(AsyncBroker):
@@ -76,3 +80,106 @@ async def test_task() -> None: ...
     assert "another_label" in test_kicker.labels
 
     assert test_task.labels == old_labels
+
+
+@pytest.mark.anyio
+async def test_async_context_manager_enter() -> None:
+    """Test that __aenter__ calls startup."""
+    broker = _TestBroker()
+    startup_called = False
+
+    @broker.on_event(TaskiqEvents.CLIENT_STARTUP)
+    async def track_startup(state: TaskiqState) -> None:
+        nonlocal startup_called
+        startup_called = True
+
+    async with broker:
+        assert startup_called is True
+
+
+@pytest.mark.anyio
+async def test_async_context_manager_exit() -> None:
+    """Test that __aexit__ calls shutdown."""
+    broker = _TestBroker()
+    shutdown_called = False
+
+    @broker.on_event(TaskiqEvents.CLIENT_SHUTDOWN)
+    async def track_shutdown(state: TaskiqState) -> None:
+        nonlocal shutdown_called
+        shutdown_called = True
+
+    async with broker:
+        pass
+
+    assert shutdown_called is True
+
+
+@pytest.mark.anyio
+async def test_async_context_manager_enter_worker() -> None:
+    """Test that __aenter__ calls worker startup when is_worker_process is True."""
+    broker = _TestBroker()
+    broker.is_worker_process = True
+    startup_called = False
+
+    @broker.on_event(TaskiqEvents.WORKER_STARTUP)
+    async def track_startup(state: TaskiqState) -> None:
+        nonlocal startup_called
+        startup_called = True
+
+    async with broker:
+        assert startup_called is True
+
+
+@pytest.mark.anyio
+async def test_async_context_manager_exit_worker() -> None:
+    """Test that __aexit__ calls worker shutdown when is_worker_process is True."""
+    broker = _TestBroker()
+    broker.is_worker_process = True
+    shutdown_called = False
+
+    @broker.on_event(TaskiqEvents.WORKER_SHUTDOWN)
+    async def track_shutdown(state: TaskiqState) -> None:
+        nonlocal shutdown_called
+        shutdown_called = True
+
+    async with broker:
+        pass
+
+    assert shutdown_called is True
+
+
+@pytest.mark.anyio
+async def test_async_context_manager_exit_on_exception() -> None:
+    """Test that __aexit__ calls shutdown even if exception is raised."""
+    broker = _TestBroker()
+    shutdown_called = False
+
+    @broker.on_event(TaskiqEvents.CLIENT_SHUTDOWN)
+    async def track_shutdown(state: TaskiqState) -> None:
+        nonlocal shutdown_called
+        shutdown_called = True
+
+    with pytest.raises(ValueError, match="Test exception"):
+        async with broker:
+            raise ValueError("Test exception")
+
+    assert shutdown_called is True
+
+
+@pytest.mark.anyio
+async def test_async_context_manager_exit_worker_on_exception() -> None:
+    """Test that __aexit__ calls worker shutdown even if exception is raised."""
+    broker = _TestBroker()
+    broker.is_worker_process = True
+    shutdown_called = False
+
+    @broker.on_event(TaskiqEvents.WORKER_SHUTDOWN)
+    async def track_shutdown(state: TaskiqState) -> None:
+        nonlocal shutdown_called
+        shutdown_called = True
+
+    with pytest.raises(ValueError, match="Test exception"):
+        async with broker:
+            raise ValueError("Test exception")
+
+    assert shutdown_called is True
@@ -0,0 +1,93 @@
+"""Test for ProcessPoolExecutor sync task execution."""
+import asyncio
+from concurrent.futures import ProcessPoolExecutor
+
+import pytest
+from taskiq.message import TaskiqMessage
+from taskiq.receiver import Receiver
+from tests.utils import AsyncQueueBroker
+
+
+def get_process_pool_receiver(broker) -> Receiver:
+    """Get a receiver with ProcessPoolExecutor."""
+    return Receiver(
+        broker,
+        executor=ProcessPoolExecutor(max_workers=2),
+        validate_params=True,
+    )
+
+
+# Module-level functions that can be pickled
+def my_test_func(param: int) -> int:
+    """A simple test function for ProcessPoolExecutor."""
+    return param
+
+
+def my_test_func_with_error() -> None:
+    """A test function that raises an error."""
+    raise ValueError("test error")
+
+
+def my_test_func_with_args_and_kwargs(a: int, b: str, c: float = 3.14) -> str:
+    """A test function with args and kwargs."""
+    return f"a={a}, b={b}, c={c}"
+
+
+async def test_run_task_successful_sync_with_process_pool():
+    """Tests that run_task can run sync tasks with ProcessPoolExecutor."""
+
+    receiver = get_process_pool_receiver(AsyncQueueBroker())
+
+    result = await receiver.run_task(
+        my_test_func,
+        TaskiqMessage(
+            task_id="",
+            task_name="",
+            labels={},
+            args=[42],
+            kwargs={},
+        ),
+    )
+    assert result.return_value == 42
+
+
+async def test_run_task_exception_with_process_pool():
+    """Tests that run_task can handle exceptions with ProcessPoolExecutor."""
+
+    receiver = get_process_pool_receiver(AsyncQueueBroker())
+
+    result = await receiver.run_task(
+        my_test_func_with_error,
+        TaskiqMessage(
+            task_id="",
+            task_name="",
+            labels={},
+            args=[],
+            kwargs={},
+        ),
+    )
+    assert result.return_value is None
+    assert result.is_err
+    assert isinstance(result.error, ValueError)
+
+
+async def test_sync_task_with_args_and_kwargs_process_pool():
+    """Tests sync task with both args and kwargs using ProcessPoolExecutor."""
+
+    receiver = get_process_pool_receiver(AsyncQueueBroker())
+
+    result = await receiver.run_task(
+        my_test_func_with_args_and_kwargs,
+        TaskiqMessage(
+            task_id="",
+            task_name="",
+            labels={},
+            args=[10, "hello"],
+            kwargs={"c": 2.71},
+        ),
+    )
+    assert result.return_value == "a=10, b=hello, c=2.71"
+
+
+if __name__ == "__main__":
+    pytest.main([__file__, "-v"])