client.py

# This file was auto-generated by Fern from our API Definition.

from __future__ import annotations

import os
import typing

import httpx

from supermemory._types import NOT_GIVEN, NotGiven
from .core.api_error import ApiError
from .core.client_wrapper import AsyncClientWrapper, SyncClientWrapper
from .core.logging import LogConfig, Logger
from .core.request_options import RequestOptions
from .environment import SupermemoryEnvironment
from .raw_client import AsyncRawSupermemory, RawSupermemory
from .requests.filter_expression import FilterExpressionParams
from .requests.list_request_include import ListRequestIncludeParams
from .requests.metadata_value import MetadataValueParams
from .requests.search_include import SearchIncludeParams
from .types.document_ref import DocumentRef
from .types.dreaming_mode import DreamingMode
from .types.list_response import ListResponse
from .types.list_sort import ListSort
from .types.list_type import ListType
from .types.profile_response import ProfileResponse
from .types.search_mode import SearchMode
from .types.search_request_rerank import SearchRequestRerank
from .types.search_response import SearchResponse
from .types.sort_order import SortOrder
from .types.task_type import TaskType

if typing.TYPE_CHECKING:
    from .connectors.client import AsyncConnectorsClient, ConnectorsClient
    from .documents.client import AsyncDocumentsClient, DocumentsClient
    from .memories.client import AsyncMemoriesClient, MemoriesClient
    from .namespaces.client import AsyncNamespacesClient, NamespacesClient
    from .organization.client import AsyncOrganizationClient, OrganizationClient
    from .profiles.client import AsyncProfilesClient, ProfilesClient
# this is used as the default value for optional parameters
OMIT = typing.cast(typing.Any, ...)


class Supermemory:
    """
    Use this class to access the different functions within the SDK. You can instantiate any number of clients with different configuration that will propagate to these functions.

    Parameters
    ----------
    base_url : typing.Optional[str]
        The base url to use for requests from the client.

    environment : SupermemoryEnvironment
        The environment to use for requests from the client. from .environment import SupermemoryEnvironment



        Defaults to SupermemoryEnvironment.DEFAULT



    api_key : typing.Optional[typing.Union[str, typing.Callable[[], str]]]
    headers : typing.Optional[typing.Dict[str, str]]
        Additional headers to send with every request.

    timeout : typing.Optional[float]
        The timeout to be used, in seconds, for requests. By default the timeout is 60 seconds, unless a custom httpx client is used, in which case this default is not enforced.

    max_retries : typing.Optional[int]
        The default maximum number of retries for failed requests. Defaults to 2. Per-request `max_retries` in `request_options` takes precedence over this value.

    stream_reconnection_enabled : typing.Optional[bool]
        Whether to automatically reconnect on stream disconnection for resumable streaming endpoints. Defaults to True. Per-request `stream_reconnection_enabled` in `request_options` takes precedence over this value.

    max_stream_reconnection_attempts : typing.Optional[int]
        The maximum number of reconnection attempts for resumable streaming endpoints. Defaults to no limit. Per-request `max_stream_reconnection_attempts` in `request_options` takes precedence over this value.

    follow_redirects : typing.Optional[bool]
        Whether the default httpx client follows redirects or not, this is irrelevant if a custom httpx client is passed in.

    httpx_client : typing.Optional[httpx.Client]
        The httpx client to use for making requests, a preconfigured client is used by default, however this is useful should you want to pass in any custom httpx configuration.

    logging : typing.Optional[typing.Union[LogConfig, Logger]]
        Configure logging for the SDK. Accepts a LogConfig dict with 'level' (debug/info/warn/error), 'logger' (custom logger implementation), and 'silent' (boolean, defaults to True) fields. You can also pass a pre-configured Logger instance.

    Examples
    --------
    from supermemory import Supermemory

    client = Supermemory(
        api_key="YOUR_API_KEY",
    )
    """

    def __init__(
        self,
        *,
        base_url: typing.Optional[str] = None,
        environment: SupermemoryEnvironment = SupermemoryEnvironment.DEFAULT,
        api_key: typing.Optional[typing.Union[str, typing.Callable[[], str]]] = os.getenv("SUPERMEMORY_API_KEY"),
        headers: typing.Optional[typing.Dict[str, str]] = None,
        timeout: typing.Optional[float] = None,
        max_retries: typing.Optional[int] = None,
        stream_reconnection_enabled: typing.Optional[bool] = None,
        max_stream_reconnection_attempts: typing.Optional[int] = None,
        follow_redirects: typing.Optional[bool] = True,
        httpx_client: typing.Optional[httpx.Client] = None,
        logging: typing.Optional[typing.Union[LogConfig, Logger]] = None,
    ):
        _defaulted_timeout = timeout if timeout is not None else 60 if httpx_client is None else None
        _defaulted_max_retries = max_retries if max_retries is not None else 2
        if api_key is None:
            raise ApiError(
                body="The client must be instantiated be either passing in api_key or setting SUPERMEMORY_API_KEY"
            )
        self._client_wrapper = SyncClientWrapper(
            base_url=_get_base_url(base_url=base_url, environment=environment),
            api_key=api_key,
            headers=headers,
            httpx_client=httpx_client
            if httpx_client is not None
            else httpx.Client(timeout=_defaulted_timeout, follow_redirects=follow_redirects)
            if follow_redirects is not None
            else httpx.Client(timeout=_defaulted_timeout),
            timeout=_defaulted_timeout,
            max_retries=_defaulted_max_retries,
            stream_reconnection_enabled=stream_reconnection_enabled,
            max_stream_reconnection_attempts=max_stream_reconnection_attempts,
            logging=logging,
        )
        self._raw_client = RawSupermemory(client_wrapper=self._client_wrapper)
        self._documents: typing.Optional[DocumentsClient] = None
        self._profiles: typing.Optional[ProfilesClient] = None
        self._memories: typing.Optional[MemoriesClient] = None
        self._connectors: typing.Optional[ConnectorsClient] = None
        self._namespaces: typing.Optional[NamespacesClient] = None
        self._organization: typing.Optional[OrganizationClient] = None

    @property
    def with_raw_response(self) -> RawSupermemory:
        """
        Retrieves a raw implementation of this client that returns raw responses.

        Returns
        -------
        RawSupermemory
        """
        return self._raw_client

    def add(
        self,
        namespace: str,
        *,
        content: str,
        id: typing.Optional[str] = OMIT,
        supporting_context: typing.Optional[str] = OMIT,
        metadata: typing.Optional[typing.Dict[str, MetadataValueParams]] = OMIT,
        group: typing.Optional[typing.Dict[str, MetadataValueParams]] = OMIT,
        date: typing.Optional[str] = OMIT,
        task_type: typing.Optional[TaskType] = OMIT,
        dreaming: typing.Optional[DreamingMode] = OMIT,
        request_options: typing.Optional[RequestOptions] = None,
        extra_headers: typing.Optional[typing.Mapping[str, str]] = None,
        extra_query: typing.Optional[typing.Mapping[str, typing.Any]] = None,
        extra_body: typing.Optional[typing.Mapping[str, typing.Any]] = None,
        timeout: typing.Union[float, httpx.Timeout, None, NotGiven] = NOT_GIVEN,
    ) -> DocumentRef:
        """
        Turn text or a supported URL into searchable, evolving memory. Supply a new ID to create a document, or reuse an existing ID to append new information while preserving its history.

        Parameters
        ----------
        namespace : str
            The isolated namespace. This can be an ID for your user, a project ID, or any other identifier you wish to use to scope documents.

        content : str
            The content to process. This may be plaintext or a URL to supported rich content.

        id : typing.Optional[str]
            An optional caller-defined document ID

        supporting_context : typing.Optional[str]
            Context used to guide memory extraction within this namespace. Max 1500 characters.

        metadata : typing.Optional[typing.Dict[str, MetadataValueParams]]
            Arbitrary metadata attached to the document and derived data

        group : typing.Optional[typing.Dict[str, MetadataValueParams]]
            Metadata values that isolate related context within the namespace

        date : typing.Optional[str]
            When the source content is from, in ISO 8601 format

        task_type : typing.Optional[TaskType]
            Processing pipeline. "memory" builds durable learned context; "superrag" optimizes the document for retrieval without generating memories.

        dreaming : typing.Optional[DreamingMode]
            Processing mode. "dynamic" (default) groups related documents so memories form from coherent context. "instant" processes each document independently right away and bills one extra operation per document.

        request_options : typing.Optional[RequestOptions]
            Request-specific configuration.

        Returns
        -------
        DocumentRef
            Document accepted for processing

        Examples
        --------
        from supermemory import Supermemory

        client = Supermemory(
            api_key="YOUR_API_KEY",
        )
        client.add(
            namespace="user_alex",
            content="Supermemory turns unstructured content into evolving memory.",
        )
        """
        _response = self._raw_client.add(
            namespace,
            content=content,
            id=id,
            supporting_context=supporting_context,
            metadata=metadata,
            group=group,
            date=date,
            task_type=task_type,
            dreaming=dreaming,
            request_options=request_options,
        )
        return _response.data

    def search(
        self,
        namespace: str,
        *,
        query: str,
        filter: typing.Optional[FilterExpressionParams] = OMIT,
        search_mode: typing.Optional[SearchMode] = OMIT,
        limit: typing.Optional[int] = OMIT,
        include: typing.Optional[SearchIncludeParams] = OMIT,
        threshold: typing.Optional[float] = OMIT,
        rerank: typing.Optional[SearchRequestRerank] = OMIT,
        rewrite_query: typing.Optional[bool] = OMIT,
        request_options: typing.Optional[RequestOptions] = None,
        extra_headers: typing.Optional[typing.Mapping[str, str]] = None,
        extra_query: typing.Optional[typing.Mapping[str, typing.Any]] = None,
        extra_body: typing.Optional[typing.Mapping[str, typing.Any]] = None,
        timeout: typing.Union[float, httpx.Timeout, None, NotGiven] = NOT_GIVEN,
    ) -> SearchResponse:
        """
        Recall the most relevant learned context and source passages from a namespace. Hybrid search combines memories with document chunks by default, with optional query rewriting, reranking, and supporting context attachments.

        Parameters
        ----------
        namespace : str
            The isolated namespace to search. This can be an ID for your user, a project ID, or any other identifier you wish to use to scope memories.

        query : str
            Natural-language question, topic, or phrase to retrieve relevant context for. Replaces the v4 `q` field.

        filter : typing.Optional[FilterExpressionParams]
            Type-safe metadata conditions applied before ranking results. Replaces the v4 `filters` field and unifies both legacy filter formats into one expression type.

        search_mode : typing.Optional[SearchMode]
            Search surface. "hybrid" combines learned memories with source chunks, "memories" returns learned context, and "chunks" returns source passages.

        limit : typing.Optional[int]
            Maximum number of results to return

        include : typing.Optional[SearchIncludeParams]
            Optional context to include alongside each matching result

        threshold : typing.Optional[float]
            Minimum relevance score from 0 to 1. Raise it for precision (fewer, accurate results) or lower it for broader recall (more results).

        rerank : typing.Optional[SearchRequestRerank]
            Post-retrieval ranking. "order" improves result ordering; "aggregate" also combines overlapping context into cleaner answers. This is helpful if you want to ensure the most relevant results are returned.

        rewrite_query : typing.Optional[bool]
            Expand and clarify the query before retrieval to improve recall for conversational or underspecified prompts. This increases the latency by about 400ms.

        request_options : typing.Optional[RequestOptions]
            Request-specific configuration.

        Returns
        -------
        SearchResponse
            Search results

        Examples
        --------
        from supermemory import Supermemory

        client = Supermemory(
            api_key="YOUR_API_KEY",
        )
        client.search(
            namespace="user_alex",
            query="what are the API rate limits",
        )
        """
        _response = self._raw_client.search(
            namespace,
            query=query,
            filter=filter,
            search_mode=search_mode,
            limit=limit,
            include=include,
            threshold=threshold,
            rerank=rerank,
            rewrite_query=rewrite_query,
            request_options=request_options,
        )
        return _response.data

    def profile(
        self,
        namespace: str,
        *,
        filter: typing.Optional[FilterExpressionParams] = OMIT,
        buckets: typing.Optional[typing.Sequence[str]] = OMIT,
        request_options: typing.Optional[RequestOptions] = None,
        extra_headers: typing.Optional[typing.Mapping[str, str]] = None,
        extra_query: typing.Optional[typing.Mapping[str, typing.Any]] = None,
        extra_body: typing.Optional[typing.Mapping[str, typing.Any]] = None,
        timeout: typing.Union[float, httpx.Timeout, None, NotGiven] = NOT_GIVEN,
    ) -> ProfileResponse:
        """
        Read a continuously maintained understanding of the subject represented by this namespace. Stable facts, evolving context, and selected custom buckets are returned together without requiring a search query.

        Parameters
        ----------
        namespace : str
            The isolated namespace to search. This can be an ID for your user, a project ID, or any other identifier you wish to use to scope memories.

        filter : typing.Optional[FilterExpressionParams]
            Type-safe metadata conditions that limit which memories contribute to the profile

        buckets : typing.Optional[typing.Sequence[str]]
            Custom buckets to return. Omit to return every effective bucket.

        request_options : typing.Optional[RequestOptions]
            Request-specific configuration.

        Returns
        -------
        ProfileResponse
            Namespace profile. Send `Accept: text/markdown` to receive it as a markdown document instead of JSON.

        Examples
        --------
        from supermemory import Supermemory

        client = Supermemory(
            api_key="YOUR_API_KEY",
        )
        client.profile(
            namespace="user_alex",
        )
        """
        _response = self._raw_client.profile(namespace, filter=filter, buckets=buckets, request_options=request_options)
        return _response.data

    def list(
        self,
        namespace: str,
        type: ListType,
        *,
        page: typing.Optional[int] = None,
        limit: typing.Optional[int] = None,
        sort: typing.Optional[ListSort] = None,
        order: typing.Optional[SortOrder] = None,
        filter: typing.Optional[FilterExpressionParams] = OMIT,
        include: typing.Optional[ListRequestIncludeParams] = OMIT,
        request_options: typing.Optional[RequestOptions] = None,
        extra_headers: typing.Optional[typing.Mapping[str, str]] = None,
        extra_query: typing.Optional[typing.Mapping[str, typing.Any]] = None,
        extra_body: typing.Optional[typing.Mapping[str, typing.Any]] = None,
        timeout: typing.Union[float, httpx.Timeout, None, NotGiven] = NOT_GIVEN,
    ) -> ListResponse:
        """
        Browse documents, source chunks, or learned memories through one predictable paginated contract. Choose the collection in the path; the other collection arrays remain empty for a stable response shape.

        Parameters
        ----------
        namespace : str
            Namespace containing the resources to list. This can be an ID for your user, a project ID, or any other identifier you wish to use to scope resources.

        type : ListType
            Resource collection to list: source documents, extracted chunks, or learned memories

        page : typing.Optional[int]
            One-based page number

        limit : typing.Optional[int]
            Maximum resources to return per page

        sort : typing.Optional[ListSort]
            Field used to order results. Position is available only when listing chunks.

        order : typing.Optional[SortOrder]
            Ascending or descending sort direction

        filter : typing.Optional[FilterExpressionParams]
            Type-safe metadata conditions applied before pagination

        include : typing.Optional[ListRequestIncludeParams]
            Optional extras, matching search's include

        request_options : typing.Optional[RequestOptions]
            Request-specific configuration.

        Returns
        -------
        ListResponse
            Namespace resources retrieved successfully

        Examples
        --------
        from supermemory import Supermemory

        client = Supermemory(
            api_key="YOUR_API_KEY",
        )
        client.list(
            namespace="user_alex",
            type="documents",
            page=1,
            limit=10,
        )
        """
        _response = self._raw_client.list(
            namespace,
            type,
            page=page,
            limit=limit,
            sort=sort,
            order=order,
            filter=filter,
            include=include,
            request_options=request_options,
        )
        return _response.data

    @property
    def documents(self):
        if self._documents is None:
            from .documents.client import DocumentsClient  # noqa: E402

            self._documents = DocumentsClient(client_wrapper=self._client_wrapper)
        return self._documents

    @property
    def profiles(self):
        if self._profiles is None:
            from .profiles.client import ProfilesClient  # noqa: E402

            self._profiles = ProfilesClient(client_wrapper=self._client_wrapper)
        return self._profiles

    @property
    def memories(self):
        if self._memories is None:
            from .memories.client import MemoriesClient  # noqa: E402

            self._memories = MemoriesClient(client_wrapper=self._client_wrapper)
        return self._memories

    @property
    def connectors(self):
        if self._connectors is None:
            from .connectors.client import ConnectorsClient  # noqa: E402

            self._connectors = ConnectorsClient(client_wrapper=self._client_wrapper)
        return self._connectors

    @property
    def namespaces(self):
        if self._namespaces is None:
            from .namespaces.client import NamespacesClient  # noqa: E402

            self._namespaces = NamespacesClient(client_wrapper=self._client_wrapper)
        return self._namespaces

    @property
    def organization(self):
        if self._organization is None:
            from .organization.client import OrganizationClient  # noqa: E402

            self._organization = OrganizationClient(client_wrapper=self._client_wrapper)
        return self._organization


def _make_default_async_client(
    timeout: typing.Optional[float],
    follow_redirects: typing.Optional[bool],
) -> httpx.AsyncClient:
    try:
        import httpx_aiohttp  # type: ignore[import-not-found]
    except ImportError:
        pass
    else:
        if follow_redirects is not None:
            return httpx_aiohttp.HttpxAiohttpClient(timeout=timeout, follow_redirects=follow_redirects)
        return httpx_aiohttp.HttpxAiohttpClient(timeout=timeout)

    if follow_redirects is not None:
        return httpx.AsyncClient(timeout=timeout, follow_redirects=follow_redirects)
    return httpx.AsyncClient(timeout=timeout)


class AsyncSupermemory:
    """
    Use this class to access the different functions within the SDK. You can instantiate any number of clients with different configuration that will propagate to these functions.

    Parameters
    ----------
    base_url : typing.Optional[str]
        The base url to use for requests from the client.

    environment : SupermemoryEnvironment
        The environment to use for requests from the client. from .environment import SupermemoryEnvironment



        Defaults to SupermemoryEnvironment.DEFAULT



    api_key : typing.Optional[typing.Union[str, typing.Callable[[], str]]]
    headers : typing.Optional[typing.Dict[str, str]]
        Additional headers to send with every request.

    async_token : typing.Optional[typing.Callable[[], typing.Awaitable[str]]]
        An async callable that returns a bearer token. Use this when token acquisition involves async I/O (e.g., refreshing tokens via an async HTTP client). When provided, this is used instead of the synchronous token for async requests.

    timeout : typing.Optional[float]
        The timeout to be used, in seconds, for requests. By default the timeout is 60 seconds, unless a custom httpx client is used, in which case this default is not enforced.

    max_retries : typing.Optional[int]
        The default maximum number of retries for failed requests. Defaults to 2. Per-request `max_retries` in `request_options` takes precedence over this value.

    stream_reconnection_enabled : typing.Optional[bool]
        Whether to automatically reconnect on stream disconnection for resumable streaming endpoints. Defaults to True. Per-request `stream_reconnection_enabled` in `request_options` takes precedence over this value.

    max_stream_reconnection_attempts : typing.Optional[int]
        The maximum number of reconnection attempts for resumable streaming endpoints. Defaults to no limit. Per-request `max_stream_reconnection_attempts` in `request_options` takes precedence over this value.

    follow_redirects : typing.Optional[bool]
        Whether the default httpx client follows redirects or not, this is irrelevant if a custom httpx client is passed in.

    httpx_client : typing.Optional[httpx.AsyncClient]
        The httpx client to use for making requests, a preconfigured client is used by default, however this is useful should you want to pass in any custom httpx configuration.

    logging : typing.Optional[typing.Union[LogConfig, Logger]]
        Configure logging for the SDK. Accepts a LogConfig dict with 'level' (debug/info/warn/error), 'logger' (custom logger implementation), and 'silent' (boolean, defaults to True) fields. You can also pass a pre-configured Logger instance.

    Examples
    --------
    from supermemory import AsyncSupermemory

    client = AsyncSupermemory(
        api_key="YOUR_API_KEY",
    )
    """

    def __init__(
        self,
        *,
        base_url: typing.Optional[str] = None,
        environment: SupermemoryEnvironment = SupermemoryEnvironment.DEFAULT,
        api_key: typing.Optional[typing.Union[str, typing.Callable[[], str]]] = os.getenv("SUPERMEMORY_API_KEY"),
        headers: typing.Optional[typing.Dict[str, str]] = None,
        async_token: typing.Optional[typing.Callable[[], typing.Awaitable[str]]] = None,
        timeout: typing.Optional[float] = None,
        max_retries: typing.Optional[int] = None,
        stream_reconnection_enabled: typing.Optional[bool] = None,
        max_stream_reconnection_attempts: typing.Optional[int] = None,
        follow_redirects: typing.Optional[bool] = True,
        httpx_client: typing.Optional[httpx.AsyncClient] = None,
        logging: typing.Optional[typing.Union[LogConfig, Logger]] = None,
    ):
        _defaulted_timeout = timeout if timeout is not None else 60 if httpx_client is None else None
        _defaulted_max_retries = max_retries if max_retries is not None else 2
        if api_key is None:
            raise ApiError(
                body="The client must be instantiated be either passing in api_key or setting SUPERMEMORY_API_KEY"
            )
        self._client_wrapper = AsyncClientWrapper(
            base_url=_get_base_url(base_url=base_url, environment=environment),
            api_key=api_key,
            headers=headers,
            async_token=async_token,
            httpx_client=httpx_client
            if httpx_client is not None
            else _make_default_async_client(timeout=_defaulted_timeout, follow_redirects=follow_redirects),
            timeout=_defaulted_timeout,
            max_retries=_defaulted_max_retries,
            stream_reconnection_enabled=stream_reconnection_enabled,
            max_stream_reconnection_attempts=max_stream_reconnection_attempts,
            logging=logging,
        )
        self._raw_client = AsyncRawSupermemory(client_wrapper=self._client_wrapper)
        self._documents: typing.Optional[AsyncDocumentsClient] = None
        self._profiles: typing.Optional[AsyncProfilesClient] = None
        self._memories: typing.Optional[AsyncMemoriesClient] = None
        self._connectors: typing.Optional[AsyncConnectorsClient] = None
        self._namespaces: typing.Optional[AsyncNamespacesClient] = None
        self._organization: typing.Optional[AsyncOrganizationClient] = None

    @property
    def with_raw_response(self) -> AsyncRawSupermemory:
        """
        Retrieves a raw implementation of this client that returns raw responses.

        Returns
        -------
        AsyncRawSupermemory
        """
        return self._raw_client

    async def add(
        self,
        namespace: str,
        *,
        content: str,
        id: typing.Optional[str] = OMIT,
        supporting_context: typing.Optional[str] = OMIT,
        metadata: typing.Optional[typing.Dict[str, MetadataValueParams]] = OMIT,
        group: typing.Optional[typing.Dict[str, MetadataValueParams]] = OMIT,
        date: typing.Optional[str] = OMIT,
        task_type: typing.Optional[TaskType] = OMIT,
        dreaming: typing.Optional[DreamingMode] = OMIT,
        request_options: typing.Optional[RequestOptions] = None,
        extra_headers: typing.Optional[typing.Mapping[str, str]] = None,
        extra_query: typing.Optional[typing.Mapping[str, typing.Any]] = None,
        extra_body: typing.Optional[typing.Mapping[str, typing.Any]] = None,
        timeout: typing.Union[float, httpx.Timeout, None, NotGiven] = NOT_GIVEN,
    ) -> DocumentRef:
        """
        Turn text or a supported URL into searchable, evolving memory. Supply a new ID to create a document, or reuse an existing ID to append new information while preserving its history.

        Parameters
        ----------
        namespace : str
            The isolated namespace. This can be an ID for your user, a project ID, or any other identifier you wish to use to scope documents.

        content : str
            The content to process. This may be plaintext or a URL to supported rich content.

        id : typing.Optional[str]
            An optional caller-defined document ID

        supporting_context : typing.Optional[str]
            Context used to guide memory extraction within this namespace. Max 1500 characters.

        metadata : typing.Optional[typing.Dict[str, MetadataValueParams]]
            Arbitrary metadata attached to the document and derived data

        group : typing.Optional[typing.Dict[str, MetadataValueParams]]
            Metadata values that isolate related context within the namespace

        date : typing.Optional[str]
            When the source content is from, in ISO 8601 format

        task_type : typing.Optional[TaskType]
            Processing pipeline. "memory" builds durable learned context; "superrag" optimizes the document for retrieval without generating memories.

        dreaming : typing.Optional[DreamingMode]
            Processing mode. "dynamic" (default) groups related documents so memories form from coherent context. "instant" processes each document independently right away and bills one extra operation per document.

        request_options : typing.Optional[RequestOptions]
            Request-specific configuration.

        Returns
        -------
        DocumentRef
            Document accepted for processing

        Examples
        --------
        import asyncio

        from supermemory import AsyncSupermemory

        client = AsyncSupermemory(
            api_key="YOUR_API_KEY",
        )


        async def main() -> None:
            await client.add(
                namespace="user_alex",
                content="Supermemory turns unstructured content into evolving memory.",
            )


        asyncio.run(main())
        """
        _response = await self._raw_client.add(
            namespace,
            content=content,
            id=id,
            supporting_context=supporting_context,
            metadata=metadata,
            group=group,
            date=date,
            task_type=task_type,
            dreaming=dreaming,
            request_options=request_options,
        )
        return _response.data

    async def search(
        self,
        namespace: str,
        *,
        query: str,
        filter: typing.Optional[FilterExpressionParams] = OMIT,
        search_mode: typing.Optional[SearchMode] = OMIT,
        limit: typing.Optional[int] = OMIT,
        include: typing.Optional[SearchIncludeParams] = OMIT,
        threshold: typing.Optional[float] = OMIT,
        rerank: typing.Optional[SearchRequestRerank] = OMIT,
        rewrite_query: typing.Optional[bool] = OMIT,
        request_options: typing.Optional[RequestOptions] = None,
        extra_headers: typing.Optional[typing.Mapping[str, str]] = None,
        extra_query: typing.Optional[typing.Mapping[str, typing.Any]] = None,
        extra_body: typing.Optional[typing.Mapping[str, typing.Any]] = None,
        timeout: typing.Union[float, httpx.Timeout, None, NotGiven] = NOT_GIVEN,
    ) -> SearchResponse:
        """
        Recall the most relevant learned context and source passages from a namespace. Hybrid search combines memories with document chunks by default, with optional query rewriting, reranking, and supporting context attachments.

        Parameters
        ----------
        namespace : str
            The isolated namespace to search. This can be an ID for your user, a project ID, or any other identifier you wish to use to scope memories.

        query : str
            Natural-language question, topic, or phrase to retrieve relevant context for. Replaces the v4 `q` field.

        filter : typing.Optional[FilterExpressionParams]
            Type-safe metadata conditions applied before ranking results. Replaces the v4 `filters` field and unifies both legacy filter formats into one expression type.

        search_mode : typing.Optional[SearchMode]
            Search surface. "hybrid" combines learned memories with source chunks, "memories" returns learned context, and "chunks" returns source passages.

        limit : typing.Optional[int]
            Maximum number of results to return

        include : typing.Optional[SearchIncludeParams]
            Optional context to include alongside each matching result

        threshold : typing.Optional[float]
            Minimum relevance score from 0 to 1. Raise it for precision (fewer, accurate results) or lower it for broader recall (more results).

        rerank : typing.Optional[SearchRequestRerank]
            Post-retrieval ranking. "order" improves result ordering; "aggregate" also combines overlapping context into cleaner answers. This is helpful if you want to ensure the most relevant results are returned.

        rewrite_query : typing.Optional[bool]
            Expand and clarify the query before retrieval to improve recall for conversational or underspecified prompts. This increases the latency by about 400ms.

        request_options : typing.Optional[RequestOptions]
            Request-specific configuration.

        Returns
        -------
        SearchResponse
            Search results

        Examples
        --------
        import asyncio

        from supermemory import AsyncSupermemory

        client = AsyncSupermemory(
            api_key="YOUR_API_KEY",
        )


        async def main() -> None:
            await client.search(
                namespace="user_alex",
                query="what are the API rate limits",
            )


        asyncio.run(main())
        """
        _response = await self._raw_client.search(
            namespace,
            query=query,
            filter=filter,
            search_mode=search_mode,
            limit=limit,
            include=include,
            threshold=threshold,
            rerank=rerank,
            rewrite_query=rewrite_query,
            request_options=request_options,
        )
        return _response.data

    async def profile(
        self,
        namespace: str,
        *,
        filter: typing.Optional[FilterExpressionParams] = OMIT,
        buckets: typing.Optional[typing.Sequence[str]] = OMIT,
        request_options: typing.Optional[RequestOptions] = None,
        extra_headers: typing.Optional[typing.Mapping[str, str]] = None,
        extra_query: typing.Optional[typing.Mapping[str, typing.Any]] = None,
        extra_body: typing.Optional[typing.Mapping[str, typing.Any]] = None,
        timeout: typing.Union[float, httpx.Timeout, None, NotGiven] = NOT_GIVEN,
    ) -> ProfileResponse:
        """
        Read a continuously maintained understanding of the subject represented by this namespace. Stable facts, evolving context, and selected custom buckets are returned together without requiring a search query.

        Parameters
        ----------
        namespace : str
            The isolated namespace to search. This can be an ID for your user, a project ID, or any other identifier you wish to use to scope memories.

        filter : typing.Optional[FilterExpressionParams]
            Type-safe metadata conditions that limit which memories contribute to the profile

        buckets : typing.Optional[typing.Sequence[str]]
            Custom buckets to return. Omit to return every effective bucket.

        request_options : typing.Optional[RequestOptions]
            Request-specific configuration.

        Returns
        -------
        ProfileResponse
            Namespace profile. Send `Accept: text/markdown` to receive it as a markdown document instead of JSON.

        Examples
        --------
        import asyncio

        from supermemory import AsyncSupermemory

        client = AsyncSupermemory(
            api_key="YOUR_API_KEY",
        )


        async def main() -> None:
            await client.profile(
                namespace="user_alex",
            )


        asyncio.run(main())
        """
        _response = await self._raw_client.profile(
            namespace, filter=filter, buckets=buckets, request_options=request_options
        )
        return _response.data

    async def list(
        self,
        namespace: str,
        type: ListType,
        *,
        page: typing.Optional[int] = None,
        limit: typing.Optional[int] = None,
        sort: typing.Optional[ListSort] = None,
        order: typing.Optional[SortOrder] = None,
        filter: typing.Optional[FilterExpressionParams] = OMIT,
        include: typing.Optional[ListRequestIncludeParams] = OMIT,
        request_options: typing.Optional[RequestOptions] = None,
        extra_headers: typing.Optional[typing.Mapping[str, str]] = None,
        extra_query: typing.Optional[typing.Mapping[str, typing.Any]] = None,
        extra_body: typing.Optional[typing.Mapping[str, typing.Any]] = None,
        timeout: typing.Union[float, httpx.Timeout, None, NotGiven] = NOT_GIVEN,
    ) -> ListResponse:
        """
        Browse documents, source chunks, or learned memories through one predictable paginated contract. Choose the collection in the path; the other collection arrays remain empty for a stable response shape.

        Parameters
        ----------
        namespace : str
            Namespace containing the resources to list. This can be an ID for your user, a project ID, or any other identifier you wish to use to scope resources.

        type : ListType
            Resource collection to list: source documents, extracted chunks, or learned memories

        page : typing.Optional[int]
            One-based page number

        limit : typing.Optional[int]
            Maximum resources to return per page

        sort : typing.Optional[ListSort]
            Field used to order results. Position is available only when listing chunks.

        order : typing.Optional[SortOrder]
            Ascending or descending sort direction

        filter : typing.Optional[FilterExpressionParams]
            Type-safe metadata conditions applied before pagination

        include : typing.Optional[ListRequestIncludeParams]
            Optional extras, matching search's include

        request_options : typing.Optional[RequestOptions]
            Request-specific configuration.

        Returns
        -------
        ListResponse
            Namespace resources retrieved successfully

        Examples
        --------
        import asyncio

        from supermemory import AsyncSupermemory

        client = AsyncSupermemory(
            api_key="YOUR_API_KEY",
        )


        async def main() -> None:
            await client.list(
                namespace="user_alex",
                type="documents",
                page=1,
                limit=10,
            )


        asyncio.run(main())
        """
        _response = await self._raw_client.list(
            namespace,
            type,
            page=page,
            limit=limit,
            sort=sort,
            order=order,
            filter=filter,
            include=include,
            request_options=request_options,
        )
        return _response.data

    @property
    def documents(self):
        if self._documents is None:
            from .documents.client import AsyncDocumentsClient  # noqa: E402

            self._documents = AsyncDocumentsClient(client_wrapper=self._client_wrapper)
        return self._documents

    @property
    def profiles(self):
        if self._profiles is None:
            from .profiles.client import AsyncProfilesClient  # noqa: E402

            self._profiles = AsyncProfilesClient(client_wrapper=self._client_wrapper)
        return self._profiles

    @property
    def memories(self):
        if self._memories is None:
            from .memories.client import AsyncMemoriesClient  # noqa: E402

            self._memories = AsyncMemoriesClient(client_wrapper=self._client_wrapper)
        return self._memories

    @property
    def connectors(self):
        if self._connectors is None:
            from .connectors.client import AsyncConnectorsClient  # noqa: E402

            self._connectors = AsyncConnectorsClient(client_wrapper=self._client_wrapper)
        return self._connectors

    @property
    def namespaces(self):
        if self._namespaces is None:
            from .namespaces.client import AsyncNamespacesClient  # noqa: E402

            self._namespaces = AsyncNamespacesClient(client_wrapper=self._client_wrapper)
        return self._namespaces

    @property
    def organization(self):
        if self._organization is None:
            from .organization.client import AsyncOrganizationClient  # noqa: E402

            self._organization = AsyncOrganizationClient(client_wrapper=self._client_wrapper)
        return self._organization


def _get_base_url(*, base_url: typing.Optional[str] = None, environment: SupermemoryEnvironment) -> str:
    if base_url is not None:
        return base_url
    elif environment is not None:
        return environment.value
    else:
        raise Exception("Please pass in either base_url or environment to construct the client")