Add create_namespace method to Index and IndexAsyncio

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":

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":

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:

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:

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:
         """

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:
         """

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

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:
         """

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