# 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,
)