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

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

@@ -1,23 +1,28 @@
 import asyncio
-from fastmcp import MCPServer, Tool
+from fastmcp import FastMCP
-from typing import Dict, Any
+from typing import Any
 
 # Внутреннее хранилище памяти
-memory_store: Dict[str, Any] = {}
+memory_store: dict[str, Any] = {}
-namespace_store: Dict[str, Dict[str, Any]] = {}
+namespace_store: dict[str, dict[str, Any]] = {}
+
+mcp = FastMCP("memory_server")
 
 
-def save(key: str, value: Any) -> str:
+@mcp.tool()
+def save(key: str, value: str) -> str:
     """Сохраняет значение по ключу."""
     memory_store[key] = value
     return f"Значение сохранено под ключом '{key}'."
 
 
-def get(key: str) -> Any:
+@mcp.tool()
-    """Возвращает значение по ключу. Если ключ не найден, возвращается None."""
+def get(key: str) -> str | None:
+    """Возвращает значение по ключу. Если ключ не найден, возвращает None."""
     return memory_store.get(key)
 
 
+@mcp.tool()
 def delete(key: str) -> str:
     """Удаляет значение по ключу."""
     if key in memory_store:
@@ -26,37 +31,26 @@ def delete(key: str) -> str:
     return f"Ключ '{key}' не найден."
 
 
-def list_keys() -> Dict[str, Any]:
+@mcp.tool()
+def list_keys() -> dict[str, list[str]]:
     """Возвращает список всех сохранённых ключей."""
     return {"keys": list(memory_store.keys())}
 
 
-def save_with_namespace(namespace: str, key: str, value: Any) -> str:
+@mcp.tool()
+def save_with_namespace(namespace: str, key: str, value: str) -> str:
     """Сохраняет значение в указанном пространстве имён."""
     ns = namespace_store.setdefault(namespace, {})
     ns[key] = value
     return f"Значение сохранено под ключом '{key}' в пространстве имён '{namespace}'."
 
-def get_by_namespace(namespace: str, key: str) -> Any:
+
+@mcp.tool()
+def get_by_namespace(namespace: str, key: str) -> str | None:
     """Получает значение из указанного пространства имён."""
     ns = namespace_store.get(namespace, {})
     return ns.get(key)
 
 
-# Создаём сервер MCP
-server = MCPServer(
-    name="memory_server",
-    description="Сервер для управления памятью агента.",
-)
-
-# Регистрируем инструменты
-server.register_tool(Tool(name="save", func=save))
-server.register_tool(Tool(name="get", func=get))
-server.register_tool(Tool(name="delete", func=delete))
-server.register_tool(Tool(name="list_keys", func=list_keys))
-server.register_tool(Tool(name="save_with_namespace", func=save_with_namespace))
-server.register_tool(Tool(name="get_by_namespace", func=get_by_namespace))
-
 if __name__ == "__main__":
-    # Запускаем сервер
+    mcp.run()
-    asyncio.run(server.serve())