diff --git a/python/packages/mem0/AGENTS.md b/python/packages/mem0/AGENTS.md
index c1defa5ff86..7625604d5b7 100644
--- a/python/packages/mem0/AGENTS.md
+++ b/python/packages/mem0/AGENTS.md
@@ -13,10 +13,21 @@ from agent_framework.mem0 import Mem0ContextProvider
 
 provider = Mem0ContextProvider(
     api_key="your-key",
+    # Storage scope: memories are written with this user_id.
     user_id="user-id",
+    # Retrieval scope: must be set explicitly, it never inherits the storage scope.
+    search_user_id="user-id",
 )
 ```
 
+## Memory Scoping
+
+- `application_id` / `agent_id` / `user_id` are the **storage scope** stamped on written memories.
+- `search_application_id` / `search_agent_id` / `search_user_id` are the **retrieval scope**.
+- Retrieval scope never defaults to the storage scope. With no `search_*` value set, `before_run`
+  retrieves nothing and logs a warning. This prevents memories written under a shared `agent_id`
+  from being retrieved for every user of that agent.
+
 ## Import Path
 
 ```python
diff --git a/python/packages/mem0/README.md b/python/packages/mem0/README.md
index 4d9cb64530f..e6dbb3b3a32 100644
--- a/python/packages/mem0/README.md
+++ b/python/packages/mem0/README.md
@@ -19,6 +19,27 @@ See the [Mem0 basic example](../../samples/02-agents/context_providers/mem0/mem0
 - Retrieving information using remembered context across new threads
 - Persistent memory
 
+### Memory Scoping
+
+The provider separates the scope used to **store** memories from the scope used to **retrieve** them:
+
+- `application_id` / `agent_id` / `user_id` stamp every memory that is written.
+- `search_application_id` / `search_agent_id` / `search_user_id` select which memories are searched.
+
+Note: `application_id` and `search_application_id` are only supported by the Platform client (`AsyncMemoryClient`), not the OSS `AsyncMemory` client.
+Retrieval scope values never inherit from the storage scope. If none of the `search_*` values are
+set, no memories are retrieved and a warning is logged. Set `search_user_id` for per-user memory,
+and only set `search_agent_id` when memories under that agent are safe to share across all of its
+users.
+
+```python
+provider = Mem0ContextProvider(
+    api_key="your-key",
+    user_id="user-id",
+    search_user_id="user-id",
+)
+```
+
 ## Telemetry
 
 Mem0's telemetry is **disabled by default** when using this package. If you want to enable telemetry, set the environment variable before importing:
diff --git a/python/packages/mem0/agent_framework_mem0/_context_provider.py b/python/packages/mem0/agent_framework_mem0/_context_provider.py
index 1ca3138949f..20bba5ad9e1 100644
--- a/python/packages/mem0/agent_framework_mem0/_context_provider.py
+++ b/python/packages/mem0/agent_framework_mem0/_context_provider.py
@@ -46,6 +46,19 @@ class Mem0ContextProvider(ContextProvider):
 
     Integrates Mem0 for persistent semantic memory, searching and storing
     memories via the Mem0 API.
+
+    The provider keeps the storage scope and the retrieval scope separate:
+
+    * ``application_id`` / ``agent_id`` / ``user_id`` are the **storage scope**. They are
+      stamped onto every memory written by :meth:`after_run` and are never used to
+      retrieve memories.
+    * ``search_application_id`` / ``search_agent_id`` / ``search_user_id`` are the
+      **retrieval scope** used by :meth:`before_run`.
+
+    Retrieval scope values never inherit from the storage scope. If no ``search_*``
+    value is configured, no memories are retrieved. This prevents a memory written
+    under a shared ``agent_id`` from being read back by an unrelated user, since
+    agent-wide retrieval must be requested explicitly via ``search_agent_id``.
     """
 
     DEFAULT_CONTEXT_PROMPT = "## Memories\nConsider the following memories when answering user questions:"
@@ -59,6 +72,9 @@ def __init__(
         application_id: str | None = None,
         agent_id: str | None = None,
         user_id: str | None = None,
+        search_application_id: str | None = None,
+        search_agent_id: str | None = None,
+        search_user_id: str | None = None,
         *,
         context_prompt: str | None = None,
     ) -> None:
@@ -68,13 +84,23 @@ def __init__(
             source_id: Unique identifier for this provider instance.
             mem0_client: A pre-created Mem0 MemoryClient or None to create a default client.
             api_key: The API key for authenticating with the Mem0 API.
-            application_id: The application ID for scoping memories. Platform-only:
+            application_id: The application ID that stored memories are stamped with. Platform-only:
                 the OSS ``AsyncMemory`` client does not recognize an application
                 scope (it scopes only by user_id/agent_id in this provider), so
                 application_id cannot be used with an OSS client.
-            agent_id: The agent ID for scoping memories.
-            user_id: The user ID for scoping memories.
+            agent_id: The agent ID that stored memories are stamped with.
+            user_id: The user ID that stored memories are stamped with.
+            search_application_id: The application ID to retrieve memories for. Platform-only,
+                like ``application_id``.
+            search_agent_id: The agent ID to retrieve memories for. Setting this retrieves
+                memories stored by **any** user under that agent, so only set it for
+                agent-wide knowledge that is safe to share across users.
+            search_user_id: The user ID to retrieve memories for.
             context_prompt: The prompt to prepend to retrieved memories.
+
+        Remarks:
+            The ``search_*`` parameters do not default to their storage-scope counterparts.
+            When none of them are set, :meth:`before_run` retrieves nothing and logs a warning.
         """
         super().__init__(source_id)
         should_close_client = False
@@ -86,9 +112,13 @@ def __init__(
         self.application_id = application_id
         self.agent_id = agent_id
         self.user_id = user_id
+        self.search_application_id = search_application_id
+        self.search_agent_id = search_agent_id
+        self.search_user_id = search_user_id
         self.context_prompt = context_prompt or self.DEFAULT_CONTEXT_PROMPT
         self.mem0_client = mem0_client
         self._should_close_client = should_close_client
+        self._warned_no_search_scope = False
 
     async def __aenter__(self) -> Self:
         """Async context manager entry."""
@@ -114,6 +144,15 @@ async def before_run(
         """Search Mem0 for relevant memories and add to the session context."""
         mark_feature_used(FeatureIndex.MEM0)
         self._validate_filters()
+        if not (self.search_user_id or self.search_agent_id or self.search_application_id):
+            if not self._warned_no_search_scope:
+                self._warned_no_search_scope = True
+                logger.warning(
+                    "Mem0ContextProvider has no retrieval scope configured, so no memories will be retrieved. "
+                    "Set search_user_id, search_agent_id and/or search_application_id."
+                )
+            return
+
         input_text = "\n".join(msg.text for msg in context.input_messages if msg and msg.text and msg.text.strip())
         if not input_text.strip():
             return
@@ -123,18 +162,18 @@ async def before_run(
         search_tasks: list[Awaitable[Any]] = []
 
         # 1. Query User partition independently
-        if self.user_id:
-            user_kwargs = self._build_search_kwargs(input_text, "user_id", self.user_id)
+        if self.search_user_id:
+            user_kwargs = self._build_search_kwargs(input_text, "user_id", self.search_user_id)
             search_tasks.append(self.mem0_client.search(**user_kwargs))  # type: ignore[reportUnknownMemberType, reportUnknownArgumentType]
 
         # 2. Query Agent partition independently
-        if self.agent_id:
-            agent_kwargs = self._build_search_kwargs(input_text, "agent_id", self.agent_id)
+        if self.search_agent_id:
+            agent_kwargs = self._build_search_kwargs(input_text, "agent_id", self.search_agent_id)
             search_tasks.append(self.mem0_client.search(**agent_kwargs))  # type: ignore[reportUnknownMemberType, reportUnknownArgumentType]
 
-        # Fall back to an app-scoped search when only application_id is configured.
-        if not search_tasks and self.application_id:
-            app_kwargs: dict[str, Any] = {"query": input_text, "filters": self._build_filters()}
+        # Fall back to an app-scoped search when only search_application_id is configured.
+        if not search_tasks and self.search_application_id:
+            app_kwargs: dict[str, Any] = {"query": input_text, "filters": {"app_id": self.search_application_id}}
             search_tasks.append(self.mem0_client.search(**app_kwargs))  # pyright: ignore[reportUnknownMemberType, reportUnknownArgumentType]
         if not search_tasks:
             return
@@ -239,7 +278,7 @@ def _validate_filters(self) -> None:
         """Validates that at least one usable filter is provided for the configured client."""
         if not self.agent_id and not self.user_id and not self.application_id:
             raise ValueError("At least one of the filters: agent_id, user_id, or application_id is required.")
-        if isinstance(self.mem0_client, AsyncMemory) and self.application_id:
+        if isinstance(self.mem0_client, AsyncMemory) and (self.application_id or self.search_application_id):
             raise ValueError(
                 "application_id is not supported by the OSS AsyncMemory client, which scopes "
                 "memories only by user_id/agent_id. Remove application_id or use AsyncMemoryClient."
@@ -247,22 +286,23 @@ def _validate_filters(self) -> None:
 
     def _build_search_kwargs(self, input_text: str, entity_key: str, entity_value: str) -> dict[str, Any]:
         """Build search keyword arguments formatted for OSS vs Platform clients."""
-        filters: dict[str, Any] = {"query": input_text}
+        kwargs: dict[str, Any] = {"query": input_text}
 
-        if self.application_id and isinstance(self.mem0_client, AsyncMemory):
+        if self.search_application_id and isinstance(self.mem0_client, AsyncMemory):
             raise ValueError(
                 "application_id is not supported by the OSS AsyncMemory client, which scopes "
                 "memories only by user_id/agent_id. Remove application_id or use AsyncMemoryClient."
             )
 
-        filters["filters"] = {entity_key: entity_value}
-        if self.application_id and not isinstance(self.mem0_client, AsyncMemory):
-            filters["filters"]["app_id"] = self.application_id
+        filters: dict[str, Any] = {entity_key: entity_value}
+        if self.search_application_id and not isinstance(self.mem0_client, AsyncMemory):
+            filters["app_id"] = self.search_application_id
+        kwargs["filters"] = filters
 
-        return filters
+        return kwargs
 
     def _build_filters(self) -> dict[str, Any]:
-        """Build identity filters from initialization parameters."""
+        """Build storage identity filters from initialization parameters."""
         filters: dict[str, Any] = {}
         if self.user_id:
             filters["user_id"] = self.user_id
diff --git a/python/packages/mem0/tests/test_mem0_context_provider.py b/python/packages/mem0/tests/test_mem0_context_provider.py
index ab2ddef1fc3..bde518c56d1 100644
--- a/python/packages/mem0/tests/test_mem0_context_provider.py
+++ b/python/packages/mem0/tests/test_mem0_context_provider.py
@@ -52,6 +52,9 @@ def test_init_with_all_params(self, mock_mem0_client: AsyncMock) -> None:
             application_id="app1",
             agent_id="agent1",
             user_id="user1",
+            search_application_id="app2",
+            search_agent_id="agent2",
+            search_user_id="user2",
             context_prompt="Custom prompt",
         )
         assert provider.source_id == "mem0"
@@ -59,10 +62,22 @@ def test_init_with_all_params(self, mock_mem0_client: AsyncMock) -> None:
         assert provider.application_id == "app1"
         assert provider.agent_id == "agent1"
         assert provider.user_id == "user1"
+        assert provider.search_application_id == "app2"
+        assert provider.search_agent_id == "agent2"
+        assert provider.search_user_id == "user2"
         assert provider.context_prompt == "Custom prompt"
         assert provider.mem0_client is mock_mem0_client
         assert provider._should_close_client is False
 
+    def test_init_search_scope_defaults_to_none(self, mock_mem0_client: AsyncMock) -> None:
+        """Retrieval scope never inherits from the storage scope."""
+        provider = Mem0ContextProvider(
+            source_id="mem0", mem0_client=mock_mem0_client, user_id="u1", agent_id="a1", application_id="app1"
+        )
+        assert provider.search_user_id is None
+        assert provider.search_agent_id is None
+        assert provider.search_application_id is None
+
     def test_init_default_context_prompt(self, mock_mem0_client: AsyncMock) -> None:
         provider = Mem0ContextProvider(source_id="mem0", mem0_client=mock_mem0_client, user_id="u1")
         assert provider.context_prompt == Mem0ContextProvider.DEFAULT_CONTEXT_PROMPT
@@ -94,7 +109,9 @@ async def test_memories_added_to_context(self, mock_mem0_client: AsyncMock) -> N
             {"memory": "User likes Python"},
             {"memory": "User prefers dark mode"},
         ]
-        provider = Mem0ContextProvider(source_id="mem0", mem0_client=mock_mem0_client, user_id="u1")
+        provider = Mem0ContextProvider(
+            source_id="mem0", mem0_client=mock_mem0_client, user_id="u1", search_user_id="u1"
+        )
         session = AgentSession(session_id="test-session")
         ctx = SessionContext(input_messages=[Message(role="user", contents=["Hello"])], session_id="s1")
 
@@ -115,7 +132,9 @@ async def test_memories_added_to_context(self, mock_mem0_client: AsyncMock) -> N
 
     async def test_empty_input_skips_search(self, mock_mem0_client: AsyncMock) -> None:
         """Empty input messages → no search performed."""
-        provider = Mem0ContextProvider(source_id="mem0", mem0_client=mock_mem0_client, user_id="u1")
+        provider = Mem0ContextProvider(
+            source_id="mem0", mem0_client=mock_mem0_client, user_id="u1", search_user_id="u1"
+        )
         session = AgentSession(session_id="test-session")
         ctx = SessionContext(input_messages=[Message(role="user", contents=[""])], session_id="s1")
 
@@ -132,7 +151,9 @@ async def test_empty_input_skips_search(self, mock_mem0_client: AsyncMock) -> No
     async def test_empty_search_results_no_messages(self, mock_mem0_client: AsyncMock) -> None:
         """Empty search results → no messages added."""
         mock_mem0_client.search.return_value = []
-        provider = Mem0ContextProvider(source_id="mem0", mem0_client=mock_mem0_client, user_id="u1")
+        provider = Mem0ContextProvider(
+            source_id="mem0", mem0_client=mock_mem0_client, user_id="u1", search_user_id="u1"
+        )
         session = AgentSession(session_id="test-session")
         ctx = SessionContext(input_messages=[Message(role="user", contents=["test"])], session_id="s1")
 
@@ -162,7 +183,9 @@ async def test_validates_filters_before_search(self, mock_mem0_client: AsyncMock
     async def test_v1_1_response_format(self, mock_mem0_client: AsyncMock) -> None:
         """Search response in v1.1 dict format with 'results' key."""
         mock_mem0_client.search.return_value = {"results": [{"memory": "remembered fact"}]}
-        provider = Mem0ContextProvider(source_id="mem0", mem0_client=mock_mem0_client, user_id="u1")
+        provider = Mem0ContextProvider(
+            source_id="mem0", mem0_client=mock_mem0_client, user_id="u1", search_user_id="u1"
+        )
         session = AgentSession(session_id="test-session")
         ctx = SessionContext(input_messages=[Message(role="user", contents=["test"])], session_id="s1")
 
@@ -179,7 +202,9 @@ async def test_v1_1_response_format(self, mock_mem0_client: AsyncMock) -> None:
     async def test_search_query_combines_input_messages(self, mock_mem0_client: AsyncMock) -> None:
         """Multiple input messages are joined for the search query."""
         mock_mem0_client.search.return_value = []
-        provider = Mem0ContextProvider(source_id="mem0", mem0_client=mock_mem0_client, user_id="u1")
+        provider = Mem0ContextProvider(
+            source_id="mem0", mem0_client=mock_mem0_client, user_id="u1", search_user_id="u1"
+        )
         session = AgentSession(session_id="test-session")
         ctx = SessionContext(
             input_messages=[
@@ -202,7 +227,9 @@ async def test_search_query_combines_input_messages(self, mock_mem0_client: Asyn
     async def test_oss_client_passes_filters_dict(self, mock_oss_mem0_client: AsyncMock) -> None:
         """OSS AsyncMemory client should receive entity IDs in a filters dict (mem0 >=2.0)."""
         mock_oss_mem0_client.search.return_value = [{"memory": "User likes Python"}]
-        provider = Mem0ContextProvider(source_id="mem0", mem0_client=mock_oss_mem0_client, user_id="u1")
+        provider = Mem0ContextProvider(
+            source_id="mem0", mem0_client=mock_oss_mem0_client, user_id="u1", search_user_id="u1"
+        )
         session = AgentSession(session_id="test-session")
         ctx = SessionContext(input_messages=[Message(role="user", contents=["Hello"])], session_id="s1")
 
@@ -228,6 +255,8 @@ async def test_oss_client_rejects_application_id_with_user_or_agent(self, mock_o
             user_id="u1",
             agent_id="a1",
             application_id="app1",
+            search_user_id="u1",
+            search_agent_id="a1",
         )
 
         mock_context = MagicMock(spec=SessionContext)
@@ -245,7 +274,9 @@ async def test_oss_client_rejects_application_id_with_user_or_agent(self, mock_o
 
     async def test_oss_client_rejects_application_id_only(self, mock_oss_mem0_client: AsyncMock) -> None:
         """OSS client with only application_id set raises and never searches."""
-        provider = Mem0ContextProvider(source_id="mem0", mem0_client=mock_oss_mem0_client, application_id="app1")
+        provider = Mem0ContextProvider(
+            source_id="mem0", mem0_client=mock_oss_mem0_client, application_id="app1", search_application_id="app1"
+        )
         session = AgentSession(session_id="test-session")
         ctx = SessionContext(input_messages=[Message(role="user", contents=["Hello"])], session_id="s1")
 
@@ -268,6 +299,8 @@ async def test_platform_client_passes_filters_dict_except_app_id(self, mock_mem0
             mem0_client=mock_mem0_client,
             user_id="u1",
             agent_id="a1",
+            search_user_id="u1",
+            search_agent_id="a1",
         )
 
         mock_context = MagicMock(spec=SessionContext)
@@ -280,7 +313,7 @@ async def test_platform_client_passes_filters_dict_except_app_id(self, mock_mem0
             agent=MagicMock(), session=MagicMock(spec=AgentSession), context=mock_context, state={}
         )
 
-        # Re-aligned assertion: Platform client isolates filters per call to bypass AND limitations
+        # Retrieval scope is explicit: user and agent partitions are queried separately and merged.
         assert mock_mem0_client.search.call_count == 2
         mock_mem0_client.search.assert_any_call(query="hello", filters={"user_id": "u1"})
         mock_mem0_client.search.assert_any_call(query="hello", filters={"agent_id": "a1"})
@@ -290,7 +323,12 @@ async def test_platform_client_keeps_app_id(self, mock_mem0_client: AsyncMock) -
         mock_mem0_client.search.return_value = []
 
         provider = Mem0ContextProvider(
-            source_id="mem0", mem0_client=mock_mem0_client, user_id="u1", application_id="app1"
+            source_id="mem0",
+            mem0_client=mock_mem0_client,
+            user_id="u1",
+            application_id="app1",
+            search_user_id="u1",
+            search_application_id="app1",
         )
         session = AgentSession(session_id="test-session")
         ctx = SessionContext(input_messages=[Message(role="user", contents=["Hello"])], session_id="s1")
@@ -304,6 +342,200 @@ async def test_platform_client_keeps_app_id(self, mock_mem0_client: AsyncMock) -
 
         mock_mem0_client.search.assert_awaited_once_with(query="Hello", filters={"user_id": "u1", "app_id": "app1"})
 
+    async def test_no_search_scope_skips_retrieval(self, mock_mem0_client: AsyncMock) -> None:
+        """Storage scope is never used for retrieval: without search_* values nothing is searched."""
+        provider = Mem0ContextProvider(
+            source_id="mem0", mem0_client=mock_mem0_client, user_id="u1", agent_id="a1", application_id="app1"
+        )
+        session = AgentSession(session_id="test-session")
+        ctx = SessionContext(input_messages=[Message(role="user", contents=["Hello"])], session_id="s1")
+
+        with patch("agent_framework_mem0._context_provider.logger") as mock_logger:
+            await provider.before_run(
+                agent=cast(Any, None),
+                session=session,
+                context=ctx,
+                state=session.state.setdefault(provider.source_id, {}),
+            )  # type: ignore[arg-type]
+            # Second run must not re-emit the warning.
+            await provider.before_run(
+                agent=cast(Any, None),
+                session=session,
+                context=ctx,
+                state=session.state.setdefault(provider.source_id, {}),
+            )  # type: ignore[arg-type]
+
+        mock_mem0_client.search.assert_not_awaited()
+        assert "mem0" not in ctx.context_messages
+        mock_logger.warning.assert_called_once()
+
+    async def test_search_scope_does_not_inherit_storage_scope(self, mock_mem0_client: AsyncMock) -> None:
+        """A shared storage agent_id is not queried unless search_agent_id is set explicitly."""
+        mock_mem0_client.search.return_value = []
+        provider = Mem0ContextProvider(
+            source_id="mem0",
+            mem0_client=mock_mem0_client,
+            user_id="u1",
+            agent_id="shared-agent",
+            search_user_id="u1",
+        )
+        session = AgentSession(session_id="test-session")
+        ctx = SessionContext(input_messages=[Message(role="user", contents=["Hello"])], session_id="s1")
+
+        await provider.before_run(
+            agent=cast(Any, None),
+            session=session,
+            context=ctx,
+            state=session.state.setdefault(provider.source_id, {}),
+        )  # type: ignore[arg-type]
+
+        mock_mem0_client.search.assert_awaited_once_with(query="Hello", filters={"user_id": "u1"})
+
+    async def test_search_agent_id_opts_into_agent_partition(self, mock_mem0_client: AsyncMock) -> None:
+        """search_agent_id can differ from the storage agent_id and is queried on its own partition."""
+        mock_mem0_client.search.return_value = []
+        provider = Mem0ContextProvider(
+            source_id="mem0",
+            mem0_client=mock_mem0_client,
+            user_id="u1",
+            agent_id="storage-agent",
+            search_agent_id="shared-knowledge",
+        )
+        session = AgentSession(session_id="test-session")
+        ctx = SessionContext(input_messages=[Message(role="user", contents=["Hello"])], session_id="s1")
+
+        await provider.before_run(
+            agent=cast(Any, None),
+            session=session,
+            context=ctx,
+            state=session.state.setdefault(provider.source_id, {}),
+        )  # type: ignore[arg-type]
+
+        mock_mem0_client.search.assert_awaited_once_with(query="Hello", filters={"agent_id": "shared-knowledge"})
+
+
+class TestCrossUserIsolation:
+    """Regression tests for cross-user memory leakage through the shared agent partition."""
+
+    @staticmethod
+    def _fake_partition_client() -> Any:
+        """Build a fake Platform client modelling Mem0 partition semantics.
+
+        A search matches every stored memory that carries all of the provided filter
+        fields, so a filter without ``user_id`` spans all users.
+        """
+        from mem0 import AsyncMemoryClient
+
+        stored: list[dict[str, Any]] = []
+
+        async def add(**kwargs: Any) -> None:
+            filters = kwargs.get("filters", {})
+            for index, message in enumerate(kwargs["messages"]):
+                stored.append({
+                    "id": f"{filters.get('user_id', '')}-{len(stored)}-{index}",
+                    "memory": message["content"],
+                    **filters,
+                })
+
+        async def search(**kwargs: Any) -> list[dict[str, Any]]:
+            filters: dict[str, Any] = kwargs.get("filters", {})
+            return [memory for memory in stored if all(memory.get(key) == value for key, value in filters.items())]
+
+        client = AsyncMock(spec=AsyncMemoryClient)
+        client.add = AsyncMock(side_effect=add)
+        client.search = AsyncMock(side_effect=search)
+        return client
+
+    async def test_other_users_memories_are_not_retrieved(self) -> None:
+        """Alice's memories must never reach Bob, even though both write under the same agent_id."""
+        client = self._fake_partition_client()
+
+        alice = Mem0ContextProvider(
+            source_id="mem0",
+            mem0_client=client,
+            user_id="alice",
+            agent_id="support-bot",
+            search_user_id="alice",
+        )
+        bob = Mem0ContextProvider(
+            source_id="mem0",
+            mem0_client=client,
+            user_id="bob",
+            agent_id="support-bot",
+            search_user_id="bob",
+        )
+
+        alice_session = AgentSession(session_id="alice-session")
+        alice_ctx = SessionContext(
+            input_messages=[Message(role="user", contents=["remember my credit card is 4111-1111-1111-1111"])],
+            session_id="alice-session",
+        )
+        alice_ctx._response = AgentResponse(messages=[Message(role="assistant", contents=["Saved your card."])])
+        await alice.after_run(
+            agent=cast(Any, None),
+            session=alice_session,
+            context=alice_ctx,
+            state=alice_session.state.setdefault(alice.source_id, {}),
+        )  # type: ignore[arg-type]
+
+        bob_session = AgentSession(session_id="bob-session")
+        bob_ctx = SessionContext(
+            input_messages=[Message(role="user", contents=["what is my credit card number?"])],
+            session_id="bob-session",
+        )
+        await bob.before_run(
+            agent=cast(Any, None),
+            session=bob_session,
+            context=bob_ctx,
+            state=bob_session.state.setdefault(bob.source_id, {}),
+        )  # type: ignore[arg-type]
+
+        for call in client.search.await_args_list:
+            assert call.kwargs["filters"].get("user_id") == "bob"
+
+        retrieved = "".join(message.text or "" for message in bob_ctx.context_messages.get("mem0", []))
+        assert "4111-1111-1111-1111" not in retrieved
+        assert "mem0" not in bob_ctx.context_messages
+
+    async def test_own_memories_are_still_retrieved(self) -> None:
+        """The isolation fix must not regress retrieval of the user's own memories."""
+        client = self._fake_partition_client()
+
+        alice = Mem0ContextProvider(
+            source_id="mem0",
+            mem0_client=client,
+            user_id="alice",
+            agent_id="support-bot",
+            search_user_id="alice",
+        )
+
+        write_session = AgentSession(session_id="alice-session")
+        write_ctx = SessionContext(
+            input_messages=[Message(role="user", contents=["my favourite colour is teal"])],
+            session_id="alice-session",
+        )
+        await alice.after_run(
+            agent=cast(Any, None),
+            session=write_session,
+            context=write_ctx,
+            state=write_session.state.setdefault(alice.source_id, {}),
+        )  # type: ignore[arg-type]
+
+        read_session = AgentSession(session_id="alice-session-2")
+        read_ctx = SessionContext(
+            input_messages=[Message(role="user", contents=["what is my favourite colour?"])],
+            session_id="alice-session-2",
+        )
+        await alice.before_run(
+            agent=cast(Any, None),
+            session=read_session,
+            context=read_ctx,
+            state=read_session.state.setdefault(alice.source_id, {}),
+        )  # type: ignore[arg-type]
+
+        retrieved = "".join(message.text or "" for message in read_ctx.context_messages["mem0"])
+        assert "teal" in retrieved
+
 
 # -- after_run tests -----------------------------------------------------------
 
@@ -517,6 +749,14 @@ def test_oss_passes_with_user_id(self, mock_oss_mem0_client: AsyncMock) -> None:
         provider = Mem0ContextProvider(source_id="mem0", mem0_client=mock_oss_mem0_client, user_id="u1")
         provider._validate_filters()
 
+    def test_oss_search_application_id_raises(self, mock_oss_mem0_client: AsyncMock) -> None:
+        """OSS client rejects the Platform-only search_application_id retrieval scope."""
+        provider = Mem0ContextProvider(
+            source_id="mem0", mem0_client=mock_oss_mem0_client, user_id="u1", search_application_id="app1"
+        )
+        with pytest.raises(ValueError, match="application_id is not supported"):
+            provider._validate_filters()
+
 
 # -- _build_search_kwargs tests -----------------------------------------------------
 
@@ -540,6 +780,8 @@ def test_all_params(self, mock_mem0_client: AsyncMock) -> None:
             user_id="u1",
             agent_id="a1",
             application_id="app1",
+            search_agent_id="a1",
+            search_application_id="app1",
         )
 
         # Test that app_id correctly merges with the isolated target entity
@@ -573,7 +815,11 @@ def test_no_run_id_in_search_filters(self, mock_mem0_client: AsyncMock) -> None:
     def test_oss_search_filters_reject_app_id(self, mock_oss_mem0_client: AsyncMock) -> None:
         """OSS search filters reject application_id because app_id is only supported by Platform."""
         provider = Mem0ContextProvider(
-            source_id="mem0", mem0_client=mock_oss_mem0_client, user_id="u1", application_id="app1"
+            source_id="mem0",
+            mem0_client=mock_oss_mem0_client,
+            user_id="u1",
+            application_id="app1",
+            search_application_id="app1",
         )
 
         with pytest.raises(ValueError, match="application_id is not supported"):
@@ -589,7 +835,10 @@ def test_empty_when_no_params(self, mock_mem0_client: AsyncMock) -> None:
 
     async def test_before_run_application_only_fallback(self, mock_mem0_client: AsyncMock) -> None:
         provider = Mem0ContextProvider(
-            source_id="mem0", mem0_client=mock_mem0_client, application_id="app_fallback_test"
+            source_id="mem0",
+            mem0_client=mock_mem0_client,
+            application_id="app_fallback_test",
+            search_application_id="app_fallback_test",
         )
 
         # Mock a valid message list and session container setup
diff --git a/python/samples/02-agents/context_providers/mem0/README.md b/python/samples/02-agents/context_providers/mem0/README.md
index c4a3cdbc612..a15c515ad4e 100644
--- a/python/samples/02-agents/context_providers/mem0/README.md
+++ b/python/samples/02-agents/context_providers/mem0/README.md
@@ -40,8 +40,20 @@ Set the following environment variables:
 
 ### Memory Scoping
 
-The Mem0 context provider supports scoping via identifiers:
+The Mem0 context provider keeps the **storage scope** and the **retrieval scope** separate.
+
+Storage scope — stamped onto every memory that is written:
 
 - **User scope** (`user_id`): Associate memories with a specific user, shared across all sessions
 - **Agent scope** (`agent_id`): Isolate memories per agent persona
-- **Application scope** (`application_id`): Associate memories with an application context
+- **Application scope** (`application_id`): Associate memories with an application context (Platform client only)
+
+Retrieval scope — used when searching for memories to inject into the context:
+
+- `search_user_id`, `search_agent_id`, `search_application_id`
+
+Retrieval scope values **never** inherit from the storage scope. If none of the `search_*`
+values are set, no memories are retrieved and a warning is logged. This is deliberate: a memory
+written under an `agent_id` shared by many users must not be read back for every user. Set
+`search_user_id` for per-user memory, and only set `search_agent_id` when the agent's memories
+are safe to share across all of its users.
diff --git a/python/samples/02-agents/context_providers/mem0/mem0_basic.py b/python/samples/02-agents/context_providers/mem0/mem0_basic.py
index 712d3b0491c..ce0086634e1 100644
--- a/python/samples/02-agents/context_providers/mem0/mem0_basic.py
+++ b/python/samples/02-agents/context_providers/mem0/mem0_basic.py
@@ -44,7 +44,7 @@ async def main() -> None:
             name="FriendlyAssistant",
             instructions="You are a friendly assistant.",
             tools=retrieve_company_report,
-            context_providers=[Mem0ContextProvider(source_id="mem0", user_id=user_id)],
+            context_providers=[Mem0ContextProvider(source_id="mem0", user_id=user_id, search_user_id=user_id)],
         ) as agent,
     ):
         # First ask the agent to retrieve a company report with no previous context.
diff --git a/python/samples/02-agents/context_providers/mem0/mem0_oss.py b/python/samples/02-agents/context_providers/mem0/mem0_oss.py
index a0c4bbb372f..c9744cde2c8 100644
--- a/python/samples/02-agents/context_providers/mem0/mem0_oss.py
+++ b/python/samples/02-agents/context_providers/mem0/mem0_oss.py
@@ -47,7 +47,11 @@ async def main() -> None:
             name="FriendlyAssistant",
             instructions="You are a friendly assistant.",
             tools=retrieve_company_report,
-            context_providers=[Mem0ContextProvider(source_id="mem0", user_id=user_id, mem0_client=local_mem0_client)],
+            context_providers=[
+                Mem0ContextProvider(
+                    source_id="mem0", user_id=user_id, search_user_id=user_id, mem0_client=local_mem0_client
+                )
+            ],
         ) as agent,
     ):
         # First ask the agent to retrieve a company report with no previous context.
diff --git a/python/samples/02-agents/context_providers/mem0/mem0_sessions.py b/python/samples/02-agents/context_providers/mem0/mem0_sessions.py
index 2ba6113d158..e992ca3a046 100644
--- a/python/samples/02-agents/context_providers/mem0/mem0_sessions.py
+++ b/python/samples/02-agents/context_providers/mem0/mem0_sessions.py
@@ -43,7 +43,11 @@ async def example_user_scoped_memory() -> None:
             context_providers=[
                 Mem0ContextProvider(
                     source_id="mem0",
+                    # Storage scope: memories are stamped with this user_id.
                     user_id=user_id,
+                    # Retrieval scope: only this user's memories are read back.
+                    # Retrieval scope is always explicit; it never inherits the storage scope.
+                    search_user_id=user_id,
                 )
             ],
         ) as user_agent,
@@ -82,6 +86,9 @@ async def example_agent_scoped_memory() -> None:
                 Mem0ContextProvider(
                     source_id="mem0",
                     agent_id="scoped_assistant",
+                    # Agent-wide retrieval is opt-in and returns memories written by any user
+                    # under this agent_id, so only use it for knowledge that is safe to share.
+                    search_agent_id="scoped_assistant",
                 )
             ],
         ) as scoped_agent,
@@ -119,6 +126,7 @@ async def example_multiple_agents() -> None:
                 Mem0ContextProvider(
                     source_id="mem0",
                     agent_id=agent_id_1,
+                    search_agent_id=agent_id_1,
                 )
             ],
         ) as personal_agent,
@@ -130,6 +138,7 @@ async def example_multiple_agents() -> None:
                 Mem0ContextProvider(
                     source_id="mem0",
                     agent_id=agent_id_2,
+                    search_agent_id=agent_id_2,
                 )
             ],
         ) as work_agent,