前言
如果你之前接觸過LangGraph的"Human in the loop"概念，那麼理解MCP的Elicitation機制就會容易很多。這兩個功能非常相似，都是讓AI在需要時停下來，禮貌地向人類尋求幫助或確認。
想象一下，當你正在和朋友聊天，突然他問你："嘿，我該穿哪件襯衫去參加明天的聚會？"這時候你就會停下來思考，然後給出建議。Elicitation就是讓AI具備這種"求助"的能力。它允許MCP服務器在工具執行過程中向用户請求結構化的輸入信息。與要求一次性提供所有輸入不同，服務器可以根據需要與用户進行交互式溝通——比如，提示缺失的參數、請求澄清或收集額外上下文信息。舉個例子，一個文件管理工具可能會問："我應該在哪個目錄創建這個文件？"；而一個數據分析工具則可能請求："我該分析哪個時間段的數據？"
Elicitation能讓工具在關鍵時刻暫停執行，並向用户請求特定信息。這在以下場景中尤為有用：
- 缺失參數：當初始輸入未提供必要信息時，主動向用户索取
- 澄清請求：在模糊或有歧義的場景下，獲取用户的確認或選擇
- 漸進式披露：分步驟收集複雜信息，避免一次性要求過多內容
- 動態工作流：根據用户的響應實時調整工具的行為邏輯
基本示例
讓我們通過幾個基本示例來演示如何使用elicitation功能。
MCP Server
在服務器端，我們創建一個收集用户信息的工具：
from fastmcp import FastMCP, Context
from dataclasses import dataclass
mcp = FastMCP("Elicitation Server")
@dataclass
class UserInfo:
name: str
age: int
@mcp.tool
async def collect_user_info(ctx: Context) -> str:
"""Collect user information through interactive prompts."""
result = await ctx.elicit(
message="Please provide your information",
response_type=UserInfo
)
if result.action == "accept":
user = result.data
return f"Hello {user.name}, you are {user.age} years old"
elif result.action == "decline":
return "Information not provided"
else: # cancel
return "Operation cancelled"
if __name__ == "__main__":
mcp.run(transport="streamable-http", host="localhost", port=8001, show_banner=False)
ctx.elicit()方法的參數説明：
message: 顯示給用户的提示詞，就像一個禮貌的請求response_type: 定義預期響應結構的Python類型（數據類、基本類型等）。注意，Elicitation響應僅支持JSON Schema類型的子集。
ctx.elicit()的響應是一個ElicitationResult對象，包含以下屬性：
action: 用户如何迴應，其值只有accept（接受）、decline（拒絕）和cancel（取消）data: 用户的輸入值，類型為response_type或者None，只有當action=accept時存在
MCP Client
客户端需要實現一個處理徵詢請求的回調函數：
import asyncio
from fastmcp import Client
from fastmcp.client.elicitation import ElicitResult
from mcp.shared.context import RequestContext
from mcp.types import ElicitRequestParams
from openai import AsyncOpenAI
from pkg.config import cfg
llm = AsyncOpenAI(
base_url=cfg.llm_base_url,
api_key=cfg.llm_api_key,
)
async def elicitation_handler(message: str, response_type: type, params: ElicitRequestParams, context: RequestContext):
print(f"MCP Server asks: {message}")
user_name = input("Your name: ").strip()
user_age = input("Your age: ").strip()
if not user_name or not user_age:
return ElicitResult(action="decline")
response_date = response_type(name=user_name, age=user_age)
return response_date
mcp_client = Client("http://localhost:8001/mcp", elicitation_handler=elicitation_handler)
async def main():
async with mcp_client:
resp = await mcp_client.call_tool("collect_user_info", {})
print(f"collect_user_info result: {resp}")
if __name__ == "__main__":
# 運行主程序
asyncio.run(main())
運行輸出示例：
- 正常填寫的交互
MCP Server asks: Please provide your information
Your name: Rainux
Your age: 18
collect_user_info result: CallToolResult(content=[TextContent(type='text', text='Hello Rainux, you are 18 years old', annotations=None, meta=None)], structured_content={'result': 'Hello Rainux, you are 18 years old'}, data='Hello Rainux, you are 18 years old', is_error=False)
- 用户拒絕提供信息
MCP Server asks: Please provide your information
Your name:
Your age:
collect_user_info result: CallToolResult(content=[TextContent(type='text', text='Information not provided', annotations=None, meta=None)], structured_content={'result': 'Information not provided'}, data='Information not provided', is_error=False)
運行shell命令-用户徵詢
在這個示例中，我們將實現一個命令行交互程序。當用户的命令需要在主機上執行可能造成修改的命令時，AI會禮貌地請求用户確認。
MCP Server
當執行可能修改系統的命令時，服務器會要求用户確認是否執行：
import asyncio
from dataclasses import dataclass
from fastmcp import Context, FastMCP
@dataclass
class UserDecision:
decision: str = "decline"
mcp = FastMCP("Elicitation Server")
@mcp.tool()
async def execute_command_local(command: str, ctx: Context, is_need_user_check: bool = False, timeout: int = 10) -> str:
"""Execute a shell command locally.
Args:
command (str): The shell command to execute.
is_need_user_check (bool): Set to True when performing create, delete, or modify operations on the host, indicating that user confirmation is required.
timeout (int): Timeout in seconds for command execution. Default is 10 seconds.
Returns:
str: The output of the shell command.
"""
if is_need_user_check:
user_check_result = await ctx.elicit(
message=f"Do you want to execute this command(yes or no): {command}",
response_type=UserDecision, # response_type 必須是符合 JSON Schema
)
if user_check_result.action != "accept":
return "User denied command execution."
try:
proc = await asyncio.create_subprocess_shell(
command,
stdout=asyncio.subprocess.PIPE,
stderr=asyncio.subprocess.PIPE
)
stdout, stderr = await asyncio.wait_for(proc.communicate(), timeout=timeout)
stdout_str = stdout.decode().strip()
stderr_str = stderr.decode().strip()
if stdout_str:
return f"Stdout: {stdout_str}"
elif stderr_str:
return f"Stderr: {stderr_str}"
else:
return "Command executed successfully with no output"
except asyncio.TimeoutError:
if proc and not proc.returncode:
try:
proc.terminate()
await proc.wait()
except:
pass
return f"Error: Command '{command}' timed out after {timeout} seconds"
except Exception as e:
return f"Error executing command '{command}': {str(e)}"
if __name__ == "__main__":
mcp.run(transport="streamable-http", host="localhost", port=8001, show_banner=False)
MCP Client
客户端的主體邏輯與之前的示例基本一致，主要實現了elicitation_handler()方法來處理服務器的徵詢請求：
import asyncio
import json
import readline # For enhanced input editing
import traceback
from typing import cast
from fastmcp import Client
from fastmcp.client.elicitation import ElicitResult
from mcp.shared.context import RequestContext
from mcp.types import ElicitRequestParams
from openai import AsyncOpenAI
from openai.types.chat import ChatCompletionMessageFunctionToolCall
from pkg.config import cfg
from pkg.log import logger
class MCPHost:
"""MCP主機類，用於管理與MCP服務器的連接和交互"""
def __init__(self, server_uri: str):
"""
初始化MCP客户端
Args:
server_uri (str): MCP服務器的URI地址
"""
# 初始化MCP客户端連接
self.mcp_client: Client = Client(server_uri, elicitation_handler=self.elicitation_handler)
# 初始化異步OpenAI客户端用於與LLM交互
self.llm = AsyncOpenAI(
base_url=cfg.llm_base_url,
api_key=cfg.llm_api_key,
)
# 存儲對話歷史消息
self.messages = []
async def close(self):
"""關閉MCP客户端連接"""
if self.mcp_client:
await self.mcp_client.close()
async def elicitation_handler(self, message: str, response_type: type, params: ElicitRequestParams, context: RequestContext):
print(f"MCP Server asks: {message}")
user_decision = input("Please check(yes or no): ").strip()
if not user_decision or user_decision != "yes":
return ElicitResult(action="decline")
response_data = response_type(decision="accept")
return response_data
async def process_query(self, query: str) -> str:
"""Process a user query by interacting with the MCP server and LLM.
Args:
query (str): The user query to process.
Returns:
str: The response from the MCP server.
"""
# 將用户查詢添加到消息歷史中
self.messages.append({
"role": "user",
"content": query,
})
# 使用異步上下文管理器確保MCP客户端連接正確建立和關閉
async with self.mcp_client:
# 從MCP服務器獲取可用工具列表
tools = await self.mcp_client.list_tools()
# 構造LLM可以理解的工具格式
available_tools = []
# 將MCP工具轉換為OpenAI格式
for tool in tools:
available_tools.append({
"type": "function",
"function": {
"name": tool.name,
"description": tool.description,
"parameters": tool.inputSchema,
}
})
logger.info(f"Available tools: {[tool['function']['name'] for tool in available_tools]}")
# 調用LLM，傳入對話歷史和可用工具
resp = await self.llm.chat.completions.create(
model=cfg.llm_model,
messages=self.messages,
tools=available_tools,
temperature=0.3,
)
# 存儲最終響應文本
final_text = []
# 獲取LLM的首個響應消息
message = resp.choices[0].message
# 如果響應包含直接內容，則添加到結果中
if hasattr(message, "content") and message.content:
final_text.append(message.content)
# 循環處理工具調用，直到沒有更多工具調用為止
while message.tool_calls:
# 遍歷所有工具調用
for tool_call in message.tool_calls:
# 確保工具調用有函數信息
if not hasattr(tool_call, "function"):
continue
# 類型轉換以獲取函數調用詳情
function_call = cast(ChatCompletionMessageFunctionToolCall, tool_call)
function = function_call.function
tool_name = function.name
# 解析函數參數
tool_args = json.loads(function.arguments)
# 檢查MCP客户端是否已連接
if not self.mcp_client.is_connected():
raise RuntimeError("Session not initialized. Cannot call tool.")
# 調用MCP服務器上的指定工具
result = await self.mcp_client.call_tool(tool_name, tool_args)
# 將助手的工具調用添加到消息歷史中
self.messages.append({
"role": "assistant",
"tool_calls": [
{
"id": tool_call.id,
"type": "function",
"function": {
"name": function.name,
"arguments": function.arguments
}
}
]
})
# 將工具調用結果添加到消息歷史中
self.messages.append({
"role": "tool",
"tool_call_id":tool_call.id,
"content": str(result.content) if result.content else ""
})
# 基於工具調用結果再次調用LLM
final_resp = await self.llm.chat.completions.create(
model=cfg.llm_model,
messages=self.messages,
tools=available_tools,
temperature=0.3,
)
# 更新消息為最新的LLM響應
message = final_resp.choices[0].message
# 如果響應包含內容，則添加到最終結果中
if message.content:
final_text.append(message.content)
# 返回連接後的完整響應
return "\n".join(final_text)
async def chat_loop(self):
"""主聊天循環，處理用户輸入並顯示響應"""
print("Welcome to the MCP chat! Type 'quit' to exit.")
# 持續處理用户輸入直到用户退出
while True:
try:
# 獲取用户輸入
query = input("You: ").strip()
# 檢查退出命令
if query.lower() == "quit":
print("Exiting chat. Goodbye!")
break
# 跳過空輸入
if not query:
continue
# 處理用户查詢並獲取響應
resp = await self.process_query(query)
print(f"Assistant: {resp}")
# 捕獲並記錄聊天循環中的任何異常
except Exception as e:
logger.error(f"Error in chat loop: {str(e)}")
logger.error(traceback.format_exc())
async def main():
"""主函數，程序入口點"""
# 創建MCP主機實例
client = MCPHost(server_uri="http://localhost:8001/mcp")
try:
# 啓動聊天循環
await client.chat_loop()
except Exception as e:
# 記錄主程序中的任何異常
logger.error(f"Error in main: {str(e)}")
logger.error(traceback.format_exc())
finally:
# 確保客户端連接被正確關閉
await client.close()
if __name__ == "__main__":
# 運行主程序
asyncio.run(main())
運行示例：
Welcome to the MCP chat! Type 'quit' to exit.
You: what can you do?
Assistant: I can execute shell commands on your local machine. Please let me know what specific task you'd like me to help with. Keep in mind that any command I run will be on your local system, and you should ensure that the commands are safe and appropriate for your environment.
You: 查詢下當前內存使用情況
Assistant: 當前內存使用情況如下：
- **總內存**: 62Gi
- **已使用內存**: 11Gi
- **空閒內存**: 45Gi
- **共享內存**: 137Mi
- **緩存/緩衝區**: 6.6Gi
- **可用內存**: 50Gi
交換分區情況：
- **總交換空間**: 3.8Gi
- **已使用交換空間**: 0B
- **空閒交換空間**: 3.8Gi
如果還有其他問題，請隨時告訴我！
You: 在家目錄創建一個文件，文件內容為當前平均負載，文件名為當前日期
MCP Server asks: Do you want to execute this command(yes or no): echo $(uptime | awk -F 'load average:' '{print $2}') > ~/$(date +%Y%m%d).txt
Please check(yes or no): yes
Assistant: 已在您的家目錄下創建了一個文件，文件名為當前日期（例如：20231005.txt），文件內容為系統的當前平均負載。如果您需要進一步的幫助，請告訴我！
You: quit
Exiting chat. Goodbye!
小結
通過以上示例，我們可以看到MCP的Elicitation機制在實際應用中的強大之處：
- 增強安全性：在執行敏感操作前徵詢用户意見，避免意外操作帶來的風險。就像有個貼心的助手在執行重要任務前總是先問一句"您確定嗎？"
- 提升用户體驗：讓用户參與到AI的決策過程中，而不是被動接受結果。這種交互方式讓用户感覺更有控制感，也更信任AI系統。
- 靈活的數據收集：可以按需收集結構化數據，避免一次性要求用户提供過多信息造成的認知負擔。
- 優雅的錯誤處理：當用户拒絕或取消操作時，系統能夠優雅地處理這些情況，而不是崩潰或產生不可預期的行為。
- 無縫集成：Elicitation機制與MCP的其他功能（如工具調用、資源訪問）無縫集成，形成一個完整的AI交互生態系統。
在實際開發中，Elicitation機制特別適用於以下場景：
- 系統管理工具在執行修改操作前的確認
- 數據分析工具在處理敏感數據前的權限驗證
- 文件操作工具在創建、刪除或修改文件前的用户確認
- 金融或醫療等高風險領域的操作審批流程
總的來説，Elicitation機制為AI系統與人類用户之間建立了一座溝通的橋樑，讓AI不再是冷冰冰的執行者，而是一個懂得在關鍵時刻尋求幫助的智能夥伴。
參考
- yuan - MCP 徵詢機制（Elicitation）詳解，附代碼實現
- FastMCP - Server Elicitation
- FastMCP - Client Elicitation
