From ce36785e56a020f6a084eb40e97af0b529584c97 Mon Sep 17 00:00:00 2001 From: lonpatovaadelina Date: Wed, 27 May 2026 10:34:07 +0000 Subject: [PATCH] =?UTF-8?q?Human-in-the-Loop=20=D1=87=D0=B5=D1=80=D0=B5?= =?UTF-8?q?=D0=B7=20middleware:=20solution.py?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../solution.py | 100 ++++++++++++++++++ 1 file changed, 100 insertions(+) create mode 100644 solutions/69a86305c46fd26feae6bcaa_Human-in-the-Loop_через_middleware/solution.py diff --git a/solutions/69a86305c46fd26feae6bcaa_Human-in-the-Loop_через_middleware/solution.py b/solutions/69a86305c46fd26feae6bcaa_Human-in-the-Loop_через_middleware/solution.py new file mode 100644 index 0000000..8a701ae --- /dev/null +++ b/solutions/69a86305c46fd26feae6bcaa_Human-in-the-Loop_через_middleware/solution.py @@ -0,0 +1,100 @@ +# solution.py + +""" +Human-in-the-Loop (HIL) через middleware на основе FastMCP. +В этом примере реализован простой сервер, который имеет один эндпоинт /process. +Перед выполнением запроса к этому эндпоинту пользователь должен подтвердить действие, +что демонстрирует принцип HIL: человек вмешивается в поток выполнения. + +Требования: +- Используется только стек FastMCP (не FastAPI). +- Middleware реализует запрос подтверждения через консольный ввод. +- Если пользователь отвечает "yes" (любое регистровое сочетание), запрос + продолжается, иначе возвращается сообщение об отмене. +""" + +import sys +from fastmcp import FastMCP, Request, Response + +# -------------------------------------------------------------------- +# Middleware для Human-in-the-Loop +# -------------------------------------------------------------------- +class HILMiddleware: + """ + Middleware, который перехватывает запросы к /process и запрашивает подтверждение у пользователя. + """ + + def __init__(self, app: FastMCP): + self.app = app + + async def __call__(self, request: Request) -> Response: + # Проверяем путь запроса + if request.path == "/process": + # Запрос подтверждения + print("\n=== Human-in-the-Loop ===") + print(f"Запрос к {request.method} {request.path}") + confirm = input("Вы действительно хотите выполнить это действие? (yes/no): ").strip().lower() + if confirm != "yes": + return Response( + status_code=403, + body="Действие отменено пользователем.", + headers={"Content-Type": "text/plain"}, + ) + # Если подтверждение получено или путь не /process, продолжаем цепочку + return await self.app(request) + +# -------------------------------------------------------------------- +# Основное приложение FastMCP +# -------------------------------------------------------------------- +app = FastMCP() + +@app.route("/process", methods=["POST"]) +async def process_handler(request: Request) -> Response: + """ + Пример обработчика, который выполняет некоторую задачу. + В реальном приложении здесь могла бы быть бизнес-логика. + """ + # Здесь можно добавить любую логику + return Response( + status_code=200, + body="Действие выполнено успешно.", + headers={"Content-Type": "text/plain"}, + ) + +@app.route("/", methods=["GET"]) +async def index(request: Request) -> Response: + """ + Статическая страница приветствия. + """ + html = ( + "" + "HIL Middleware Demo" + "" + "

Добро пожаловать!

" + "

Отправьте POST запрос на /process, чтобы увидеть HIL.

" + "" + "" + ) + return Response( + status_code=200, + body=html, + headers={"Content-Type": "text/html"}, + ) + +# Оборачиваем приложение в middleware +app = HILMiddleware(app) + +# -------------------------------------------------------------------- +# Запуск сервера +# -------------------------------------------------------------------- +if __name__ == "__main__": + # FastMCP использует uvicorn под капотом, но запускаем напрямую + import uvicorn + + uvicorn.run( + "solution:app", + host="127.0.0.1", + port=8000, + log_level="info", + reload=False, + ) \ No newline at end of file