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

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

@@ -1,104 +1,87 @@
-import json
+<|channel|>final code<|message|>import json
+import sys
 from pathlib import Path
 
 import httpx
-from rich.console import Console
-from rich.table import Table
-
-console = Console()
-BASE_URL = "http://127.0.0.1:8000"
 
 
-def save(key: str, value: str) -> None:
+BASE_URL = "http://localhost:8000"
+
+
+def save(key: str, value: str) -> dict:
     """Сохраняет значение по ключу."""
-    response = httpx.post(f"{BASE_URL}/save", json={"key": key, "value": value})
+    payload = {"key": key, "value": value}
-    response.raise_for_status()
+    response = httpx.post(f"{BASE_URL}/save", json=payload)
-    console.print(f"[green]✅[/green] Сохранено: {key} → {value}")
+    return response.json()
 
 
-def get(key: str) -> None:
+def get(key: str) -> dict:
     """Получает значение по ключу."""
-    response = httpx.get(f"{BASE_URL}/get", params={"key": key})
+    params = {"key": key}
-    if response.status_code == 404:
+    response = httpx.get(f"{BASE_URL}/get", params=params)
-        console.print(f"[red]❌[/red] Ключ не найден: {key}")
+    return response.json()
-        return
-    response.raise_for_status()
-    data = response.json()
-    console.print(f"[blue]📥[/blue] Получено: {data['value']}")
 
 
-def delete(key: str) -> None:
+def delete(key: str) -> dict:
-    """Удаляет ключ."""
+    """Удаляет запись по ключу."""
-    response = httpx.delete(f"{BASE_URL}/delete", params={"key": key})
+    payload = {"key": key}
-    if response.status_code == 404:
+    response = httpx.delete(f"{BASE_URL}/delete", json=payload)
-        console.print(f"[red]❌[/red] Ключ не найден: {key}")
+    return response.json()
-        return
-    response.raise_for_status()
-    console.print(f"[yellow]⚠️[/yellow] Удалено: {key}")
 
 
-def list_keys() -> None:
+def list_keys() -> dict:
-    """Выводит список всех ключей."""
+    """Возвращает список всех ключей."""
     response = httpx.get(f"{BASE_URL}/list_keys")
-    response.raise_for_status()
+    return response.json()
-    keys = response.json().get("keys", [])
-    table = Table(title="Список ключей")
-    table.add_column("№", justify="right")
-    table.add_column("Ключ")
-    for idx, key in enumerate(keys, 1):
-        table.add_row(str(idx), key)
-    console.print(table)
 
 
-def save_with_namespace(namespace: str, key: str, value: str) -> None:
+def save_with_namespace(namespace: str, key: str, value: str) -> dict:
     """Сохраняет значение в указанном пространстве имён."""
-    response = httpx.post(
+    payload = {"namespace": namespace, "key": key, "value": value}
-        f"{BASE_URL}/save_with_namespace",
+    response = httpx.post(f"{BASE_URL}/save_with_namespace", json=payload)
-        json={"namespace": namespace, "key": key, "value": value},
+    return response.json()
-    )
-    response.raise_for_status()
-    console.print(f"[green]✅[/green] Сохранено: {namespace}:{key} → {value}")
 
 
-def get_by_namespace(namespace: str, key: str) -> None:
+def get_by_namespace(namespace: str, key: str) -> dict:
     """Получает значение из указанного пространства имён."""
-    response = httpx.get(
+    params = {"namespace": namespace, "key": key}
-        f"{BASE_URL}/get_by_namespace",
+    response = httpx.get(f"{BASE_URL}/get_by_namespace", params=params)
-        params={"namespace": namespace, "key": key},
+    return response.json()
-    )
-    if response.status_code == 404:
-        console.print(f"[red]❌[/red] Ключ не найден: {namespace}:{key}")
-        return
-    response.raise_for_status()
-    data = response.json()
-    console.print(f"[blue]📥[/blue] Получено из {namespace}: {data['value']}")
 
 
 def main() -> None:
-    """Тестовый сценарий работы с сервером."""
+    """Простой тестовый сценарий для проверки работы сервера."""
-    # Очистка тестовых данных (если они уже есть)
+    print("=== Тестирование MCP-сервера ===")
-    delete("test_key")
-    delete("user1:session")
 
-    console.print("[bold underline]=== Тестирование сервера памяти ===\n[/bold underline]")
+    # 1. Сохраняем простое значение
+    print("\nСохраняем ключ 'foo' со значением 'bar'")
+    print(json.dumps(save("foo", "bar"), indent=2))
 
-    # 1. Сохраняем и получаем обычный ключ
+    # 2. Получаем сохранённое значение
-    save("test_key", "Hello, World!")
+    print("\nПолучаем ключ 'foo'")
-    get("test_key")
+    print(json.dumps(get("foo"), indent=2))
 
-    # 2. Удаляем ключ и проверяем отсутствие
+    # 3. Сохраняем в пространстве имён
-    delete("test_key")
+    ns = "experiment"
-    get("test_key")
+    print(f"\nСохраняем ключ '{ns}:baz' со значением 'qux' в namespace '{ns}'")
+    print(json.dumps(save_with_namespace(ns, "baz", "qux"), indent=2))
 
-    # 3. Работа с пространством имён
+    # 4. Получаем из пространства имён
-    save_with_namespace("user1", "session", "abc123")
+    print(f"\nПолучаем ключ '{ns}:baz' из namespace '{ns}'")
-    get_by_namespace("user1", "session")
+    print(json.dumps(get_by_namespace(ns, "baz"), indent=2))
 
-    # 4. Список всех ключей (должен включать только user1:session)
+    # 5. Список всех ключей
-    list_keys()
+    print("\nСписок всех ключей:")
+    print(json.dumps(list_keys(), indent=2))
 
-    console.print("\n[bold green]✅ Тесты завершены успешно![/bold green]")
+    # 6. Удаляем ключ 'foo'
+    print("\nУдаляем ключ 'foo'")
+    print(json.dumps(delete("foo"), indent=2))
+
+    # 7. Проверяем, что ключ удалён
+    print("\nПроверка удаления ключа 'foo':")
+    print(json.dumps(get("foo"), indent=2))
 
 
 if __name__ == "__main__":
-    main()
+    sys.exit(main())