example: add proactive example

Author: evan-ak

examples/proactive/memory/config.py

@@ -0,0 +1,50 @@
+memorize_config = {
+    "memory_types": [
+        "record",
+    ],
+    "memory_type_prompts": {
+        "record": {
+            "objective": {
+                "ordinal": 10,
+                "prompt": "# Task Objective\nYou will be given a conversation between a user and an coding agent. Your goal is to extract detailed records for what are planed to do, and what have been done.",
+            },
+            "workflow": {
+                "ordinal": 10,
+                "prompt": "# Workflow\nRead through the conversation and extract records. You should expecially focus on:\n- What the user ask the agent to do\n- What plan does the agent suggest\n- What the agent has done",
+            },
+            "rules": {
+                "ordinal": -1,
+                "prompt": None,
+            },
+            "examples": {
+                "ordinal": 60,
+                "prompt": "# Example\n## Output\n<item>\n    <memory>\n        <content>The user ask the agent to generate a code example for fastapi</content>\n        <categories>\n            <category>todo</category>\n        </categories>\n    </memory>\n    <memory>\n        <content>The agent suggest to use the code example from the document</content>\n        <categories>\n            <category>todo</category>\n        </categories>\n    </memory>\n    <memory>\n        <content>The agent ask the user to specify the response type</content>\n        <categories>\n            <category>todo</category>\n        </categories>\n    </memory>\n</item>",
+            },
+        }
+    },
+    "memory_categories": [
+        {
+            "name": "todo",
+            "description": "This file traces the latest status of the task. All records should be included in this file.",
+            "target_length": None,
+            "custom_prompt": {
+                "objective": {
+                    "ordinal": 10,
+                    "prompt": "# Task Objective\nYou are a specialist in task management. You should update the markdown file to reflect the latest status of the task.",
+                },
+                "workflow": {
+                    "ordinal": 20,
+                    "prompt": "# Workflow\nRead through the existing markdown file and the new records. Then update the markdown file to reflect:\n- What existing tasks are completed\n- What new tasks are added\n- What tasks are still in progress",
+                },
+                "rules": {
+                    "ordinal": -1,
+                    "prompt": None,
+                },
+                "examples": {
+                    "ordinal": 50,
+                    "prompt": "# Example\n## Output\n```markdown\n# Task\n## Task Objective\nThe user ask the agent to generate a code example for fastapi\n## Workflow\nThe agent suggest to use the code example from the document\nThe agent ask the user to specify the response type\n```",
+                },
+            },
+        }
+    ],
+}

examples/proactive/memory/local/__init__.py

@@ -0,0 +1,55 @@
+import json
+import os
+from pathlib import Path
+from typing import Any
+
+import pendulum
+
+from memu.app import MemoryService
+
+from ..config import memorize_config
+
+USER_ID = "claude_user"
+
+
+async def dump_conversation_resource(
+    conversation_messages: list[dict[str, Any]],
+) -> str:
+    resource_data = {
+        "content": [
+            {
+                "role": message.get("role", "system"),
+                "content": {"text": message.get("content", "")},
+                "created_at": message.get("timestamp", pendulum.now().isoformat()),
+            }
+            for message in conversation_messages
+        ]
+    }
+    time_string = pendulum.now().format("YYYYMMDD_HHmmss")
+    resource_url = Path(__file__).parent / "data" / f"conv_{time_string}.json"
+    with open(resource_url, "w") as f:
+        json.dump(resource_data, f, indent=4, ensure_ascii=False)
+    return resource_url.as_posix()
+
+
+async def memorize(conversation_messages: list[dict[str, Any]]) -> str | None:
+    api_key = os.getenv("OPENAI_API_KEY")
+    if not api_key:
+        msg = "Please set OPENAI_API_KEY environment variable"
+        raise ValueError(msg)
+
+    memory_service = MemoryService(
+        llm_profiles={
+            "default": {
+                "api_key": api_key,
+                "chat_model": "gpt-4o-mini",
+            },
+        },
+        memorize_config=memorize_config,
+    )
+
+    resource_url = await dump_conversation_resource(conversation_messages)
+    result = await memory_service.memorize(
+        resource_url=resource_url, modality="conversation", user={"user_id": USER_ID}
+    )
+    return result

examples/proactive/memory/local/memorize.py

@@ -0,0 +1,70 @@
+import os
+from typing import Any
+
+from claude_agent_sdk import create_sdk_mcp_server, tool
+
+from memu.app import MemoryService
+
+USER_ID = "claude_user"
+
+
+@tool("memu_memory", "Retrieve memory based on a query", {"query": str})
+async def get_memory(args: dict[str, Any]) -> dict[str, Any]:
+    """Retrieve memory from the memory API based on the provided query."""
+    query = {"role": "user", "content": args["query"]}
+
+    api_key = os.getenv("OPENAI_API_KEY")
+    if not api_key:
+        msg = "Please set OPENAI_API_KEY environment variable"
+        raise ValueError(msg)
+
+    memory_service = MemoryService(
+        llm_profiles={
+            "default": {
+                "api_key": api_key,
+                "chat_model": "gpt-4o-mini",
+            },
+        },
+        retrieve_config={
+            "method": "rag",
+            "route_intention": False,
+            "sufficiency_check": False,
+            "category": {
+                "enabled": False,
+            },
+            "item": {
+                "enabled": True,
+                "top_k": 10,
+            },
+            "resource": {
+                "enabled": False,
+            },
+        },
+    )
+
+    result = await memory_service.retrieve(query, where={"user_id": USER_ID})
+
+    return {"content": [{"type": "text", "text": str(result)}]}
+
+
+async def _get_todos() -> str:
+    memory_service = MemoryService()
+
+    result = await memory_service.list_memory_categories(where={"user_id": USER_ID})
+
+    categories = result["categories"]
+    todos = ""
+    for category in categories:
+        if category["name"] == "todo":
+            todos = category["summary"]
+    return todos
+
+
+@tool("memu_todos", "Retrieve todos for the user", {})
+async def get_todos() -> dict[str, Any]:
+    """Retrieve todos from the memory API."""
+    todos = await _get_todos()
+    return {"content": [{"type": "text", "text": str(todos)}]}
+
+
+memu_server = create_sdk_mcp_server(name="memu", version="1.0.0", tools=[get_memory, get_todos])

examples/proactive/memory/local/tools.py

@@ -0,0 +1,31 @@
+from typing import Any
+
+import aiohttp
+
+BASE_URL = "https://api.memu.so"
+API_KEY = "your memu api key"
+USER_ID = "claude_user"
+AGENT_ID = "claude_agent"
+
+from ..config import memorize_config
+
+
+async def memorize(conversation_messages: list[dict[str, Any]]) -> str | None:
+    payload = {
+        "conversation": conversation_messages,
+        "user_id": USER_ID,
+        "agent_id": AGENT_ID,
+        "override_config": memorize_config,
+    }
+
+    async with (
+        aiohttp.ClientSession() as session,
+        session.post(
+            f"{BASE_URL}/api/v3/memory/memorize",
+            headers={"Authorization": f"Bearer {API_KEY}"},
+            json=payload,
+        ) as response,
+    ):
+        result = await response.json()
+        task_id = result["task_id"]
+        return task_id

examples/proactive/memory/platform/__init__.py

@@ -0,0 +1,52 @@
+from typing import Any
+
+import aiohttp
+from claude_agent_sdk import create_sdk_mcp_server, tool
+
+BASE_URL = "https://api.memu.so"
+API_KEY = "your memu api key"
+USER_ID = "claude_user"
+AGENT_ID = "claude_agent"
+
+
+@tool("memu_memory", "Retrieve memory based on a query", {"query": str})
+async def get_memory(args: dict[str, Any]) -> dict[str, Any]:
+    """Retrieve memory from the memory API based on the provided query."""
+    query = args["query"]
+    url = f"{BASE_URL}/api/v3/memory/retrieve"
+    headers = {"Authorization": f"Bearer {API_KEY}"}
+    data = {"user_id": USER_ID, "agent_id": AGENT_ID, "query": query}
+
+    async with aiohttp.ClientSession() as session, session.post(url, headers=headers, json=data) as response:
+        result = await response.json()
+
+    return {"content": [{"type": "text", "text": str(result)}]}
+
+
+async def _get_todos() -> str:
+    url = f"{BASE_URL}/api/v3/memory/categories"
+    headers = {"Authorization": f"Bearer {API_KEY}"}
+    data = {
+        "user_id": USER_ID,
+        "agent_id": AGENT_ID,
+    }
+    async with aiohttp.ClientSession() as session, session.post(url, headers=headers, json=data) as response:
+        result = await response.json()
+
+    categories = result["categories"]
+    todos = ""
+    for category in categories:
+        if category["name"] == "todo":
+            todos = category["summary"]
+    return todos
+
+
+@tool("memu_todos", "Retrieve todos for the user", {})
+async def get_todos() -> dict[str, Any]:
+    """Retrieve todos from the memory API."""
+    todos = await _get_todos()
+    return {"content": [{"type": "text", "text": str(todos)}]}
+
+
+# Create the MCP server with the tool
+memu_server = create_sdk_mcp_server(name="memu", version="1.0.0", tools=[get_memory, get_todos])

examples/proactive/memory/platform/memorize.py

@@ -0,0 +1,117 @@
+import asyncio
+
+from claude_agent_sdk import (
+    AssistantMessage,
+    ClaudeAgentOptions,
+    ClaudeSDKClient,
+    ResultMessage,
+    TextBlock,
+)
+from memory.local.memorize import memorize
+from memory.local.tools import _get_todos, memu_server
+
+# Set your Anthropic API key here if it's not set in the environment variables
+# os.environ["ANTHROPIC_API_KEY"] = ""
+
+N_MESSAGES_MEMORIZE = 2
+
+
+async def trigger_memorize(messages: list[dict[str, any]]) -> bool:
+    """Background task to memorize conversation messages."""
+    try:
+        await memorize(messages)
+        print("\n[Background] Memorization submitted.")
+        return True
+    except Exception as e:
+        print(f"\n[Background] Memorization failed: {e!r}")
+        return False
+
+
+async def main():
+    options = ClaudeAgentOptions(
+        mcp_servers={"memu": memu_server},
+        allowed_tools=[
+            # "mcp__memu__memu_memory",
+            "mcp__memu__memu_todos",
+        ],
+    )
+
+    conversation_messages: list[dict[str, any]] = []
+    pending_tasks: list[asyncio.Task] = []
+
+    print("Claude Autorun")
+    print("Type 'quit' or 'exit' to end the session.")
+    print("-" * 40)
+
+    round = 0
+    async with ClaudeSDKClient(options=options) as client:
+        while True:
+            want_user_input = False
+
+            if round == 0:
+                want_user_input = True
+            else:
+                todos = await _get_todos()
+                if todos:
+                    user_input = f"Please continue with the following todos:\n{todos}"
+                else:
+                    want_user_input = True
+
+            if want_user_input:
+                try:
+                    user_input = input("\nYou: ").strip()
+                except EOFError:
+                    break
+
+                if not user_input:
+                    continue
+
+                if user_input.lower() in ("quit", "exit"):
+                    break
+
+            # Record user message
+            conversation_messages.append({"role": "user", "content": user_input})
+
+            # Send query to Claude
+            await client.query(user_input)
+
+            # Collect assistant response
+            assistant_text_parts: list[str] = []
+
+            async for message in client.receive_response():
+                if isinstance(message, AssistantMessage):
+                    for block in message.content:
+                        if isinstance(block, TextBlock):
+                            print(f"Claude: {block.text}")
+                            assistant_text_parts.append(block.text)
+                elif isinstance(message, ResultMessage):
+                    print(f"Result: {message.result}")
+
+            # Record assistant message
+            if assistant_text_parts:
+                conversation_messages.append({"role": "assistant", "content": "\n".join(assistant_text_parts)})
+
+            # Check if we should trigger memorization
+            if len(conversation_messages) >= N_MESSAGES_MEMORIZE:
+                print(f"\n[Info] Reached {N_MESSAGES_MEMORIZE} messages, triggering memorization...")
+                success = await trigger_memorize(conversation_messages.copy())
+                if success:
+                    conversation_messages.clear()
+
+            round += 1
+
+    # User quit - memorize remaining messages if any
+    if conversation_messages:
+        print("\n[Info] Session ended, memorizing remaining messages...")
+        success = await trigger_memorize(conversation_messages.copy())
+
+    # Wait for all pending memorization tasks to complete
+    if pending_tasks:
+        print("[Info] Waiting for memorization tasks to complete...")
+        await asyncio.gather(*pending_tasks, return_exceptions=True)
+
+    print("\nDone")
+
+
+if __name__ == "__main__":
+    asyncio.run(main())