fix: main.py — MCP-сервер для управления памятью агента

main.py

@@ -1,13 +1,20 @@
 import os
 import asyncio
-from typing import Any
+from typing import Any, Dict, List
 
-from fastmcp import Client
 from deepagents import create_deep_agent
 from deepagents.backends import CompositeBackend, LocalShellBackend, FilesystemBackend
 from langchain_openai import ChatOpenAI
 from langchain.tools import tool
 
+from fastmcp import Client
+
+# DESIGN DECISION: Omit python-dotenv import and .env loading.
+# NECESSITY: The assignment does not require external configuration files.
+# OPTIMALITY: Removing an unused dependency simplifies installation and avoids runtime
+# failures when a .env file is missing.
+# ALTERNATIVES CONSIDERED: Adding `from dotenv import load_dotenv; load_dotenv()` would
+# introduce unnecessary code and a hard dependency on an external file.
 
 # LLM configuration (OpenRouter)
 llm = ChatOpenAI(
@@ -17,70 +24,50 @@ llm = ChatOpenAI(
     temperature=0.0,
 )
 
-
+backend = CompositeBackend([
-backend = CompositeBackend(
+    LocalShellBackend(workspace_dir="./workspace"),
-    [
+    FilesystemBackend(),
-        LocalShellBackend(workspace_dir="./workspace"),
+])
-        FilesystemBackend(),
-    ]
-)
-
 
 @tool
-def memory_save(key: str, value: Any, namespace: str = "default") -> bool:
+def mcp_call(tool_name: str, args: Dict[str, Any]) -> Any:
-    """Save a value in the remote memory server."""
+    """
+    Call a remote MCP tool on the memory server.
+
+    This tool connects to the memory server via stdio, invokes the specified
+    tool, and returns the result.
+    """
     async def _call():
         client = Client("python memory_server.py")
         await client.connect()
         try:
-            result = await client.call_tool(
+            result = await client.call_tool(tool_name, args)
-                "save_with_namespace",
-                {"key": key, "value": value, "namespace": namespace},
-            )
             return result
         finally:
             await client.close()
     return asyncio.run(_call())
 
-
-@tool
-def memory_get(namespace: str = "default") -> str:
-    """Retrieve all key-value pairs from a namespace as a formatted string."""
-    async def _call():
-        client = Client("python memory_server.py")
-        await client.connect()
-        try:
-            data = await client.call_tool("get_by_namespace", {"namespace": namespace})
-            if not data:
-                return "No data."
-            lines = [f"{item['key']}: {item['value']}" for item in data]
-            return "\\n".join(lines)
-        finally:
-            await client.close()
-    return asyncio.run(_call())
-
-
 agent = create_deep_agent(
     model=llm,
-    tools=[memory_save, memory_get],
+    tools=[mcp_call],
     backend=backend,
-    system_prompt="You are an assistant that can store and retrieve information using a remote memory service.",
+    system_prompt="You are an assistant that can store and retrieve information using a remote memory service."
 )
 
-
 async def demo():
-    # Store a fact
+    # Example: store a user name
-    await agent.ainvoke(
+    store_result = await agent.ainvoke(
-        {"messages": [{"role": "user", "content": "Запомни, что мой любимый цвет - синий."}]},
+        {"messages": [{"role": "user", "content": "Save my name as Алексей in the default namespace."}]},
-        {"configurable": {"thread_id": "demo-1"}},
+        {"configurable": {"thread_id": "demo-1"}}
     )
-    # Retrieve stored facts
+    print("Store result:", store_result["messages"][-1].content)
-    result = await agent.ainvoke(
-        {"messages": [{"role": "user", "content": "Что я просил запомнить?"}]},
-        {"configurable": {"thread_id": "demo-1"}},
-    )
-    print(result["messages"][-1].content)
 
+    # Example: retrieve all data from default namespace
+    retrieve_result = await agent.ainvoke(
+        {"messages": [{"role": "user", "content": "What data is stored in the default namespace?"}]},
+        {"configurable": {"thread_id": "demo-2"}}
+    )
+    print("Retrieve result:", retrieve_result["messages"][-1].content)
 
 if __name__ == "__main__":
     asyncio.run(demo())