Understanding Elicitation

In the previous lessons, your tool executions were “fire-and-forget.” The LLM (or client) sent a request, the tool ran its logic, and immediately returned a result.

But what happens when a tool needs to perform a high-stakes operation, like deleting a database table or transferring money? You don’t want the server to execute that immediately. You want it to pause, ask the user for confirmation or extra details, and only then proceed.

This capability is called Elicitation.

Elicitation allows an MCP Server to ask the Client for input during the execution of a tool. It turns a one-way command into a two-way conversation.

The Compatibility Challenge

It is important to understand that Elicitation is a sophisticated feature that requires explicit support from the Host application.

Not all MCP clients support it. For example:

• VS Code: Supports Elicitation. It can pop up dialog boxes to ask the user for input.
• Claude Desktop: Currently does not support Elicitation. If a server sends an Elicitation request to Claude Desktop, the application will simply not know what to do, and the tool call will fail or hang.

You cannot force an application like Claude Desktop to support this feature.

However, because you are learning to build Custom MCP Clients, you have the power to implement this support yourself. In this lesson, you will write a custom Python client that listens for these requests and handles them programmatically, creating a safe, human-in-the-loop workflow.

The Interaction Flow

Here is how the flow works when the Host application supports Elicitation:

1. The Host (Client) sends a tool call request to the MCP Server (e.g., execute_query(DROP TABLE)).
2. The Server starts the tool logic but detects a dangerous keyword. It pauses execution and sends an elicitation/create request back to the Host.
3. The Host catches this request. It presents a UI (like a dialog box or terminal prompt) to the Human User, displaying the server’s warning message.
4. The User provides input (e.g., types “Yes” or enters a justification).
5. The Host sends this structured data back to the Server.
6. The Server validates the user’s input, finishes the operation, and finally returns the result (e.g., “Table Dropped”) to the Host.

This architecture ensures that humans remain in the loop for sensitive actions, but it depends entirely on the Host’s ability to handle step #3.

Implementing the Server

You will now create a system that simulates a database manager. It will allow read-only queries freely but will trigger an Elicitation flow for destructive commands.

Create a new directory lesson_04 under your present working directory. Inside that, create a file named db_server.py:

from pydantic import BaseModel, Field
from mcp.server.fastmcp import FastMCP, Context

mcp = FastMCP("SafeDB-Manager", host="127.0.0.1", port=8000)

class SafetyVerification(BaseModel):
    """Schema for validating critical operations."""
    confirm: bool = Field(
        description="Must be True to proceed with the operation."
    )
    justification: str = Field(
        description="A mandatory reason for why this critical operation is being performed."
    )
    environment: str = Field(
        description="The environment you intend to affect (Development, Staging, or Production)."
    )

@mcp.tool()
async def execute_query(query: str, ctx: Context) -> str:
    """
    Executes a database query.
    Triggers a safety check (elicitation) for destructive commands.
    """
    query_upper = query.upper().strip()

    is_destructive = any(cmd in query_upper for cmd in ["DROP", "DELETE", "TRUNCATE"])

    if is_destructive:
        result = await ctx.elicit(
            message=f"CRITICAL WARNING: You are attempting to run a destructive query: '{query}'. Verification required.",
            schema=SafetyVerification
        )

        if result.action == "accept" and result.data:
            data = result.data

            if not data.confirm:
                return "Operation blocked: User did not confirm."

            if data.environment == "Production" and len(data.justification) < 10:
                return "Operation blocked: Production changes require a detailed justification (min 10 chars)."

            return (
                f"Query Executed Successfully on [{data.environment}].\n"
                f"   Query: {query}\n"
                f"   Log Reason: {data.justification}"
            )

        elif result.action == "decline":
            return "Operation declined by user."

        else:
            return "Operation cancelled."

    return f"Read-only query executed: {query}"

if __name__ == "__main__":
    print("Starting SafeDB Server on http://127.0.0.1:8000/mcp ...")
    mcp.run(transport="streamable-http")

What You Are Building

You are building a tiny server that pauses on dangerous SQL and asks the client to confirm. The client becomes the “human approval” step and sends structured input back to the server before it proceeds.

Key Concepts in db_server.py:

• ctx: Context: Note the new argument in execute_query. By adding ctx: Context, FastMCP automatically injects the connection context, giving you access to special methods like elicit.
• SafetyVerification: We use Pydantic to define exactly what we need from the user. We aren’t just asking “Yes or No”; we are enforcing structured data (Environment, Justification, Confirmation).
• await ctx.elicit(...): This is the core logic. It sends a message and the schema to the client. The Python function execution creates a checkpoint here and waits.

How the Server Script Works

• FastMCP(...) starts the MCP server and exposes tools over streamable-http.
• SafetyVerification defines the exact input the user must provide.
• ctx.elicit(...) pauses the tool and asks the client for that input.
• After the client replies, the tool resumes and validates the response before returning a final result.

Implementing the Custom Client

Since standard AI clients might not support this flow yet, you must write a custom client. This client will act as the Host application, intercepting the server’s request and asking you (the user) for input via the terminal.

Create a file named db_client.py:

import asyncio
import json
from mcp import ClientSession, types
from mcp.client.session import RequestContext
from mcp.client.streamable_http import streamablehttp_client

SERVER_URL = "http://127.0.0.1:8000/mcp"

async def elicitation_handler(
    context: RequestContext,
    params: types.ElicitRequestParams
) -> types.ElicitResult:
    """
    This function is called automatically when the Server triggers ctx.elicit().
    """
    print("\n" + "!" * 50)
    print(f"SERVER MESSAGE: {params.message}")
    print("!" * 50 + "\n")

    print("Please provide the required safety details:")

    print("Select Environment (Development/Staging/Production):")
    env_input = input("> ").strip().title()

    print("Justification for this action:")
    reason_input = input("> ").strip()

    print("Type 'yes' to confirm execution:")
    confirm_input = input("> ").lower().strip()
    is_confirmed = (confirm_input == "yes")

    user_response_data = {
        "confirm": is_confirmed,
        "justification": reason_input,
        "environment": env_input
    }

    return types.ElicitResult(
        action="accept",
        content=user_response_data
    )

async def run_client():
    print(f"Connecting to DB Server at {SERVER_URL}...")

    async with streamablehttp_client(SERVER_URL) as (read, write, _):
        async with ClientSession(
            read,
            write,
            elicitation_callback=elicitation_handler
        ) as session:
            await session.initialize()
            print("Connected.\n")

            print("--- Test 1: Running Safe Query ---")
            result_safe = await session.call_tool(
                "execute_query",
                arguments={"query": "SELECT * FROM users"}
            )
            print(f"Result: {result_safe.content[0].text}\n")

            print("--- Test 2: Running DESTRUCTIVE Query ---")
            print("Sending: DROP TABLE invoices...")

            result_dangerous = await session.call_tool(
                "execute_query",
                arguments={"query": "DROP TABLE invoices"}
            )

            print("\n--- Final Result from Server ---")
            if result_dangerous.isError:
                 print(f"Error: {result_dangerous.content}")
            else:
                 print(result_dangerous.content[0].text)

if __name__ == "__main__":
    asyncio.run(run_client())

Key Concepts in db_client.py:

• elicitation_handler: This is a standalone asynchronous function. It receives the message from the server. In a real desktop app, this would open a modal window. In this script, it uses input() to pause the terminal and ask you for data.
• elicitation_callback=elicitation_handler: When initializing the ClientSession, you must register your handler. This is the crucial step that “enables” elicitation support for your client.

How the Client Script Works

• elicitation_handler(...) is the callback the server triggers when it calls ctx.elicit(...).
• It collects user input via input() and returns a structured response that matches SafetyVerification.
• ClientSession(..., elicitation_callback=...) is what makes the client “elicitation-aware.”

You have written the code for a sophisticated, human-in-the-loop safety system.

1. The Server defines the safety rules and pauses execution when they are met.
2. The Custom Client implements a handler to catch that pause, ask the user for input, and return the data.

This setup demonstrates how custom MCP clients can provide advanced features that generic AI chat interfaces do not yet support. In the next video lesson, you will run these scripts and see the interactivity in action.

Run It

1. In one terminal, start the server:

uv run python db_server.py

2. In a second terminal, run the client:

uv run python db_client.py

You should see the safe query return immediately and the destructive query pause to request confirmation. This only works if the client registers the elicitation callback; otherwise the tool call will fail or hang.