100 lines
4.2 KiB
Python
100 lines
4.2 KiB
Python
# 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 = (
|
|
"<html>"
|
|
"<head><title>HIL Middleware Demo</title></head>"
|
|
"<body>"
|
|
"<h1>Добро пожаловать!</h1>"
|
|
"<p>Отправьте POST запрос на <code>/process</code>, чтобы увидеть HIL.</p>"
|
|
"</body>"
|
|
"</html>"
|
|
)
|
|
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,
|
|
) |