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

solutions/69f8e929da860fb4533faa2a_MCP-сервер_для_управления_памятью_агента/memory_server.py

@@ -1,109 +1,62 @@
-import json
+import asyncio
-from typing import Any, Dict, List
+from fastmcp import MCPServer, Tool
+from typing import Dict, Any
 
-from fastmcp import FastMCP, tool
+# Внутреннее хранилище памяти
+memory_store: Dict[str, Any] = {}
+namespace_store: Dict[str, Dict[str, Any]] = {}
 
 
-class MemoryStore:
+def save(key: str, value: Any) -> str:
-    """
+    """Сохраняет значение по ключу."""
-    Простое хранилище в памяти с поддержкой пространств имён.
+    memory_store[key] = value
-    """
+    return f"Значение сохранено под ключом '{key}'."
-
-    def __init__(self) -> None:
-        # основной словарь: ключ → значение
-        self._store: Dict[str, Any] = {}
-        # словари по namespace: namespace → {key: value}
-        self._namespaces: Dict[str, Dict[str, Any]] = {}
-
-    # ---------- Основные операции ----------
-    def save(self, key: str, value: Any) -> None:
-        """Сохраняет значение под ключом."""
-        self._store[key] = value
-
-    def get(self, key: str) -> Any | None:
-        """Возвращает значение по ключу или None, если его нет."""
-        return self._store.get(key)
-
-    def delete(self, key: str) -> bool:
-        """Удаляет ключ. Возвращает True, если удалено."""
-        if key in self._store:
-            del self._store[key]
-            return True
-        return False
-
-    def list_keys(self) -> List[str]:
-        """Возвращает список всех ключей."""
-        return list(self._store.keys())
-
-    # ---------- Операции с namespace ----------
-    def save_with_namespace(self, namespace: str, key: str, value: Any) -> None:
-        """Сохраняет значение в указанном пространстве имён."""
-        ns = self._namespaces.setdefault(namespace, {})
-        ns[key] = value
-
-    def get_by_namespace(self, namespace: str, key: str) -> Any | None:
-        """Получает значение из пространства имён."""
-        return self._namespaces.get(namespace, {}).get(key)
 
 
-# Создаём экземпляр хранилища
+def get(key: str) -> Any:
-memory_store = MemoryStore()
+    """Возвращает значение по ключу. Если ключ не найден, возвращается None."""
+    return memory_store.get(key)
 
 
-@tool(name="save", description="Сохраняет значение по ключу")
+def delete(key: str) -> str:
-def tool_save(key: str, value: Any) -> None:
+    """Удаляет значение по ключу."""
-    memory_store.save(key, value)
+    if key in memory_store:
-    return f"Ключ '{key}' сохранён."
+        del memory_store[key]
-
+        return f"Ключ '{key}' удалён."
-
-@tool(name="get", description="Получает значение по ключу")
-def tool_get(key: str) -> Any | None:
-    result = memory_store.get(key)
-    if result is None:
     return f"Ключ '{key}' не найден."
-    return result
 
 
-@tool(name="delete", description="Удаляет ключ из памяти")
+def list_keys() -> Dict[str, Any]:
-def tool_delete(key: str) -> bool:
+    """Возвращает список всех сохранённых ключей."""
-    deleted = memory_store.delete(key)
+    return {"keys": list(memory_store.keys())}
-    return deleted
 
 
-@tool(name="list_keys", description="Возвращает список всех ключей в памяти")
+def save_with_namespace(namespace: str, key: str, value: Any) -> str:
-def tool_list_keys() -> List[str]:
+    """Сохраняет значение в указанном пространстве имён."""
-    return memory_store.list_keys()
+    ns = namespace_store.setdefault(namespace, {})
+    ns[key] = value
+    return f"Значение сохранено под ключом '{key}' в пространстве имён '{namespace}'."
+
+def get_by_namespace(namespace: str, key: str) -> Any:
+    """Получает значение из указанного пространства имён."""
+    ns = namespace_store.get(namespace, {})
+    return ns.get(key)
 
 
-@tool(name="save_with_namespace", description="Сохраняет значение под ключом внутри namespace")
+# Создаём сервер MCP
-def tool_save_with_namespace(namespace: str, key: str, value: Any) -> None:
+server = MCPServer(
-    memory_store.save_with_namespace(namespace, key, value)
+    name="memory_server",
-    return f"Ключ '{key}' в namespace '{namespace}' сохранён."
+    description="Сервер для управления памятью агента.",
-
-
-@tool(name="get_by_namespace", description="Получает значение по ключу из namespace")
-def tool_get_by_namespace(namespace: str, key: str) -> Any | None:
-    result = memory_store.get_by_namespace(namespace, key)
-    if result is None:
-        return f"Ключ '{key}' в namespace '{namespace}' не найден."
-    return result
-
-
-# Создаём FastMCP приложение
-app = FastMCP(
-    name="MemoryServer",
-    description="FastMCP сервер для управления памятью агента",
 )
 
-# Регистрируем все инструменты
+# Регистрируем инструменты
-app.register_tool(tool_save)
+server.register_tool(Tool(name="save", func=save))
-app.register_tool(tool_get)
+server.register_tool(Tool(name="get", func=get))
-app.register_tool(tool_delete)
+server.register_tool(Tool(name="delete", func=delete))
-app.register_tool(tool_list_keys)
+server.register_tool(Tool(name="list_keys", func=list_keys))
-app.register_tool(tool_save_with_namespace)
+server.register_tool(Tool(name="save_with_namespace", func=save_with_namespace))
-app.register_tool(tool_get_by_namespace)
+server.register_tool(Tool(name="get_by_namespace", func=get_by_namespace))
-
 
 if __name__ == "__main__":
-    # Запускаем сервер на порту 8000
+    # Запускаем сервер
-    app.run(host="0.0.0.0", port=8000)
+    asyncio.run(server.serve())