fix(strands): keep example on default extraction cadence

Use production-default extraction=True with a session-boundary flush
instead of an every-turn InvocationTrigger in the example and live tests.

Co-authored-by: Cursor <cursoragent@cursor.com>

Author: mehulp93

integrations/strands/python/CHANGELOG.md

@@ -21,8 +21,8 @@
 
 - Documented that Strands' default extraction cadence is every **5 turns**, so conversation batches only reach Zep when the trigger fires (or on `flush()`), delaying graph building relative to turn-by-turn persistence — in addition to Zep's asynchronous ingestion after messages arrive.
 - Documented the failure-handling contract: Zep SDK errors propagate out of `search`/`add`/`add_messages` by design, because `MemoryManager` and `ExtractionCoordinator` own failure isolation (skip-and-log, `AggregateMemoryError`, and high-water-mark rollback for retry). Added tests pinning that behavior.
-- Example and live tests use every-turn extraction (`InvocationTrigger`) plus
-  `flush()` after `invoke_async` so demos are not flaky on the default cadence.
+- Example and live tests flush at the session boundary after `invoke_async`,
+  which `MemoryManager` requires to persist buffered turns on that path.
 
 ## 0.1.0 (2026-08-01)
 

integrations/strands/python/examples/basic_agent.py

@@ -1,15 +1,15 @@
 """
 Basic Strands Agents agent with Zep long-term memory via MemoryManager.
 
-Wires a ``ZepMemoryStore`` into Strands' ``MemoryManager`` so the agent:
+This example wires a ``ZepMemoryStore`` into Strands' ``MemoryManager`` so the
+agent automatically:
 
 * injects relevant Zep context before each model call
-* extracts conversation turns into the user graph (server-side via
-  ``add_messages``) on an every-turn trigger, then ``flush()`` after each
-  ``invoke_async`` so nothing is left buffered
+* extracts conversation turns into the user graph (server-side, via
+  ``add_messages``) on the manager's default cadence
 
-Earlier turns seed facts about the user; a later turn on a *new* thread shows
-cross-session recall from Zep's user graph.
+Earlier turns seed facts about the user; a later turn -- in a *new* conversation
+thread -- shows the agent recalling those facts from Zep's user graph.
 
 Prerequisites:
     pip install zep-strands 'strands-agents[openai]'
@@ -26,13 +26,27 @@
 
 from strands import Agent
 from strands.memory import MemoryManager
-from strands.memory.extraction.triggers import InvocationTrigger
-from strands.memory.extraction.types import ExtractionConfig
 from strands.types.content import Message
 from zep_cloud.client import AsyncZep
 
 from zep_strands import ZepMemoryStore, ensure_thread, ensure_user
 
+
+def _message_text(message: Message | object) -> str:
+    """Extract joined text blocks from an agent result message."""
+    if not isinstance(message, dict):
+        return str(message)
+    parts = [
+        block["text"]
+        for block in message.get("content") or []
+        if isinstance(block, dict) and "text" in block
+    ]
+    return "\n".join(parts) if parts else str(message)
+
+
+# ---------------------------------------------------------------------------
+# Configuration
+# ---------------------------------------------------------------------------
 ZEP_API_KEY = os.environ.get("ZEP_API_KEY", "")
 OPENAI_API_KEY = os.environ.get("OPENAI_API_KEY", "")
 
@@ -47,18 +61,6 @@
 THREAD_2 = f"strands-example-thread2-{_suffix}"
 
 
-def _message_text(message: Message | object) -> str:
-    """Extract joined text blocks from an agent result message."""
-    if not isinstance(message, dict):
-        return str(message)
-    parts = [
-        block["text"]
-        for block in message.get("content") or []
-        if isinstance(block, dict) and "text" in block
-    ]
-    return "\n".join(parts) if parts else str(message)
-
-
 async def build_agent(zep: AsyncZep, thread_id: str) -> Agent:
     """Build an agent whose memory is scoped to USER_ID on the given thread."""
     await ensure_user(
@@ -78,9 +80,7 @@ async def build_agent(zep: AsyncZep, thread_id: str) -> Agent:
         last_name="Nguyen",
         email="alice@example.com",
         writable=True,
-        # Every-turn extraction so the demo does not wait on the default
-        # 5-turn cadence; still flush after invoke_async (see main).
-        extraction=ExtractionConfig(trigger=InvocationTrigger()),
+        extraction=True,
         expose_search_tool=True,
         search_pinned_params={"scope": "auto"},
     )
@@ -94,14 +94,6 @@ async def build_agent(zep: AsyncZep, thread_id: str) -> Agent:
     )
 
 
-async def chat(agent: Agent, message: str) -> str:
-    """Send one message and flush pending extraction writes."""
-    result = await agent.invoke_async(message)
-    if agent.memory_manager is not None:
-        await agent.memory_manager.flush()
-    return _message_text(result.message)
-
-
 async def main() -> None:
     zep = AsyncZep(api_key=ZEP_API_KEY)
 
@@ -115,12 +107,19 @@ async def main() -> None:
 
     print("--- Conversation 1: seeding facts ---\n")
     agent1 = await build_agent(zep, THREAD_1)
-    for message in (
+    seed_messages = [
         "Hi! I'm Alice, a data scientist living in Portland, Oregon.",
         "On weekends I love hiking and landscape photography.",
-    ):
+    ]
+    for message in seed_messages:
         print(f"User:  {message}")
-        print(f"Agent: {await chat(agent1, message)}\n")
+        result = await agent1.invoke_async(message)
+        print(f"Agent: {_message_text(result.message)}\n")
+
+    # Flush at the session boundary: invoke_async does not flush, and flushing
+    # every turn would defeat the extraction trigger's schedule.
+    if agent1.memory_manager is not None:
+        await agent1.memory_manager.flush()
 
     wait_seconds = 20
     print(f"--- Waiting {wait_seconds}s for Zep to process the graph ---\n")
@@ -130,7 +129,8 @@ async def main() -> None:
     agent2 = await build_agent(zep, THREAD_2)
     recall = "Where do I live, and what do I like to do on weekends?"
     print(f"User:  {recall}")
-    print(f"Agent: {await chat(agent2, recall)}\n")
+    result = await agent2.invoke_async(recall)
+    print(f"Agent: {_message_text(result.message)}\n")
 
 
 if __name__ == "__main__":

integrations/strands/python/tests/test_integration.py

@@ -45,8 +45,6 @@
 
 from strands import Agent  # noqa: E402
 from strands.memory import MemoryManager  # noqa: E402
-from strands.memory.extraction.triggers import InvocationTrigger  # noqa: E402
-from strands.memory.extraction.types import ExtractionConfig  # noqa: E402
 from strands.types.content import Message  # noqa: E402
 from zep_cloud.client import AsyncZep  # noqa: E402
 
@@ -129,8 +127,8 @@ async def build_agent(
 ) -> Agent:
     """Build an agent whose memory is scoped to USER_ID on the given thread.
 
-    Uses every-turn extraction so the test is not flaky on Strands' default
-    5-turn cadence; callers still ``flush()`` after ``invoke_async``.
+    Configured exactly as the example is: default extraction cadence, with
+    callers flushing at the session boundary.
     """
     await ensure_user(
         zep,
@@ -150,7 +148,7 @@ async def build_agent(
         last_name=LAST_NAME,
         email=EMAIL,
         writable=True,
-        extraction=ExtractionConfig(trigger=InvocationTrigger()),
+        extraction=True,
         expose_search_tool=True,
         search_pinned_params={"scope": "auto"},
     )
@@ -165,11 +163,19 @@ async def build_agent(
 
 
 async def chat(agent: Agent, message: str) -> str:
-    """Send one message and flush pending extraction writes."""
+    """Send one message to the agent and return its text reply."""
     result = await agent.invoke_async(message)
+    return _message_text(result.message)
+
+
+async def flush(agent: Agent) -> None:
+    """Force buffered extraction writes out at a session boundary.
+
+    ``invoke_async`` never flushes on its own, and the default trigger only
+    fires every 5 turns, so without this the seeded turns never reach Zep.
+    """
     if agent.memory_manager is not None:
         await agent.memory_manager.flush()
-    return _message_text(result.message)
 
 
 # ---------------------------------------------------------------------------
@@ -261,6 +267,7 @@ async def on_user_created(client: AsyncZep, user_id: str) -> None:
             agent1, "My name is IntegTest. I work at Acme Corp as a data scientist."
         )
         reply2 = await chat(agent1, "I live in Portland, Oregon and I love hiking and photography.")
+        await flush(agent1)
         assert reply1
         assert reply2
         assert hook_calls == [USER_ID]
@@ -325,6 +332,8 @@ async def on_user_created(client: AsyncZep, user_id: str) -> None:
             print(f"  Agent: {reply}\n")
             passed &= check("Agent returned a non-empty response", len(reply) > 0)
 
+        await flush(agent1)
+
         passed &= check(
             "on_user_created hook fired exactly once",
             len(hook_calls) == 1 and hook_calls[0] == USER_ID,