Implement elicitation feature for FastMCP following sampling pattern

Added comprehensive elicitation support including:
- Client-side handler infrastructure following sampling pattern
- Server-side Context.elicit() method with type safety and validation
- Support for primitive types (str, int, float, bool) with automatic wrapping
- Support for dataclasses and other complex types
- Pattern matching with AcceptedElicitation, DeclinedElicitation, CancelledElicitation
- Comprehensive atomic test coverage for all scenarios

🤖 Generated with [Claude Code](https://claude.ai/code)

Co-Authored-By: Claude <noreply@anthropic.com>

src/fastmcp/client/client.py

@@ -24,6 +24,7 @@ from fastmcp.client.roots import (
     RootsList,
     create_roots_callback,
 )
+from fastmcp.client.elicitation import ElicitationHandler, create_elicitation_callback
 from fastmcp.client.sampling import SamplingHandler, create_sampling_callback
 from fastmcp.exceptions import ToolError
 from fastmcp.server import FastMCP
@@ -52,6 +53,7 @@ __all__ = [
     "LogHandler",
     "MessageHandler",
     "SamplingHandler",
+    "ElicitationHandler",
     "ProgressHandler",
 ]
 
@@ -141,6 +143,7 @@ class Client(Generic[ClientTransportT]):
         # Common args
         roots: RootsList | RootsHandler | None = None,
         sampling_handler: SamplingHandler | None = None,
+        elicitation_handler: ElicitationHandler | None = None,
         log_handler: LogHandler | None = None,
         message_handler: MessageHandler | None = None,
         progress_handler: ProgressHandler | None = None,
@@ -193,6 +196,11 @@ class Client(Generic[ClientTransportT]):
                 sampling_handler
             )
 
+        if elicitation_handler is not None:
+            self._session_kwargs["elicitation_callback"] = create_elicitation_callback(
+                elicitation_handler
+            )
+
         # session context management
         self._session: ClientSession | None = None
         self._exit_stack: AsyncExitStack | None = None
@@ -231,6 +239,12 @@ class Client(Generic[ClientTransportT]):
             sampling_callback
         )
 
+    def set_elicitation_callback(self, elicitation_callback: ElicitationHandler) -> None:
+        """Set the elicitation callback for the client."""
+        self._session_kwargs["elicitation_callback"] = create_elicitation_callback(
+            elicitation_callback
+        )
+
     def is_connected(self) -> bool:
         """Check if the client is currently connected."""
         return self._session is not None

src/fastmcp/client/elicitation.py

@@ -0,0 +1,47 @@
+from __future__ import annotations
+
+import inspect
+from collections.abc import Awaitable, Callable
+from typing import Any, TypeAlias
+
+import mcp.types
+from mcp import ClientSession
+from mcp.client.session import ElicitationFnT
+from mcp.shared.context import LifespanContextT, RequestContext
+from mcp.types import ElicitRequestParams, ElicitResult
+
+__all__ = ["ElicitRequestParams", "ElicitResult", "ElicitationHandler"]
+
+
+ElicitationHandler: TypeAlias = Callable[
+    [
+        str,  # message
+        dict[str, Any],  # requested_schema
+        RequestContext[ClientSession, LifespanContextT],
+    ],
+    ElicitResult | Awaitable[ElicitResult],
+]
+
+
+def create_elicitation_callback(
+    elicitation_handler: ElicitationHandler,
+) -> ElicitationFnT:
+    async def _elicitation_handler(
+        context: RequestContext[ClientSession, LifespanContextT],
+        params: ElicitRequestParams,
+    ) -> ElicitResult | mcp.types.ErrorData:
+        try:
+            result = elicitation_handler(
+                params.message, params.requestedSchema, context
+            )
+            if inspect.isawaitable(result):
+                result = await result
+
+            return result
+        except Exception as e:
+            return mcp.types.ErrorData(
+                code=mcp.types.INTERNAL_ERROR,
+                message=str(e),
+            )
+
+    return _elicitation_handler

src/fastmcp/server/context.py

@@ -1,12 +1,13 @@
-from __future__ import annotations as _annotations
+from __future__ import annotations
 
 import warnings
 from collections.abc import Generator
 from contextlib import contextmanager
 from contextvars import ContextVar, Token
 from dataclasses import dataclass
+from typing import TypeVar, cast
 
-from mcp import LoggingLevel
+from mcp import LoggingLevel, ServerSession
 from mcp.server.lowlevel.helper_types import ReadResourceContents
 from mcp.server.lowlevel.server import request_ctx
 from mcp.shared.context import RequestContext
@@ -23,12 +24,20 @@ from starlette.requests import Request
 
 import fastmcp.server.dependencies
 from fastmcp import settings
+from fastmcp.server.elicitation import (
+    AcceptedElicitation,
+    CancelledElicitation,
+    DeclinedElicitation,
+    PrimitiveElicitationType,
+    get_elicitation_schema,
+)
 from fastmcp.server.server import FastMCP
 from fastmcp.utilities.logging import get_logger
-from fastmcp.utilities.types import MCPContent
+from fastmcp.utilities.types import MCPContent, get_cached_typeadapter
 
 logger = get_logger(__name__)
 
+T = TypeVar("T")
 _current_context: ContextVar[Context | None] = ContextVar("context", default=None)
 
 
@@ -105,6 +114,56 @@ class Context:
         except LookupError:
             raise ValueError("Context is not available outside of a request")
 
+    @property
+    def session(self) -> ServerSession:
+        """Access to the underlying session for advanced usage."""
+        return self.request_context.session
+
+    @property
+    def client_id(self) -> str | None:
+        """Get the client ID if available."""
+        return (
+            getattr(self.request_context.meta, "client_id", None)
+            if self.request_context.meta
+            else None
+        )
+
+    @property
+    def request_id(self) -> str:
+        """Get the unique ID for this request."""
+        return str(self.request_context.request_id)
+
+    @property
+    def session_id(self) -> str | None:
+        """Get the MCP session ID for HTTP transports.
+
+        Returns the session ID that can be used as a key for session-based
+        data storage (e.g., Redis) to share data between tool calls within
+        the same client session.
+
+        Returns:
+            The session ID for HTTP transports (SSE, StreamableHTTP), or None
+            for stdio and in-memory transports which don't use session IDs.
+
+        Example:
+            ```python
+            @server.tool
+            def store_data(data: dict, ctx: Context) -> str:
+                if session_id := ctx.session_id:
+                    redis_client.set(f"session:{session_id}:data", json.dumps(data))
+                    return f"Data stored for session {session_id}"
+                return "No session ID available (stdio/memory transport)"
+            ```
+        """
+        try:
+            from fastmcp.server.dependencies import get_http_headers
+
+            headers = get_http_headers(include_all=True)
+            return headers.get("mcp-session-id")
+        except RuntimeError:
+            # No HTTP context available (stdio/in-memory transport)
+            return None
+
     async def report_progress(
         self, progress: float, total: float | None = None, message: str | None = None
     ) -> None:
@@ -124,7 +183,7 @@ class Context:
         if progress_token is None:
             return
 
-        await self.request_context.session.send_progress_notification(
+        await self.session.send_progress_notification(
             progress_token=progress_token,
             progress=progress,
             total=total,
@@ -160,60 +219,13 @@ class Context:
         """
         if level is None:
             level = "info"
-        await self.request_context.session.send_log_message(
-            level=level, data=message, logger=logger_name
-        )
-
-    @property
-    def client_id(self) -> str | None:
-        """Get the client ID if available."""
-        return (
-            getattr(self.request_context.meta, "client_id", None)
-            if self.request_context.meta
-            else None
+        await self.session.send_log_message(
+            level=level,
+            data=message,
+            logger=logger_name,
+            related_request_id=self.request_id,
         )
 
-    @property
-    def request_id(self) -> str:
-        """Get the unique ID for this request."""
-        return str(self.request_context.request_id)
-
-    @property
-    def session_id(self) -> str | None:
-        """Get the MCP session ID for HTTP transports.
-
-        Returns the session ID that can be used as a key for session-based
-        data storage (e.g., Redis) to share data between tool calls within
-        the same client session.
-
-        Returns:
-            The session ID for HTTP transports (SSE, StreamableHTTP), or None
-            for stdio and in-memory transports which don't use session IDs.
-
-        Example:
-            ```python
-            @server.tool
-            def store_data(data: dict, ctx: Context) -> str:
-                if session_id := ctx.session_id:
-                    redis_client.set(f"session:{session_id}:data", json.dumps(data))
-                    return f"Data stored for session {session_id}"
-                return "No session ID available (stdio/memory transport)"
-            ```
-        """
-        try:
-            from fastmcp.server.dependencies import get_http_headers
-
-            headers = get_http_headers(include_all=True)
-            return headers.get("mcp-session-id")
-        except RuntimeError:
-            # No HTTP context available (stdio/in-memory transport)
-            return None
-
-    @property
-    def session(self):
-        """Access to the underlying session for advanced usage."""
-        return self.request_context.session
-
     # Convenience methods for common log levels
     async def debug(self, message: str, logger_name: str | None = None) -> None:
         """Send a debug log message."""
@@ -233,7 +245,7 @@ class Context:
 
     async def list_roots(self) -> list[Root]:
         """List the roots available to the server, as indicated by the client."""
-        result = await self.request_context.session.list_roots()
+        result = await self.session.list_roots()
         return result.roots
 
     async def sample(
@@ -269,16 +281,73 @@ class Context:
                 for m in messages
             ]
 
-        result: CreateMessageResult = await self.request_context.session.create_message(
+        result: CreateMessageResult = await self.session.create_message(
             messages=sampling_messages,
             system_prompt=system_prompt,
             temperature=temperature,
             max_tokens=max_tokens,
             model_preferences=self._parse_model_preferences(model_preferences),
+            related_request_id=self.request_id,
         )
 
         return result.content
 
+    async def elicit(
+        self,
+        message: str,
+        response_type: type[T] | None = None,
+    ) -> AcceptedElicitation[T] | DeclinedElicitation | CancelledElicitation:
+        """
+        Send an elicitation request to the client and await the response.
+
+        Call this method at any time to request additional information from
+        the user through the client. The client must support elicitation,
+        or the request will error.
+
+        Note that the MCP protocol only supports simple object schemas with
+        primitive types. You can provide a dataclass, TypedDict, or BaseModel to
+        comply. If you provide a primitive type, an object schema with a single
+        "value" field will be generated for the MCP interaction and
+        automatically deconstructed into the primitive type upon response.
+
+        Args:
+            message: A human-readable message explaining what information is needed
+            response_type: The type of the response, which should be a primitive
+                type or dataclass or BaseModel. If it is a primitive type, an
+                object schema with a single "value" field will be generated.
+        """
+        if response_type is None:
+            response_type = str  # type: ignore
+
+        if response_type in {bool, int, float, str}:
+            response_type = PrimitiveElicitationType[response_type]  # type: ignore
+
+        requested_schema = get_elicitation_schema(response_type)  # type: ignore
+
+        result = await self.session.elicit(
+            message=message,
+            requestedSchema=requested_schema,
+            related_request_id=self.request_id,
+        )
+
+        if result.action == "accept" and result.content:
+            type_adapter = get_cached_typeadapter(response_type)
+            validated_data = cast(
+                T | PrimitiveElicitationType[T],
+                type_adapter.validate_python(result.content),
+            )
+            if isinstance(validated_data, PrimitiveElicitationType):
+                return AcceptedElicitation[T](data=validated_data.value)
+            else:
+                return AcceptedElicitation[T](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}")
+
     def get_http_request(self) -> Request:
         """Get the active starlette request."""
 

src/fastmcp/server/elicitation.py

@@ -0,0 +1,133 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import Any, Generic, Literal, TypeVar
+
+from mcp.server.elicitation import (
+    CancelledElicitation,
+    DeclinedElicitation,
+)
+from pydantic import BaseModel
+
+from fastmcp.utilities.json_schema import compress_schema
+from fastmcp.utilities.logging import get_logger
+from fastmcp.utilities.types import get_cached_typeadapter
+
+__all__ = [
+    "AcceptedElicitation",
+    "CancelledElicitation",
+    "DeclinedElicitation",
+    "get_elicitation_schema",
+    "PrimitiveElicitationType",
+]
+
+logger = get_logger(__name__)
+
+T = TypeVar("T")
+
+
+# we can't use the low-level AcceptedElicitation because it only works with BaseModels
+class AcceptedElicitation(BaseModel, Generic[T]):
+    """Result when user accepts the elicitation."""
+
+    action: Literal["accept"] = "accept"
+    data: T
+
+
+@dataclass
+class PrimitiveElicitationType(Generic[T]):
+    value: T
+
+
+def get_elicitation_schema(response_type: type[T]) -> dict[str, Any]:
+    """Get the schema for an elicitation response.
+
+    Args:
+        response_type: The type of the response
+    """
+
+    schema = get_cached_typeadapter(response_type).json_schema()
+    schema = compress_schema(schema)
+
+    # Validate the schema to ensure it follows MCP elicitation requirements
+    validate_elicitation_json_schema(schema)
+
+    return schema
+
+
+def validate_elicitation_json_schema(schema: dict[str, Any]) -> None:
+    """Validate that a JSON schema follows MCP elicitation requirements.
+
+    This ensures the schema is compatible with MCP elicitation requirements:
+    - Must be an object schema
+    - Must only contain primitive field types (string, number, integer, boolean)
+    - Must be flat (no nested objects or arrays of objects)
+    - Only primitive types and their nullable variants are allowed
+
+    Args:
+        schema: The JSON schema to validate
+
+    Raises:
+        TypeError: If the schema doesn't meet MCP elicitation requirements
+    """
+    ALLOWED_TYPES = {"string", "number", "integer", "boolean"}
+
+    # Check that the schema is an object
+    if schema.get("type") != "object":
+        raise TypeError(
+            f"Elicitation schema must be an object schema, got type '{schema.get('type')}'. "
+            "Elicitation schemas are limited to flat objects with primitive properties only."
+        )
+
+    properties = schema.get("properties", {})
+    if not properties:
+        raise TypeError(
+            "Elicitation schema must have at least one property. "
+            "Empty object schemas are not allowed."
+        )
+
+    for prop_name, prop_schema in properties.items():
+        prop_type = prop_schema.get("type")
+
+        # Handle nullable types
+        if isinstance(prop_type, list):
+            if "null" in prop_type:
+                prop_type = [t for t in prop_type if t != "null"]
+                if len(prop_type) == 1:
+                    prop_type = prop_type[0]
+        elif prop_schema.get("nullable", False):
+            continue  # Nullable with no other type is fine
+
+        # Handle union types (oneOf/anyOf)
+        if "oneOf" in prop_schema or "anyOf" in prop_schema:
+            union_schemas = prop_schema.get("oneOf", []) + prop_schema.get("anyOf", [])
+            for union_schema in union_schemas:
+                union_type = union_schema.get("type")
+                if union_type not in ALLOWED_TYPES:
+                    raise TypeError(
+                        f"Elicitation schema field '{prop_name}' has union type '{union_type}' which is not "
+                        f"a primitive type. Only {ALLOWED_TYPES} are allowed in elicitation schemas."
+                    )
+            continue
+
+        # Check if it's a primitive type
+        if prop_type not in ALLOWED_TYPES:
+            raise TypeError(
+                f"Elicitation schema field '{prop_name}' has type '{prop_type}' which is not "
+                f"a primitive type. Only {ALLOWED_TYPES} are allowed in elicitation schemas."
+            )
+
+        # Check for nested objects or arrays of objects (not allowed)
+        if prop_type == "object":
+            raise TypeError(
+                f"Elicitation schema field '{prop_name}' is an object, but nested objects are not allowed. "
+                "Elicitation schemas must be flat objects with primitive properties only."
+            )
+
+        if prop_type == "array":
+            items_schema = prop_schema.get("items", {})
+            if items_schema.get("type") == "object":
+                raise TypeError(
+                    f"Elicitation schema field '{prop_name}' is an array of objects, but arrays of objects are not allowed. "
+                    "Elicitation schemas must be flat objects with primitive properties only."
+                )

src/fastmcp/utilities/json_schema_to_type.py

@@ -0,0 +1,492 @@
+from __future__ import annotations
+
+import hashlib
+import json
+import re
+from collections.abc import Callable, Mapping
+from copy import deepcopy
+from dataclasses import MISSING, field, make_dataclass
+from datetime import datetime
+from enum import Enum
+from typing import (
+    Annotated,
+    Any,
+    ForwardRef,
+    Literal,
+    Optional,
+    Union,
+)
+
+from pydantic import (
+    AnyUrl,
+    EmailStr,
+    Field,
+    Json,
+    StringConstraints,
+    model_validator,
+)
+from typing_extensions import NotRequired, TypedDict
+
+__all__ = ["jsonschema_to_type", "JSONSchema"]
+
+
+FORMAT_TYPES: dict[str, Any] = {
+    "date-time": datetime,
+    "email": EmailStr,
+    "uri": AnyUrl,
+    "json": Json,
+}
+
+_classes: dict[tuple[str, Any], type | None] = {}
+
+
+def jsonschema_to_type(
+    schema: Mapping[str, Any],
+    name: str | None = None,
+) -> type:
+    """Convert JSON schema to appropriate Python type with validation.
+
+    Args:
+        schema: A JSON Schema dictionary defining the type structure and validation rules
+        name: Optional name for object schemas. Only allowed when schema type is "object".
+            If not provided for objects, name will be inferred from schema's "title"
+            property or default to "Root".
+
+    Returns:
+        A Python type (typically a dataclass for objects) with Pydantic validation
+
+    Raises:
+        ValueError: If a name is provided for a non-object schema
+
+    Examples:
+        Create a dataclass from an object schema:
+        ```python
+        schema = {
+            "type": "object",
+            "title": "Person",
+            "properties": {
+                "name": {"type": "string", "minLength": 1},
+                "age": {"type": "integer", "minimum": 0},
+                "email": {"type": "string", "format": "email"}
+            },
+            "required": ["name", "age"]
+        }
+
+        Person = jsonschema_to_type(schema)
+        # Creates a dataclass with name, age, and optional email fields:
+        # @dataclass
+        # class Person:
+        #     name: str
+        #     age: int
+        #     email: str | None = None
+        ```
+        Person(name="John", age=30)
+
+        Create a scalar type with constraints:
+        ```python
+        schema = {
+            "type": "string",
+            "minLength": 3,
+            "pattern": "^[A-Z][a-z]+$"
+        }
+
+        NameType = jsonschema_to_type(schema)
+        # Creates Annotated[str, StringConstraints(min_length=3, pattern="^[A-Z][a-z]+$")]
+
+        @dataclass
+        class Name:
+            name: NameType
+        ```
+    """
+    # Always use the top-level schema for references
+    if schema.get("type") == "object":
+        return _create_dataclass(schema, name, schemas=schema)
+    elif name:
+        raise ValueError(f"Can not apply name to non-object schema: {name}")
+    return _schema_to_type(schema, schemas=schema)
+
+
+def _hash_schema(schema: Mapping[str, Any]) -> str:
+    """Generate a deterministic hash for schema caching."""
+    return hashlib.sha256(json.dumps(schema, sort_keys=True).encode()).hexdigest()
+
+
+def _resolve_ref(ref: str, schemas: Mapping[str, Any]) -> Mapping[str, Any]:
+    """Resolve JSON Schema reference to target schema."""
+    path = ref.replace("#/", "").split("/")
+    current = schemas
+    for part in path:
+        current = current.get(part, {})
+    return current
+
+
+def _create_string_type(schema: Mapping[str, Any]) -> type | Annotated[Any, ...]:
+    """Create string type with optional constraints."""
+    if "const" in schema:
+        return Literal[schema["const"]]  # type: ignore
+
+    if fmt := schema.get("format"):
+        if fmt == "uri":
+            return AnyUrl
+        elif fmt == "uri-reference":
+            return str
+        return FORMAT_TYPES.get(fmt, str)
+
+    constraints = {
+        k: v
+        for k, v in {
+            "min_length": schema.get("minLength"),
+            "max_length": schema.get("maxLength"),
+            "pattern": schema.get("pattern"),
+        }.items()
+        if v is not None
+    }
+
+    return Annotated[str, StringConstraints(**constraints)] if constraints else str
+
+
+def _create_numeric_type(
+    base: type[int | float], schema: Mapping[str, Any]
+) -> type | Annotated[Any, ...]:
+    """Create numeric type with optional constraints."""
+    if "const" in schema:
+        return Literal[schema["const"]]  # type: ignore
+
+    constraints = {
+        k: v
+        for k, v in {
+            "gt": schema.get("exclusiveMinimum"),
+            "ge": schema.get("minimum"),
+            "lt": schema.get("exclusiveMaximum"),
+            "le": schema.get("maximum"),
+            "multiple_of": schema.get("multipleOf"),
+        }.items()
+        if v is not None
+    }
+
+    return Annotated[base, Field(**constraints)] if constraints else base
+
+
+def _create_enum(name: str, values: list[Any]) -> type | Enum:
+    """Create enum type from list of values."""
+    if all(isinstance(v, str) for v in values):
+        return Enum(name, {v.upper(): v for v in values})
+    return Literal[tuple(values)]  # type: ignore
+
+
+def _create_array_type(
+    schema: Mapping[str, Any], schemas: Mapping[str, Any]
+) -> type | Annotated[Any, ...]:
+    """Create list/set type with optional constraints."""
+    items = schema.get("items", {})
+    if isinstance(items, list):
+        # Handle positional item schemas
+        item_types = [_schema_to_type(s, schemas) for s in items]
+        combined = Union[tuple(item_types)]
+        base = list[combined]
+    else:
+        # Handle single item schema
+        item_type = _schema_to_type(items, schemas)
+        base = set if schema.get("uniqueItems") else list
+        base = base[item_type]
+
+    constraints = {
+        k: v
+        for k, v in {
+            "min_length": schema.get("minItems"),
+            "max_length": schema.get("maxItems"),
+        }.items()
+        if v is not None
+    }
+
+    return Annotated[base, Field(**constraints)] if constraints else base
+
+
+def _return_Any() -> Any:
+    return Any
+
+
+def _get_from_type_handler(
+    schema: Mapping[str, Any], schemas: Mapping[str, Any]
+) -> Callable[..., Any]:
+    """Get the appropriate type handler for the schema."""
+
+    type_handlers: dict[str, Callable[..., Any]] = {  # TODO
+        "string": lambda s: _create_string_type(s),  # type: ignore
+        "integer": lambda s: _create_numeric_type(int, s),  # type: ignore
+        "number": lambda s: _create_numeric_type(float, s),  # type: ignore
+        "boolean": lambda _: bool,  # type: ignore
+        "null": lambda _: type(None),  # type: ignore
+        "array": lambda s: _create_array_type(s, schemas),  # type: ignore
+        "object": lambda s: _create_dataclass(s, s.get("title"), schemas),  # type: ignore
+    }
+    return type_handlers.get(schema.get("type", None), _return_Any)
+
+
+def _schema_to_type(
+    schema: Mapping[str, Any],
+    schemas: Mapping[str, Any],
+) -> type:
+    """Convert schema to appropriate Python type."""
+    if not schema:
+        return object
+
+    if "type" not in schema and "properties" in schema:
+        return _create_dataclass(schema, schema.get("title", "<unknown>"), schemas)
+
+    # Handle references first
+    if "$ref" in schema:
+        ref = schema["$ref"]
+        # Handle self-reference
+        if ref == "#":
+            return ForwardRef(schema.get("title", "Root"))
+        return _schema_to_type(_resolve_ref(ref, schemas), schemas)
+
+    if "const" in schema:
+        return Literal[schema["const"]]  # type: ignore
+
+    if "enum" in schema:
+        return _create_enum(f"Enum_{len(_classes)}", schema["enum"])
+
+    schema_type = schema.get("type")
+    if not schema_type:
+        return Any
+
+    if isinstance(schema_type, list):
+        # Create a copy of the schema for each type, but keep all constraints
+        types: list[type | Any] = []
+        for t in schema_type:
+            type_schema = schema.copy()
+            type_schema["type"] = t
+            types.append(_schema_to_type(type_schema, schemas))
+        has_null = type(None) in types
+        types = [t for t in types if t is not type(None)]
+        if has_null:
+            return Optional[tuple(types) if len(types) > 1 else types[0]]
+        return Union[tuple(types)]
+
+    return _get_from_type_handler(schema, schemas)(schema)
+
+
+def _sanitize_name(name: str) -> str:
+    """Convert string to valid Python identifier."""
+    # Step 1: replace everything except [0-9a-zA-Z_] with underscores
+    cleaned = re.sub(r"[^0-9a-zA-Z_]", "_", name)
+    # Step 2: deduplicate underscores
+    cleaned = re.sub(r"__+", "_", cleaned)
+    # Step 3: if the first char of original name isn't a letter, prepend field_
+    if not name or not re.match(r"[a-zA-Z]", name[0]):
+        cleaned = f"field_{cleaned}"
+    # Step 4: deduplicate again and strip trailing underscores
+    cleaned = re.sub(r"__+", "_", cleaned).strip("_")
+    return cleaned
+
+
+def _get_default_value(
+    schema: dict[str, Any],
+    prop_name: str,
+    parent_default: dict[str, Any] | None = None,
+) -> Any:
+    """Get default value with proper priority ordering.
+    1. Value from parent's default if it exists
+    2. Property's own default if it exists
+    3. None
+    """
+    if parent_default is not None and prop_name in parent_default:
+        return parent_default[prop_name]
+    return schema.get("default")
+
+
+def _create_field_with_default(
+    field_type: type,
+    default_value: Any,
+    schema: dict[str, Any],
+) -> Any:
+    """Create a field with simplified default handling."""
+    # Always use None as default for complex types
+    if isinstance(default_value, (dict, list)) or default_value is None:
+        return field(default=None)
+
+    # For simple types, use the value directly
+    return field(default=default_value)
+
+
+def _create_dataclass(
+    schema: Mapping[str, Any],
+    name: str | None = None,
+    schemas: Mapping[str, Any] | None = None,
+) -> type:
+    """Create dataclass from object schema."""
+    name = name or schema.get("title", "Root")
+    # Sanitize name for class creation
+    sanitized_name = _sanitize_name(name)
+    schema_hash = _hash_schema(schema)
+    cache_key = (schema_hash, sanitized_name)
+    original_schema = dict(schema)  # Store copy for validator
+
+    # Return existing class if already built
+    if cache_key in _classes:
+        existing = _classes[cache_key]
+        if existing is None:
+            return ForwardRef(sanitized_name)
+        return existing
+
+    # Place placeholder for recursive references
+    _classes[cache_key] = None
+
+    if "$ref" in schema:
+        ref = schema["$ref"]
+        if ref == "#":
+            return ForwardRef(sanitized_name)
+        schema = _resolve_ref(ref, schemas or {})
+
+    properties = schema.get("properties", {})
+    required = schema.get("required", [])
+
+    fields: list[tuple[Any, ...]] = []
+    for prop_name, prop_schema in properties.items():
+        field_name = _sanitize_name(prop_name)
+
+        # Check for self-reference in property
+        if prop_schema.get("$ref") == "#":
+            field_type = ForwardRef(sanitized_name)
+        else:
+            field_type = _schema_to_type(prop_schema, schemas)
+
+        default_val = prop_schema.get("default", MISSING)
+        is_required = prop_name in required
+
+        # Include alias in field metadata
+        meta = {"alias": prop_name}
+
+        if default_val is not MISSING:
+            if isinstance(default_val, (dict, list)):
+                field_def = field(
+                    default_factory=lambda d=default_val: deepcopy(d), metadata=meta
+                )
+            else:
+                field_def = field(default=default_val, metadata=meta)
+        else:
+            if is_required:
+                field_def = field(metadata=meta)
+            else:
+                field_def = field(default=None, metadata=meta)
+
+        if is_required and default_val is not MISSING:
+            fields.append((field_name, field_type, field_def))
+        elif is_required:
+            fields.append((field_name, field_type, field_def))
+        else:
+            fields.append((field_name, Optional[field_type], field_def))
+
+    cls = make_dataclass(sanitized_name, fields, kw_only=True)
+
+    # Add model validator for defaults
+    @model_validator(mode="before")
+    @classmethod
+    def _apply_defaults(cls, data: Mapping[str, Any]):
+        if isinstance(data, dict):
+            return _merge_defaults(data, original_schema)
+        return data
+
+    setattr(cls, "_apply_defaults", _apply_defaults)
+
+    # Store completed class
+    _classes[cache_key] = cls
+    return cls
+
+
+def _merge_defaults(
+    data: Mapping[str, Any],
+    schema: Mapping[str, Any],
+    parent_default: Mapping[str, Any] | None = None,
+) -> dict[str, Any]:
+    """Merge defaults with provided data at all levels."""
+    # If we have no data
+    if not data:
+        # Start with parent default if available
+        if parent_default:
+            result = dict(parent_default)
+        # Otherwise use schema default if available
+        elif "default" in schema:
+            result = dict(schema["default"])
+        # Otherwise start empty
+        else:
+            result = {}
+    # If we have data and a parent default, merge them
+    elif parent_default:
+        result = dict(parent_default)
+        for key, value in data.items():
+            if (
+                isinstance(value, dict)
+                and key in result
+                and isinstance(result[key], dict)
+            ):
+                # recursively merge nested dicts
+                result[key] = _merge_defaults(value, {"properties": {}}, result[key])
+            else:
+                result[key] = value
+    # Otherwise just use the data
+    else:
+        result = dict(data)
+
+    # For each property in the schema
+    for prop_name, prop_schema in schema.get("properties", {}).items():
+        # If property is missing, apply defaults in priority order
+        if prop_name not in result:
+            if parent_default and prop_name in parent_default:
+                result[prop_name] = parent_default[prop_name]
+            elif "default" in prop_schema:
+                result[prop_name] = prop_schema["default"]
+
+        # If property exists and is an object, recursively merge
+        if (
+            prop_name in result
+            and isinstance(result[prop_name], dict)
+            and prop_schema.get("type") == "object"
+        ):
+            # Get the appropriate default for this nested object
+            nested_default = None
+            if parent_default and prop_name in parent_default:
+                nested_default = parent_default[prop_name]
+            elif "default" in prop_schema:
+                nested_default = prop_schema["default"]
+
+            result[prop_name] = _merge_defaults(
+                result[prop_name], prop_schema, nested_default
+            )
+
+    return result
+
+
+class JSONSchema(TypedDict):
+    type: NotRequired[str | list[str]]
+    properties: NotRequired[dict[str, JSONSchema]]
+    required: NotRequired[list[str]]
+    additionalProperties: NotRequired[bool | JSONSchema]
+    items: NotRequired[JSONSchema | list[JSONSchema]]
+    enum: NotRequired[list[Any]]
+    const: NotRequired[Any]
+    default: NotRequired[Any]
+    description: NotRequired[str]
+    title: NotRequired[str]
+    examples: NotRequired[list[Any]]
+    format: NotRequired[str]
+    allOf: NotRequired[list[JSONSchema]]
+    anyOf: NotRequired[list[JSONSchema]]
+    oneOf: NotRequired[list[JSONSchema]]
+    not_: NotRequired[JSONSchema]
+    definitions: NotRequired[dict[str, JSONSchema]]
+    dependencies: NotRequired[dict[str, JSONSchema | list[str]]]
+    pattern: NotRequired[str]
+    minLength: NotRequired[int]
+    maxLength: NotRequired[int]
+    minimum: NotRequired[int | float]
+    maximum: NotRequired[int | float]
+    exclusiveMinimum: NotRequired[int | float]
+    exclusiveMaximum: NotRequired[int | float]
+    multipleOf: NotRequired[int | float]
+    uniqueItems: NotRequired[bool]
+    minItems: NotRequired[int]
+    maxItems: NotRequired[int]
+    additionalItems: NotRequired[bool | JSONSchema]

src/fastmcp/utilities/types.py

@@ -42,7 +42,7 @@ def get_cached_typeadapter(cls: T) -> TypeAdapter[T]:
     The _parent_depth is set to 3 to look at an additional frame, since this
     function is in its own scope.
     """
-    return TypeAdapter(cls, _parent_depth=3)
+    return TypeAdapter(cls)
 
 
 def issubclass_safe(cls: type, base: type) -> bool:

tests/client/test_elicitation.py

@@ -0,0 +1,509 @@
+from dataclasses import dataclass
+
+import pytest
+from mcp.types import ElicitResult
+
+from fastmcp import Context, FastMCP
+from fastmcp.client.client import Client
+from fastmcp.server.elicitation import (
+    AcceptedElicitation,
+    CancelledElicitation,
+    DeclinedElicitation,
+)
+
+
+@pytest.fixture
+def fastmcp_server():
+    mcp = FastMCP("TestServer")
+
+    @dataclass
+    class Person:
+        name: str
+
+    @mcp.tool
+    async def ask_for_name(context: Context) -> str:
+        result = await context.elicit(
+            message="What is your name?",
+            response_type=Person,
+        )
+        if result.action == "accept":
+            return f"Hello, {result.data.name}!"
+        else:
+            return "No name provided."
+
+    @mcp.tool
+    def simple_test() -> str:
+        return "Hello!"
+
+    return mcp
+
+
+async def test_elicitation_accept_content(fastmcp_server):
+    """Test basic elicitation functionality."""
+
+    async def elicitation_handler(message, schema, ctx):
+        # Mock user providing their name
+        return ElicitResult(action="accept", content={"name": "Alice"})
+
+    async with Client(
+        fastmcp_server, elicitation_handler=elicitation_handler
+    ) as client:
+        result = await client.call_tool("ask_for_name", {})
+        assert result[0].text == "Hello, Alice!"  # type: ignore[attr-defined]
+
+
+async def test_elicitation_decline(fastmcp_server):
+    """Test that elicitation handler receives correct parameters."""
+
+    async def elicitation_handler(message, schema, ctx):
+        return ElicitResult(action="decline")
+
+    async with Client(
+        fastmcp_server, elicitation_handler=elicitation_handler
+    ) as client:
+        result = await client.call_tool("ask_for_name", {})
+        assert result[0].text == "No name provided."  # type: ignore[attr-defined]
+
+
+async def test_default_response_type(fastmcp_server):
+    """Test elicitation with string content."""
+    mcp = FastMCP("TestServer")
+
+    @mcp.tool
+    async def ask_for_color(context: Context) -> str:
+        result = await context.elicit(
+            message="What is your favorite color?"
+            # Default schema should be string
+        )
+        if result.action == "accept":
+            assert isinstance(result.data, str)
+            return f"Your favorite color is {result.data}!"
+        return "No color provided"
+
+    async def elicitation_handler(message, schema, ctx):
+        # Mock user providing their favorite color as string in content dict
+        return ElicitResult(action="accept", content={"value": "blue"})
+
+    async with Client(mcp, elicitation_handler=elicitation_handler) as client:
+        result = await client.call_tool("ask_for_color", {})
+        assert result[0].text == "Your favorite color is blue!"  # type: ignore[attr-defined]
+
+
+async def test_elicitation_handler_parameters():
+    """Test that elicitation handler receives correct parameters."""
+    mcp = FastMCP("TestServer")
+    captured_params = {}
+
+    @mcp.tool
+    async def test_tool(context: Context) -> str:
+        await context.elicit(
+            message="Test message",
+            response_type=int,
+        )
+        return "done"
+
+    async def elicitation_handler(message, schema, ctx):
+        captured_params["message"] = message
+        captured_params["schema"] = schema
+        captured_params["ctx"] = ctx
+        return ElicitResult(action="accept", content={"value": 42})
+
+    async with Client(mcp, elicitation_handler=elicitation_handler) as client:
+        await client.call_tool("test_tool", {})
+
+        assert captured_params["message"] == "Test message"
+        assert captured_params["schema"] == {
+            "properties": {"value": {"title": "Value", "type": "integer"}},
+            "required": ["value"],
+            "title": "PrimitiveElicitationType",
+            "type": "object",
+        }
+        assert captured_params["ctx"] is not None
+
+
+async def test_elicitation_default_string_schema():
+    """Test elicitation with default string schema."""
+    mcp = FastMCP("TestServer")
+
+    @mcp.tool
+    async def ask_for_input(context: Context) -> str:
+        result = await context.elicit(
+            message="Please provide some input"
+            # No schema provided - should default to string
+        )
+        if result.action == "accept":
+            return f"You said: {result.data}"
+        return "No input provided"
+
+    async def elicitation_handler(message, schema, ctx):
+        # Verify default schema is wrapped string object
+        expected_schema = {
+            "properties": {"value": {"title": "Value", "type": "string"}},
+            "required": ["value"],
+            "title": "PrimitiveElicitationType",
+            "type": "object",
+        }
+        assert schema == expected_schema
+        return ElicitResult(action="accept", content={"value": "Hello world!"})
+
+    async with Client(mcp, elicitation_handler=elicitation_handler) as client:
+        result = await client.call_tool("ask_for_input", {})
+        assert result[0].text == "You said: Hello world!"  # type: ignore[attr-defined]
+
+
+async def test_elicitation_cancel_action():
+    """Test user canceling elicitation request."""
+    mcp = FastMCP("TestServer")
+
+    @mcp.tool
+    async def ask_for_optional_info(context: Context) -> str:
+        result = await context.elicit(
+            message="Optional: What's your age?", response_type=int
+        )
+        if result.action == "cancel":
+            return "Request was canceled"
+        elif result.action == "accept":
+            return f"Age: {result.data}"
+        else:
+            return "No response provided"
+
+    async def elicitation_handler(message, schema, ctx):
+        return ElicitResult(action="cancel")
+
+    async with Client(mcp, elicitation_handler=elicitation_handler) as client:
+        result = await client.call_tool("ask_for_optional_info", {})
+        assert result[0].text == "Request was canceled"  # type: ignore[attr-defined]
+
+
+async def test_elicitation_number_schema():
+    """Test elicitation with number schema."""
+    mcp = FastMCP("TestServer")
+
+    @mcp.tool
+    async def get_age(context: Context) -> str:
+        result = await context.elicit(message="How old are you?", response_type=int)
+        if result.action == "accept":
+            return f"You are {result.data} years old"
+        return "No age provided"
+
+    async def elicitation_handler(message, schema, ctx):
+        expected_schema = {
+            "properties": {"value": {"title": "Value", "type": "integer"}},
+            "required": ["value"],
+            "title": "PrimitiveElicitationType",
+            "type": "object",
+        }
+        assert schema == expected_schema
+        return ElicitResult(action="accept", content={"value": 25})
+
+    async with Client(mcp, elicitation_handler=elicitation_handler) as client:
+        result = await client.call_tool("get_age", {})
+        assert result[0].text == "You are 25 years old"  # type: ignore[attr-defined]
+
+
+async def test_elicitation_handler_error():
+    """Test error handling in elicitation handler."""
+    mcp = FastMCP("TestServer")
+
+    @mcp.tool
+    async def failing_elicit(context: Context) -> str:
+        try:
+            result = await context.elicit(message="This will fail", response_type=str)
+            assert result.action == "accept"
+            return f"Got: {result.data}"
+        except Exception as e:
+            return f"Error: {str(e)}"
+
+    async def elicitation_handler(message, schema, ctx):
+        raise ValueError("Handler failed!")
+
+    async with Client(mcp, elicitation_handler=elicitation_handler) as client:
+        result = await client.call_tool("failing_elicit", {})
+        assert "Error:" in result[0].text  # type: ignore[attr-defined]
+
+
+async def test_elicitation_multiple_calls():
+    """Test multiple elicitation calls in sequence."""
+    mcp = FastMCP("TestServer")
+
+    @mcp.tool
+    async def multi_step_form(context: Context) -> str:
+        # First question
+        name_result = await context.elicit(
+            message="What's your name?", response_type=str
+        )
+        if name_result.action != "accept":
+            return "Form abandoned"
+
+        # Second question
+        age_result = await context.elicit(message="What's your age?", response_type=int)
+        if age_result.action != "accept":
+            return f"Hello {name_result.data}, form incomplete"
+
+        return f"Hello {name_result.data}, you are {age_result.data} years old"
+
+    call_count = 0
+
+    async def elicitation_handler(message, schema, ctx):
+        nonlocal call_count
+        call_count += 1
+        if call_count == 1:
+            assert "name" in message.lower()
+            expected_schema = {
+                "properties": {"value": {"title": "Value", "type": "string"}},
+                "required": ["value"],
+                "title": "PrimitiveElicitationType",
+                "type": "object",
+            }
+            assert schema == expected_schema
+            return ElicitResult(action="accept", content={"value": "Bob"})
+        elif call_count == 2:
+            assert "age" in message.lower()
+            expected_schema = {
+                "properties": {"value": {"title": "Value", "type": "integer"}},
+                "required": ["value"],
+                "title": "PrimitiveElicitationType",
+                "type": "object",
+            }
+            assert schema == expected_schema
+            return ElicitResult(action="accept", content={"value": 25})
+        else:
+            raise ValueError("Unexpected call")
+
+    async with Client(mcp, elicitation_handler=elicitation_handler) as client:
+        result = await client.call_tool("multi_step_form", {})
+        assert result[0].text == "Hello Bob, you are 25 years old"  # type: ignore[attr-defined]
+        assert call_count == 2
+
+
+async def test_dataclass_response_type():
+    """Test elicitation with dataclass response type."""
+    mcp = FastMCP("TestServer")
+
+    @dataclass
+    class UserInfo:
+        name: str
+        age: int
+
+    @mcp.tool
+    async def get_user_info(context: Context) -> str:
+        result = await context.elicit(
+            message="Please provide your information", response_type=UserInfo
+        )
+        if result.action == "accept":
+            user = result.data
+            return f"User: {user.name}, age: {user.age}"
+        return "No user info provided"
+
+    async def elicitation_handler(message, schema, ctx):
+        # Verify the schema has the dataclass fields
+        assert schema["type"] == "object"
+        assert "name" in schema["properties"]
+        assert "age" in schema["properties"]
+        assert schema["properties"]["name"]["type"] == "string"
+        assert schema["properties"]["age"]["type"] == "integer"
+
+        return ElicitResult(action="accept", content={"name": "Alice", "age": 30})
+
+    async with Client(mcp, elicitation_handler=elicitation_handler) as client:
+        result = await client.call_tool("get_user_info", {})
+        assert result[0].text == "User: Alice, age: 30"  # type: ignore[attr-defined]
+
+
+async def test_primitive_type_string():
+    """Test elicitation with string primitive type."""
+    mcp = FastMCP("TestServer")
+
+    @mcp.tool
+    async def test_string(context: Context) -> str:
+        result = await context.elicit("Enter text:", response_type=str)
+        assert result.action == "accept"
+        return f"Got: {result.data}"
+
+    async def elicitation_handler(message, schema, ctx):
+        assert schema["properties"]["value"]["type"] == "string"
+        return ElicitResult(action="accept", content={"value": "hello"})
+
+    async with Client(mcp, elicitation_handler=elicitation_handler) as client:
+        result = await client.call_tool("test_string", {})
+        assert result[0].text == "Got: hello"  # type: ignore[attr-defined]
+
+
+async def test_primitive_type_int():
+    """Test elicitation with integer primitive type."""
+    mcp = FastMCP("TestServer")
+
+    @mcp.tool
+    async def test_int(context: Context) -> str:
+        result = await context.elicit("Enter number:", response_type=int)
+        assert result.action == "accept"
+        return f"Got: {result.data}"
+
+    async def elicitation_handler(message, schema, ctx):
+        assert schema["properties"]["value"]["type"] == "integer"
+        return ElicitResult(action="accept", content={"value": 42})
+
+    async with Client(mcp, elicitation_handler=elicitation_handler) as client:
+        result = await client.call_tool("test_int", {})
+        assert result[0].text == "Got: 42"  # type: ignore[attr-defined]
+
+
+async def test_primitive_type_float():
+    """Test elicitation with float primitive type."""
+    mcp = FastMCP("TestServer")
+
+    @mcp.tool
+    async def test_float(context: Context) -> str:
+        result = await context.elicit("Enter decimal:", response_type=float)
+        assert result.action == "accept"
+        return f"Got: {result.data}"
+
+    async def elicitation_handler(message, schema, ctx):
+        assert schema["properties"]["value"]["type"] == "number"
+        return ElicitResult(action="accept", content={"value": 3.14})
+
+    async with Client(mcp, elicitation_handler=elicitation_handler) as client:
+        result = await client.call_tool("test_float", {})
+        assert result[0].text == "Got: 3.14"  # type: ignore[attr-defined]
+
+
+async def test_primitive_type_bool():
+    """Test elicitation with boolean primitive type."""
+    mcp = FastMCP("TestServer")
+
+    @mcp.tool
+    async def test_bool(context: Context) -> str:
+        result = await context.elicit("Enter true/false:", response_type=bool)
+        assert result.action == "accept"
+        return f"Got: {result.data}"
+
+    async def elicitation_handler(message, schema, ctx):
+        assert schema["properties"]["value"]["type"] == "boolean"
+        return ElicitResult(action="accept", content={"value": True})
+
+    async with Client(mcp, elicitation_handler=elicitation_handler) as client:
+        result = await client.call_tool("test_bool", {})
+        assert result[0].text == "Got: True"  # type: ignore[attr-defined]
+
+
+async def test_schema_validation_rejects_non_object():
+    """Test that non-object schemas are rejected."""
+    from fastmcp.server.elicitation import validate_elicitation_json_schema
+
+    with pytest.raises(TypeError, match="must be an object schema"):
+        validate_elicitation_json_schema({"type": "string"})
+
+
+async def test_schema_validation_rejects_empty_object():
+    """Test that object schemas without properties are rejected."""
+    from fastmcp.server.elicitation import validate_elicitation_json_schema
+
+    with pytest.raises(TypeError, match="must have at least one property"):
+        validate_elicitation_json_schema({"type": "object"})
+
+
+async def test_schema_validation_rejects_nested_objects():
+    """Test that nested object schemas are rejected."""
+    from fastmcp.server.elicitation import validate_elicitation_json_schema
+
+    with pytest.raises(
+        TypeError, match="has type 'object' which is not a primitive type"
+    ):
+        validate_elicitation_json_schema(
+            {
+                "type": "object",
+                "properties": {
+                    "user": {
+                        "type": "object",
+                        "properties": {"name": {"type": "string"}},
+                    }
+                },
+            }
+        )
+
+
+async def test_schema_validation_rejects_arrays():
+    """Test that array schemas are rejected."""
+    from fastmcp.server.elicitation import validate_elicitation_json_schema
+
+    with pytest.raises(
+        TypeError, match="has type 'array' which is not a primitive type"
+    ):
+        validate_elicitation_json_schema(
+            {
+                "type": "object",
+                "properties": {"users": {"type": "array", "items": {"type": "string"}}},
+            }
+        )
+
+
+async def test_pattern_matching_accept():
+    """Test pattern matching with AcceptedElicitation."""
+    mcp = FastMCP("TestServer")
+
+    @mcp.tool
+    async def pattern_match_tool(context: Context) -> str:
+        result = await context.elicit("Enter your name:", response_type=str)
+
+        match result:
+            case AcceptedElicitation(data=name):
+                return f"Hello {name}!"
+            case DeclinedElicitation():
+                return "You declined"
+            case CancelledElicitation():
+                return "Cancelled"
+
+    async def elicitation_handler(message, schema, ctx):
+        return ElicitResult(action="accept", content={"value": "Alice"})
+
+    async with Client(mcp, elicitation_handler=elicitation_handler) as client:
+        result = await client.call_tool("pattern_match_tool", {})
+        assert result[0].text == "Hello Alice!"  # type: ignore[attr-defined]
+
+
+async def test_pattern_matching_decline():
+    """Test pattern matching with DeclinedElicitation."""
+    mcp = FastMCP("TestServer")
+
+    @mcp.tool
+    async def pattern_match_tool(context: Context) -> str:
+        result = await context.elicit("Enter your name:", response_type=str)
+
+        match result:
+            case AcceptedElicitation(data=name):
+                return f"Hello {name}!"
+            case DeclinedElicitation():
+                return "You declined"
+            case CancelledElicitation():
+                return "Cancelled"
+
+    async def elicitation_handler(message, schema, ctx):
+        return ElicitResult(action="decline")
+
+    async with Client(mcp, elicitation_handler=elicitation_handler) as client:
+        result = await client.call_tool("pattern_match_tool", {})
+        assert result[0].text == "You declined"  # type: ignore[attr-defined]
+
+
+async def test_pattern_matching_cancel():
+    """Test pattern matching with CancelledElicitation."""
+    mcp = FastMCP("TestServer")
+
+    @mcp.tool
+    async def pattern_match_tool(context: Context) -> str:
+        result = await context.elicit("Enter your name:", response_type=str)
+
+        match result:
+            case AcceptedElicitation(data=name):
+                return f"Hello {name}!"
+            case DeclinedElicitation():
+                return "You declined"
+            case CancelledElicitation():
+                return "Cancelled"
+
+    async def elicitation_handler(message, schema, ctx):
+        return ElicitResult(action="cancel")
+
+    async with Client(mcp, elicitation_handler=elicitation_handler) as client:
+        result = await client.call_tool("pattern_match_tool", {})
+        assert result[0].text == "Cancelled"  # type: ignore[attr-defined]