diff --git a/foreign/python/apache_iggy.pyi b/foreign/python/apache_iggy.pyi
index 6c3715b54b..954299a3b1 100644
--- a/foreign/python/apache_iggy.pyi
+++ b/foreign/python/apache_iggy.pyi
@@ -1194,6 +1194,48 @@ class IggyClient:
         Raises:
             RuntimeError: If an identifier is invalid or the request fails.
         """
+    def create_partitions(
+        self,
+        stream_id: builtins.str | builtins.int,
+        topic_id: builtins.str | builtins.int,
+        partitions_count: builtins.int,
+    ) -> collections.abc.Awaitable[None]:
+        r"""
+        Create partitions for a topic.
+
+        Args:
+            stream_id: Stream identifier as `str | int`.
+            topic_id: Topic identifier as `str | int`.
+            partitions_count: Number of partitions to create.
+
+        Returns:
+            An awaitable that resolves to `None` when the partitions are created.
+
+        Raises:
+            ValueError: If an identifier is invalid.
+            RuntimeError: If the request fails.
+        """
+    def delete_partitions(
+        self,
+        stream_id: builtins.str | builtins.int,
+        topic_id: builtins.str | builtins.int,
+        partitions_count: builtins.int,
+    ) -> collections.abc.Awaitable[None]:
+        r"""
+        Delete the last partitions from a topic, including all messages stored in them.
+
+        Args:
+            stream_id: Stream identifier as `str | int`.
+            topic_id: Topic identifier as `str | int`.
+            partitions_count: Number of partitions to delete from the end of the topic.
+
+        Returns:
+            An awaitable that resolves to `None` when the partitions are deleted.
+
+        Raises:
+            ValueError: If an identifier is invalid.
+            RuntimeError: If the request fails.
+        """
     def create_consumer_group(
         self,
         stream_id: builtins.str | builtins.int,
diff --git a/foreign/python/src/client.rs b/foreign/python/src/client.rs
index f669a87484..b2f2372dca 100644
--- a/foreign/python/src/client.rs
+++ b/foreign/python/src/client.rs
@@ -777,6 +777,74 @@ impl IggyClient {
         })
     }
 
+    /// Create partitions for a topic.
+    ///
+    /// Args:
+    ///     stream_id: Stream identifier as `str | int`.
+    ///     topic_id: Topic identifier as `str | int`.
+    ///     partitions_count: Number of partitions to create.
+    ///
+    /// Returns:
+    ///     An awaitable that resolves to `None` when the partitions are created.
+    ///
+    /// Raises:
+    ///     ValueError: If an identifier is invalid.
+    ///     RuntimeError: If the request fails.
+    #[gen_stub(override_return_type(type_repr="collections.abc.Awaitable[None]", imports=("collections.abc")))]
+    fn create_partitions<'a>(
+        &self,
+        py: Python<'a>,
+        stream_id: PyIdentifier,
+        topic_id: PyIdentifier,
+        partitions_count: u32,
+    ) -> PyResult<Bound<'a, PyAny>> {
+        let stream_id = Identifier::try_from(stream_id)?;
+        let topic_id = Identifier::try_from(topic_id)?;
+        let inner = self.inner.clone();
+
+        future_into_py(py, async move {
+            inner
+                .create_partitions(&stream_id, &topic_id, partitions_count)
+                .await
+                .map_err(|e| PyErr::new::<pyo3::exceptions::PyRuntimeError, _>(e.to_string()))?;
+            Ok(())
+        })
+    }
+
+    /// Delete the last partitions from a topic, including all messages stored in them.
+    ///
+    /// Args:
+    ///     stream_id: Stream identifier as `str | int`.
+    ///     topic_id: Topic identifier as `str | int`.
+    ///     partitions_count: Number of partitions to delete from the end of the topic.
+    ///
+    /// Returns:
+    ///     An awaitable that resolves to `None` when the partitions are deleted.
+    ///
+    /// Raises:
+    ///     ValueError: If an identifier is invalid.
+    ///     RuntimeError: If the request fails.
+    #[gen_stub(override_return_type(type_repr="collections.abc.Awaitable[None]", imports=("collections.abc")))]
+    fn delete_partitions<'a>(
+        &self,
+        py: Python<'a>,
+        stream_id: PyIdentifier,
+        topic_id: PyIdentifier,
+        partitions_count: u32,
+    ) -> PyResult<Bound<'a, PyAny>> {
+        let stream_id = Identifier::try_from(stream_id)?;
+        let topic_id = Identifier::try_from(topic_id)?;
+        let inner = self.inner.clone();
+
+        future_into_py(py, async move {
+            inner
+                .delete_partitions(&stream_id, &topic_id, partitions_count)
+                .await
+                .map_err(|e| PyErr::new::<pyo3::exceptions::PyRuntimeError, _>(e.to_string()))?;
+            Ok(())
+        })
+    }
+
     /// Create a consumer group for a stream and topic.
     ///
     /// Args:
diff --git a/foreign/python/tests/test_partition.py b/foreign/python/tests/test_partition.py
new file mode 100644
index 0000000000..72d44e0344
--- /dev/null
+++ b/foreign/python/tests/test_partition.py
@@ -0,0 +1,124 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+import pytest
+
+from apache_iggy import IggyClient
+
+
+@pytest.mark.asyncio
+async def test_create_and_delete_partitions(iggy_client: IggyClient, unique_name):
+    stream_name = unique_name()
+    topic_name = unique_name()
+
+    await iggy_client.create_stream(stream_name)
+    await iggy_client.create_topic(
+        stream=stream_name, name=topic_name, partitions_count=2
+    )
+
+    await iggy_client.create_partitions(stream_name, topic_name, 2)
+    topic = await iggy_client.get_topic(stream_name, topic_name)
+    assert topic is not None
+    assert topic.partitions_count == 4
+    assert [partition.id for partition in topic.partitions] == [0, 1, 2, 3]
+
+    await iggy_client.delete_partitions(stream_name, topic_name, 2)
+    topic = await iggy_client.get_topic(stream_name, topic_name)
+    assert topic is not None
+    assert topic.partitions_count == 2
+    assert [partition.id for partition in topic.partitions] == [0, 1]
+
+
+@pytest.mark.asyncio
+async def test_partition_management_accepts_numeric_ids(
+    iggy_client: IggyClient, unique_name
+):
+    stream_name = unique_name()
+    topic_name = unique_name()
+
+    await iggy_client.create_stream(stream_name)
+    stream = await iggy_client.get_stream(stream_name)
+    assert stream is not None
+    await iggy_client.create_topic(
+        stream=stream.id, name=topic_name, partitions_count=2
+    )
+    topic = await iggy_client.get_topic(stream.id, topic_name)
+    assert topic is not None
+
+    await iggy_client.create_partitions(stream.id, topic.id, 1)
+    await iggy_client.delete_partitions(stream.id, topic.id, 1)
+
+    topic = await iggy_client.get_topic(stream.id, topic.id)
+    assert topic is not None
+    assert [partition.id for partition in topic.partitions] == [0, 1]
+
+
+@pytest.mark.asyncio
+async def test_partition_management_rejects_zero_count(
+    iggy_client: IggyClient, unique_name
+):
+    stream_name = unique_name()
+    topic_name = unique_name()
+
+    await iggy_client.create_stream(stream_name)
+    await iggy_client.create_topic(
+        stream=stream_name, name=topic_name, partitions_count=2
+    )
+
+    with pytest.raises(RuntimeError):
+        await iggy_client.create_partitions(stream_name, topic_name, 0)
+    with pytest.raises(RuntimeError):
+        await iggy_client.delete_partitions(stream_name, topic_name, 0)
+
+
+@pytest.mark.asyncio
+async def test_delete_partitions_rejects_count_larger_than_topic(
+    iggy_client: IggyClient, unique_name
+):
+    stream_name = unique_name()
+    topic_name = unique_name()
+
+    await iggy_client.create_stream(stream_name)
+    await iggy_client.create_topic(
+        stream=stream_name, name=topic_name, partitions_count=2
+    )
+
+    with pytest.raises(RuntimeError):
+        await iggy_client.delete_partitions(stream_name, topic_name, 3)
+
+
+@pytest.mark.asyncio
+async def test_partition_management_rejects_missing_stream_or_topic(
+    iggy_client: IggyClient, unique_name
+):
+    stream_name = unique_name()
+    topic_name = unique_name()
+    missing_name = unique_name()
+
+    await iggy_client.create_stream(stream_name)
+    await iggy_client.create_topic(
+        stream=stream_name, name=topic_name, partitions_count=2
+    )
+
+    with pytest.raises(RuntimeError):
+        await iggy_client.create_partitions(missing_name, topic_name, 1)
+    with pytest.raises(RuntimeError):
+        await iggy_client.create_partitions(stream_name, missing_name, 1)
+    with pytest.raises(RuntimeError):
+        await iggy_client.delete_partitions(missing_name, topic_name, 1)
+    with pytest.raises(RuntimeError):
+        await iggy_client.delete_partitions(stream_name, missing_name, 1)