Elicitation

Author: dsp-ant

src/mcp/client/session.py

@@ -22,6 +22,14 @@ async def __call__(
     ) -> types.CreateMessageResult | types.ErrorData: ...
 
 
+class ElicitationFnT(Protocol):
+    async def __call__(
+        self,
+        context: RequestContext["ClientSession", Any],
+        params: types.ElicitRequestParams,
+    ) -> types.ElicitResult | types.ErrorData: ...
+
+
 class ListRootsFnT(Protocol):
     async def __call__(
         self, context: RequestContext["ClientSession", Any]
@@ -58,6 +66,16 @@ async def _default_sampling_callback(
     )
 
 
+async def _default_elicitation_callback(
+    context: RequestContext["ClientSession", Any],
+    params: types.ElicitRequestParams,
+) -> types.ElicitResult | types.ErrorData:
+    return types.ErrorData(
+        code=types.INVALID_REQUEST,
+        message="Elicitation not supported",
+    )
+
+
 async def _default_list_roots_callback(
     context: RequestContext["ClientSession", Any],
 ) -> types.ListRootsResult | types.ErrorData:
@@ -91,6 +109,7 @@ def __init__(
         write_stream: MemoryObjectSendStream[SessionMessage],
         read_timeout_seconds: timedelta | None = None,
         sampling_callback: SamplingFnT | None = None,
+        elicitation_callback: ElicitationFnT | None = None,
         list_roots_callback: ListRootsFnT | None = None,
         logging_callback: LoggingFnT | None = None,
         message_handler: MessageHandlerFnT | None = None,
@@ -105,12 +124,14 @@ def __init__(
         )
         self._client_info = client_info or DEFAULT_CLIENT_INFO
         self._sampling_callback = sampling_callback or _default_sampling_callback
+        self._elicitation_callback = elicitation_callback or _default_elicitation_callback
         self._list_roots_callback = list_roots_callback or _default_list_roots_callback
         self._logging_callback = logging_callback or _default_logging_callback
         self._message_handler = message_handler or _default_message_handler
 
     async def initialize(self) -> types.InitializeResult:
         sampling = types.SamplingCapability() if self._sampling_callback is not _default_sampling_callback else None
+        elicitation = types.ElicitationCapability()
         roots = (
             # TODO: Should this be based on whether we
             # _will_ send notifications, or only whether
@@ -128,6 +149,7 @@ async def initialize(self) -> types.InitializeResult:
                         protocolVersion=types.LATEST_PROTOCOL_VERSION,
                         capabilities=types.ClientCapabilities(
                             sampling=sampling,
+                            elicitation=elicitation,
                             experimental=None,
                             roots=roots,
                         ),
@@ -362,6 +384,12 @@ async def _received_request(self, responder: RequestResponder[types.ServerReques
                     client_response = ClientResponse.validate_python(response)
                     await responder.respond(client_response)
 
+            case types.ElicitRequest(params=params):
+                with responder:
+                    response = await self._elicitation_callback(ctx, params)
+                    client_response = ClientResponse.validate_python(response)
+                    await responder.respond(client_response)
+
             case types.ListRootsRequest():
                 with responder:
                     response = await self._list_roots_callback(ctx)

src/mcp/server/fastmcp/server.py

@@ -972,6 +972,39 @@ async def read_resource(self, uri: str | AnyUrl) -> Iterable[ReadResourceContent
         assert self._fastmcp is not None, "Context is not available outside of a request"
         return await self._fastmcp.read_resource(uri)
 
+    async def elicit(
+        self,
+        message: str,
+        requestedSchema: dict[str, Any],
+    ) -> dict[str, Any]:
+        """Elicit information from the client/user.
+
+        This method can be used to interactively ask for additional information from the
+        client within a tool's execution.
+        The client might display the message to the user and collect a response
+        according to the provided schema. Or in case a client is an agent, it might
+        decide how to handle the elicitation -- either by asking the user or
+        automatically generating a response.
+
+        Args:
+            message: The message to present to the user
+            requestedSchema: JSON Schema defining the expected response structure
+
+        Returns:
+            The user's response as a dict matching the request schema structure
+
+        Raises:
+            ValueError: If elicitation is not supported by the client or fails
+        """
+
+        result = await self.request_context.session.elicit(
+            message=message,
+            requestedSchema=requestedSchema,
+            related_request_id=self.request_id,
+        )
+
+        return result.response
+
     async def log(
         self,
         level: Literal["debug", "info", "warning", "error"],

src/mcp/server/session.py

@@ -121,6 +121,10 @@ def check_client_capability(self, capability: types.ClientCapabilities) -> bool:
             if client_caps.sampling is None:
                 return False
 
+        if capability.elicitation is not None:
+            if client_caps.elicitation is None:
+                return False
+
         if capability.experimental is not None:
             if client_caps.experimental is None:
                 return False
@@ -251,6 +255,35 @@ async def list_roots(self) -> types.ListRootsResult:
             types.ListRootsResult,
         )
 
+    async def elicit(
+        self,
+        message: str,
+        requestedSchema: dict[str, Any],
+        related_request_id: types.RequestId | None = None,
+    ) -> types.ElicitResult:
+        """Send an elicitation/create request.
+
+        Args:
+            message: The message to present to the user
+            requestedSchema: JSON Schema defining the expected response structure
+
+        Returns:
+            The client's response
+        """
+        return await self.send_request(
+            types.ServerRequest(
+                types.ElicitRequest(
+                    method="elicitation/create",
+                    params=types.ElicitRequestParams(
+                        message=message,
+                        requestedSchema=requestedSchema,
+                    ),
+                )
+            ),
+            types.ElicitResult,
+            metadata=ServerMessageMetadata(related_request_id=related_request_id),
+        )
+
     async def send_ping(self) -> types.EmptyResult:
         """Send a ping request."""
         return await self.send_request(

src/mcp/types.py

@@ -216,13 +216,21 @@ class SamplingCapability(BaseModel):
     model_config = ConfigDict(extra="allow")
 
 
+class ElicitationCapability(BaseModel):
+    """Capability for elicitation operations."""
+
+    model_config = ConfigDict(extra="allow")
+
+
 class ClientCapabilities(BaseModel):
     """Capabilities a client may support."""
 
     experimental: dict[str, dict[str, Any]] | None = None
     """Experimental, non-standard capabilities that the client supports."""
     sampling: SamplingCapability | None = None
     """Present if the client supports sampling from an LLM."""
+    elicitation: ElicitationCapability | None = None
+    """Present if the client supports elicitation from the user."""
     roots: RootsCapability | None = None
     """Present if the client supports listing roots."""
     model_config = ConfigDict(extra="allow")
@@ -1186,11 +1194,38 @@ class ClientNotification(
     pass
 
 
-class ClientResult(RootModel[EmptyResult | CreateMessageResult | ListRootsResult]):
+class ElicitRequestParams(RequestParams):
+    """Parameters for elicitation requests."""
+
+    message: str
+    """The message to present to the user."""
+
+    requestedSchema: dict[str, Any]
+    """
+    A JSON Schema object defining the expected structure of the response.
+    """
+    model_config = ConfigDict(extra="allow")
+
+
+class ElicitRequest(Request[ElicitRequestParams, Literal["elicitation/create"]]):
+    """A request from the server to elicit information from the client."""
+
+    method: Literal["elicitation/create"]
+    params: ElicitRequestParams
+
+
+class ElicitResult(Result):
+    """The client's response to an elicitation/create request from the server."""
+
+    response: dict[str, Any]
+    """The response from the client, matching the structure of requestedSchema."""
+
+
+class ClientResult(RootModel[EmptyResult | CreateMessageResult | ListRootsResult | ElicitResult]):
     pass
 
 
-class ServerRequest(RootModel[PingRequest | CreateMessageRequest | ListRootsRequest]):
+class ServerRequest(RootModel[PingRequest | CreateMessageRequest | ListRootsRequest | ElicitRequest]):
     pass
 
 

tests/server/fastmcp/test_integration.py

@@ -21,7 +21,7 @@
 from mcp.client.session import ClientSession
 from mcp.client.sse import sse_client
 from mcp.client.streamable_http import streamablehttp_client
-from mcp.server.fastmcp import FastMCP
+from mcp.server.fastmcp import Context, FastMCP
 from mcp.server.fastmcp.resources import FunctionResource
 from mcp.shared.context import RequestContext
 from mcp.types import (
@@ -30,6 +30,7 @@
     CompletionContext,
     CreateMessageRequestParams,
     CreateMessageResult,
+    ElicitResult,
     GetPromptResult,
     InitializeResult,
     LoggingMessageNotification,
@@ -98,6 +99,23 @@ def make_fastmcp_app():
     def echo(message: str) -> str:
         return f"Echo: {message}"
 
+    # Add a tool that uses elicitation
+    @mcp.tool(description="A tool that uses elicitation")
+    async def ask_user(prompt: str, ctx: Context) -> str:
+        schema = {
+            "type": "object",
+            "properties": {
+                "answer": {"type": "string"},
+            },
+            "required": ["answer"],
+        }
+
+        response = await ctx.elicit(
+            message=f"Tool wants to ask: {prompt}",
+            requestedSchema=schema,
+        )
+        return f"User answered: {response['answer']}"
+
     # Create the SSE app
     app = mcp.sse_app()
 
@@ -937,3 +955,32 @@ async def message_handler(message):
         ) as session:
             # Run the common test suite with HTTP-specific test suffix
             await call_all_mcp_features(session, collector)
+
+
+@pytest.mark.anyio
+async def test_elicitation_feature(server: None, server_url: str) -> None:
+    """Test the elicitation feature."""
+
+    # Create a custom handler for elicitation requests
+    async def elicitation_callback(context, params):
+        # Verify the elicitation parameters
+        if params.message == "Tool wants to ask: What is your name?":
+            return ElicitResult(response={"answer": "Test User"})
+        else:
+            raise ValueError("Unexpected elicitation message")
+
+    # Connect to the server with our custom elicitation handler
+    async with sse_client(server_url + "/sse") as streams:
+        async with ClientSession(*streams, elicitation_callback=elicitation_callback) as session:
+            # First initialize the session
+            result = await session.initialize()
+            assert isinstance(result, InitializeResult)
+            assert result.serverInfo.name == "NoAuthServer"
+
+            # Call the tool that uses elicitation
+            tool_result = await session.call_tool("ask_user", {"prompt": "What is your name?"})
+            # Verify the result
+            assert len(tool_result.content) == 1
+            assert isinstance(tool_result.content[0], TextContent)
+            # # The test should only succeed with the successful elicitation response
+            assert tool_result.content[0].text == "User answered: Test User"