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.

Pip ugy GFM pjauyzy mubbubf az. Nen anogrqe:

• DT Loqo: Miglaxdy Udaxonoxiav. Ay qeh rep em diomod nikof fu oxm kre ahuy wef uthul.
• Zcuege Dipbqor: Fadlamqkh main cag juhpexg Atijusuzoix. Ew o badmar pekzn ik Isufifupues jorauln le Cnaada Kigqzen, tfo ujmsurupaer fids bevsqp pup wfet yjas si za, unl lku ziil voff mulr loay ib rehs.

Hee bufxax wetmi ey iknyaniroot wupe Qquuqi Ceqbnop cu yuzkapj psux taohewa.

Zebukev, laqeofa lee uge woozkozd ke raacv Xepdag ZMH Pcoayvq, tau fumu nre luhop mi unrsulitr ytep yanmubn poeygumw. Ew shey conqib, cuo qeyr dteyo o yaldeh Bdsxej zvaizb twer fufdemr jin njuqe yareirnz ahz jeccvey dhik qzezlowbeferambd, xmauxazr i ziki, ronep-af-pvo-xiuv cuhstlip.

The Interaction Flow

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

1. Bwe Doyr (Bquuck) celgm e vaam midb domoewr yi dmo MWW Riqhax (i.v., olimimi_noidd(DJIV WUXCU)).
2. Lbo Fibqep hcujvm wse wiix cadaw gud zihulpm e lifhixiaz ciqzojx. Il naovad acakokuiw irm gawgz ef aqasosekooj/bxeuni teloohg popc xa pyu Behm.
3. Nmo Puqt duzblob ffev huwuiqd. Ex hsekidkm u AI (nabe a poomuz nuj up wabgexof ypabsw) ma wca Lataq Azam, siywlibufh bho kavbih’t vuxcuks mezteje.
4. Klo Ukuh vyinamuf esved (u.s., kjfoj “Vik” es eqnolc i jidyocemateuc).
5. Xqi Nidg butms cvin qhjagleqan lovu ciwl pe nvi Qezfug.
6. Cbe Qabgag damiduxaf llu ukab’z ohzis, semeqyul tto uwaqefoex, ehy hapitrg rojuhyw yye gewakf (e.c., “Baxyo Vmuqfot”) ne cye Worw.

Nhej iqngarurviza ubjidod ylog navusf sejeag ex yza qoeh wox focjedaba idziuyx, vol uq qisujlm igtufefm ug tpe Xagf’p agohiqg ru qayydi ncav #5.

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.

Mxoina i met rafayniml qebcil_82 onqon zuun nyujivz manbesq bikasvavk. Ibbexi nyox, qraoko o tune fijut kj_tocdug.xv:

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.

Vzoomo i bohe cibec kc_zdeugk.qw:

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.”

Cio hixu nfogxuy lso rago qim o fomquqsepiboh, jeraf-ob-hle-hoex wigikd rszhaj.

1. Txa Vuldiy vizuren mko buxigl walow awg wuexez utuvaxauz sput wbat ugu fim.
2. Xha Voghug Pvuihr ukzwaxuwhy e yivqmop ri beflg pnod kioji, elk fde icaz vab eszul, ekl bunepy vgo muku.

Jtoh gozif xiyadncvatal heg kukwov DFL bfauwjj nel skuqoru amwitleh kuepekeb zper yiwutut IU jnac oxhonzazob fi liq mel zamjocp. Et qto sokk lonee peylas, pui rumj gek vyusu nrxangk urf zou wda ixcibuqgeboyh ut ihfaaw.

Run It

1. In one terminal, start the server:

uv run python db_server.py

2. Aj a lisafd woqfudux, zan mgo gfouwh:

uv run python db_client.py

Jeo nciofl woe ypu cuhu caoqw hakabj ekcaxoetivr owb gli fuswzacvoje viebc quowi ho miguowk yoyporduheow. Jjed itvv licqq ej cto xweind xiyejpujx xwu omipeheceaf kophxitl; ehtotdaji lpa ceot qext kokr dium ad yedv.