diff --git a/examples/python/basic/consumer.py b/examples/python/basic/consumer.py
index e20cf3175b..cf35faefb7 100644
--- a/examples/python/basic/consumer.py
+++ b/examples/python/basic/consumer.py
@@ -19,7 +19,7 @@
 import asyncio
 from typing import NamedTuple
 
-from apache_iggy import IggyClient, PollingStrategy, ReceiveMessage
+from apache_iggy import IggyClient, IggyError, PollingStrategy, ReceiveMessage
 from loguru import logger
 
 STREAM_NAME = "sample-stream"
@@ -88,6 +88,12 @@ async def consume_messages(client: IggyClient):
                 handle_message(message)
             n_consumed_batches += 1
             await asyncio.sleep(interval)
+        except IggyError as error:
+            logger.error(
+                "Iggy error while consuming messages: "
+                f"[{error.code}] {error.name}: {error.message}"
+            )
+            break
         except Exception as error:
             logger.exception(f"Exception occurred while consuming messages: {error}")
             break
diff --git a/examples/python/basic/producer.py b/examples/python/basic/producer.py
index b04ca8fc86..5566f4d08d 100644
--- a/examples/python/basic/producer.py
+++ b/examples/python/basic/producer.py
@@ -19,7 +19,7 @@
 import asyncio
 from typing import NamedTuple
 
-from apache_iggy import IggyClient, StreamDetails, TopicDetails
+from apache_iggy import IggyClient, IggyError, StreamDetails, TopicDetails
 from apache_iggy import SendMessage as Message
 from loguru import logger
 
@@ -69,6 +69,10 @@ async def init_system(client: IggyClient):
         else:
             logger.warning(f"Stream {stream.name} already exists with ID {stream.id}")
 
+    except IggyError as error:
+        logger.error(
+            f"Error creating stream: [{error.code}] {error.name}: {error.message}"
+        )
     except Exception as error:
         logger.error(f"Error creating stream: {error}")
         logger.exception(error)
@@ -86,6 +90,10 @@ async def init_system(client: IggyClient):
             logger.info("Topic was created successfully.")
         else:
             logger.warning(f"Topic {topic.name} already exists with ID {topic.id}")
+    except IggyError as error:
+        logger.error(
+            f"Error creating topic: [{error.code}] {error.name}: {error.message}"
+        )
     except Exception as error:
         logger.error(f"Error creating topic {error}")
         logger.exception(error)
@@ -124,6 +132,11 @@ async def produce_messages(client: IggyClient):
                 f"Successfully sent batch of {messages_per_batch} messages. "
                 f"Batch ID: {current_id // messages_per_batch}"
             )
+        except IggyError as error:
+            logger.error(
+                "Iggy error while sending messages: "
+                f"[{error.code}] {error.name}: {error.message}"
+            )
         except Exception as error:
             logger.error(f"Exception type: {type(error).__name__}, message: {error}")
             logger.exception(error)
diff --git a/examples/python/getting-started/consumer.py b/examples/python/getting-started/consumer.py
index bb10e91382..b60f01386d 100755
--- a/examples/python/getting-started/consumer.py
+++ b/examples/python/getting-started/consumer.py
@@ -20,7 +20,7 @@
 import typing
 import urllib.parse
 
-from apache_iggy import IggyClient, PollingStrategy, ReceiveMessage
+from apache_iggy import IggyClient, IggyError, PollingStrategy, ReceiveMessage
 from loguru import logger
 
 STREAM_NAME = "sample-stream"
@@ -122,6 +122,10 @@ async def main():
         await client.connect()
         logger.info("Connected.")
         await consume_messages(client)
+    except IggyError as error:
+        logger.error(
+            f"Iggy error in main: [{error.code}] {error.name}: {error.message}"
+        )
     except Exception as error:
         logger.exception(f"Exception occurred in main function: {error}")
 
@@ -157,6 +161,12 @@ async def consume_messages(client: IggyClient):
                 handle_message(message)
             n_consumed_batches += 1
             await asyncio.sleep(interval)
+        except IggyError as error:
+            logger.error(
+                "Iggy error while consuming messages: "
+                f"[{error.code}] {error.name}: {error.message}"
+            )
+            break
         except Exception as error:
             logger.exception(f"Exception occurred while consuming messages: {error}")
             break
diff --git a/examples/python/getting-started/producer.py b/examples/python/getting-started/producer.py
index c05c0317cd..6cb95f2231 100755
--- a/examples/python/getting-started/producer.py
+++ b/examples/python/getting-started/producer.py
@@ -20,7 +20,7 @@
 import typing
 import urllib.parse
 
-from apache_iggy import IggyClient, StreamDetails, TopicDetails
+from apache_iggy import IggyClient, IggyError, StreamDetails, TopicDetails
 from apache_iggy import SendMessage as Message
 from loguru import logger
 
@@ -135,6 +135,10 @@ async def init_system(client: IggyClient):
         else:
             logger.warning(f"Stream {stream.name} already exists with ID {stream.id}")
 
+    except IggyError as error:
+        logger.error(
+            f"Error creating stream: [{error.code}] {error.name}: {error.message}"
+        )
     except Exception as error:
         logger.error(f"Error creating stream: {error}")
         logger.exception(error)
@@ -152,6 +156,10 @@ async def init_system(client: IggyClient):
             logger.info("Topic was created successfully.")
         else:
             logger.warning(f"Topic {topic.name} already exists with ID {topic.id}")
+    except IggyError as error:
+        logger.error(
+            f"Error creating topic: [{error.code}] {error.name}: {error.message}"
+        )
     except Exception as error:
         logger.error(f"Error creating topic {error}")
         logger.exception(error)
@@ -190,6 +198,11 @@ async def produce_messages(client: IggyClient):
                 f"Successfully sent batch of {messages_per_batch} messages. "
                 f"Batch ID: {current_id // messages_per_batch}"
             )
+        except IggyError as error:
+            logger.error(
+                "Iggy error while sending messages: "
+                f"[{error.code}] {error.name}: {error.message}"
+            )
         except Exception as error:
             logger.error(f"Exception type: {type(error).__name__}, message: {error}")
             logger.exception(error)
diff --git a/foreign/python/README.md b/foreign/python/README.md
index f01754f6dc..2e72c82bf7 100644
--- a/foreign/python/README.md
+++ b/foreign/python/README.md
@@ -48,7 +48,8 @@ cargo run --bin iggy-server -- --with-default-root-credentials --fresh
 # Using uv:
 uv sync --all-extras
 uv run maturin develop
-uv run pytest tests/ -v # Run tests (requires iggy-server running)
+uv run pytest tests/ -v # Run all tests (for release versions, requires iggy-server running)
+uv run --no-sync pytest tests/ -v # Run tests without syncing (for development, implicit uv run's syncing overwrites the installations made by maturin develop)
 
 # Using pip:
 python3 -m venv .venv
diff --git a/foreign/python/apache_iggy.pyi b/foreign/python/apache_iggy.pyi
index 5b53669384..0f0d1e207a 100644
--- a/foreign/python/apache_iggy.pyi
+++ b/foreign/python/apache_iggy.pyi
@@ -33,6 +33,7 @@ __all__ = [
     "ConsumerGroupMember",
     "IggyClient",
     "IggyConsumer",
+    "IggyError",
     "PollingStrategy",
     "ReceiveMessage",
     "SendMessage",
@@ -709,6 +710,24 @@ class IggyConsumer:
         Returns an awaitable that completes when shutdown is signaled or a PyRuntimeError on failure.
         """
 
+@typing.final
+class IggyError(builtins.Exception):
+    r"""
+    A Python class representing the Rust's IggyError.
+    Allows transparent representation of Iggy-specific errors.
+    """
+    @property
+    def code(self) -> builtins.int: ...
+    @property
+    def name(self) -> builtins.str: ...
+    @property
+    def message(self) -> builtins.str: ...
+    def __new__(
+        cls, code: builtins.int, name: builtins.str, message: builtins.str
+    ) -> IggyError: ...
+    def __str__(self) -> builtins.str: ...
+    def __repr__(self) -> builtins.str: ...
+
 class PollingStrategy:
     @typing.final
     class Offset(PollingStrategy):
diff --git a/foreign/python/src/client.rs b/foreign/python/src/client.rs
index 2a36ef3335..50fc2e244f 100644
--- a/foreign/python/src/client.rs
+++ b/foreign/python/src/client.rs
@@ -32,6 +32,7 @@ use crate::consumer::{
     AutoCommit, ConsumerGroup as PyConsumerGroup, ConsumerGroupDetails as PyConsumerGroupDetails,
     IggyConsumer, py_delta_to_iggy_duration,
 };
+use crate::error::IggyError as PyIggyError;
 use crate::identifier::PyIdentifier;
 use crate::receive_message::{PollingStrategy, ReceiveMessage};
 use crate::send_message::SendMessage;
@@ -63,7 +64,7 @@ impl IggyClient {
             .with_tcp()
             .with_server_address(conn.unwrap_or("127.0.0.1:8090".to_string()))
             .build()
-            .map_err(|e| PyErr::new::<pyo3::exceptions::PyRuntimeError, _>(e.to_string()))?;
+            .map_err(PyIggyError::new_err_from_rust)?;
         Ok(IggyClient {
             inner: Arc::new(client),
         })
@@ -80,7 +81,7 @@ impl IggyClient {
         connection_string: String,
     ) -> PyResult<Self> {
         let client = RustIggyClient::from_connection_string(&connection_string)
-            .map_err(|e| PyErr::new::<pyo3::exceptions::PyRuntimeError, _>(e.to_string()))?;
+            .map_err(PyIggyError::new_err_from_rust)?;
         Ok(Self {
             inner: Arc::new(client),
         })
@@ -93,10 +94,7 @@ impl IggyClient {
     fn ping<'a>(&self, py: Python<'a>) -> PyResult<Bound<'a, PyAny>> {
         let inner = self.inner.clone();
         future_into_py(py, async move {
-            inner
-                .ping()
-                .await
-                .map_err(|e| PyErr::new::<pyo3::exceptions::PyRuntimeError, _>(e.to_string()))
+            inner.ping().await.map_err(PyIggyError::new_err_from_rust)
         })
     }
 
@@ -114,7 +112,7 @@ impl IggyClient {
             inner
                 .login_user(&username, &password)
                 .await
-                .map_err(|e| PyErr::new::<pyo3::exceptions::PyRuntimeError, _>(e.to_string()))?;
+                .map_err(PyIggyError::new_err_from_rust)?;
             Ok(())
         })
     }
@@ -128,7 +126,7 @@ impl IggyClient {
             inner
                 .connect()
                 .await
-                .map_err(|e| PyErr::new::<pyo3::exceptions::PyRuntimeError, _>(e.to_string()))?;
+                .map_err(PyIggyError::new_err_from_rust)?;
             Ok(())
         })
     }
@@ -143,7 +141,7 @@ impl IggyClient {
             inner
                 .create_stream(&name)
                 .await
-                .map_err(|e| PyErr::new::<pyo3::exceptions::PyRuntimeError, _>(e.to_string()))?;
+                .map_err(PyIggyError::new_err_from_rust)?;
             Ok(())
         })
     }
@@ -163,7 +161,7 @@ impl IggyClient {
             let stream = inner
                 .get_stream(&stream_id)
                 .await
-                .map_err(|e| PyErr::new::<pyo3::exceptions::PyRuntimeError, _>(e.to_string()))?;
+                .map_err(PyIggyError::new_err_from_rust)?;
             Ok(stream.map(StreamDetails::from))
         })
     }
@@ -219,7 +217,7 @@ impl IggyClient {
                     max_size,
                 )
                 .await
-                .map_err(|e| PyErr::new::<pyo3::exceptions::PyRuntimeError, _>(e.to_string()))?;
+                .map_err(PyIggyError::new_err_from_rust)?;
             Ok(())
         })
     }
@@ -241,7 +239,7 @@ impl IggyClient {
             let topic = inner
                 .get_topic(&stream_id, &topic_id)
                 .await
-                .map_err(|e| PyErr::new::<pyo3::exceptions::PyRuntimeError, _>(e.to_string()))?;
+                .map_err(PyIggyError::new_err_from_rust)?;
             Ok(topic.map(TopicDetails::from))
         })
     }
@@ -269,7 +267,7 @@ impl IggyClient {
             let topics = inner
                 .get_topics(&stream_id)
                 .await
-                .map_err(|e| PyErr::new::<pyo3::exceptions::PyRuntimeError, _>(e.to_string()))?;
+                .map_err(PyIggyError::new_err_from_rust)?;
             Ok(topics.into_iter().map(Topic::from).collect::<Vec<_>>())
         })
     }
@@ -343,7 +341,7 @@ impl IggyClient {
                     max_size,
                 )
                 .await
-                .map_err(|e| PyErr::new::<pyo3::exceptions::PyRuntimeError, _>(e.to_string()))?;
+                .map_err(PyIggyError::new_err_from_rust)?;
             Ok(())
         })
     }
@@ -374,7 +372,7 @@ impl IggyClient {
             inner
                 .delete_topic(&stream_id, &topic_id)
                 .await
-                .map_err(|e| PyErr::new::<pyo3::exceptions::PyRuntimeError, _>(e.to_string()))?;
+                .map_err(PyIggyError::new_err_from_rust)?;
             Ok(())
         })
     }
@@ -405,7 +403,7 @@ impl IggyClient {
             inner
                 .purge_topic(&stream_id, &topic_id)
                 .await
-                .map_err(|e| PyErr::new::<pyo3::exceptions::PyRuntimeError, _>(e.to_string()))?;
+                .map_err(PyIggyError::new_err_from_rust)?;
             Ok(())
         })
     }
@@ -439,7 +437,7 @@ impl IggyClient {
             inner
                 .create_consumer_group(&stream_id, &topic_id, &name)
                 .await
-                .map_err(|e| PyErr::new::<pyo3::exceptions::PyRuntimeError, _>(e.to_string()))?;
+                .map_err(PyIggyError::new_err_from_rust)?;
             Ok(())
         })
     }
@@ -475,7 +473,7 @@ impl IggyClient {
             let group = inner
                 .get_consumer_group(&stream_id, &topic_id, &group_id)
                 .await
-                .map_err(|e| PyErr::new::<pyo3::exceptions::PyRuntimeError, _>(e.to_string()))?;
+                .map_err(PyIggyError::new_err_from_rust)?;
             Ok(group.map(PyConsumerGroupDetails::from))
         })
     }
@@ -507,7 +505,7 @@ impl IggyClient {
             let groups = inner
                 .get_consumer_groups(&stream_id, &topic_id)
                 .await
-                .map_err(|e| PyErr::new::<pyo3::exceptions::PyRuntimeError, _>(e.to_string()))?;
+                .map_err(PyIggyError::new_err_from_rust)?;
             Ok(groups
                 .into_iter()
                 .map(PyConsumerGroup::from)
@@ -545,7 +543,7 @@ impl IggyClient {
             inner
                 .delete_consumer_group(&stream_id, &topic_id, &group_id)
                 .await
-                .map_err(|e| PyErr::new::<pyo3::exceptions::PyRuntimeError, _>(e.to_string()))?;
+                .map_err(PyIggyError::new_err_from_rust)?;
             Ok(())
         })
     }
@@ -583,7 +581,7 @@ impl IggyClient {
             inner
                 .join_consumer_group(&stream_id, &topic_id, &group_id)
                 .await
-                .map_err(|e| PyErr::new::<pyo3::exceptions::PyRuntimeError, _>(e.to_string()))?;
+                .map_err(PyIggyError::new_err_from_rust)?;
             Ok(())
         })
     }
@@ -623,7 +621,7 @@ impl IggyClient {
             inner
                 .leave_consumer_group(&stream_id, &topic_id, &group_id)
                 .await
-                .map_err(|e| PyErr::new::<pyo3::exceptions::PyRuntimeError, _>(e.to_string()))?;
+                .map_err(PyIggyError::new_err_from_rust)?;
             Ok(())
         })
     }
@@ -660,7 +658,7 @@ impl IggyClient {
             inner
                 .send_messages(&stream, &topic, &partitioning, messages.as_mut())
                 .await
-                .map_err(|e| PyErr::new::<pyo3::exceptions::PyRuntimeError, _>(e.to_string()))?;
+                .map_err(PyIggyError::new_err_from_rust)?;
             Ok(())
         })
     }
@@ -698,7 +696,7 @@ impl IggyClient {
                     auto_commit,
                 )
                 .await
-                .map_err(|e| PyErr::new::<pyo3::exceptions::PyRuntimeError, _>(e.to_string()))?;
+                .map_err(PyIggyError::new_err_from_rust)?;
             let messages = polled_messages
                 .messages
                 .into_iter()
@@ -759,7 +757,7 @@ impl IggyClient {
         let mut builder = self
             .inner
             .consumer_group(name, stream, topic)
-            .map_err(|e| PyErr::new::<pyo3::exceptions::PyRuntimeError, _>(e.to_string()))?
+            .map_err(PyIggyError::new_err_from_rust)?
             .without_encryptor()
             .partition(partition_id);
 
@@ -817,7 +815,7 @@ impl IggyClient {
             consumer
                 .init()
                 .await
-                .map_err(|e| PyErr::new::<pyo3::exceptions::PyRuntimeError, _>(e.to_string()))?;
+                .map_err(PyIggyError::new_err_from_rust)?;
             Ok(IggyConsumer {
                 inner: Arc::new(Mutex::new(consumer)),
             })
diff --git a/foreign/python/src/consumer.rs b/foreign/python/src/consumer.rs
index 0b6066b686..e493c4e56e 100644
--- a/foreign/python/src/consumer.rs
+++ b/foreign/python/src/consumer.rs
@@ -25,7 +25,7 @@ use iggy::prelude::{
     AutoCommitWhen as RustAutoCommitWhen, ConsumerGroup as RustConsumerGroup,
     ConsumerGroupDetails as RustConsumerGroupDetails,
     ConsumerGroupMember as RustConsumerGroupMember, IggyConsumer as RustIggyConsumer, IggyDuration,
-    IggyError, ReceivedMessage,
+    IggyError as RustIggyError, ReceivedMessage,
 };
 use pyo3::exceptions::PyStopAsyncIteration;
 use pyo3::types::{PyDelta, PyDeltaAccess};
@@ -39,6 +39,7 @@ use tokio::sync::Mutex;
 use tokio::sync::oneshot::Sender;
 use tokio::task::JoinHandle;
 
+use crate::error::IggyError as PyIggyError;
 use crate::identifier::PyIdentifier;
 use crate::receive_message::ReceiveMessage;
 
@@ -110,7 +111,7 @@ impl IggyConsumer {
                 .await
                 .store_offset(offset, partition_id)
                 .await
-                .map_err(|e| PyErr::new::<pyo3::exceptions::PyRuntimeError, _>(e.to_string()))
+                .map_err(PyIggyError::new_err_from_rust)
         })
     }
 
@@ -131,7 +132,7 @@ impl IggyConsumer {
                 .await
                 .delete_offset(partition_id)
                 .await
-                .map_err(|e| PyErr::new::<pyo3::exceptions::PyRuntimeError, _>(e.to_string()))
+                .map_err(PyIggyError::new_err_from_rust)
         })
     }
 
@@ -167,8 +168,8 @@ impl IggyConsumer {
             let (shutdown_tx, shutdown_rx) = tokio::sync::oneshot::channel::<()>();
 
             let task_locals = Python::attach(pyo3_async_runtimes::tokio::get_current_locals)?;
-            let handle_consume: JoinHandle<PyResult<Result<(), IggyError>>> =
-                get_runtime().spawn(scope(task_locals, async move {
+            let handle_consume: JoinHandle<PyResult<Result<(), RustIggyError>>> = get_runtime()
+                .spawn(scope(task_locals, async move {
                     let task_locals =
                         Python::attach(pyo3_async_runtimes::tokio::get_current_locals)?;
                     let consumer = PyCallbackConsumer {
@@ -212,7 +213,7 @@ impl IggyConsumer {
 
             consume_result
                 .map_err(|e| PyErr::new::<pyo3::exceptions::PyRuntimeError, _>(e.to_string()))??
-                .map_err(|e| PyErr::new::<pyo3::exceptions::PyRuntimeError, _>(e.to_string()))?;
+                .map_err(PyIggyError::new_err_from_rust)?;
             Ok(())
         })
     }
@@ -365,9 +366,7 @@ impl ReceiveMessageIterator {
                         inner: m.message,
                         partition_id: m.partition_id,
                     })
-                    .map_err(|e| {
-                        PyErr::new::<pyo3::exceptions::PyRuntimeError, _>(e.to_string())
-                    })?)
+                    .map_err(PyIggyError::new_err_from_rust)?)
             } else {
                 Err(PyStopAsyncIteration::new_err("No more messages"))
             }
@@ -385,7 +384,7 @@ struct PyCallbackConsumer {
 }
 
 impl MessageConsumer for PyCallbackConsumer {
-    async fn consume(&self, received: ReceivedMessage) -> Result<(), IggyError> {
+    async fn consume(&self, received: ReceivedMessage) -> Result<(), RustIggyError> {
         let callback = self.callback.clone();
         let task_locals = self.task_locals.clone().lock_owned().await;
         let task_locals = task_locals.clone();
@@ -402,10 +401,10 @@ impl MessageConsumer for PyCallbackConsumer {
                 })
             }))
             .await
-            .map_err(|_| IggyError::CannotReadMessage)?
-            .map_err(|_| IggyError::CannotReadMessage)?
+            .map_err(|_| RustIggyError::CannotReadMessage)?
+            .map_err(|_| RustIggyError::CannotReadMessage)?
             .await
-            .map_err(|_| IggyError::CannotReadMessage)?;
+            .map_err(|_| RustIggyError::CannotReadMessage)?;
         Ok(())
     }
 }
diff --git a/foreign/python/src/error.rs b/foreign/python/src/error.rs
new file mode 100644
index 0000000000..c0208913c4
--- /dev/null
+++ b/foreign/python/src/error.rs
@@ -0,0 +1,78 @@
+// 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.
+
+//! Python-facing error type bridging [`iggy::prelude::IggyError`] to a
+//! `PyException` subclass so callers can `except IggyError as e` and
+//! inspect `e.code`, `e.name`, and `e.message` directly.
+//!
+//! Error codes and names are defined in
+//! `core/common/src/error/iggy_error.rs`.
+
+use iggy::prelude::IggyError as RustIggyError;
+use pyo3::exceptions::PyException;
+use pyo3::prelude::*;
+use pyo3_stub_gen::derive::{gen_stub_pyclass, gen_stub_pymethods};
+
+/// Iggy-specific exception exposed to Python.
+#[gen_stub_pyclass]
+#[pyclass(skip_from_py_object, extends=PyException)]
+#[derive(Clone, Debug)]
+pub struct IggyError {
+    /// Numeric error code from [`iggy::prelude::IggyError::as_code`].
+    #[pyo3(get)]
+    code: u32,
+    /// Snake-case error variant name from [`iggy::prelude::IggyError::as_string`]
+    /// (e.g. `"topic_name_already_exists"`).
+    #[pyo3(get)]
+    name: String,
+    /// Human-readable description from the variant's `#[error(...)]` template
+    /// (e.g. `"Topic with name: foo for stream with ID: 1 already exists."`).
+    #[pyo3(get)]
+    message: String,
+}
+
+#[gen_stub_pymethods]
+#[pymethods]
+impl IggyError {
+    #[new]
+    fn new(code: u32, name: String, message: String) -> Self {
+        Self {
+            code,
+            name,
+            message,
+        }
+    }
+
+    fn __str__(&self) -> String {
+        self.message.clone()
+    }
+
+    fn __repr__(&self) -> String {
+        format!(
+            "IggyError(code={}, name='{}', message='{}')",
+            self.code, self.name, self.message
+        )
+    }
+}
+
+impl IggyError {
+    /// Convert a Rust [`RustIggyError`] into a [`PyErr`] carrying this type.
+    /// Use this at every SDK boundary instead of mapping to a generic `PyRuntimeError`.
+    pub fn new_err_from_rust(e: RustIggyError) -> PyErr {
+        PyErr::new::<IggyError, _>((e.as_code(), e.as_string().to_string(), e.to_string()))
+    }
+}
diff --git a/foreign/python/src/identifier.rs b/foreign/python/src/identifier.rs
index a113e5e542..4b1d6d9192 100644
--- a/foreign/python/src/identifier.rs
+++ b/foreign/python/src/identifier.rs
@@ -18,12 +18,11 @@
 use std::str::FromStr;
 
 use iggy::prelude::{IdKind, Identifier};
-use pyo3::{
-    exceptions::{PyRuntimeError, PyValueError},
-    prelude::*,
-};
+use pyo3::prelude::*;
 use pyo3_stub_gen::impl_stub_type;
 
+use crate::error::IggyError as PyIggyError;
+
 #[derive(FromPyObject, IntoPyObject)]
 pub(crate) enum PyIdentifier {
     #[pyo3(transparent, annotation = "str")]
@@ -39,11 +38,9 @@ impl TryFrom<PyIdentifier> for Identifier {
     fn try_from(py_identifier: PyIdentifier) -> Result<Self, Self::Error> {
         match py_identifier {
             PyIdentifier::String(s) => {
-                Identifier::from_str(&s).map_err(|e| PyErr::new::<PyValueError, _>(e.to_string()))
-            }
-            PyIdentifier::Int(i) => {
-                Identifier::numeric(i).map_err(|e| PyErr::new::<PyValueError, _>(e.to_string()))
+                Identifier::from_str(&s).map_err(PyIggyError::new_err_from_rust)
             }
+            PyIdentifier::Int(i) => Identifier::numeric(i).map_err(PyIggyError::new_err_from_rust),
         }
     }
 }
@@ -56,11 +53,11 @@ impl TryFrom<&Identifier> for PyIdentifier {
             IdKind::String => val
                 .get_string_value()
                 .map(PyIdentifier::String)
-                .map_err(|e| PyErr::new::<PyRuntimeError, _>(e.to_string())),
+                .map_err(PyIggyError::new_err_from_rust),
             IdKind::Numeric => val
                 .get_u32_value()
                 .map(PyIdentifier::Int)
-                .map_err(|e| PyErr::new::<PyRuntimeError, _>(e.to_string())),
+                .map_err(PyIggyError::new_err_from_rust),
         }
     }
 }
diff --git a/foreign/python/src/lib.rs b/foreign/python/src/lib.rs
index 8c78456062..fea4e3d2a2 100644
--- a/foreign/python/src/lib.rs
+++ b/foreign/python/src/lib.rs
@@ -17,6 +17,7 @@
 
 pub mod client;
 mod consumer;
+mod error;
 mod identifier;
 mod receive_message;
 mod send_message;
@@ -28,6 +29,7 @@ use consumer::{
     AutoCommit, AutoCommitAfter, AutoCommitWhen, ConsumerGroup, ConsumerGroupDetails,
     ConsumerGroupMember, IggyConsumer, ReceiveMessageIterator,
 };
+use error::IggyError;
 use pyo3::prelude::*;
 use receive_message::{PollingStrategy, ReceiveMessage};
 use send_message::SendMessage;
@@ -40,6 +42,7 @@ fn apache_iggy(_py: Python, m: &Bound<'_, PyModule>) -> PyResult<()> {
     m.add_class::<SendMessage>()?;
     m.add_class::<ReceiveMessage>()?;
     m.add_class::<IggyClient>()?;
+    m.add_class::<IggyError>()?;
     m.add_class::<StreamDetails>()?;
     m.add_class::<Topic>()?;
     m.add_class::<TopicDetails>()?;
diff --git a/foreign/python/src/send_message.rs b/foreign/python/src/send_message.rs
index 021125d7e6..fc24da37d2 100644
--- a/foreign/python/src/send_message.rs
+++ b/foreign/python/src/send_message.rs
@@ -24,6 +24,8 @@ use pyo3_stub_gen::{
 };
 use std::str::FromStr;
 
+use crate::error::IggyError as PyIggyError;
+
 /// A Python class representing a message to be sent.
 /// This class wraps a Rust message meant for sending, facilitating
 /// the creation of such messages from Python and their subsequent use in Rust.
@@ -63,14 +65,15 @@ impl SendMessage {
     #[new]
     pub fn new(py: Python, data: PyMessagePayload) -> PyResult<Self> {
         let inner = match data {
-            PyMessagePayload::String(data) => RustIggyMessage::from_str(&data)
-                .map_err(|e| PyErr::new::<pyo3::exceptions::PyValueError, _>(e.to_string()))?,
+            PyMessagePayload::String(data) => {
+                RustIggyMessage::from_str(&data).map_err(PyIggyError::new_err_from_rust)?
+            }
             PyMessagePayload::Bytes(data) => {
                 let bytes = Bytes::from(data.extract::<Vec<u8>>(py)?);
                 RustIggyMessage::builder()
                     .payload(bytes)
                     .build()
-                    .map_err(|e| PyErr::new::<pyo3::exceptions::PyValueError, _>(e.to_string()))?
+                    .map_err(PyIggyError::new_err_from_rust)?
             }
         };
         Ok(Self { inner })
diff --git a/foreign/python/tests/test_connectivity.py b/foreign/python/tests/test_connectivity.py
index 69516d74c1..c0c1ab50bb 100644
--- a/foreign/python/tests/test_connectivity.py
+++ b/foreign/python/tests/test_connectivity.py
@@ -15,9 +15,10 @@
 # specific language governing permissions and limitations
 # under the License.
 
+
 import pytest
 
-from apache_iggy import IggyClient
+from apache_iggy import IggyClient, IggyError
 
 from .utils import get_server_config, wait_for_ping, wait_for_server
 
@@ -51,46 +52,98 @@ async def test_valid_connection_string(self, connection_string: str):
         await wait_for_ping(client, timeout=5, interval=1)
 
     @pytest.mark.parametrize(
-        ("invalid_value", "expected_error"),
+        (
+            "invalid_value",
+            "expected_iggy_error_name",
+            "expected_iggy_error_code",
+        ),
         [
-            ("", "Invalid connection string"),
-            ("bad address", "Invalid connection string"),
-            ("http://{host}:{port}", "Invalid connection string"),
-            ("tcp://iggy:iggy@{host}:{port}", "Invalid connection string"),
-            ("{host}:", "Invalid connection string"),
-            (":{port}", "Invalid connection string"),
-            ("{host}:not-a-port", "Invalid connection string"),
-            ("{host}:70000", "Invalid connection string"),
-            ("iggy+tcp://", "Invalid connection string"),
-            ("iggy+tcp://iggy:iggy@", "Invalid connection string"),
+            ("", "invalid_connection_string", 8000),
+            (
+                "bad address",
+                "invalid_connection_string",
+                8000,
+            ),
+            (
+                "http://{host}:{port}",
+                "invalid_connection_string",
+                8000,
+            ),
+            (
+                "tcp://iggy:iggy@{host}:{port}",
+                "invalid_connection_string",
+                8000,
+            ),
+            ("{host}:", "invalid_connection_string", 8000),
+            (":{port}", "invalid_connection_string", 8000),
+            (
+                "{host}:not-a-port",
+                "invalid_connection_string",
+                8000,
+            ),
+            (
+                "{host}:70000",
+                "invalid_connection_string",
+                8000,
+            ),
+            (
+                "iggy+tcp://",
+                "invalid_connection_string",
+                8000,
+            ),
+            (
+                "iggy+tcp://iggy:iggy@",
+                "invalid_connection_string",
+                8000,
+            ),
             (
                 "iggy+tcp://iggy:iggy@{host}:not-a-port",
-                "Invalid connection string",
+                "invalid_connection_string",
+                8000,
+            ),
+            (
+                "iggy+tcp://iggy:iggy@{host}:-1",
+                "invalid_connection_string",
+                8000,
+            ),
+            (
+                "iggy+tcp://iggy:iggy@{host}:70000",
+                "invalid_connection_string",
+                8000,
             ),
-            ("iggy+tcp://iggy:iggy@{host}:-1", "Invalid connection string"),
-            ("iggy+tcp://iggy:iggy@{host}:70000", "Invalid connection string"),
             (
                 "iggy+tcp://iggy:bad:format@{host}:{port}",
-                "Invalid connection string",
+                "invalid_connection_string",
+                8000,
             ),
             (
                 "iggy+tcp://iggy:iggy@{host}:{port}?invalid_option=value",
-                "Invalid connection string",
+                "invalid_connection_string",
+                8000,
+            ),
+            (
+                "iggy+quic://iggy:iggy@127.0.0.1:8080",
+                "cannot_create_endpoint",
+                305,
             ),
-            ("iggy+quic://iggy:iggy@127.0.0.1:8080", "Cannot create endpoint"),
         ],
     )
-    def test_invalid_connection_string(self, invalid_value: str, expected_error: str):
+    def test_invalid_connection_string(
+        self,
+        invalid_value: str,
+        expected_iggy_error_name: str,
+        expected_iggy_error_code: int,
+    ):
         """Test malformed server addresses and connection strings are rejected."""
         host, port = get_server_config()
         value = invalid_value.format(host=host, port=port)
 
-        with pytest.raises(RuntimeError):
-            IggyClient(value)
-
-        with pytest.raises(RuntimeError, match=expected_error):
+        with pytest.raises(IggyError) as excinfo_from_connection_string:
             IggyClient.from_connection_string(value)
 
+        assert excinfo_from_connection_string.value.name == expected_iggy_error_name
+        assert excinfo_from_connection_string.value.code == expected_iggy_error_code
+
     @pytest.mark.asyncio
     async def test_repeated_connect_does_not_error(self):
         """Test calling connect twice on the same client succeeds."""
@@ -112,19 +165,46 @@ async def test_login_before_connect_does_not_error(self):
         await wait_for_ping(client)
 
     @pytest.mark.parametrize(
-        ("username", "password", "expected_exception"),
+        (
+            "username",
+            "password",
+            "expected_exception",
+            "expected_iggy_error_name",
+            "expected_iggy_error_code",
+            "expected_error_message",
+        ),
         [
-            ("invalid-user", "iggy", RuntimeError),
-            ("iggy", "invalid-password", RuntimeError),
-            ("", "iggy", RuntimeError),
-            ("iggy", "", RuntimeError),
-            (None, "iggy", TypeError),
-            ("iggy", None, TypeError),
+            (
+                "invalid-user",
+                "iggy",
+                IggyError,
+                "invalid_credentials",
+                42,
+                "Invalid credentials",
+            ),
+            (
+                "iggy",
+                "invalid-password",
+                IggyError,
+                "invalid_credentials",
+                42,
+                "Invalid credentials",
+            ),
+            ("", "iggy", IggyError, "invalid_username", 43, None),
+            ("iggy", "", IggyError, "invalid_password", 44, None),
+            (None, "iggy", TypeError, None, None, "'None' is not an instance of 'str'"),
+            ("iggy", None, TypeError, None, None, "'None' is not an instance of 'str'"),
         ],
     )
     @pytest.mark.asyncio
     async def test_login_with_invalid_credentials_fails(
-        self, username, password, expected_exception
+        self,
+        username: str,
+        password: str,
+        expected_exception: type[Exception],
+        expected_iggy_error_name: str,
+        expected_iggy_error_code: int,
+        expected_error_message: str,
     ):
         """Test login rejects invalid credentials and invalid argument values."""
         host, port = get_server_config()
@@ -134,10 +214,16 @@ async def test_login_with_invalid_credentials_fails(
         await client.connect()
         await wait_for_ping(client)
 
-        with pytest.raises(expected_exception):
+        with pytest.raises(expected_exception) as exc_info:
             result = client.login_user(username, password)
             await result
 
+        if isinstance(exc_info.value, IggyError):
+            assert exc_info.value.name == expected_iggy_error_name
+            assert exc_info.value.code == expected_iggy_error_code
+        else:
+            assert exc_info.value.args[0] == expected_error_message
+
     @pytest.mark.asyncio
     async def test_relogin_with_invalid_credentials_fails(self):
         """Test re-login with invalid credentials fails after a successful login."""
@@ -149,9 +235,12 @@ async def test_relogin_with_invalid_credentials_fails(self):
         await wait_for_ping(client)
         await client.login_user("iggy", "iggy")
 
-        with pytest.raises(RuntimeError):
+        with pytest.raises(IggyError) as exc_info:
             await client.login_user("iggy", "invalid-password")
 
+        assert exc_info.value.name == "invalid_credentials"
+        assert exc_info.value.code == 42
+
     @pytest.mark.asyncio
     async def test_relogin_with_same_valid_credentials_does_not_error(self):
         """Test repeated login with the same valid credentials is allowed."""
diff --git a/foreign/python/tests/test_consumer_group.py b/foreign/python/tests/test_consumer_group.py
index acb1260485..8915525722 100644
--- a/foreign/python/tests/test_consumer_group.py
+++ b/foreign/python/tests/test_consumer_group.py
@@ -26,6 +26,7 @@
 from apache_iggy import (
     AutoCommit,
     IggyClient,
+    IggyError,
     PollingStrategy,
     ReceiveMessage,
 )
@@ -162,13 +163,16 @@ async def test_duplicate_consumer_group_creation_fails(
             group_name,
         )
 
-        with pytest.raises(RuntimeError):
+        with pytest.raises(IggyError) as exc_info:
             await iggy_client.create_consumer_group(
                 stream_name,
                 topic_name,
                 group_name,
             )
 
+        assert exc_info.value.name == "consumer_group_name_already_exists"
+        assert exc_info.value.code == 5004
+
     @pytest.mark.asyncio
     async def test_create_consumer_group_requires_connection_and_auth(
         self, unique_name
@@ -178,21 +182,27 @@ async def test_create_consumer_group_requires_connection_and_auth(
         wait_for_server(host, port)
 
         client = IggyClient(f"{host}:{port}")
-        with pytest.raises(RuntimeError):
+        with pytest.raises(IggyError) as exc_info_disconnected:
             await client.create_consumer_group(
                 unique_name(),
                 unique_name(),
                 unique_name(),
             )
 
+        assert exc_info_disconnected.value.name == "disconnected"
+        assert exc_info_disconnected.value.code == 8
+
         await client.connect()
-        with pytest.raises(RuntimeError):
+        with pytest.raises(IggyError) as exc_info_unauthenticated:
             await client.create_consumer_group(
                 unique_name(),
                 unique_name(),
                 unique_name(),
             )
 
+        assert exc_info_unauthenticated.value.name == "unauthenticated"
+        assert exc_info_unauthenticated.value.code == 40
+
 
 class TestGetConsumerGroup:
     """Test consumer group retrieval via get_consumer_group."""
@@ -318,21 +328,27 @@ async def test_get_consumer_group_requires_connection_and_auth(self, unique_name
         wait_for_server(host, port)
 
         client = IggyClient(f"{host}:{port}")
-        with pytest.raises(RuntimeError):
+        with pytest.raises(IggyError) as exc_info_disconnected:
             await client.get_consumer_group(
                 unique_name(),
                 unique_name(),
                 unique_name(),
             )
 
+        assert exc_info_disconnected.value.name == "disconnected"
+        assert exc_info_disconnected.value.code == 8
+
         await client.connect()
-        with pytest.raises(RuntimeError):
+        with pytest.raises(IggyError) as exc_info_unauthenticated:
             await client.get_consumer_group(
                 unique_name(),
                 unique_name(),
                 unique_name(),
             )
 
+        assert exc_info_unauthenticated.value.name == "unauthenticated"
+        assert exc_info_unauthenticated.value.code == 40
+
 
 class TestGetConsumerGroups:
     """Test listing consumer groups via get_consumer_groups."""
@@ -458,13 +474,19 @@ async def test_get_consumer_groups_requires_connection_and_auth(self, unique_nam
         wait_for_server(host, port)
 
         client = IggyClient(f"{host}:{port}")
-        with pytest.raises(RuntimeError):
+        with pytest.raises(IggyError) as exc_info_disconnected:
             await client.get_consumer_groups(unique_name(), unique_name())
 
+        assert exc_info_disconnected.value.name == "disconnected"
+        assert exc_info_disconnected.value.code == 8
+
         await client.connect()
-        with pytest.raises(RuntimeError):
+        with pytest.raises(IggyError) as exc_info_unauthenticated:
             await client.get_consumer_groups(unique_name(), unique_name())
 
+        assert exc_info_unauthenticated.value.name == "unauthenticated"
+        assert exc_info_unauthenticated.value.code == 40
+
 
 class TestDeleteConsumerGroup:
     """Test deleting consumer groups via delete_consumer_group."""
@@ -551,13 +573,16 @@ async def test_delete_nonexistent_consumer_group_fails(
             partitions_count=1,
         )
 
-        with pytest.raises(RuntimeError, match="Consumer group with ID:"):
+        with pytest.raises(IggyError) as exc_info:
             await iggy_client.delete_consumer_group(
                 stream_name,
                 topic_name,
                 unique_name(),
             )
 
+        assert exc_info.value.code == 5000
+        assert exc_info.value.name == "consumer_group_id_not_found"
+
     @pytest.mark.asyncio
     async def test_delete_consumer_group_removes_live_member(
         self, iggy_client: IggyClient, unique_name
@@ -674,13 +699,16 @@ async def test_join_nonexistent_consumer_group_fails(
             partitions_count=1,
         )
 
-        with pytest.raises(RuntimeError, match="Consumer group with ID:"):
+        with pytest.raises(IggyError) as exc_info:
             await iggy_client.join_consumer_group(
                 stream_name,
                 topic_name,
                 unique_name(),
             )
 
+        assert exc_info.value.code == 5000
+        assert exc_info.value.name == "consumer_group_id_not_found"
+
 
 class TestLeaveConsumerGroup:
     """Test leaving consumer groups via leave_consumer_group."""
@@ -764,11 +792,13 @@ async def test_leave_consumer_group_twice_fails_second_time(
 
         await iggy_client.leave_consumer_group(stream_name, topic_name, group_name)
         with pytest.raises(
-            RuntimeError,
-            match="Consumer group member with client ID:",
-        ):
+            IggyError,
+        ) as exc_info:
             await iggy_client.leave_consumer_group(stream_name, topic_name, group_name)
 
+        assert exc_info.value.code == 5006
+        assert exc_info.value.name == "consumer_group_member_not_found"
+
     @pytest.mark.asyncio
     async def test_leave_nonexistent_consumer_group_fails(
         self, iggy_client: IggyClient, unique_name
@@ -784,13 +814,16 @@ async def test_leave_nonexistent_consumer_group_fails(
             partitions_count=1,
         )
 
-        with pytest.raises(RuntimeError, match="Consumer group with ID:"):
+        with pytest.raises(IggyError) as exc_info:
             await iggy_client.leave_consumer_group(
                 stream_name,
                 topic_name,
                 unique_name(),
             )
 
+        assert exc_info.value.code == 5000
+        assert exc_info.value.name == "consumer_group_id_not_found"
+
 
 @pytest.mark.parametrize(
     "method_name",
@@ -812,13 +845,19 @@ async def test_consumer_group_lifecycle_requires_connection_and_auth(
     method = getattr(client, method_name)
     identifiers = (unique_name(), unique_name(), unique_name())
 
-    with pytest.raises(RuntimeError, match="Disconnected"):
+    with pytest.raises(IggyError) as exc_info_before_conn:
         await method(*identifiers)
 
+    assert exc_info_before_conn.value.code == 8
+    assert exc_info_before_conn.value.name == "disconnected"
+
     await client.connect()
-    with pytest.raises(RuntimeError, match="Unauthenticated"):
+    with pytest.raises(IggyError) as exc_info_after_conn:
         await method(*identifiers)
 
+    assert exc_info_after_conn.value.code == 40
+    assert exc_info_after_conn.value.name == "unauthenticated"
+
 
 class TestConsumerGroup:
     """Test consumer group consumers."""
@@ -1036,11 +1075,11 @@ async def test_delete_offset_behavior(self, iggy_client: IggyClient, unique_name
             poll_interval=timedelta(milliseconds=25),
         )
 
-        with pytest.raises(
-            RuntimeError,
-            match=r"Consumer offset for consumer with ID: 0 was not found\.",
-        ):
+        with pytest.raises(IggyError) as exc_info:
             await consumer.delete_offset(partition_id)
+
+        assert exc_info.value.name == "consumer_offset_not_found"
+        assert exc_info.value.code == 3021
         assert consumer.get_last_stored_offset(partition_id) is None
 
         await iggy_client.send_messages(
@@ -1931,10 +1970,10 @@ async def test_consumer_group_can_disable_auto_create_when_group_exists(
 
     @pytest.mark.asyncio
     @pytest.mark.parametrize(
-        ("create_stream", "expected_error"),
+        ("create_stream", "expected_iggy_error_name", "expected_iggy_error_code"),
         [
-            (False, "Stream with name:"),
-            (True, "Topic with name:"),
+            (False, "stream_name_not_found", 1010),
+            (True, "topic_name_not_found", 2011),
         ],
     )
     async def test_consumer_group_missing_stream_or_topic_fails(
@@ -1942,7 +1981,8 @@ async def test_consumer_group_missing_stream_or_topic_fails(
         iggy_client: IggyClient,
         unique_name,
         create_stream: bool,
-        expected_error: str,
+        expected_iggy_error_name: str,
+        expected_iggy_error_code: int,
     ):
         """Test initialization fails when the target stream or topic is missing."""
         consumer_name = unique_name()
@@ -1952,7 +1992,7 @@ async def test_consumer_group_missing_stream_or_topic_fails(
         if create_stream:
             await iggy_client.create_stream(stream_name)
 
-        with pytest.raises(RuntimeError, match=expected_error):
+        with pytest.raises(IggyError) as exc_info:
             await iggy_client.consumer_group(
                 consumer_name,
                 stream_name,
@@ -1964,6 +2004,9 @@ async def test_consumer_group_missing_stream_or_topic_fails(
                 poll_interval=timedelta(milliseconds=25),
             )
 
+        assert exc_info.value.name == expected_iggy_error_name
+        assert exc_info.value.code == expected_iggy_error_code
+
     @pytest.mark.asyncio
     async def test_consumer_group_auto_create_disabled_fails_for_missing_group(
         self, iggy_client: IggyClient, unique_name
@@ -1979,7 +2022,7 @@ async def test_consumer_group_auto_create_disabled_fails_for_missing_group(
             partitions_count=1,
         )
 
-        with pytest.raises(RuntimeError, match="Consumer group with name:"):
+        with pytest.raises(IggyError) as exc_info:
             await iggy_client.consumer_group(
                 unique_name(),
                 stream_name,
@@ -1992,6 +2035,9 @@ async def test_consumer_group_auto_create_disabled_fails_for_missing_group(
                 poll_interval=timedelta(milliseconds=25),
             )
 
+        assert exc_info.value.name == "consumer_group_name_not_found"
+        assert exc_info.value.code == 5003
+
     @pytest.mark.asyncio
     @pytest.mark.parametrize(
         ("kwargs", "expected_error"),
@@ -2039,7 +2085,7 @@ async def test_consumer_group_before_connect_fails(self, unique_name):
         host, port = get_server_config()
         client = IggyClient(f"{host}:{port}")
 
-        with pytest.raises(RuntimeError):
+        with pytest.raises(IggyError) as exc_info:
             await client.consumer_group(
                 unique_name(),
                 unique_name(),
@@ -2051,6 +2097,9 @@ async def test_consumer_group_before_connect_fails(self, unique_name):
                 poll_interval=timedelta(milliseconds=25),
             )
 
+        assert exc_info.value.name == "disconnected"
+        assert exc_info.value.code == 8
+
     @pytest.mark.asyncio
     async def test_consumer_group_before_login_fails(self, unique_name):
         """Test consumer_group requires authentication."""
@@ -2061,7 +2110,7 @@ async def test_consumer_group_before_login_fails(self, unique_name):
         await client.connect()
         await wait_for_ping(client)
 
-        with pytest.raises(RuntimeError):
+        with pytest.raises(IggyError) as exc_info:
             await client.consumer_group(
                 unique_name(),
                 unique_name(),
@@ -2073,6 +2122,9 @@ async def test_consumer_group_before_login_fails(self, unique_name):
                 poll_interval=timedelta(milliseconds=25),
             )
 
+        assert exc_info.value.name == "unauthenticated"
+        assert exc_info.value.code == 40
+
     @pytest.mark.asyncio
     async def test_shutdown(self, iggy_client: IggyClient, unique_name):
         """Test that the consumer group can be signaled to shutdown."""
diff --git a/foreign/python/tests/test_message_operations.py b/foreign/python/tests/test_message_operations.py
index da57637c05..f9b36bfa20 100644
--- a/foreign/python/tests/test_message_operations.py
+++ b/foreign/python/tests/test_message_operations.py
@@ -20,7 +20,7 @@
 
 import pytest
 
-from apache_iggy import IggyClient, PollingStrategy
+from apache_iggy import IggyClient, IggyError, PollingStrategy
 from apache_iggy import SendMessage as Message
 
 
@@ -146,9 +146,12 @@ async def test_message_properties(self, iggy_client: IggyClient, unique_name):
     )
     async def test_empty_payload_is_rejected(self, payload):
         """Test empty string and bytes payloads are rejected."""
-        with pytest.raises(ValueError, match="Invalid message payload length"):
+        with pytest.raises(IggyError) as exc_info:
             Message(payload)
 
+        assert exc_info.value.name == "invalid_message_payload_length"
+        assert exc_info.value.code == 4025
+
     @pytest.mark.asyncio
     @pytest.mark.parametrize(
         "payload",
@@ -361,7 +364,7 @@ async def test_poll_messages_with_count_zero_is_rejected(
             messages=[Message(message) for message in test_messages],
         )
 
-        with pytest.raises(RuntimeError, match="Invalid messages count"):
+        with pytest.raises(IggyError) as exc_info:
             await iggy_client.poll_messages(
                 stream=stream_name,
                 topic=topic_name,
@@ -371,6 +374,9 @@ async def test_poll_messages_with_count_zero_is_rejected(
                 auto_commit=False,
             )
 
+        assert exc_info.value.name == "invalid_messages_count"
+        assert exc_info.value.code == 4009
+
     @pytest.mark.asyncio
     async def test_poll_messages_with_invalid_partition_id_raises(
         self, iggy_client: IggyClient, unique_name
@@ -392,13 +398,7 @@ async def test_poll_messages_with_invalid_partition_id_raises(
             messages=[Message(f"Partition test - {unique_name()}")],
         )
 
-        with pytest.raises(
-            RuntimeError,
-            match=(
-                r"Partition with ID: 0 for topic with ID: 0 "
-                r"for stream with ID: 0 was not found\."
-            ),
-        ):
+        with pytest.raises(IggyError) as exc_info:
             await iggy_client.poll_messages(
                 stream=stream_name,
                 topic=topic_name,
@@ -408,6 +408,9 @@ async def test_poll_messages_with_invalid_partition_id_raises(
                 auto_commit=False,
             )
 
+        assert exc_info.value.name == "partition_not_found"
+        assert exc_info.value.code == 3007
+
     @pytest.mark.asyncio
     async def test_polling_strategy_last_with_count_one_returns_last_message(
         self, iggy_client: IggyClient, unique_name
diff --git a/foreign/python/tests/test_stream.py b/foreign/python/tests/test_stream.py
index 5e71a543a4..a5b0db8fcc 100644
--- a/foreign/python/tests/test_stream.py
+++ b/foreign/python/tests/test_stream.py
@@ -17,7 +17,7 @@
 
 import pytest
 
-from apache_iggy import IggyClient
+from apache_iggy import IggyClient, IggyError
 
 from .utils import get_server_config, wait_for_ping, wait_for_server
 
@@ -91,9 +91,12 @@ async def test_create_stream_invalid_names(
         """Test create_stream enforces byte-length validation."""
         stream_name = unique_name(prefix, min_bytes=min_bytes, max_bytes=max_bytes)
 
-        with pytest.raises(RuntimeError):
+        with pytest.raises(IggyError) as exc_info:
             await iggy_client.create_stream(stream_name)
 
+        assert exc_info.value.name == "invalid_format"
+        assert exc_info.value.code == 4
+
     @pytest.mark.asyncio
     async def test_get_stream_by_name_and_id(
         self, iggy_client: IggyClient, unique_name
@@ -146,10 +149,11 @@ async def test_duplicate_stream_creation(
 
         await iggy_client.create_stream(stream_name)
 
-        with pytest.raises(RuntimeError) as exc_info:
+        with pytest.raises(IggyError) as exc_info:
             await iggy_client.create_stream(stream_name)
 
-        assert "already exists" in str(exc_info.value)
+        assert exc_info.value.name == "stream_name_already_exists"
+        assert exc_info.value.code == 1012
 
     @pytest.mark.asyncio
     async def test_get_nonexistent_stream(self, iggy_client: IggyClient, unique_name):
@@ -168,9 +172,12 @@ async def test_get_stream_before_connect_fails(self, unique_name):
         host, port = get_server_config()
         client = IggyClient(f"{host}:{port}")
 
-        with pytest.raises(RuntimeError):
+        with pytest.raises(IggyError) as exc_info:
             await client.get_stream(unique_name())
 
+        assert exc_info.value.name == "disconnected"
+        assert exc_info.value.code == 8
+
     @pytest.mark.asyncio
     async def test_get_stream_before_login_fails(self, unique_name):
         """Test get_stream requires authentication."""
@@ -180,18 +187,24 @@ async def test_get_stream_before_login_fails(self, unique_name):
         client = IggyClient(f"{host}:{port}")
         await client.connect()
 
-        with pytest.raises(RuntimeError):
+        with pytest.raises(IggyError) as exc_info:
             await client.get_stream(unique_name())
 
+        assert exc_info.value.name == "unauthenticated"
+        assert exc_info.value.code == 40
+
     @pytest.mark.asyncio
     async def test_create_stream_before_connect_fails(self, unique_name):
         """Test create_stream requires an established connection."""
         host, port = get_server_config()
         client = IggyClient(f"{host}:{port}")
 
-        with pytest.raises(RuntimeError):
+        with pytest.raises(IggyError) as exc_info:
             await client.create_stream(unique_name())
 
+        assert exc_info.value.name == "disconnected"
+        assert exc_info.value.code == 8
+
     @pytest.mark.asyncio
     async def test_create_stream_before_login_fails(self, unique_name):
         """Test create_stream requires authentication."""
@@ -201,5 +214,8 @@ async def test_create_stream_before_login_fails(self, unique_name):
         client = IggyClient(f"{host}:{port}")
         await client.connect()
 
-        with pytest.raises(RuntimeError):
+        with pytest.raises(IggyError) as exc_info:
             await client.create_stream(unique_name())
+
+        assert exc_info.value.name == "unauthenticated"
+        assert exc_info.value.code == 40
diff --git a/foreign/python/tests/test_topic.py b/foreign/python/tests/test_topic.py
index 62bb37a5b9..8d22312580 100644
--- a/foreign/python/tests/test_topic.py
+++ b/foreign/python/tests/test_topic.py
@@ -19,7 +19,7 @@
 
 import pytest
 
-from apache_iggy import IggyClient, SendMessage
+from apache_iggy import IggyClient, IggyError, SendMessage
 
 from .utils import get_server_config, wait_for_ping, wait_for_server
 
@@ -104,11 +104,14 @@ async def test_create_topic_invalid_names(
 
         await iggy_client.create_stream(stream_name)
 
-        with pytest.raises(RuntimeError):
+        with pytest.raises(IggyError) as exc_info:
             await iggy_client.create_topic(
                 stream=stream_name, name=topic_name, partitions_count=1
             )
 
+        assert exc_info.value.name == "invalid_format"
+        assert exc_info.value.code == 4
+
     @pytest.mark.asyncio
     async def test_create_and_get_topic_with_numeric_stream_id(
         self, iggy_client: IggyClient, unique_name
@@ -146,12 +149,13 @@ async def test_duplicate_topic_creation(self, iggy_client: IggyClient, unique_na
             stream=stream_name, name=topic_name, partitions_count=1
         )
 
-        with pytest.raises(RuntimeError) as exc_info:
+        with pytest.raises(IggyError) as exc_info:
             await iggy_client.create_topic(
                 stream=stream_name, name=topic_name, partitions_count=1
             )
 
-        assert "already exists" in str(exc_info.value)
+        assert exc_info.value.name == "topic_name_already_exists"
+        assert exc_info.value.code == 2013
 
     @pytest.mark.asyncio
     async def test_topic_names_can_repeat_across_different_streams(
@@ -306,7 +310,7 @@ async def test_create_topic_with_valid_max_topic_size(
     @pytest.mark.parametrize(
         ("max_topic_size", "expected_exception"),
         [
-            (4563, RuntimeError),
+            (4563, IggyError),
             (-1, OverflowError),
             (2e64, TypeError),
         ],
@@ -324,7 +328,7 @@ async def test_create_topic_invalid_max_topic_size(
 
         await iggy_client.create_stream(stream_name)
 
-        with pytest.raises(expected_exception):
+        with pytest.raises(expected_exception) as exc_info:
             await iggy_client.create_topic(
                 stream=stream_name,
                 name=topic_name,
@@ -332,6 +336,10 @@ async def test_create_topic_invalid_max_topic_size(
                 max_topic_size=max_topic_size,
             )
 
+        if expected_exception is IggyError:
+            assert exc_info.value.name == "invalid_topic_size"
+            assert exc_info.value.code == 1019
+
     @pytest.mark.asyncio
     @pytest.mark.parametrize(
         "replication_factor",
@@ -403,12 +411,13 @@ async def test_create_topic_invalid_partitions_count(
 
         await iggy_client.create_stream(stream_name)
 
-        with pytest.raises(RuntimeError) as exc_info:
+        with pytest.raises(IggyError) as exc_info:
             await iggy_client.create_topic(
                 stream=stream_name, name=topic_name, partitions_count=partitions_count
             )
 
-        assert "Too many partitions" in str(exc_info.value)
+        assert exc_info.value.name == "too_many_partitions"
+        assert exc_info.value.code == 2015
 
     @pytest.mark.asyncio
     @pytest.mark.parametrize("partitions_count", [0, 1, 1000])
@@ -489,11 +498,14 @@ async def test_create_topic_in_nonexistent_stream(
         nonexistent_stream = unique_name()
         topic_name = unique_name()
 
-        with pytest.raises(RuntimeError):
+        with pytest.raises(IggyError) as exc_info:
             await iggy_client.create_topic(
                 stream=nonexistent_stream, name=topic_name, partitions_count=1
             )
 
+        assert exc_info.value.name == "stream_id_not_found"
+        assert exc_info.value.code == 1009
+
     @pytest.mark.asyncio
     async def test_create_topic_requires_connection_and_auth(self, unique_name):
         """Test create_topic fails both before connecting and before logging in."""
@@ -501,17 +513,23 @@ async def test_create_topic_requires_connection_and_auth(self, unique_name):
         wait_for_server(host, port)
 
         client = IggyClient(f"{host}:{port}")
-        with pytest.raises(RuntimeError):
+        with pytest.raises(IggyError) as exc_info_disconnected:
             await client.create_topic(
                 stream=unique_name(), name=unique_name(), partitions_count=1
             )
 
+        assert exc_info_disconnected.value.name == "disconnected"
+        assert exc_info_disconnected.value.code == 8
+
         await client.connect()
-        with pytest.raises(RuntimeError):
+        with pytest.raises(IggyError) as exc_info_unauthenticated:
             await client.create_topic(
                 stream=unique_name(), name=unique_name(), partitions_count=1
             )
 
+        assert exc_info_unauthenticated.value.name == "unauthenticated"
+        assert exc_info_unauthenticated.value.code == 40
+
 
 class TestGetTopic:
     """Test topic retrieval via get_topic."""
@@ -572,13 +590,19 @@ async def test_get_topic_requires_connection_and_auth(self, unique_name):
         wait_for_server(host, port)
 
         client = IggyClient(f"{host}:{port}")
-        with pytest.raises(RuntimeError):
+        with pytest.raises(IggyError) as exc_info_disconnected:
             await client.get_topic(unique_name(), unique_name())
 
+        assert exc_info_disconnected.value.name == "disconnected"
+        assert exc_info_disconnected.value.code == 8
+
         await client.connect()
-        with pytest.raises(RuntimeError):
+        with pytest.raises(IggyError) as exc_info_unauthenticated:
             await client.get_topic(unique_name(), unique_name())
 
+        assert exc_info_unauthenticated.value.name == "unauthenticated"
+        assert exc_info_unauthenticated.value.code == 40
+
 
 class TestGetTopics:
     """Test listing topics in a stream via get_topics."""
@@ -679,13 +703,19 @@ async def test_get_topics_requires_connection_and_auth(self, unique_name):
         wait_for_server(host, port)
 
         client = IggyClient(f"{host}:{port}")
-        with pytest.raises(RuntimeError):
+        with pytest.raises(IggyError) as exc_info_disconnected:
             await client.get_topics(unique_name())
 
+        assert exc_info_disconnected.value.name == "disconnected"
+        assert exc_info_disconnected.value.code == 8
+
         await client.connect()
-        with pytest.raises(RuntimeError):
+        with pytest.raises(IggyError) as exc_info_unauthenticated:
             await client.get_topics(unique_name())
 
+        assert exc_info_unauthenticated.value.name == "unauthenticated"
+        assert exc_info_unauthenticated.value.code == 40
+
 
 class TestUpdateTopic:
     """Test updating topics via update_topic."""
@@ -1018,21 +1048,27 @@ async def test_update_nonexistent_topic_fails(
 
         await iggy_client.create_stream(stream_name)
 
-        with pytest.raises(RuntimeError):
+        with pytest.raises(IggyError) as exc_info:
             await iggy_client.update_topic(
                 stream_id=stream_name, topic_id=unique_name(), name=unique_name()
             )
 
+        assert exc_info.value.code == 2010
+        assert exc_info.value.name == "topic_id_not_found"
+
     @pytest.mark.asyncio
     async def test_update_topic_in_nonexistent_stream_fails(
         self, iggy_client: IggyClient, unique_name
     ):
         """Test update_topic raises when the stream does not exist."""
-        with pytest.raises(RuntimeError):
+        with pytest.raises(IggyError) as exc_info:
             await iggy_client.update_topic(
                 stream_id=unique_name(), topic_id=unique_name(), name=unique_name()
             )
 
+        assert exc_info.value.code == 1009
+        assert exc_info.value.name == "stream_id_not_found"
+
     @pytest.mark.asyncio
     async def test_update_topic_to_existing_name_fails(
         self, iggy_client: IggyClient, unique_name
@@ -1050,11 +1086,14 @@ async def test_update_topic_to_existing_name_fails(
             stream=stream_name, name=second_topic, partitions_count=1
         )
 
-        with pytest.raises(RuntimeError):
+        with pytest.raises(IggyError) as exc_info:
             await iggy_client.update_topic(
                 stream_id=stream_name, topic_id=second_topic, name=first_topic
             )
 
+        assert exc_info.value.code == 2013
+        assert exc_info.value.name == "topic_name_already_exists"
+
     @pytest.mark.asyncio
     async def test_update_topic_requires_connection_and_auth(self, unique_name):
         """Test update_topic fails both before connecting and before logging in."""
@@ -1062,17 +1101,23 @@ async def test_update_topic_requires_connection_and_auth(self, unique_name):
         wait_for_server(host, port)
 
         client = IggyClient(f"{host}:{port}")
-        with pytest.raises(RuntimeError):
+        with pytest.raises(IggyError) as exc_info_before_conn:
             await client.update_topic(
                 stream_id=unique_name(), topic_id=unique_name(), name=unique_name()
             )
 
+        assert exc_info_before_conn.value.code == 8
+        assert exc_info_before_conn.value.name == "disconnected"
+
         await client.connect()
-        with pytest.raises(RuntimeError):
+        with pytest.raises(IggyError) as exc_info_after_conn:
             await client.update_topic(
                 stream_id=unique_name(), topic_id=unique_name(), name=unique_name()
             )
 
+        assert exc_info_after_conn.value.code == 40
+        assert exc_info_after_conn.value.name == "unauthenticated"
+
 
 class TestDeleteTopic:
     """Test deleting topics via delete_topic."""
@@ -1150,9 +1195,12 @@ async def test_delete_nonexistent_topic_fails(
 
         await iggy_client.create_stream(stream_name)
 
-        with pytest.raises(RuntimeError):
+        with pytest.raises(IggyError) as exc_info:
             await iggy_client.delete_topic(stream_name, unique_name())
 
+        assert exc_info.value.name == "topic_id_not_found"
+        assert exc_info.value.code == 2010
+
     @pytest.mark.asyncio
     async def test_delete_topic_twice_fails_second_time(
         self, iggy_client: IggyClient, unique_name
@@ -1167,9 +1215,12 @@ async def test_delete_topic_twice_fails_second_time(
         )
 
         await iggy_client.delete_topic(stream_name, topic_name)
-        with pytest.raises(RuntimeError):
+        with pytest.raises(IggyError) as exc_info:
             await iggy_client.delete_topic(stream_name, topic_name)
 
+        assert exc_info.value.name == "topic_id_not_found"
+        assert exc_info.value.code == 2010
+
     @pytest.mark.asyncio
     async def test_delete_topic_requires_connection_and_auth(self, unique_name):
         """Test delete_topic fails both before connecting and before logging in."""
@@ -1177,13 +1228,19 @@ async def test_delete_topic_requires_connection_and_auth(self, unique_name):
         wait_for_server(host, port)
 
         client = IggyClient(f"{host}:{port}")
-        with pytest.raises(RuntimeError):
+        with pytest.raises(IggyError) as exc_info_disconnected:
             await client.delete_topic(unique_name(), unique_name())
 
+        assert exc_info_disconnected.value.name == "disconnected"
+        assert exc_info_disconnected.value.code == 8
+
         await client.connect()
-        with pytest.raises(RuntimeError):
+        with pytest.raises(IggyError) as exc_info_unauthenticated:
             await client.delete_topic(unique_name(), unique_name())
 
+        assert exc_info_unauthenticated.value.name == "unauthenticated"
+        assert exc_info_unauthenticated.value.code == 40
+
 
 class TestPurgeTopic:
     """Test purging topic messages via purge_topic."""
@@ -1271,17 +1328,23 @@ async def test_purge_nonexistent_topic_fails(
 
         await iggy_client.create_stream(stream_name)
 
-        with pytest.raises(RuntimeError):
+        with pytest.raises(IggyError) as exc_info:
             await iggy_client.purge_topic(stream_name, unique_name())
 
+        assert exc_info.value.name == "topic_id_not_found"
+        assert exc_info.value.code == 2010
+
     @pytest.mark.asyncio
     async def test_purge_topic_in_nonexistent_stream_fails(
         self, iggy_client: IggyClient, unique_name
     ):
         """Test purge_topic raises when the stream does not exist."""
-        with pytest.raises(RuntimeError):
+        with pytest.raises(IggyError) as exc_info:
             await iggy_client.purge_topic(unique_name(), unique_name())
 
+        assert exc_info.value.name == "stream_id_not_found"
+        assert exc_info.value.code == 1009
+
     @pytest.mark.asyncio
     async def test_purge_topic_requires_connection_and_auth(self, unique_name):
         """Test purge_topic fails both before connecting and before logging in."""
@@ -1289,9 +1352,15 @@ async def test_purge_topic_requires_connection_and_auth(self, unique_name):
         wait_for_server(host, port)
 
         client = IggyClient(f"{host}:{port}")
-        with pytest.raises(RuntimeError):
+        with pytest.raises(IggyError) as exc_info_disconnected:
             await client.purge_topic(unique_name(), unique_name())
 
+        assert exc_info_disconnected.value.name == "disconnected"
+        assert exc_info_disconnected.value.code == 8
+
         await client.connect()
-        with pytest.raises(RuntimeError):
+        with pytest.raises(IggyError) as exc_info_unauthenticated:
             await client.purge_topic(unique_name(), unique_name())
+
+        assert exc_info_unauthenticated.value.name == "unauthenticated"
+        assert exc_info_unauthenticated.value.code == 40