1. 引导提问（Elicitation） #

本章主要介绍如何使用 MCP 的引导提问机制，包括在工具执行中向用户发起结构化信息收集，以及客户端如何处理引导提问并返回用户响应。你将学会：

• 理解 Elicitation 机制：在工具执行中向用户发起结构化信息收集
• 在服务端使用 ctx.elicit() 发起引导提问，定义提问的 Schema
• 在客户端侧处理 Elicitation 请求并返回用户响应

2. 服务器：在工具执行中发起引导提问 #

新建 C:\mcp-project\elicitation_server.py：

# 导入异步编程相关模块
import asyncio

# 导入 Pydantic，用于数据模型和校验
from pydantic import BaseModel, Field, field_validator

# 导入 MCP 服务器主类和上下文类型
from mcp.server.fastmcp import FastMCP, Context
from mcp.server.session import ServerSession

# 创建 FastMCP 服务器实例，命名为 "Elicitation  Server"
mcp = FastMCP(name="Elicitation  Server")


# 定义任务信息的数据结构
class TaskInfo(BaseModel):
    """任务信息数据结构"""

    # 任务标题，字符串类型
    title: str = Field(description="任务标题")
    # 任务优先级，字符串类型，限定 low/medium/high
    priority: str = Field(description="任务优先级 (low/medium/high)")

    # 校验器：确保 priority 字段值合法
    @field_validator("priority")
    def validate_priority(cls, v):
        # 如果优先级不在允许的范围内，抛出异常
        if v.lower() not in ["low", "medium", "high"]:
            raise ValueError("优先级必须是 low、medium 或 high")
        # 返回小写形式
        return v.lower()


# 定义用户偏好设置的数据结构
class UserPreferences(BaseModel):
    """用户偏好设置数据结构"""

    # 是否检查替代选项，布尔类型
    checkAlternative: bool = Field(description="是否检查替代选项？")
    # 替代选项名称，字符串类型，默认值为 "option_b"
    alternativeOption: str = Field(default="option_b", description="替代选项名称")


# 定义确认操作的数据结构
class ConfirmationData(BaseModel):
    """确认操作数据结构"""

    # 是否继续执行，布尔类型
    proceed: bool = Field(description="是否继续执行？")
    # 继续或取消的原因，字符串类型，默认为空
    reason: str = Field(default="", description="继续或取消的原因（可选）")


# 工具1：创建任务（使用 elicitation 收集任务信息）
@mcp.tool()
async def create_task(ctx: Context[ServerSession, None]) -> str:
    """创建任务，使用 elicitation 收集任务信息"""
    try:
        # 发起用户引导，收集任务信息
        result = await ctx.elicit(message="请提供任务信息：", schema=TaskInfo)

        # 如果用户接受并填写了数据
        if result.action == "accept" and result.data:
            task_info = result.data
            # 返回任务创建成功信息
            return f" 任务创建成功：{task_info.title} (优先级: {task_info.priority})"
        # 如果用户拒绝
        elif result.action == "decline":
            return " 用户拒绝了任务创建请求"
        # 其他情况（如取消）
        else:
            return " 任务创建被取消"

    # 捕获异常并返回错误信息
    except Exception as e:
        return f" 任务创建失败：{str(e)}"


# 工具2：收集用户偏好
@mcp.tool()
async def collect_preferences(ctx: Context[ServerSession, None]) -> str:
    """收集用户偏好设置"""
    try:
        # 发起用户引导，收集偏好设置
        result = await ctx.elicit(
            message="请告诉我你的偏好设置：", schema=UserPreferences
        )

        # 如果用户接受并填写了数据
        if result.action == "accept" and result.data:
            prefs = result.data
            # 返回收集到的偏好设置
            return f" 偏好设置已收集：检查替代={prefs.checkAlternative}, 替代选项={prefs.alternativeOption}"
        # 如果用户拒绝
        elif result.action == "decline":
            return " 用户拒绝了偏好设置收集"
        # 其他情况（如取消）
        else:
            return " 偏好设置收集被取消"

    # 捕获异常并返回错误信息
    except Exception as e:
        return f" 偏好设置收集失败：{str(e)}"


# 工具3：确认操作
@mcp.tool()
async def confirm_action(ctx: Context[ServerSession, None]) -> str:
    """确认重要操作"""
    try:
        # 发起用户引导，收集确认信息
        result = await ctx.elicit(
            message="这是一个重要操作，请确认是否继续：", schema=ConfirmationData
        )

        # 如果用户接受并填写了数据
        if result.action == "accept" and result.data:
            confirm_data = result.data
            # 如果用户选择继续
            if confirm_data.proceed:
                reason = confirm_data.reason or "未提供原因"
                return f" 操作已确认，继续执行。原因：{reason}"
            # 如果用户选择不继续
            else:
                reason = confirm_data.reason or "未提供原因"
                return f" 操作被拒绝。原因：{reason}"
        # 如果用户拒绝
        elif result.action == "decline":
            return " 用户拒绝了确认请求"
        # 其他情况（如取消）
        else:
            return " 确认请求被取消"

    # 捕获异常并返回错误信息
    except Exception as e:
        return f" 确认操作失败：{str(e)}"


# 主函数：启动服务器
async def main():
    """启动 MCP 服务器"""
    # 打印服务器启动信息
    print(" 启动 Elicitation  Server...")
    print(" 可用工具：")
    print("  - create_task: 创建任务")
    print("  - collect_preferences: 收集用户偏好")
    print("  - confirm_action: 确认操作")

    # 启动服务器，使用 stdio 传输方式
    mcp.run(transport="stdio")


# 程序入口
if __name__ == "__main__":
    try:
        # 运行主函数，启动事件循环
        asyncio.run(main())
    except KeyboardInterrupt:
        # 捕获 Ctrl+C，优雅退出
        print("\n 服务器已停止")
    except Exception as e:
        # 捕获其他异常，打印错误信息和建议
        print(f" 服务器启动失败：{e}")
        print(" 建议：检查 MCP 版本是否支持 elicitation 功能")

要点：

• 使用 ctx.elicit() 发起引导提问
• 定义引导提问的数据结构（Pydantic Schema）
• 处理 ElicitationResult 的不同响应状态

3. 运行与验证 #

mcp dev elicitation_server.py