feat: add proactive example (#268)

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
+
+from ..config import memorize_config
+
+BASE_URL = "https://api.memu.so"
+API_KEY = "your memu api key"
+USER_ID = "claude_user"
+AGENT_ID = "claude_agent"
+
+
+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])