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

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

@@ -1,104 +1,109 @@
 import json
-from typing import Dict, List, Optional
+from typing import Any, Dict, List
 
-from fastmcp import FastMCP
+from fastmcp import FastMCP, tool
-from fastmcp.exceptions import RPCError
-from pydantic import BaseModel
 
 
-class SaveRequest(BaseModel):
+class MemoryStore:
-    key: str
+    """
-    value: str
+    Простое хранилище в памяти с поддержкой пространств имён.
+    """
+
+    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)
 
 
-class GetRequest(BaseModel):
+# Создаём экземпляр хранилища
-    key: str
+memory_store = MemoryStore()
 
 
-class DeleteRequest(BaseModel):
+@tool(name="save", description="Сохраняет значение по ключу")
-    key: str
+def tool_save(key: str, value: Any) -> None:
+    memory_store.save(key, value)
+    return f"Ключ '{key}' сохранён."
 
 
-class ListKeysResponse(BaseModel):
+@tool(name="get", description="Получает значение по ключу")
-    keys: List[str]
+def tool_get(key: str) -> Any | None:
+    result = memory_store.get(key)
+    if result is None:
+        return f"Ключ '{key}' не найден."
+    return result
 
 
-class NamespaceSaveRequest(BaseModel):
+@tool(name="delete", description="Удаляет ключ из памяти")
-    namespace: str
+def tool_delete(key: str) -> bool:
-    key: str
+    deleted = memory_store.delete(key)
-    value: str
+    return deleted
 
 
-class NamespaceGetRequest(BaseModel):
+@tool(name="list_keys", description="Возвращает список всех ключей в памяти")
-    namespace: str
+def tool_list_keys() -> List[str]:
-    key: str
+    return memory_store.list_keys()
 
 
-# In‑memory store.  Keys are strings, values are arbitrary JSON serialisable objects.
+@tool(name="save_with_namespace", description="Сохраняет значение под ключом внутри namespace")
-_memory_store: Dict[str, any] = {}
+def tool_save_with_namespace(namespace: str, key: str, value: Any) -> None:
-# Namespaces are dictionaries inside the main store.
+    memory_store.save_with_namespace(namespace, key, value)
-_namespaces: Dict[str, Dict[str, any]] = {}
+    return f"Ключ '{key}' в namespace '{namespace}' сохранён."
 
 
-def _ensure_namespace(namespace: str):
+@tool(name="get_by_namespace", description="Получает значение по ключу из namespace")
-    if namespace not in _namespaces:
+def tool_get_by_namespace(namespace: str, key: str) -> Any | None:
-        _namespaces[namespace] = {}
+    result = memory_store.get_by_namespace(namespace, key)
+    if result is None:
+        return f"Ключ '{key}' в namespace '{namespace}' не найден."
+    return result
 
 
-app = FastMCP()
+# Создаём FastMCP приложение
+app = FastMCP(
+    name="MemoryServer",
+    description="FastMCP сервер для управления памятью агента",
+)
 
-
+# Регистрируем все инструменты
-@app.rpc("save")
+app.register_tool(tool_save)
-async def save(request: SaveRequest) -> None:
+app.register_tool(tool_get)
-    """Сохраняет значение по ключу."""
+app.register_tool(tool_delete)
-    _memory_store[request.key] = request.value
+app.register_tool(tool_list_keys)
-
+app.register_tool(tool_save_with_namespace)
-
+app.register_tool(tool_get_by_namespace)
-@app.rpc("get")
-async def get(request: GetRequest):
-    """Возвращает сохранённое значение. Если ключа нет – ошибка."""
-    if request.key not in _memory_store:
-        raise RPCError(f"Key '{request.key}' not found")
-    return _memory_store[request.key]
-
-
-@app.rpc("delete")
-async def delete(request: DeleteRequest) -> None:
-    """Удаляет ключ из памяти."""
-    if request.key in _memory_store:
-        del _memory_store[request.key]
-    else:
-        raise RPCError(f"Key '{request.key}' not found")
-
-
-@app.rpc("list_keys")
-async def list_keys() -> ListKeysResponse:
-    """Возвращает список всех ключей."""
-    return ListKeysResponse(keys=list(_memory_store.keys()))
-
-
-@app.rpc("save_with_namespace")
-async def save_with_namespace(request: NamespaceSaveRequest) -> None:
-    """Сохраняет значение в указанном пространстве имён."""
-    _ensure_namespace(request.namespace)
-    _namespaces[request.namespace][request.key] = request.value
-
-
-@app.rpc("get_by_namespace")
-async def get_by_namespace(request: NamespaceGetRequest):
-    """Получает значение из пространства имён. Ошибка, если не найдено."""
-    if request.namespace not in _namespaces:
-        raise RPCError(f"Namespace '{request.namespace}' does not exist")
-    ns = _namespaces[request.namespace]
-    if request.key not in ns:
-        raise RPCError(
-            f"Key '{request.key}' not found in namespace '{request.namespace}'"
-        )
-    return ns[request.key]
 
 
 if __name__ == "__main__":
-    # Запускаем сервер FastMCP
+    # Запускаем сервер на порту 8000
-    import uvicorn
+    app.run(host="0.0.0.0", port=8000)
-
-    uvicorn.run(app, host="0.0.0.0", port=8000)