rpatil524 · pull · Jun 17, 2025 · Jun 17, 2025 · Jun 17, 2025 · Jun 17, 2025
diff --git a/README.md b/README.md
@@ -30,6 +30,9 @@
     - [Prompts](#prompts)
     - [Images](#images)
     - [Context](#context)
+    - [Completions](#completions)
+    - [Elicitation](#elicitation)
+    - [Authentication](#authentication)
   - [Running Your Server](#running-your-server)
     - [Development Mode](#development-mode)
     - [Claude Desktop Integration](#claude-desktop-integration)
@@ -73,7 +76,7 @@ The Model Context Protocol allows applications to provide context for LLMs in a
 
 ### Adding MCP to your python project
 
-We recommend using [uv](https://docs.astral.sh/uv/) to manage your Python projects. 
+We recommend using [uv](https://docs.astral.sh/uv/) to manage your Python projects.
 
 If you haven't created a uv-managed project yet, create one:
 
@@ -310,6 +313,112 @@ async def long_task(files: list[str], ctx: Context) -> str:
     return "Processing complete"
 ```
 
+### Completions
+
+MCP supports providing completion suggestions for prompt arguments and resource template parameters. With the context parameter, servers can provide completions based on previously resolved values:
+
+Client usage:
+```python
+from mcp.client.session import ClientSession
+from mcp.types import ResourceTemplateReference
+
+
+async def use_completion(session: ClientSession):
+    # Complete without context
+    result = await session.complete(
+        ref=ResourceTemplateReference(
+            type="ref/resource", uri="github://repos/{owner}/{repo}"
+        ),
+        argument={"name": "owner", "value": "model"},
+    )
+
+    # Complete with context - repo suggestions based on owner
+    result = await session.complete(
+        ref=ResourceTemplateReference(
+            type="ref/resource", uri="github://repos/{owner}/{repo}"
+        ),
+        argument={"name": "repo", "value": "test"},
+        context_arguments={"owner": "modelcontextprotocol"},
+    )
+```
+
+Server implementation:
+```python
+from mcp.server import Server
+from mcp.types import (
+    Completion,
+    CompletionArgument,
+    CompletionContext,
+    PromptReference,
+    ResourceTemplateReference,
+)
+
+server = Server("example-server")
+
+
+@server.completion()
+async def handle_completion(
+    ref: PromptReference | ResourceTemplateReference,
+    argument: CompletionArgument,
+    context: CompletionContext | None,
+) -> Completion | None:
+    if isinstance(ref, ResourceTemplateReference):
+        if ref.uri == "github://repos/{owner}/{repo}" and argument.name == "repo":
+            # Use context to provide owner-specific repos
+            if context and context.arguments:
+                owner = context.arguments.get("owner")
+                if owner == "modelcontextprotocol":
+                    repos = ["python-sdk", "typescript-sdk", "specification"]
+                    # Filter based on partial input
+                    filtered = [r for r in repos if r.startswith(argument.value)]
+                    return Completion(values=filtered)
+    return None
+```
+### Elicitation
+
+Request additional information from users during tool execution:
+
+```python
+from mcp.server.fastmcp import FastMCP, Context
+from mcp.server.elicitation import (
+    AcceptedElicitation,
+    DeclinedElicitation,
+    CancelledElicitation,
+)
+from pydantic import BaseModel, Field
+
+mcp = FastMCP("Booking System")
+
+
+@mcp.tool()
+async def book_table(date: str, party_size: int, ctx: Context) -> str:
+    """Book a table with confirmation"""
+
+    # Schema must only contain primitive types (str, int, float, bool)
+    class ConfirmBooking(BaseModel):
+        confirm: bool = Field(description="Confirm booking?")
+        notes: str = Field(default="", description="Special requests")
+
+    result = await ctx.elicit(
+        message=f"Confirm booking for {party_size} on {date}?", schema=ConfirmBooking
+    )
+
+    match result:
+        case AcceptedElicitation(data=data):
+            if data.confirm:
+                return f"Booked! Notes: {data.notes or 'None'}"
+            return "Booking cancelled"
+        case DeclinedElicitation():
+            return "Booking declined"
+        case CancelledElicitation():
+            return "Booking cancelled"
+```
+
+The `elicit()` method returns an `ElicitationResult` with:
+- `action`: "accept", "decline", or "cancel"
+- `data`: The validated response (only when accepted)
+- `validation_error`: Any validation error message
+
 ### Authentication
 
 Authentication can be used by servers that want to expose tools accessing protected resources.

diff --git a/src/mcp/client/session.py b/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,16 @@ 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() if self._elicitation_callback is not _default_elicitation_callback else None
+        )
         roots = (
             # TODO: Should this be based on whether we
             # _will_ send notifications, or only whether
@@ -128,6 +151,7 @@ async def initialize(self) -> types.InitializeResult:
                         protocolVersion=types.LATEST_PROTOCOL_VERSION,
                         capabilities=types.ClientCapabilities(
                             sampling=sampling,
+                            elicitation=elicitation,
                             experimental=None,
                             roots=roots,
                         ),
@@ -304,15 +328,21 @@ async def complete(
         self,
         ref: types.ResourceTemplateReference | types.PromptReference,
         argument: dict[str, str],
+        context_arguments: dict[str, str] | None = None,
     ) -> types.CompleteResult:
         """Send a completion/complete request."""
+        context = None
+        if context_arguments is not None:
+            context = types.CompletionContext(arguments=context_arguments)
+
         return await self.send_request(
             types.ClientRequest(
                 types.CompleteRequest(
                     method="completion/complete",
                     params=types.CompleteRequestParams(
                         ref=ref,
                         argument=types.CompletionArgument(**argument),
+                        context=context,
                     ),
                 )
             ),
@@ -356,6 +386,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)
+
             ca
A3E2
se types.ListRootsRequest():
                 with responder:
                     response = await self._list_roots_callback(ctx)

diff --git a/src/mcp/server/elicitation.py b/src/mcp/server/elicitation.py
@@ -0,0 +1,111 @@
+"""Elicitation utilities for MCP servers."""
+
+from __future__ import annotations
+
+import types
+from typing import Generic, Literal, TypeVar, Union, get_args, get_origin
+
+from pydantic import BaseModel
+from pydantic.fields import FieldInfo
+
+from mcp.server.session import ServerSession
+from mcp.types import RequestId
+
+ElicitSchemaModelT = TypeVar("ElicitSchemaModelT", bound=BaseModel)
+
+
+class AcceptedElicitation(BaseModel, Generic[ElicitSchemaModelT]):
+    """Result when user accepts the elicitation."""
+
+    action: Literal["accept"] = "accept"
+    data: ElicitSchemaModelT
+
+
+class DeclinedElicitation(BaseModel):
+    """Result when user declines the elicitation."""
+
+    action: Literal["decline"] = "decline"
+
+
+class CancelledElicitation(BaseModel):
+    """Result when user cancels the elicitation."""
+
+    action: Literal["cancel"] = "cancel"
+
+
+ElicitationResult = AcceptedElicitation[ElicitSchemaModelT] | DeclinedElicitation | CancelledElicitation
+
+
+# Primitive types allowed in elicitation schemas
+_ELICITATION_PRIMITIVE_TYPES = (str, int, float, bool)
+
+
+def _validate_elicitation_schema(schema: type[BaseModel]) -> None:
+    """Validate that a Pydantic model only contains primitive field types."""
+    for field_name, field_info in schema.model_fields.items():
+        if not _is_primitive_field(field_info):
+            raise TypeError(
+                f"Elicitation schema field '{field_name}' must be a primitive type "
+                f"{_ELICITATION_PRIMITIVE_TYPES} or Optional of these types. "
+                f"Complex types like lists, dicts, or nested models are not allowed."
+            )
+
+
+def _is_primitive_field(field_info: FieldInfo) -> bool:
+    """Check if a field is a primitive type allowed in elicitation schemas."""
+    annotation = field_info.annotation
+
+    # Handle None type
+    if annotation is types.NoneType:
+        return True
+
+    # Handle basic primitive types
+    if annotation in _ELICITATION_PRIMITIVE_TYPES:
+        return True
+
+    # Handle Union types
+    origin = get_origin(annotation)
+    if origin is Union or origin is types.UnionType:
+        args = get_args(annotation)
+        # All args must be primitive types or None
+        return all(arg is types.NoneType or arg in _ELICITATION_PRIMITIVE_TYPES for arg in args)
+
+    return False
+
+
+async def elicit_with_validation(
+    session: ServerSession,
+    message: str,
+    schema: type[ElicitSchemaModelT],
+    related_request_id: RequestId | None = None,
+) -> ElicitationResult[ElicitSchemaModelT]:
+    """Elicit information from the client/user with schema validation.
+
+    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.
+    """
+    # Validate that schema only contains primitive types and fail loudly if not
+    _validate_elicitation_schema(schema)
+
+    json_schema = schema.model_json_schema()
+
+    result = await session.elicit(
+        message=message,
+        requestedSchema=json_schema,
+        related_request_id=related_request_id,
+    )
+
+    if result.action == "accept" and result.content:
+        # Validate and parse the content using the schema
+        validated_data = schema.model_validate(result.content)
+        return AcceptedElicitation(data=validated_data)
+    elif result.action == "decline":
+        return DeclinedElicitation()
+    elif result.action == "cancel":
+        return CancelledElicitation()
+    else:
+        # This should never happen, but handle it just in case
+        raise ValueError(f"Unexpected elicitation action: {result.action}")