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