diff --git a/pinecone/db_data/index.py b/pinecone/db_data/index.py
index 29e19b699..20feab7ff 100644
--- a/pinecone/db_data/index.py
+++ b/pinecone/db_data/index.py
@@ -698,6 +698,13 @@ def cancel_import(self, id: str):
         """
         return self.bulk_import.cancel(id=id)
 
+    @validate_and_convert_errors
+    @require_kwargs
+    def create_namespace(
+        self, name: str, schema: Optional[Dict[str, Any]] = None, **kwargs
+    ) -> "NamespaceDescription":
+        return self.namespace.create(name=name, schema=schema, **kwargs)
+
     @validate_and_convert_errors
     @require_kwargs
     def describe_namespace(self, namespace: str, **kwargs) -> "NamespaceDescription":
diff --git a/pinecone/db_data/index_asyncio.py b/pinecone/db_data/index_asyncio.py
index a46573e10..b1818d7c4 100644
--- a/pinecone/db_data/index_asyncio.py
+++ b/pinecone/db_data/index_asyncio.py
@@ -752,6 +752,13 @@ async def cancel_import(self, id: str):
         """
         return await self.bulk_import.cancel(id=id)
 
+    @validate_and_convert_errors
+    @require_kwargs
+    async def create_namespace(
+        self, name: str, schema: Optional[Dict[str, Any]] = None, **kwargs
+    ) -> "NamespaceDescription":
+        return await self.namespace.create(name=name, schema=schema, **kwargs)
+
     @validate_and_convert_errors
     @require_kwargs
     async def describe_namespace(self, namespace: str, **kwargs) -> "NamespaceDescription":
diff --git a/pinecone/db_data/index_asyncio_interface.py b/pinecone/db_data/index_asyncio_interface.py
index 889ce215a..3f3838ecb 100644
--- a/pinecone/db_data/index_asyncio_interface.py
+++ b/pinecone/db_data/index_asyncio_interface.py
@@ -869,6 +869,49 @@ async def search_records(
         """Alias of the search() method."""
         pass
 
+    @abstractmethod
+    @require_kwargs
+    async def create_namespace(
+        self, name: str, schema: Optional[Dict[str, Any]] = None, **kwargs
+    ) -> NamespaceDescription:
+        """Create a namespace in a serverless index.
+
+        Args:
+            name (str): The name of the namespace to create
+            schema (Optional[Dict[str, Any]]): Optional schema configuration for the namespace as a dictionary. [optional]
+
+        Returns:
+            NamespaceDescription: Information about the created namespace including vector count
+
+        Create a namespace in a serverless index. For guidance and examples, see
+        `Manage namespaces <https://docs.pinecone.io/guides/manage-data/manage-namespaces>`_.
+
+        **Note:** This operation is not supported for pod-based indexes.
+
+        Examples:
+
+            .. code-block:: python
+
+                >>> # Create a namespace with just a name
+                >>> import asyncio
+                >>> from pinecone import Pinecone
+                >>>
+                >>> async def main():
+                ...     pc = Pinecone()
+                ...     async with pc.IndexAsyncio(host="example-index-dojoi3u.svc.eu-west1-gcp.pinecone.io") as idx:
+                ...         namespace = await idx.create_namespace(name="my-namespace")
+                ...         print(f"Created namespace: {namespace.name}, Vector count: {namespace.vector_count}")
+                >>>
+                >>> asyncio.run(main())
+
+                >>> # Create a namespace with schema configuration
+                >>> from pinecone.core.openapi.db_data.model.create_namespace_request_schema import CreateNamespaceRequestSchema
+                >>> schema = CreateNamespaceRequestSchema(fields={...})
+                >>> namespace = await idx.create_namespace(name="my-namespace", schema=schema)
+
+        """
+        pass
+
     @abstractmethod
     @require_kwargs
     async def describe_namespace(self, namespace: str, **kwargs) -> NamespaceDescription:
diff --git a/pinecone/db_data/interfaces.py b/pinecone/db_data/interfaces.py
index f486a77bb..3b1e3be68 100644
--- a/pinecone/db_data/interfaces.py
+++ b/pinecone/db_data/interfaces.py
@@ -843,6 +843,40 @@ def list(self, **kwargs):
         """
         pass
 
+    @abstractmethod
+    @require_kwargs
+    def create_namespace(
+        self, name: str, schema: Optional[Dict[str, Any]] = None, **kwargs
+    ) -> NamespaceDescription:
+        """Create a namespace in a serverless index.
+
+        Args:
+            name (str): The name of the namespace to create
+            schema (Optional[Dict[str, Any]]): Optional schema configuration for the namespace as a dictionary. [optional]
+
+        Returns:
+            NamespaceDescription: Information about the created namespace including vector count
+
+        Create a namespace in a serverless index. For guidance and examples, see
+        `Manage namespaces <https://docs.pinecone.io/guides/manage-data/manage-namespaces>`_.
+
+        **Note:** This operation is not supported for pod-based indexes.
+
+        Examples:
+
+            .. code-block:: python
+
+                >>> # Create a namespace with just a name
+                >>> namespace = index.create_namespace(name="my-namespace")
+                >>> print(f"Created namespace: {namespace.name}, Vector count: {namespace.vector_count}")
+
+                >>> # Create a namespace with schema configuration
+                >>> from pinecone.core.openapi.db_data.model.create_namespace_request_schema import CreateNamespaceRequestSchema
+                >>> schema = CreateNamespaceRequestSchema(fields={...})
+                >>> namespace = index.create_namespace(name="my-namespace", schema=schema)
+        """
+        pass
+
     @abstractmethod
     @require_kwargs
     def describe_namespace(self, namespace: str, **kwargs) -> NamespaceDescription:
diff --git a/pinecone/db_data/resources/asyncio/namespace_asyncio.py b/pinecone/db_data/resources/asyncio/namespace_asyncio.py
index f59b0cc25..13180fd77 100644
--- a/pinecone/db_data/resources/asyncio/namespace_asyncio.py
+++ b/pinecone/db_data/resources/asyncio/namespace_asyncio.py
@@ -1,4 +1,4 @@
-from typing import Optional, AsyncIterator
+from typing import Optional, AsyncIterator, Any
 
 from pinecone.core.openapi.db_data.api.namespace_operations_api import AsyncioNamespaceOperationsApi
 from pinecone.core.openapi.db_data.models import ListNamespacesResponse, NamespaceDescription
@@ -15,6 +15,26 @@ class NamespaceResourceAsyncio:
     def __init__(self, api_client) -> None:
         self.__namespace_operations_api = AsyncioNamespaceOperationsApi(api_client)
 
+    @require_kwargs
+    async def create(
+        self, name: str, schema: Optional[Any] = None, **kwargs
+    ) -> NamespaceDescription:
+        """
+        Args:
+            name (str): The name of the namespace to create
+            schema (Optional[Any]): Optional schema configuration for the namespace. Can be a dictionary or CreateNamespaceRequestSchema object. [optional]
+
+        Returns:
+            ``NamespaceDescription``: Information about the created namespace including vector count
+
+        Create a namespace in a serverless index. For guidance and examples, see
+        `Manage namespaces <https://docs.pinecone.io/guides/manage-data/manage-namespaces>`_.
+
+        **Note:** This operation is not supported for pod-based indexes.
+        """
+        args = NamespaceRequestFactory.create_namespace_args(name=name, schema=schema, **kwargs)
+        return await self.__namespace_operations_api.create_namespace(**args)
+
     @require_kwargs
     async def describe(self, namespace: str, **kwargs) -> NamespaceDescription:
         """
diff --git a/pinecone/db_data/resources/sync/namespace.py b/pinecone/db_data/resources/sync/namespace.py
index 5980ec71c..791034e0b 100644
--- a/pinecone/db_data/resources/sync/namespace.py
+++ b/pinecone/db_data/resources/sync/namespace.py
@@ -1,4 +1,4 @@
-from typing import Optional, Iterator
+from typing import Optional, Iterator, Any
 
 from pinecone.core.openapi.db_data.api.namespace_operations_api import NamespaceOperationsApi
 from pinecone.core.openapi.db_data.models import ListNamespacesResponse, NamespaceDescription
@@ -25,6 +25,24 @@ def __init__(self, api_client, config, openapi_config, pool_threads: int) -> Non
         self.__namespace_operations_api = NamespaceOperationsApi(api_client)
         super().__init__()
 
+    @require_kwargs
+    def create(self, name: str, schema: Optional[Any] = None, **kwargs) -> NamespaceDescription:
+        """
+        Args:
+            name (str): The name of the namespace to create
+            schema (Optional[Any]): Optional schema configuration for the namespace. Can be a dictionary or CreateNamespaceRequestSchema object. [optional]
+
+        Returns:
+            ``NamespaceDescription``: Information about the created namespace including vector count
+
+        Create a namespace in a serverless index. For guidance and examples, see
+        `Manage namespaces <https://docs.pinecone.io/guides/manage-data/manage-namespaces>`_.
+
+        **Note:** This operation is not supported for pod-based indexes.
+        """
+        args = NamespaceRequestFactory.create_namespace_args(name=name, schema=schema, **kwargs)
+        return self.__namespace_operations_api.create_namespace(**args)
+
     @require_kwargs
     def describe(self, namespace: str, **kwargs) -> NamespaceDescription:
         """
diff --git a/pinecone/db_data/resources/sync/namespace_request_factory.py b/pinecone/db_data/resources/sync/namespace_request_factory.py
index 30ae54981..468dd8a7a 100644
--- a/pinecone/db_data/resources/sync/namespace_request_factory.py
+++ b/pinecone/db_data/resources/sync/namespace_request_factory.py
@@ -1,6 +1,10 @@
-from typing import Optional, TypedDict, Any, cast
+from typing import Optional, TypedDict, Any, cast, Dict, Union
 
 from pinecone.utils import parse_non_empty_args
+from pinecone.core.openapi.db_data.model.create_namespace_request import CreateNamespaceRequest
+from pinecone.core.openapi.db_data.model.create_namespace_request_schema import (
+    CreateNamespaceRequestSchema,
+)
 
 
 class DescribeNamespaceArgs(TypedDict, total=False):
@@ -11,6 +15,10 @@ class DeleteNamespaceArgs(TypedDict, total=False):
     namespace: str
 
 
+class CreateNamespaceArgs(TypedDict, total=False):
+    create_namespace_request: CreateNamespaceRequest
+
+
 class NamespaceRequestFactory:
     @staticmethod
     def describe_namespace_args(namespace: str, **kwargs) -> DescribeNamespaceArgs:
@@ -26,6 +34,30 @@ def delete_namespace_args(namespace: str, **kwargs) -> DeleteNamespaceArgs:
         base_args = {"namespace": namespace}
         return cast(DeleteNamespaceArgs, {**base_args, **kwargs})
 
+    @staticmethod
+    def create_namespace_args(
+        name: str,
+        schema: Optional[Union[CreateNamespaceRequestSchema, Dict[str, Any]]] = None,
+        **kwargs,
+    ) -> CreateNamespaceArgs:
+        if not isinstance(name, str):
+            raise ValueError("name must be string")
+        if name.strip() == "":
+            raise ValueError("name must not be empty")
+
+        request_kwargs: Dict[str, Any] = {"name": name}
+        if schema is not None:
+            if isinstance(schema, dict):
+                schema_obj = CreateNamespaceRequestSchema(**schema)
+                request_kwargs["schema"] = schema_obj
+            else:
+                # schema is already CreateNamespaceRequestSchema
+                request_kwargs["schema"] = cast(CreateNamespaceRequestSchema, schema)
+
+        create_namespace_request = CreateNamespaceRequest(**request_kwargs)
+        base_args = {"create_namespace_request": create_namespace_request}
+        return cast(CreateNamespaceArgs, {**base_args, **kwargs})
+
     @staticmethod
     def list_namespaces_args(
         limit: Optional[int] = None, pagination_token: Optional[str] = None, **kwargs
diff --git a/pinecone/grpc/index_grpc.py b/pinecone/grpc/index_grpc.py
index adf6cc4e7..a3ac23d76 100644
--- a/pinecone/grpc/index_grpc.py
+++ b/pinecone/grpc/index_grpc.py
@@ -50,6 +50,9 @@
     DescribeNamespaceRequest,
     DeleteNamespaceRequest,
     ListNamespacesRequest,
+    CreateNamespaceRequest,
+    MetadataSchema,
+    MetadataFieldProperties,
 )
 from pinecone.core.grpc.protos.db_data_2025_10_pb2_grpc import VectorServiceStub
 from pinecone import Vector, SparseValues
@@ -769,6 +772,65 @@ def describe_index_stats(
         json_response = json_format.MessageToDict(response)
         return parse_stats_response(json_response)
 
+    @require_kwargs
+    def create_namespace(
+        self, name: str, schema: Optional[Dict[str, Any]] = None, async_req: bool = False, **kwargs
+    ) -> Union[NamespaceDescription, PineconeGrpcFuture]:
+        """
+        The create_namespace operation creates a namespace in a serverless index.
+
+        Examples:
+
+        .. code-block:: python
+
+            >>> index.create_namespace(name='my_namespace')
+
+            >>> # Create namespace asynchronously
+            >>> future = index.create_namespace(name='my_namespace', async_req=True)
+            >>> namespace = future.result()
+
+        Args:
+            name (str): The name of the namespace to create.
+            schema (Optional[Dict[str, Any]]): Optional schema configuration for the namespace as a dictionary. [optional]
+            async_req (bool): If True, the create_namespace operation will be performed asynchronously. [optional]
+
+        Returns: NamespaceDescription object which contains information about the created namespace, or a PineconeGrpcFuture object if async_req is True.
+        """
+        timeout = kwargs.pop("timeout", None)
+
+        # Build MetadataSchema from dict if provided
+        metadata_schema = None
+        if schema is not None:
+            if isinstance(schema, dict):
+                # Convert dict to MetadataSchema
+                fields = {}
+                for key, value in schema.get("fields", {}).items():
+                    if isinstance(value, dict):
+                        filterable = value.get("filterable", False)
+                        fields[key] = MetadataFieldProperties(filterable=filterable)
+                    else:
+                        # If value is already a MetadataFieldProperties, use it directly
+                        fields[key] = value
+                metadata_schema = MetadataSchema(fields=fields)
+            else:
+                # Assume it's already a MetadataSchema
+                metadata_schema = schema
+
+        request_kwargs: Dict[str, Any] = {"name": name}
+        if metadata_schema is not None:
+            request_kwargs["schema"] = metadata_schema
+
+        request = CreateNamespaceRequest(**request_kwargs)
+
+        if async_req:
+            future = self.runner.run(self.stub.CreateNamespace.future, request, timeout=timeout)
+            return PineconeGrpcFuture(
+                future, timeout=timeout, result_transformer=parse_namespace_description
+            )
+
+        response = self.runner.run(self.stub.CreateNamespace, request, timeout=timeout)
+        return parse_namespace_description(response)
+
     @require_kwargs
     def describe_namespace(self, namespace: str, **kwargs) -> NamespaceDescription:
         """
diff --git a/tests/integration/data/test_namespace.py b/tests/integration/data/test_namespace.py
index 2bf9d6353..8065550c2 100644
--- a/tests/integration/data/test_namespace.py
+++ b/tests/integration/data/test_namespace.py
@@ -43,6 +43,77 @@ def delete_all_namespaces(index):
 
 
 class TestNamespaceOperations:
+    def test_create_namespace(self, idx):
+        """Test creating a namespace"""
+        test_namespace = "test_create_namespace_sync"
+
+        try:
+            # Ensure namespace doesn't exist first
+            if verify_namespace_exists(idx, test_namespace):
+                idx.delete_namespace(namespace=test_namespace)
+                time.sleep(10)
+
+            # Create namespace
+            description = idx.create_namespace(name=test_namespace)
+
+            # Verify namespace was created
+            assert isinstance(description, NamespaceDescription)
+            assert description.name == test_namespace
+            # New namespace should have 0 records (record_count may be None, 0, or "0" as string)
+            assert (
+                description.record_count is None
+                or description.record_count == 0
+                or description.record_count == "0"
+            )
+
+            # Verify namespace exists by describing it
+            # Namespace may not be immediately available after creation, so retry with backoff
+            max_retries = 5
+            retry_delay = 2
+            for attempt in range(max_retries):
+                try:
+                    verify_description = idx.describe_namespace(namespace=test_namespace)
+                    assert verify_description.name == test_namespace
+                    break
+                except Exception:
+                    if attempt == max_retries - 1:
+                        raise
+                    time.sleep(retry_delay)
+
+        finally:
+            # Cleanup
+            if verify_namespace_exists(idx, test_namespace):
+                idx.delete_namespace(namespace=test_namespace)
+                time.sleep(10)
+
+    def test_create_namespace_duplicate(self, idx):
+        """Test creating a duplicate namespace raises an error"""
+        test_namespace = "test_create_duplicate_sync"
+
+        try:
+            # Ensure namespace doesn't exist first
+            if verify_namespace_exists(idx, test_namespace):
+                idx.delete_namespace(namespace=test_namespace)
+                time.sleep(10)
+
+            # Create namespace first time
+            description = idx.create_namespace(name=test_namespace)
+            assert description.name == test_namespace
+
+            # Try to create duplicate namespace - should raise an error
+            # GRPC errors raise PineconeException, not PineconeApiException
+            import pytest
+            from pinecone.exceptions import PineconeException
+
+            with pytest.raises(PineconeException):
+                idx.create_namespace(name=test_namespace)
+
+        finally:
+            # Cleanup
+            if verify_namespace_exists(idx, test_namespace):
+                idx.delete_namespace(namespace=test_namespace)
+                time.sleep(10)
+
     def test_describe_namespace(self, idx):
         """Test describing a namespace"""
         # Setup test data
diff --git a/tests/integration/data_asyncio/test_namespace_asyncio.py b/tests/integration/data_asyncio/test_namespace_asyncio.py
index 01ad8ece8..0591f9893 100644
--- a/tests/integration/data_asyncio/test_namespace_asyncio.py
+++ b/tests/integration/data_asyncio/test_namespace_asyncio.py
@@ -45,6 +45,69 @@ async def delete_all_namespaces(index):
 
 
 class TestNamespaceOperationsAsyncio:
+    @pytest.mark.asyncio
+    async def test_create_namespace(self, index_host):
+        """Test creating a namespace"""
+        asyncio_idx = build_asyncioindex_client(index_host)
+        test_namespace = "test_create_namespace_async"
+
+        try:
+            # Ensure namespace doesn't exist first
+            if await verify_namespace_exists(asyncio_idx, test_namespace):
+                await asyncio_idx.delete_namespace(namespace=test_namespace)
+                await asyncio.sleep(10)
+
+            # Create namespace
+            description = await asyncio_idx.create_namespace(name=test_namespace)
+
+            # Verify namespace was created
+            assert isinstance(description, NamespaceDescription)
+            assert description.name == test_namespace
+            # New namespace should have 0 records (record_count may be None, 0, or "0" as string)
+            assert (
+                description.record_count is None
+                or description.record_count == 0
+                or description.record_count == "0"
+            )
+
+            # Verify namespace exists by describing it
+            verify_description = await asyncio_idx.describe_namespace(namespace=test_namespace)
+            assert verify_description.name == test_namespace
+
+        finally:
+            # Cleanup
+            if await verify_namespace_exists(asyncio_idx, test_namespace):
+                await asyncio_idx.delete_namespace(namespace=test_namespace)
+                await asyncio.sleep(10)
+
+    @pytest.mark.asyncio
+    async def test_create_namespace_duplicate(self, index_host):
+        """Test creating a duplicate namespace raises an error"""
+        asyncio_idx = build_asyncioindex_client(index_host)
+        test_namespace = "test_create_duplicate_async"
+
+        try:
+            # Ensure namespace doesn't exist first
+            if await verify_namespace_exists(asyncio_idx, test_namespace):
+                await asyncio_idx.delete_namespace(namespace=test_namespace)
+                await asyncio.sleep(10)
+
+            # Create namespace first time
+            description = await asyncio_idx.create_namespace(name=test_namespace)
+            assert description.name == test_namespace
+
+            # Try to create duplicate namespace - should raise an error
+            from pinecone.exceptions import PineconeApiException
+
+            with pytest.raises(PineconeApiException):
+                await asyncio_idx.create_namespace(name=test_namespace)
+
+        finally:
+            # Cleanup
+            if await verify_namespace_exists(asyncio_idx, test_namespace):
+                await asyncio_idx.delete_namespace(namespace=test_namespace)
+                await asyncio.sleep(10)
+
     @pytest.mark.asyncio
     async def test_describe_namespace(self, index_host):
         """Test describing a namespace"""
diff --git a/tests/integration/data_grpc_futures/test_namespace_future.py b/tests/integration/data_grpc_futures/test_namespace_future.py
new file mode 100644
index 000000000..c030c5b9e
--- /dev/null
+++ b/tests/integration/data_grpc_futures/test_namespace_future.py
@@ -0,0 +1,130 @@
+import pytest
+import time
+from pinecone import NamespaceDescription
+from ..helpers import generate_name
+
+
+def verify_namespace_exists(idx, namespace: str) -> bool:
+    """Helper function to verify if a namespace exists"""
+    try:
+        idx.describe_namespace(namespace=namespace)
+        return True
+    except Exception:
+        return False
+
+
+class TestCreateNamespaceFuture:
+    def test_create_namespace_future(self, idx):
+        """Test creating a namespace with async_req=True"""
+        test_namespace = generate_name("TestCreateNamespaceFuture", "test-create-namespace-future")
+
+        try:
+            # Ensure namespace doesn't exist first
+            if verify_namespace_exists(idx, test_namespace):
+                idx.delete_namespace(namespace=test_namespace)
+                time.sleep(10)
+
+            # Create namespace asynchronously
+            future = idx.create_namespace(name=test_namespace, async_req=True)
+
+            # Verify it's a future
+            from pinecone.grpc import PineconeGrpcFuture
+
+            assert isinstance(future, PineconeGrpcFuture)
+
+            # Get the result
+            description = future.result(timeout=30)
+
+            # Verify namespace was created
+            assert isinstance(description, NamespaceDescription)
+            assert description.name == test_namespace
+            # New namespace should have 0 records (record_count may be None, 0, or "0" as string)
+            assert (
+                description.record_count is None
+                or description.record_count == 0
+                or description.record_count == "0"
+            )
+
+            # Verify namespace exists by describing it
+            verify_description = idx.describe_namespace(namespace=test_namespace)
+            assert verify_description.name == test_namespace
+
+        finally:
+            # Cleanup
+            if verify_namespace_exists(idx, test_namespace):
+                idx.delete_namespace(namespace=test_namespace)
+                time.sleep(10)
+
+    def test_create_namespace_future_duplicate(self, idx):
+        """Test creating a duplicate namespace raises an error with async_req=True"""
+        test_namespace = generate_name(
+            "TestCreateNamespaceFutureDuplicate", "test-create-duplicate-future"
+        )
+
+        try:
+            # Ensure namespace doesn't exist first
+            if verify_namespace_exists(idx, test_namespace):
+                idx.delete_namespace(namespace=test_namespace)
+                time.sleep(10)
+
+            # Create namespace first time
+            future1 = idx.create_namespace(name=test_namespace, async_req=True)
+            description1 = future1.result(timeout=30)
+            assert description1.name == test_namespace
+
+            # Try to create duplicate namespace - should raise an error
+            future2 = idx.create_namespace(name=test_namespace, async_req=True)
+
+            # GRPC errors are wrapped in PineconeException, not PineconeApiException
+            from pinecone.exceptions import PineconeException
+
+            with pytest.raises(PineconeException):
+                future2.result(timeout=30)
+
+        finally:
+            # Cleanup
+            if verify_namespace_exists(idx, test_namespace):
+                idx.delete_namespace(namespace=test_namespace)
+                time.sleep(10)
+
+    def test_create_namespace_future_multiple(self, idx):
+        """Test creating multiple namespaces asynchronously"""
+        test_namespaces = [
+            generate_name("TestCreateNamespaceFutureMultiple", f"test-ns-{i}") for i in range(3)
+        ]
+
+        try:
+            # Clean up any existing namespaces
+            for ns in test_namespaces:
+                if verify_namespace_exists(idx, ns):
+                    idx.delete_namespace(namespace=ns)
+                    time.sleep(5)
+
+            # Create all namespaces asynchronously
+            futures = [idx.create_namespace(name=ns, async_req=True) for ns in test_namespaces]
+
+            # Wait for all to complete
+            from concurrent.futures import as_completed
+
+            results = []
+            for future in as_completed(futures, timeout=60):
+                description = future.result()
+                results.append(description)
+
+            # Verify all were created
+            assert len(results) == len(test_namespaces)
+            namespace_names = [desc.name for desc in results]
+            for test_ns in test_namespaces:
+                assert test_ns in namespace_names
+
+            # Verify each namespace exists
+            for ns in test_namespaces:
+                verify_description = idx.describe_namespace(namespace=ns)
+                assert verify_description.name == ns
+
+        finally:
+            # Cleanup
+            for ns in test_namespaces:
+                if verify_namespace_exists(idx, ns):
+                    idx.delete_namespace(namespace=ns)
+                    time.sleep(5)
diff --git a/tests/unit_grpc/test_grpc_index_namespace.py b/tests/unit_grpc/test_grpc_index_namespace.py
index e36a3b030..44739153e 100644
--- a/tests/unit_grpc/test_grpc_index_namespace.py
+++ b/tests/unit_grpc/test_grpc_index_namespace.py
@@ -1,9 +1,11 @@
 from pinecone import Config
 from pinecone.grpc import GRPCIndex
 from pinecone.core.grpc.protos.db_data_2025_10_pb2 import (
+    CreateNamespaceRequest,
     DescribeNamespaceRequest,
     DeleteNamespaceRequest,
     ListNamespacesRequest,
+    MetadataSchema,
 )
 
 
@@ -14,6 +16,39 @@ def setup_method(self):
             config=self.config, index_name="example-name", _endpoint_override="test-endpoint"
         )
 
+    def test_create_namespace(self, mocker):
+        mocker.patch.object(self.index.runner, "run", autospec=True)
+        self.index.create_namespace(name="test_namespace")
+        self.index.runner.run.assert_called_once_with(
+            self.index.stub.CreateNamespace,
+            CreateNamespaceRequest(name="test_namespace"),
+            timeout=None,
+        )
+
+    def test_create_namespace_with_timeout(self, mocker):
+        mocker.patch.object(self.index.runner, "run", autospec=True)
+        self.index.create_namespace(name="test_namespace", timeout=30)
+        self.index.runner.run.assert_called_once_with(
+            self.index.stub.CreateNamespace,
+            CreateNamespaceRequest(name="test_namespace"),
+            timeout=30,
+        )
+
+    def test_create_namespace_with_schema(self, mocker):
+        mocker.patch.object(self.index.runner, "run", autospec=True)
+        schema_dict = {"fields": {"field1": {"filterable": True}, "field2": {"filterable": False}}}
+        self.index.create_namespace(name="test_namespace", schema=schema_dict)
+        call_args = self.index.runner.run.call_args
+        assert call_args[0][0] == self.index.stub.CreateNamespace
+        request = call_args[0][1]
+        assert isinstance(request, CreateNamespaceRequest)
+        assert request.name == "test_namespace"
+        assert isinstance(request.schema, MetadataSchema)
+        assert "field1" in request.schema.fields
+        assert "field2" in request.schema.fields
+        assert request.schema.fields["field1"].filterable is True
+        assert request.schema.fields["field2"].filterable is False
+
     def test_describe_namespace(self, mocker):
         mocker.patch.object(self.index.runner, "run", autospec=True)
         self.index.describe_namespace(namespace="test_namespace")