Merge pull request #889 from jlowin/elicitation

CLAUDE.md

@@ -1,5 +1,10 @@
 # FastMCP Development Guidelines
 
+## Documentation
+
+- Documentation uses the Mintlify framework
+- Files must be present in docs.json to be included in the documentation
+
 ## Testing and Investigation
 
 ### In-Memory Transport - Always Preferred
@@ -34,4 +39,4 @@ async with Client(transport=StreamableHttpTransport(server_url)) as client:
 - You must always run pre-commit if you open a PR, because it is run as part of a required check.
 - When opening PRs, apply labels appropriately for bugs/breaking changes/enhancements/features. Generally, improvements are enhancements (not features) unless told otherwise.
 - NEVER modify files in docs/python-sdk/**, as they are auto-generated.
-- Use # type: ignore[attr-defined] in unit tests when accessing an MCP result of indeterminate type instead of asserting its type
+- Use # type: ignore[attr-defined] in unit tests when accessing an MCP result of indeterminate type instead of asserting its type

docs/clients/elicitation.mdx

@@ -0,0 +1,120 @@
+---
+title: User Elicitation
+sidebarTitle: Elicitation
+description: Handle server-initiated user input requests with structured schemas.
+icon: message-question
+tag: NEW
+---
+
+import { VersionBadge } from "/snippets/version-badge.mdx";
+
+<VersionBadge version="2.10.0" />
+
+## What is Elicitation?
+
+Elicitation allows MCP servers to request structured input from users during tool execution. Instead of requiring all inputs upfront, servers can interactively ask users for information as needed - like prompting for missing parameters, requesting clarification, or gathering additional context.
+
+For example, a file management tool might ask "Which directory should I create?" or a data analysis tool might request "What date range should I analyze?"
+
+## How FastMCP Makes Elicitation Easy
+
+FastMCP's client provides a helpful abstraction layer that:
+
+- **Converts JSON schemas to Python types**: The raw MCP protocol uses JSON schemas, but FastMCP automatically converts these to Python dataclasses
+- **Provides structured constructors**: Instead of manually building dictionaries that match the schema, you get dataclass constructors that ensure correct structure
+- **Handles type conversion**: FastMCP takes care of converting between JSON representations and Python objects
+- **Runtime introspection**: You can inspect the generated dataclass fields to understand the expected structure
+
+When you implement an elicitation handler, FastMCP gives you a dataclass type that matches the server's schema, making it easy to create properly structured responses without having to manually parse JSON schemas.
+
+## Elicitation Handler
+
+Provide an `elicitation_handler` function when creating the client. FastMCP automatically converts the server's JSON schema into a Python dataclass type, making it easy to construct the response:
+
+```python
+from fastmcp import Client
+from fastmcp.client.elicitation import ElicitResult
+
+async def elicitation_handler(message: str, response_type: type, params, context) -> ElicitResult:
+    # Present the message to the user and collect input
+    user_input = input(f"{message}: ")
+    
+    # Create response using the provided dataclass type
+    # FastMCP converted the JSON schema to this Python type for you
+    response_data = response_type(value=user_input)
+    
+    return ElicitResult(action="accept", content=response_data)
+
+client = Client(
+    "my_mcp_server.py",
+    elicitation_handler=elicitation_handler,
+)
+```
+
+### Handler Parameters
+
+The elicitation handler receives four parameters:
+
+<Card icon="code" title="Elicitation Handler Parameters">
+<ResponseField name="message" type="str">
+  The prompt message to display to the user
+</ResponseField>
+
+<ResponseField name="response_type" type="type">
+  A Python dataclass type that FastMCP created from the server's JSON schema. Use this to construct your response with proper typing and IDE support.
+</ResponseField>
+
+<ResponseField name="params" type="ElicitRequestParams">
+  The original MCP elicitation request parameters, including the raw JSON schema in `params.requestedSchema` if you need it
+</ResponseField>
+
+<ResponseField name="context" type="RequestContext">
+  Request context containing metadata about the elicitation request
+</ResponseField>
+</Card>
+
+### Response Actions
+
+The handler must return an `ElicitResult` object that includes both an action and (when accepted) the user's input:
+
+<Card icon="code" title="ElicitResult Structure">
+<ResponseField name="action" type="Literal['accept', 'decline', 'cancel']">
+  How the user responded to the elicitation request
+</ResponseField>
+
+<ResponseField name="content" type="dataclass instance | dict | None">
+  The user's input data (required for "accept", omitted for "decline"/"cancel")
+</ResponseField>
+</Card>
+
+**Action Types:**
+- **`accept`**: User provided valid input - include their data in the `content` field
+- **`decline`**: User chose not to provide the requested information - omit `content`  
+- **`cancel`**: User cancelled the entire operation - omit `content`
+
+## Basic Example
+
+```python
+from fastmcp import Client
+from fastmcp.client.elicitation import ElicitResult
+
+async def basic_elicitation_handler(message: str, response_type: type, params, context) -> ElicitResult:
+    print(f"Server asks: {message}")
+    
+    # Simple text input for demonstration
+    user_response = input("Your response: ")
+    
+    if not user_response:
+        return ElicitResult(action="decline")
+    
+    # Use the response_type dataclass to create a properly structured response
+    # FastMCP handles the conversion from JSON schema to Python type
+    return ElicitResult(action="accept", content=response_type(value=user_response))
+
+client = Client(
+    "my_mcp_server.py", 
+    elicitation_handler=basic_elicitation_handler
+)
+```
+
+

docs/docs.json

@@ -69,8 +69,18 @@
                     "pages": [
                       "servers/tools",
                       "servers/resources",
-                      "servers/prompts",
-                      "servers/context"
+                      "servers/prompts"
+                    ]
+                  },
+                  {
+                    "group": "Advanced Features",
+                    "icon": "stars",
+                    "pages": [
+                      "servers/context",
+                      "servers/elicitation",
+                      "servers/logging",
+                      "servers/progress",
+                      "servers/sampling"
                     ]
                   },
                   {
@@ -106,6 +116,7 @@
                     "group": "Advanced Features",
                     "icon": "stars",
                     "pages": [
+                      "clients/elicitation",
                       "clients/logging",
                       "clients/progress",
                       "clients/sampling",

docs/servers/context.mdx

@@ -6,7 +6,7 @@ icon: rectangle-code
 ---
 import { VersionBadge } from '/snippets/version-badge.mdx'
 
-When defining FastMCP [tools](/servers/tools), [resources](/servers/resources), resource templates, or [prompts](/servers/prompts), your functions might need to interact with the underlying MCP session or access server capabilities. FastMCP provides the `Context` object for this purpose.
+When defining FastMCP [tools](/servers/tools), [resources](/servers/resources), resource templates, or [prompts](/servers/prompts), your functions might need to interact with the underlying MCP session or access advanced server capabilities. FastMCP provides the `Context` object for this purpose.
 
 ## What Is Context?
 
@@ -16,6 +16,7 @@ The `Context` object provides a clean interface to access MCP features within yo
 - **Progress Reporting**: Update the client on the progress of long-running operations
 - **Resource Access**: Read data from resources registered with the server
 - **LLM Sampling**: Request the client's LLM to generate text based on provided messages
+- **User Elicitation**: Request structured input from users during tool execution
 - **Request Information**: Access metadata about the current request
 - **Server Access**: When needed, access the underlying FastMCP server instance
 
@@ -36,10 +37,10 @@ To use the context object within any of your functions, simply add a parameter t
 
 #### Tools
 
-```python
+```python {1, 6}
 from fastmcp import FastMCP, Context
 
-mcp = FastMCP(name="ContextDemo")
+mcp = FastMCP(name="Context Demo")
 
 @mcp.tool
 async def process_file(file_uri: str, ctx: Context) -> str:
@@ -52,7 +53,11 @@ async def process_file(file_uri: str, ctx: Context) -> str:
 
 <VersionBadge version="2.2.5" />
 
-```python
+```python {1, 6, 12}
+from fastmcp import FastMCP, Context
+
+mcp = FastMCP(name="Context Demo")
+
 @mcp.resource("resource://user-data")
 async def get_user_data(ctx: Context) -> dict:
     """Fetch personalized user data based on the request context."""
@@ -70,7 +75,11 @@ async def get_user_profile(user_id: str, ctx: Context) -> dict:
 
 <VersionBadge version="2.2.5" />
 
-```python
+```python {1, 6}
+from fastmcp import FastMCP, Context
+
+mcp = FastMCP(name="Context Demo")
+
 @mcp.prompt
 async def data_analysis_request(dataset: str, ctx: Context) -> str:
     """Generate a request to analyze data with contextual information."""
@@ -88,10 +97,10 @@ While the simplest way to access context is through function parameter injection
 FastMCP provides dependency functions that allow you to retrieve the active context from anywhere within a server request's execution flow:
 
 ```python {2,9}
-from fastmcp import FastMCP, Context
+from fastmcp import FastMCP
 from fastmcp.server.dependencies import get_context
 
-mcp = FastMCP(name="DependencyDemo")
+mcp = FastMCP(name="Dependency Demo")
 
 # Utility function that needs context but doesn't receive it as a parameter
 async def process_data(data: list[float]) -> dict:
@@ -113,169 +122,71 @@ async def analyze_dataset(dataset_name: str) -> dict:
 
 ## Context Capabilities
 
+FastMCP provides several advanced capabilities through the context object. Each capability has dedicated documentation with comprehensive examples and best practices:
+
 ### Logging
 
-Send log messages back to the MCP client. This is useful for debugging and providing visibility into function execution during a request.
+Send debug, info, warning, and error messages back to the MCP client for visibility into function execution.
 
 ```python
-@mcp.tool
-async def analyze_data(data: list[float], ctx: Context) -> dict:
-    """Analyze numerical data with logging."""
-    await ctx.debug("Starting analysis of numerical data")
-    await ctx.info(f"Analyzing {len(data)} data points")
-    
-    try:
-        result = sum(data) / len(data)
-        await ctx.info(f"Analysis complete, average: {result}")
-        return {"average": result, "count": len(data)}
-    except ZeroDivisionError:
-        await ctx.warning("Empty data list provided")
-        return {"error": "Empty data list"}
-    except Exception as e:
-        await ctx.error(f"Analysis failed: {str(e)}")
-        raise
+await ctx.debug("Starting analysis")
+await ctx.info(f"Processing {len(data)} items") 
+await ctx.warning("Deprecated parameter used")
+await ctx.error("Processing failed")
 ```
 
-**Available Logging Methods:**
+See [Server Logging](/servers/logging) for complete documentation and examples.
+### Client Elicitation
 
-- **`ctx.debug(message: str)`**: Low-level details useful for debugging
-- **`ctx.info(message: str)`**: General information about execution
-- **`ctx.warning(message: str)`**: Potential issues that didn't prevent execution
-- **`ctx.error(message: str)`**: Errors that occurred during execution
-- **`ctx.log(level: Literal["debug", "info", "warning", "error"], message: str, logger_name: str | None = None)`**: Generic log method supporting custom logger names
+<VersionBadge version="2.10.0" />
 
-### Progress Reporting
-
-For long-running operations, notify the client about the progress. This allows clients to display progress indicators and provide a better user experience.
+Request structured input from clients during tool execution, enabling interactive workflows and progressive disclosure. This is a new feature in the 6/18/2025 MCP spec.
 
 ```python
-@mcp.tool
-async def process_items(items: list[str], ctx: Context) -> dict:
-    """Process a list of items with progress updates."""
-    total = len(items)
-    results = []
-    
-    for i, item in enumerate(items):
-        # Report progress as percentage
-        await ctx.report_progress(progress=i, total=total)
-        
-        # Process the item (simulated with a sleep)
-        await asyncio.sleep(0.1)
-        results.append(item.upper())
-    
-    # Report 100% completion
-    await ctx.report_progress(progress=total, total=total)
-    
-    return {"processed": len(results), "results": results}
+result = await ctx.elicit("Enter your name:", response_type=str)
+if result.action == "accept":
+    name = result.data
 ```
 
-**Method signature:**
-
-- **`ctx.report_progress(progress: float, total: float | None = None)`**
-  - `progress`: Current progress value (e.g., 24)
-  - `total`: Optional total value (e.g., 100). If provided, clients may interpret this as a percentage.
+See [User Elicitation](/servers/elicitation) for detailed examples and supported response types.
 
-Progress reporting requires the client to have sent a `progressToken` in the initial request. If the client doesn't support progress reporting, these calls will have no effect.
+### LLM Sampling
 
-### Resource Access
+<VersionBadge version="2.0.0" />
 
-Read data from resources registered with your FastMCP server. This allows functions to access files, configuration, or dynamically generated content.
+Request the client's LLM to generate text based on provided messages, useful for leveraging AI capabilities within your tools.
 
 ```python
-@mcp.tool
-async def summarize_document(document_uri: str, ctx: Context) -> str:
-    """Summarize a document by its resource URI."""
-    # Read the document content
-    content_list = await ctx.read_resource(document_uri)
-    
-    if not content_list:
-        return "Document is empty"
-    
-    document_text = content_list[0].content
-    
-    # Example: Generate a simple summary (length-based)
-    words = document_text.split()
-    total_words = len(words)
-    
-    await ctx.info(f"Document has {total_words} words")
-    
-    # Return a simple summary
-    if total_words > 100:
-        summary = " ".join(words[:100]) + "..."
-        return f"Summary ({total_words} words total): {summary}"
-    else:
-        return f"Full document ({total_words} words): {document_text}"
+response = await ctx.sample("Analyze this data", temperature=0.7)
 ```
 
-**Method signature:**
-
-- **`ctx.read_resource(uri: str | AnyUrl) -> list[ReadResourceContents]`**
-  - `uri`: The resource URI to read
-  - Returns a list of resource content parts (usually containing just one item)
-
-The returned content is typically accessed via `content_list[0].content` and can be text or binary data depending on the resource.
+See [LLM Sampling](/servers/sampling) for comprehensive usage and advanced techniques.
 
-### LLM Sampling
 
-<VersionBadge version="2.0.0" />
+### Progress Reporting
 
-Request the client's LLM to generate text based on provided messages. This is useful when your function needs to leverage the LLM's capabilities to process data or generate responses.
+Update clients on the progress of long-running operations, enabling progress indicators and better user experience.
 
 ```python
-@mcp.tool
-async def analyze_sentiment(text: str, ctx: Context) -> dict:
-    """Analyze the sentiment of a text using the client's LLM."""
-    # Create a sampling prompt asking for sentiment analysis
-    prompt = f"Analyze the sentiment of the following text as positive, negative, or neutral. Just output a single word - 'positive', 'negative', or 'neutral'. Text to analyze: {text}"
-    
-    # Send the sampling request to the client's LLM (provide a hint for the model you want to use)
-    response = await ctx.sample(prompt, model_preferences="claude-3-sonnet")
-    
-    # Process the LLM's response
-    sentiment = response.text.strip().lower()
-    
-    # Map to standard sentiment values
-    if "positive" in sentiment:
-        sentiment = "positive"
-    elif "negative" in sentiment:
-        sentiment = "negative"
-    else:
-        sentiment = "neutral"
-    
-    return {"text": text, "sentiment": sentiment}
+await ctx.report_progress(progress=50, total=100)  # 50% complete
 ```
 
-**Method signature:**
+See [Progress Reporting](/servers/progress) for detailed patterns and examples.
 
-- **`ctx.sample(messages: str | list[str | SamplingMessage], system_prompt: str | None = None, temperature: float | None = None, max_tokens: int | None = None, model_preferences: ModelPreferences | str | list[str] | None = None) -> TextContent | ImageContent`**
-  - `messages`: A string or list of strings/message objects to send to the LLM
-  - `system_prompt`: Optional system prompt to guide the LLM's behavior
-  - `temperature`: Optional sampling temperature (controls randomness)
-  - `max_tokens`: Optional maximum number of tokens to generate (defaults to 512)
-  - `model_preferences`: Optional model selection preferences (e.g., a model hint string, list of hints, or a ModelPreferences object)
-  - Returns the LLM's response as TextContent or ImageContent
+### Resource Access
 
-When providing a simple string, it's treated as a user message. For more complex scenarios, you can provide a list of messages with different roles.
+Read data from resources registered with your FastMCP server, allowing access to files, configuration, or dynamic content.
 
 ```python
-@mcp.tool
-async def generate_example(concept: str, ctx: Context) -> str:
-    """Generate a Python code example for a given concept."""
-    # Using a system prompt and a user message
-    response = await ctx.sample(
-        messages=f"Write a simple Python code example demonstrating '{concept}'.",
-        system_prompt="You are an expert Python programmer. Provide concise, working code examples without explanations.",
-        temperature=0.7,
-        max_tokens=300
-    )
-    
-    code_example = response.text
-    return f"```python\n{code_example}\n```"
+content_list = await ctx.read_resource("resource://config")
+content = content_list[0].content
 ```
 
-See [Client Sampling](/clients/client#llm-sampling) for more details on how clients handle these requests.
+**Method signature:**
+- **`ctx.read_resource(uri: str | AnyUrl) -> list[ReadResourceContents]`**: Returns a list of resource content parts
+
 
-### Component Changes
+### Change Notifications
 
 <VersionBadge version="2.9.1" />
 
@@ -294,7 +205,19 @@ async def custom_tool_management(ctx: Context) -> str:
 
 These methods are primarily used internally by FastMCP's automatic notification system and most users will not need to invoke them directly.
 
-### Request Information
+### FastMCP Server
+
+To access the underlying FastMCP server instance, you can use the `ctx.fastmcp` property:
+
+```python
+@mcp.tool
+async def my_tool(ctx: Context) -> None:
+    # Access the FastMCP server instance
+    server_name = ctx.fastmcp.name
+    ...
+```
+
+### MCP Request
 
 Access metadata about the current request and client.
 
@@ -314,60 +237,6 @@ async def request_info(ctx: Context) -> dict:
 - **`ctx.client_id -> str | None`**: Get the ID of the client making the request, if provided during initialization
 - **`ctx.session_id -> str | None`**: Get the MCP session ID for session-based data sharing (HTTP transports only)
 
-### Advanced Access
-
-
-#### FastMCP Server and Sessions
-
-```python
-@mcp.tool
-async def advanced_tool(ctx: Context) -> str:
-    """Demonstrate advanced context access."""
-    # Access the FastMCP server instance
-    server_name = ctx.fastmcp.name
-    
-    # Low-level session access (rarely needed)
-    session = ctx.session
-    request_context = ctx.request_context
-    
-    return f"Server: {server_name}"
-```
-
-#### HTTP Requests
-
-<VersionBadge version="2.2.7" />
-
-<Warning>
-The `ctx.get_http_request()` method is deprecated and will be removed in a future version.
-Please use the `get_http_request()` dependency function instead.
-See the [HTTP Requests pattern](/patterns/http-requests) for more details.
-</Warning>
-
-For web applications, you can access the underlying HTTP request:
-
-```python
-@mcp.tool
-async def handle_web_request(ctx: Context) -> dict:
-    """Access HTTP request information from the Starlette request."""
-    request = ctx.get_http_request()
-    
-    # Access HTTP headers, query parameters, etc.
-    user_agent = request.headers.get("user-agent", "Unknown")
-    client_ip = request.client.host if request.client else "Unknown"
-    
-    return {
-        "user_agent": user_agent,
-        "client_ip": client_ip,
-        "path": request.url.path,
-    }
-```
-
-#### Advanced Properties Reference
-
-- **`ctx.fastmcp -> FastMCP`**: Access the server instance the context belongs to
-- **`ctx.session`**: Access the raw `mcp.server.session.ServerSession` object
-- **`ctx.request_context`**: Access the raw `mcp.shared.context.RequestContext` object
-
 <Warning>
-Direct use of `session` or `request_context` requires understanding the low-level MCP Python SDK and may be less stable than using the methods provided directly on the `Context` object.
-</Warning>
+The MCP request is part of the low-level MCP SDK and intended for advanced use cases. Most users will not need to use it directly.
+</Warning>

docs/servers/elicitation.mdx

@@ -0,0 +1,304 @@
+---
+title: User Elicitation
+sidebarTitle: Elicitation
+description: Request structured input from users during tool execution through the MCP context.
+icon: message-question
+tag: NEW
+---
+
+import { VersionBadge } from '/snippets/version-badge.mdx'
+
+<VersionBadge version="2.10.0" />
+
+User elicitation allows MCP servers to request structured input from users during tool execution. Instead of requiring all inputs upfront, tools can interactively ask for missing parameters, clarification, or additional context as needed.
+
+<Tip>
+Most of the examples in this document assume you have a FastMCP server instance named `mcp` and show how to use the `ctx.elicit` method to request user input from an `@mcp.tool`-decorated function.
+</Tip>
+
+## What is Elicitation?
+
+Elicitation enables tools to pause execution and request specific information from users. This is particularly useful for:
+
+- **Missing parameters**: Ask for required information not provided initially
+- **Clarification requests**: Get user confirmation or choices for ambiguous scenarios  
+- **Progressive disclosure**: Collect complex information step-by-step
+- **Dynamic workflows**: Adapt tool behavior based on user responses
+
+For example, a file management tool might ask "Which directory should I create?" or a data analysis tool might request "What date range should I analyze?"
+
+### Basic Usage
+
+Use the `ctx.elicit()` method within any tool function to request user input:
+
+```python {14-17}
+from fastmcp import FastMCP, Context
+from dataclasses import dataclass
+
+mcp = FastMCP("Elicitation Server")
+
+@dataclass
+class UserInfo:
+    name: str
+    age: int
+
+@mcp.tool
+async def collect_user_info(ctx: Context) -> str:
+    """Collect user information through interactive prompts."""
+    result = await ctx.elicit(
+        message="Please provide your information",
+        response_type=UserInfo
+    )
+    
+    if result.action == "accept":
+        user = result.data
+        return f"Hello {user.name}, you are {user.age} years old"
+    elif result.action == "decline":
+        return "Information not provided"
+    else:  # cancel
+        return "Operation cancelled"
+```
+
+## Method Signature
+
+<Card icon="code" title="Context Elicitation Method">
+<ResponseField name="ctx.elicit" type="async method">
+  <Expandable title="Parameters">
+    <ResponseField name="message" type="str">
+      The prompt message to display to the user
+    </ResponseField>
+    
+    <ResponseField name="response_type" type="type" default="str">
+      The Python type defining the expected response structure (dataclass, primitive type, etc.) Note that elicitation responses are subject to a restricted subset of JSON Schema types. See [Supported Response Types](#supported-response-types) for more details.
+    </ResponseField>
+  </Expandable>
+  
+  <Expandable title="Response">
+    <ResponseField name="ElicitationResult" type="object">
+      Result object containing the user's response
+      
+      <Expandable title="properties">
+        <ResponseField name="action" type="Literal['accept', 'decline', 'cancel']">
+          How the user responded to the request
+        </ResponseField>
+        
+        <ResponseField name="data" type="response_type | None">
+          The user's input data (only present when action is "accept")
+        </ResponseField>
+      </Expandable>
+    </ResponseField>
+  </Expandable>
+</ResponseField>
+</Card>
+
+## Elicitation Actions
+
+The elicitation result contains an `action` field indicating how the user responded:
+
+- **`accept`**: User provided valid input - data is available in the `data` field
+- **`decline`**: User chose not to provide the requested information and the data field is `None`
+- **`cancel`**: User cancelled the entire operation and the data field is `None`
+
+```python {5, 7}
+@mcp.tool
+async def my_tool(ctx: Context) -> str:
+    result = await ctx.elicit("Choose an action")
+
+    if result.action == "accept":
+        return "Accepted!"
+    elif result.action == "decline":
+        return "Declined!"
+    else:
+        return "Cancelled!"
+```
+
+FastMCP also provides typed result classes for pattern matching on the `action` field:
+
+```python {1-5, 12, 14, 16}
+from fastmcp.server.elicitation import (
+    AcceptedElicitation, 
+    DeclinedElicitation, 
+    CancelledElicitation,
+)
+
+@mcp.tool
+async def pattern_example(ctx: Context) -> str:
+    result = await ctx.elicit("Enter your name:", response_type=str)
+    
+    match result:
+        case AcceptedElicitation(data=name):
+            return f"Hello {name}!"
+        case DeclinedElicitation():
+            return "No name provided"
+        case CancelledElicitation():
+            return "Operation cancelled"
+```
+
+## Response Types
+
+The server must send a schema to the client indicating the type of data it expects in response to the elicitation request. If the request is `accept`-ed, the client must send a response that matches the schema.
+
+The MCP spec only supports a limited subset of JSON Schema types for elicitation responses. Specifically, it only supports JSON  **objects** with **primitive** properties including `string`, `number` (or `integer`), `boolean` and `enum` fields.
+
+FastMCP makes it easy to request a broader range of types, including scalars (e.g. `str`), by automatically wrapping them in MCP-compatible object schemas.
+
+
+### Scalar Types
+
+You can request simple scalar data types for basic input, such as a string, integer, or boolean.
+
+When you request a scalar type, FastMCP automatically wraps it in an object schema for MCP spec compatibility. Clients will see a corresponding schema requesting a single "value" field of the requested type. Once clients respond, the provided object is "unwrapped" and the scalar value is returned to your tool function as the `data` field of the `ElicitationResult` object.
+
+As a developer, this means you do not have to worry about creating or accessing a structured object when you only need a scalar value.
+
+<CodeGroup>
+```python {4} title="Request a string"
+@mcp.tool
+async def get_user_name(ctx: Context) -> str:
+    """Get the user's name."""
+    result = await ctx.elicit("What's your name?", response_type=str)
+    
+    if result.action == "accept":
+        return f"Hello, {result.data}!"
+    return "No name provided"
+```
+```python {4} title="Request an integer"
+@mcp.tool
+async def pick_a_number(ctx: Context) -> str:
+    """Pick a number."""
+    result = await ctx.elicit("Pick a number!", response_type=int)
+    
+    if result.action == "accept":
+        return f"You picked {result.data}"
+    return "No number provided"
+```
+```python {4} title="Request a boolean"
+@mcp.tool
+async def pick_a_boolean(ctx: Context) -> str:
+    """Pick a boolean."""
+    result = await ctx.elicit("True or false?", response_type=bool)
+    
+    if result.action == "accept":
+        return f"You picked {result.data}"
+    return "No boolean provided"
+```
+</CodeGroup>
+
+### Constrained Options
+
+Often you'll want to constrain the user's response to a specific set of values. You can do this by using a `Literal` type or a Python enum as the response type, or by passing a list of strings to the `response_type` parameter as a convenient shortcut.
+
+<CodeGroup>
+```python {6} title="Using a list of strings"
+@mcp.tool
+async def set_priority(ctx: Context) -> str:
+    """Set task priority level."""
+    result = await ctx.elicit(
+        "What priority level?", 
+        response_type=["low", "medium", "high"],
+    )
+    
+    if result.action == "accept":
+        return f"Priority set to: {result.data}"
+```
+```python {1, 8} title="Using a Literal type"
+from typing import Literal
+
+@mcp.tool
+async def set_priority(ctx: Context) -> str:
+    """Set task priority level."""
+    result = await ctx.elicit(
+        "What priority level?", 
+        response_type=Literal["low", "medium", "high"]
+    )
+    
+    if result.action == "accept":
+        return f"Priority set to: {result.data}"
+    return "No priority set"
+```
+```python {1, 11} title="Using a Python enum"
+from enum import Enum
+
+class Priority(Enum):
+    LOW = "low"
+    MEDIUM = "medium"
+    HIGH = "high"   
+
+@mcp.tool
+async def set_priority(ctx: Context) -> str:
+    """Set task priority level."""
+    result = await ctx.elicit("What priority level?", response_type=Priority)
+    
+    if result.action == "accept":
+        return f"Priority set to: {result.data.value}"
+    return "No priority set"
+```
+</CodeGroup>
+
+
+### Structured Responses
+
+You can request structured data with multiple fields by using a dataclass, typed dict, or Pydantic model as the response type. Note that the MCP spec only supports shallow objects with scalar (string, number, boolean) or enum properties.
+
+```python {1, 16, 20}
+from dataclasses import dataclass
+from typing import Literal
+
+@dataclass
+class TaskDetails:
+    title: str
+    description: str
+    priority: Literal["low", "medium", "high"]
+    due_date: str
+
+@mcp.tool
+async def create_task(ctx: Context) -> str:
+    """Create a new task with user-provided details."""
+    result = await ctx.elicit(
+        "Please provide task details",
+        response_type=TaskDetails
+    )
+    
+    if result.action == "accept":
+        task = result.data
+        return f"Created task: {task.title} (Priority: {task.priority})"
+    return "Task creation cancelled"
+```
+
+## Multi-Turn Elicitation
+
+Tools can make multiple elicitation calls to gather information progressively:
+
+```python {6, 11, 16-19}
+@mcp.tool
+async def plan_meeting(ctx: Context) -> str:
+    """Plan a meeting by gathering details step by step."""
+    
+    # Get meeting title
+    title_result = await ctx.elicit("What's the meeting title?", response_type=str)
+    if title_result.action != "accept":
+        return "Meeting planning cancelled"
+    
+    # Get duration
+    duration_result = await ctx.elicit("Duration in minutes?", response_type=int)
+    if duration_result.action != "accept":
+        return "Meeting planning cancelled"
+    
+    # Get priority
+    priority_result = await ctx.elicit(
+        "Is this urgent?", 
+        response_type=Literal["yes", "no"]
+    )
+    if priority_result.action != "accept":
+        return "Meeting planning cancelled"
+    
+    urgent = priority_result.data == "yes"
+    return f"Meeting '{title_result.data}' planned for {duration_result.data} minutes (Urgent: {urgent})"
+```
+
+
+## Client Requirements
+
+Elicitation requires the client to implement an elicitation handler. See [Client Elicitation](/clients/elicitation) for details on how clients can handle these requests.
+
+If a client doesn't support elicitation, calls to `ctx.elicit()` will raise an error indicating that elicitation is not supported.

docs/servers/logging.mdx

@@ -0,0 +1,194 @@
+---
+title: Server Logging
+sidebarTitle: Logging
+description: Send log messages back to MCP clients through the context.
+icon: receipt
+---
+
+import { VersionBadge } from '/snippets/version-badge.mdx'
+
+<Tip>
+This documentation covers **MCP client logging** - sending messages from your server to MCP clients. For standard server-side logging (e.g., writing to files, console), use `fastmcp.utilities.logging.get_logger()` or Python's built-in `logging` module.
+</Tip>
+
+Server logging allows MCP tools to send debug, info, warning, and error messages back to the client. This provides visibility into function execution and helps with debugging during development and operation.
+
+## Why Use Server Logging?
+
+Server logging is essential for:
+
+- **Debugging**: Send detailed execution information to help diagnose issues
+- **Progress visibility**: Keep users informed about what the tool is doing
+- **Error reporting**: Communicate problems and their context to clients
+- **Audit trails**: Create records of tool execution for compliance or analysis
+
+Unlike standard Python logging, MCP server logging sends messages directly to the client, making them visible in the client's interface or logs.
+
+### Basic Usage
+
+Use the context logging methods within any tool function:
+
+```python {8-9, 13, 17, 21}
+from fastmcp import FastMCP, Context
+
+mcp = FastMCP("LoggingDemo")
+
+@mcp.tool
+async def analyze_data(data: list[float], ctx: Context) -> dict:
+    """Analyze numerical data with comprehensive logging."""
+    await ctx.debug("Starting analysis of numerical data")
+    await ctx.info(f"Analyzing {len(data)} data points")
+    
+    try:
+        if not data:
+            await ctx.warning("Empty data list provided")
+            return {"error": "Empty data list"}
+        
+        result = sum(data) / len(data)
+        await ctx.info(f"Analysis complete, average: {result}")
+        return {"average": result, "count": len(data)}
+        
+    except Exception as e:
+        await ctx.error(f"Analysis failed: {str(e)}")
+        raise
+```
+
+## Logging Methods
+
+<Card icon="code" title="Context Logging Methods">
+<ResponseField name="ctx.debug" type="async method">
+  Send debug-level messages for detailed execution information
+  
+  <Expandable title="parameters">
+    <ResponseField name="message" type="str">
+      The debug message to send to the client
+    </ResponseField>
+  </Expandable>
+</ResponseField>
+
+<ResponseField name="ctx.info" type="async method">
+  Send informational messages about normal execution
+  
+  <Expandable title="parameters">
+    <ResponseField name="message" type="str">
+      The information message to send to the client
+    </ResponseField>
+  </Expandable>
+</ResponseField>
+
+<ResponseField name="ctx.warning" type="async method">
+  Send warning messages for potential issues that didn't prevent execution
+  
+  <Expandable title="parameters">
+    <ResponseField name="message" type="str">
+      The warning message to send to the client
+    </ResponseField>
+  </Expandable>
+</ResponseField>
+
+<ResponseField name="ctx.error" type="async method">
+  Send error messages for problems that occurred during execution
+  
+  <Expandable title="parameters">
+    <ResponseField name="message" type="str">
+      The error message to send to the client
+    </ResponseField>
+  </Expandable>
+</ResponseField>
+
+<ResponseField name="ctx.log" type="async method">
+  Generic logging method with custom level and logger name
+  
+  <Expandable title="parameters">
+    <ResponseField name="level" type="Literal['debug', 'info', 'warning', 'error']">
+      The log level for the message
+    </ResponseField>
+    
+    <ResponseField name="message" type="str">
+      The message to send to the client
+    </ResponseField>
+    
+    <ResponseField name="logger_name" type="str | None" default="None">
+      Optional custom logger name for categorizing messages
+    </ResponseField>
+  </Expandable>
+</ResponseField>
+</Card>
+
+## Log Levels
+
+### Debug
+Use for detailed information that's typically only useful when diagnosing problems:
+
+```python 
+@mcp.tool
+async def process_file(file_path: str, ctx: Context) -> str:
+    """Process a file with detailed debug logging."""
+    await ctx.debug(f"Starting to process file: {file_path}")
+    await ctx.debug("Checking file permissions")
+    
+    # File processing logic
+    await ctx.debug("File processing completed successfully")
+    return "File processed"
+```
+
+### Info
+Use for general information about normal program execution:
+
+```python
+@mcp.tool
+async def backup_database(ctx: Context) -> str:
+    """Backup database with progress information."""
+    await ctx.info("Starting database backup")
+    await ctx.info("Connecting to database")
+    await ctx.info("Backup completed successfully")
+    return "Database backed up"
+```
+
+### Warning
+Use for potentially harmful situations that don't prevent execution:
+
+```python
+@mcp.tool
+async def validate_config(config: dict, ctx: Context) -> dict:
+    """Validate configuration with warnings for deprecated options."""
+    if "old_api_key" in config:
+        await ctx.warning("Using deprecated 'old_api_key' field. Please use 'api_key' instead")
+    
+    if config.get("timeout", 30) > 300:
+        await ctx.warning("Timeout value is very high (>5 minutes), this may cause issues")
+    
+    return {"status": "valid", "warnings": "see logs"}
+```
+
+### Error
+Use for error events that might still allow the application to continue:
+
+```python
+@mcp.tool
+async def batch_process(items: list[str], ctx: Context) -> dict:
+    """Process multiple items, logging errors for failed items."""
+    successful = 0
+    failed = 0
+    
+    for item in items:
+        try:
+            # Process item
+            successful += 1
+        except Exception as e:
+            await ctx.error(f"Failed to process item '{item}': {str(e)}")
+            failed += 1
+    
+    return {"successful": successful, "failed": failed}
+```
+
+
+## Client Handling
+
+Log messages are sent to the client through the MCP protocol. How clients handle these messages depends on their implementation:
+
+- **Development clients**: May display logs in real-time for debugging
+- **Production clients**: May store logs for later analysis or display to users
+- **Integration clients**: May forward logs to external logging systems
+
+See [Client Logging](/clients/logging) for details on how clients can handle server log messages.

docs/servers/progress.mdx

@@ -0,0 +1,189 @@
+---
+title: Progress Reporting
+sidebarTitle: Progress
+description: Update clients on the progress of long-running operations through the MCP context.
+icon: chart-line
+---
+
+import { VersionBadge } from '/snippets/version-badge.mdx'
+
+Progress reporting allows MCP tools to notify clients about the progress of long-running operations. This enables clients to display progress indicators and provide better user experience during time-consuming tasks.
+
+## Why Use Progress Reporting?
+
+Progress reporting is valuable for:
+
+- **User experience**: Keep users informed about long-running operations
+- **Progress indicators**: Enable clients to show progress bars or percentages
+- **Timeout prevention**: Demonstrate that operations are actively progressing
+- **Debugging**: Track execution progress for performance analysis
+
+### Basic Usage
+
+Use `ctx.report_progress()` to send progress updates to the client:
+
+```python {14, 21}
+from fastmcp import FastMCP, Context
+import asyncio
+
+mcp = FastMCP("ProgressDemo")
+
+@mcp.tool
+async def process_items(items: list[str], ctx: Context) -> dict:
+    """Process a list of items with progress updates."""
+    total = len(items)
+    results = []
+    
+    for i, item in enumerate(items):
+        # Report progress as we process each item
+        await ctx.report_progress(progress=i, total=total)
+        
+        # Simulate processing time
+        await asyncio.sleep(0.1)
+        results.append(item.upper())
+    
+    # Report 100% completion
+    await ctx.report_progress(progress=total, total=total)
+    
+    return {"processed": len(results), "results": results}
+```
+
+## Method Signature
+
+<Card icon="code" title="Context Progress Method">
+<ResponseField name="ctx.report_progress" type="async method">
+  Report progress to the client for long-running operations
+  
+  <Expandable title="Parameters">
+    <ResponseField name="progress" type="float">
+      Current progress value (e.g., 24, 0.75, 1500)
+    </ResponseField>
+    
+    <ResponseField name="total" type="float | None" default="None">
+      Optional total value (e.g., 100, 1.0, 2000). When provided, clients may interpret this as enabling percentage calculation.
+    </ResponseField>
+  </Expandable>
+</ResponseField>
+</Card>
+
+## Progress Patterns
+
+### Percentage-Based Progress
+
+Report progress as a percentage (0-100):
+
+```python {13-14}
+@mcp.tool
+async def download_file(url: str, ctx: Context) -> str:
+    """Download a file with percentage progress."""
+    total_size = 1000  # KB
+    downloaded = 0
+    
+    while downloaded < total_size:
+        # Download chunk
+        chunk_size = min(50, total_size - downloaded)
+        downloaded += chunk_size
+        
+        # Report percentage progress
+        percentage = (downloaded / total_size) * 100
+        await ctx.report_progress(progress=percentage, total=100)
+        
+        await asyncio.sleep(0.1)  # Simulate download time
+    
+    return f"Downloaded file from {url}"
+```
+
+### Absolute Progress
+
+Report progress with absolute values:
+
+```python {10}
+@mcp.tool
+async def backup_database(ctx: Context) -> str:
+    """Backup database tables with absolute progress."""
+    tables = ["users", "orders", "products", "inventory", "logs"]
+    
+    for i, table in enumerate(tables):
+        await ctx.info(f"Backing up table: {table}")
+        
+        # Report absolute progress
+        await ctx.report_progress(progress=i + 1, total=len(tables))
+        
+        # Simulate backup time
+        await asyncio.sleep(0.5)
+    
+    return "Database backup completed"
+```
+
+### Indeterminate Progress
+
+Report progress without a known total for operations where the endpoint is unknown:
+
+```python {11}
+@mcp.tool
+async def scan_directory(directory: str, ctx: Context) -> dict:
+    """Scan directory with indeterminate progress."""
+    files_found = 0
+    
+    # Simulate directory scanning
+    for i in range(10):  # Unknown number of files
+        files_found += 1
+        
+        # Report progress without total for indeterminate operations
+        await ctx.report_progress(progress=files_found)
+        
+        await asyncio.sleep(0.2)
+    
+    return {"files_found": files_found, "directory": directory}
+```
+
+### Multi-Stage Operations
+
+Break complex operations into stages with progress for each:
+
+```python
+@mcp.tool
+async def data_migration(source: str, destination: str, ctx: Context) -> str:
+    """Migrate data with multi-stage progress reporting."""
+    
+    # Stage 1: Validation (0-25%)
+    await ctx.info("Validating source data")
+    for i in range(5):
+        await ctx.report_progress(progress=i * 5, total=100)
+        await asyncio.sleep(0.1)
+    
+    # Stage 2: Export (25-60%)
+    await ctx.info("Exporting data from source")
+    for i in range(7):
+        progress = 25 + (i * 5)
+        await ctx.report_progress(progress=progress, total=100)
+        await asyncio.sleep(0.1)
+    
+    # Stage 3: Transform (60-80%)
+    await ctx.info("Transforming data format")
+    for i in range(4):
+        progress = 60 + (i * 5)
+        await ctx.report_progress(progress=progress, total=100)
+        await asyncio.sleep(0.1)
+    
+    # Stage 4: Import (80-100%)
+    await ctx.info("Importing to destination")
+    for i in range(4):
+        progress = 80 + (i * 5)
+        await ctx.report_progress(progress=progress, total=100)
+        await asyncio.sleep(0.1)
+    
+    # Final completion
+    await ctx.report_progress(progress=100, total=100)
+    
+    return f"Migration from {source} to {destination} completed"
+```
+
+
+## Client Requirements
+
+Progress reporting requires clients to support progress handling:
+
+- Clients must send a `progressToken` in the initial request to receive progress updates
+- If no progress token is provided, progress calls will have no effect (they won't error)
+- See [Client Progress](/clients/progress) for details on implementing client-side progress handling

docs/servers/sampling.mdx

@@ -0,0 +1,190 @@
+---
+title: LLM Sampling
+sidebarTitle: Sampling
+description: Request the client's LLM to generate text based on provided messages through the MCP context.
+icon: robot
+---
+
+import { VersionBadge } from '/snippets/version-badge.mdx'
+
+<VersionBadge version="2.0.0" />
+
+LLM sampling allows MCP tools to request the client's LLM to generate text based on provided messages. This is useful when tools need to leverage the LLM's capabilities to process data, generate responses, or perform text-based analysis.
+
+## Why Use LLM Sampling?
+
+LLM sampling enables tools to:
+
+- **Leverage AI capabilities**: Use the client's LLM for text generation and analysis
+- **Offload complex reasoning**: Let the LLM handle tasks requiring natural language understanding
+- **Generate dynamic content**: Create responses, summaries, or transformations based on data
+- **Maintain context**: Use the same LLM instance that the user is already interacting with
+
+### Basic Usage
+
+Use `ctx.sample()` to request text generation from the client's LLM:
+
+```python {14}
+from fastmcp import FastMCP, Context
+
+mcp = FastMCP("SamplingDemo")
+
+@mcp.tool
+async def analyze_sentiment(text: str, ctx: Context) -> dict:
+    """Analyze the sentiment of text using the client's LLM."""
+    prompt = f"""Analyze the sentiment of the following text as positive, negative, or neutral. 
+    Just output a single word - 'positive', 'negative', or 'neutral'.
+    
+    Text to analyze: {text}"""
+    
+    # Request LLM analysis
+    response = await ctx.sample(prompt)
+    
+    # Process the LLM's response
+    sentiment = response.text.strip().lower()
+    
+    # Map to standard sentiment values
+    if "positive" in sentiment:
+        sentiment = "positive"
+    elif "negative" in sentiment:
+        sentiment = "negative"
+    else:
+        sentiment = "neutral"
+    
+    return {"text": text, "sentiment": sentiment}
+```
+
+## Method Signature
+
+<Card icon="code" title="Context Sampling Method">
+<ResponseField name="ctx.sample" type="async method">
+  Request text generation from the client's LLM
+  
+  <Expandable title="Parameters">
+    <ResponseField name="messages" type="str | list[str | SamplingMessage]">
+      A string or list of strings/message objects to send to the LLM
+    </ResponseField>
+    
+    <ResponseField name="system_prompt" type="str | None" default="None">
+      Optional system prompt to guide the LLM's behavior
+    </ResponseField>
+    
+    <ResponseField name="temperature" type="float | None" default="None">
+      Optional sampling temperature (controls randomness, typically 0.0-1.0)
+    </ResponseField>
+    
+    <ResponseField name="max_tokens" type="int | None" default="512">
+      Optional maximum number of tokens to generate
+    </ResponseField>
+    
+    <ResponseField name="model_preferences" type="ModelPreferences | str | list[str] | None" default="None">
+      Optional model selection preferences (e.g., model hint string, list of hints, or ModelPreferences object)
+    </ResponseField>
+  </Expandable>
+  
+  <Expandable title="Response">
+    <ResponseField name="response" type="TextContent | ImageContent">
+      The LLM's response content (typically TextContent with a .text attribute)
+    </ResponseField>
+  </Expandable>
+</ResponseField>
+</Card>
+
+## Simple Text Generation
+
+### Basic Prompting
+
+Generate text with simple string prompts:
+
+```python {6}
+@mcp.tool
+async def generate_summary(content: str, ctx: Context) -> str:
+    """Generate a summary of the provided content."""
+    prompt = f"Please provide a concise summary of the following content:\n\n{content}"
+    
+    response = await ctx.sample(prompt)
+    return response.text
+```
+
+### System Prompt
+
+Use system prompts to guide the LLM's behavior:
+
+```python {4-9}
+@mcp.tool
+async def generate_code_example(concept: str, ctx: Context) -> str:
+    """Generate a Python code example for a given concept."""
+    response = await ctx.sample(
+        messages=f"Write a simple Python code example demonstrating '{concept}'.",
+        system_prompt="You are an expert Python programmer. Provide concise, working code examples without explanations.",
+        temperature=0.7,
+        max_tokens=300
+    )
+    
+    code_example = response.text
+    return f"```python\n{code_example}\n```"
+```
+
+
+### Model Preferences
+
+Specify model preferences for different use cases:
+
+```python {4-8, 17-22}
+@mcp.tool
+async def creative_writing(topic: str, ctx: Context) -> str:
+    """Generate creative content using a specific model."""
+    response = await ctx.sample(
+        messages=f"Write a creative short story about {topic}",
+        model_preferences="claude-3-sonnet",  # Prefer a specific model
+        temperature=0.9,  # High creativity
+        max_tokens=1000
+    )
+    
+    return response.text
+
+@mcp.tool
+async def technical_analysis(data: str, ctx: Context) -> str:
+    """Perform technical analysis with a reasoning-focused model."""
+    response = await ctx.sample(
+        messages=f"Analyze this technical data and provide insights: {data}",
+        model_preferences=["claude-3-opus", "gpt-4"],  # Prefer reasoning models
+        temperature=0.2,  # Low randomness for consistency
+        max_tokens=800
+    )
+    
+    return response.text
+```
+
+### Complex Message Structures
+
+Use structured messages for more complex interactions:
+
+```python {1, 6-10}
+from fastmcp.client.sampling import SamplingMessage
+
+@mcp.tool
+async def multi_turn_analysis(user_query: str, context_data: str, ctx: Context) -> str:
+    """Perform analysis using multi-turn conversation structure."""
+    messages = [
+        SamplingMessage(role="user", content=f"I have this data: {context_data}"),
+        SamplingMessage(role="assistant", content="I can see your data. What would you like me to analyze?"),
+        SamplingMessage(role="user", content=user_query)
+    ]
+    
+    response = await ctx.sample(
+        messages=messages,
+        system_prompt="You are a data analyst. Provide detailed insights based on the conversation context.",
+        temperature=0.3
+    )
+    
+    return response.text
+```
+
+## Client Requirements
+
+LLM sampling requires client support:
+
+- Clients must implement sampling handlers to process requests
+- If the client doesn't support sampling, calls to `ctx.sample()` will fail
+- See [Client Sampling](/clients/sampling) for details on implementing client-side sampling handlers

src/fastmcp/client/client.py

@@ -16,6 +16,7 @@ from mcp import ClientSession
 from pydantic import AnyUrl
 
 import fastmcp
+from fastmcp.client.elicitation import ElicitationHandler, create_elicitation_callback
 from fastmcp.client.logging import (
     LogHandler,
     create_log_callback,
@@ -58,6 +59,7 @@ __all__ = [
     "LogHandler",
     "MessageHandler",
     "SamplingHandler",
+    "ElicitationHandler",
     "ProgressHandler",
 ]
 
@@ -154,6 +156,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: MessageHandlerT | MessageHandler | None = None,
         progress_handler: ProgressHandler | None = None,
@@ -206,6 +209,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
@@ -244,6 +252,14 @@ 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,57 @@
+from __future__ import annotations
+
+from collections.abc import Awaitable, Callable
+from typing import Any, Generic, TypeAlias, TypeVar
+
+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
+from mcp.types import ElicitResult as MCPElicitResult
+from pydantic_core import to_jsonable_python
+
+from fastmcp.utilities.json_schema_type import json_schema_to_type
+
+__all__ = ["ElicitRequestParams", "ElicitResult", "ElicitationHandler"]
+
+T = TypeVar("T")
+
+
+class ElicitResult(MCPElicitResult, Generic[T]):
+    content: T | None = None
+
+
+ElicitationHandler: TypeAlias = Callable[
+    [
+        str,  # message
+        type[T],  # a class for creating a structured response
+        ElicitRequestParams,
+        RequestContext[ClientSession, LifespanContextT],
+    ],
+    Awaitable[ElicitResult[T | dict[str, Any]]],
+]
+
+
+def create_elicitation_callback(
+    elicitation_handler: ElicitationHandler,
+) -> ElicitationFnT:
+    async def _elicitation_handler(
+        context: RequestContext[ClientSession, LifespanContextT],
+        params: ElicitRequestParams,
+    ) -> MCPElicitResult | mcp.types.ErrorData:
+        try:
+            response_type = json_schema_to_type(params.requestedSchema)
+
+            result = await elicitation_handler(
+                params.message, response_type, params, context
+            )
+            content = to_jsonable_python(result.content)
+            return MCPElicitResult(**result.model_dump() | {"content": content})
+        except Exception as e:
+            return mcp.types.ErrorData(
+                code=mcp.types.INTERNAL_ERROR,
+                message=str(e),
+            )
+
+    return _elicitation_handler

src/fastmcp/client/transports.py

@@ -14,7 +14,13 @@ import anyio
 import httpx
 import mcp.types
 from mcp import ClientSession, StdioServerParameters
-from mcp.client.session import ListRootsFnT, LoggingFnT, MessageHandlerFnT, SamplingFnT
+from mcp.client.session import (
+    ElicitationFnT,
+    ListRootsFnT,
+    LoggingFnT,
+    MessageHandlerFnT,
+    SamplingFnT,
+)
 from mcp.server.fastmcp import FastMCP as FastMCP1Server
 from mcp.shared.memory import create_client_server_memory_streams
 from pydantic import AnyUrl
@@ -55,6 +61,7 @@ class SessionKwargs(TypedDict, total=False):
     sampling_callback: SamplingFnT | None
     list_roots_callback: ListRootsFnT | None
     logging_callback: LoggingFnT | None
+    elicitation_callback: ElicitationFnT | None
     message_handler: MessageHandlerFnT | None
     client_info: mcp.types.Implementation | None
 

src/fastmcp/server/context.py

@@ -1,4 +1,4 @@
-from __future__ import annotations as _annotations
+from __future__ import annotations
 
 import asyncio
 import warnings
@@ -6,6 +6,8 @@ from collections.abc import Generator
 from contextlib import contextmanager
 from contextvars import ContextVar, Token
 from dataclasses import dataclass
+from enum import Enum
+from typing import Literal, TypeVar, cast, get_origin
 
 from mcp import LoggingLevel, ServerSession
 from mcp.server.lowlevel.helper_types import ReadResourceContents
@@ -25,11 +27,20 @@ from starlette.requests import Request
 
 import fastmcp.server.dependencies
 from fastmcp import settings
+from fastmcp.server.elicitation import (
+    AcceptedElicitation,
+    CancelledElicitation,
+    DeclinedElicitation,
+    ScalarElicitationType,
+    get_elicitation_schema,
+)
 from fastmcp.server.server import FastMCP
 from fastmcp.utilities.logging import get_logger
+from fastmcp.utilities.types import get_cached_typeadapter
 
 logger = get_logger(__name__)
 
+T = TypeVar("T")
 _current_context: ContextVar[Context | None] = ContextVar("context", default=None)
 _flush_lock = asyncio.Lock()
 
@@ -167,7 +178,10 @@ class Context:
         if level is None:
             level = "info"
         await self.session.send_log_message(
-            level=level, data=message, logger=logger_name
+            level=level,
+            data=message,
+            logger=logger_name,
+            related_request_id=self.request_id,
         )
 
     @property
@@ -293,10 +307,86 @@ class Context:
             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] | list[str] | 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 the user provided a list of strings, treat it as a Literal
+        if isinstance(response_type, list):
+            if not all(isinstance(item, str) for item in response_type):
+                raise ValueError(
+                    "List of options must be a list of strings. Received: "
+                    f"{response_type}"
+                )
+            # Convert list of options to Literal type and wrap
+            choice_literal = Literal[tuple(response_type)]  # type: ignore
+            response_type = ScalarElicitationType[choice_literal]  # type: ignore
+        # if the user provided a primitive scalar, wrap it in an object schema
+        elif response_type in {bool, int, float, str}:
+            response_type = ScalarElicitationType[response_type]  # type: ignore
+        # if the user provided a Literal type, wrap it in an object schema
+        elif get_origin(response_type) is Literal:
+            response_type = ScalarElicitationType[response_type]  # type: ignore
+        # if the user provided an Enum type, wrap it in an object schema
+        elif isinstance(response_type, type) and issubclass(response_type, Enum):
+            response_type = ScalarElicitationType[response_type]  # type: ignore
+
+        response_type = cast(type[T], response_type)
+
+        requested_schema = get_elicitation_schema(response_type)
+
+        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 | ScalarElicitationType[T],
+                type_adapter.validate_python(result.content),
+            )
+            if isinstance(validated_data, ScalarElicitationType):
+                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,165 @@
+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",
+    "ScalarElicitationType",
+]
+
+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 ScalarElicitationType(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)
+    - Allows const fields (for Literal types) and enum fields (for Enum types)
+    - 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 const fields (Literal types)
+        if "const" in prop_schema:
+            continue  # const fields are allowed regardless of type
+
+        # Handle enum fields (Enum types)
+        if "enum" in prop_schema:
+            continue  # enum fields are allowed regardless of type
+
+        # Handle references to definitions (like Enum types)
+        if "$ref" in prop_schema:
+            # Get the referenced definition
+            ref_path = prop_schema["$ref"]
+            if ref_path.startswith("#/$defs/"):
+                def_name = ref_path[8:]  # Remove "#/$defs/" prefix
+                ref_def = schema.get("$defs", {}).get(def_name, {})
+                # If the referenced definition has an enum, it's allowed
+                if "enum" in ref_def:
+                    continue
+                # If the referenced definition has a type that's allowed, it's allowed
+                ref_type = ref_def.get("type")
+                if ref_type in ALLOWED_TYPES:
+                    continue
+            # If we can't determine what the ref points to, reject it for safety
+            raise TypeError(
+                f"Elicitation schema field '{prop_name}' contains a reference '{ref_path}' "
+                "that could not be validated. Only references to enum types or primitive types are allowed."
+            )
+
+        # 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:
+                # Allow const and enum in unions
+                if "const" in union_schema or "enum" in union_schema:
+                    continue
+                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_type.py

@@ -41,7 +41,6 @@ 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,
@@ -254,8 +253,7 @@ def _create_numeric_type(
 
 def _create_enum(name: str, values: list[Any]) -> type:
     """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})  # type: ignore[return-value]
+    # Always return Literal for enum fields to preserve the literal nature
     return Literal[tuple(values)]  # type: ignore[return-value]
 
 
@@ -399,15 +397,19 @@ def _schema_to_type(
 
 def _sanitize_name(name: str) -> str:
     """Convert string to valid Python identifier."""
+    original_name = name
     # 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]):
+    # Step 3: if the first char of original name isn't a letter or underscore, 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("_")
+    # Step 4: deduplicate again
+    cleaned = re.sub(r"__+", "_", cleaned)
+    # Step 5: only strip trailing underscores if they weren't in the original name
+    if not original_name.endswith("_"):
+        cleaned = cleaned.rstrip("_")
     return cleaned
 
 

tests/client/test_elicitation.py

@@ -0,0 +1,578 @@
+from dataclasses import asdict, dataclass
+from enum import Enum
+from typing import Literal
+
+import pytest
+from pydantic import BaseModel
+from typing_extensions import TypedDict
+
+from fastmcp import Context, FastMCP
+from fastmcp.client.client import Client
+from fastmcp.client.elicitation import ElicitResult
+from fastmcp.exceptions import ToolError
+from fastmcp.server.elicitation import (
+    AcceptedElicitation,
+    CancelledElicitation,
+    DeclinedElicitation,
+)
+from fastmcp.utilities.types import TypeAdapter
+
+
+@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_with_no_handler(fastmcp_server):
+    """Test that elicitation works without a handler."""
+
+    async with Client(fastmcp_server) as client:
+        with pytest.raises(ToolError, match="Elicitation not supported"):
+            await client.call_tool("ask_for_name", {})
+
+
+async def test_elicitation_accept_content(fastmcp_server):
+    """Test basic elicitation functionality."""
+
+    async def elicitation_handler(message, response_type, params, ctx):
+        # Mock user providing their name
+        return ElicitResult(action="accept", content=response_type(name="Alice"))
+
+    async with Client(
+        fastmcp_server, elicitation_handler=elicitation_handler
+    ) as client:
+        result = await client.call_tool("ask_for_name", {})
+        assert result.data == "Hello, Alice!"
+
+
+async def test_elicitation_decline(fastmcp_server):
+    """Test that elicitation handler receives correct parameters."""
+
+    async def elicitation_handler(message, response_type, params, 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.data == "No name provided."
+
+
+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, response_type, params, 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.data == "Your favorite color is blue!"
+
+
+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, response_type, params, ctx):
+        captured_params["message"] = message
+        captured_params["response_type"] = str(response_type)
+        captured_params["params"] = params
+        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 "ScalarElicitationType" in str(captured_params["response_type"])
+        assert captured_params["params"].requestedSchema == {
+            "properties": {"value": {"title": "Value", "type": "integer"}},
+            "required": ["value"],
+            "title": "ScalarElicitationType",
+            "type": "object",
+        }
+        assert captured_params["ctx"] is not None
+
+
+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, response_type, params, 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.data == "Request was canceled"
+
+
+class TestScalarResponseTypes:
+    async def test_elicitation_str_response(self):
+        """Test elicitation with string schema."""
+        mcp = FastMCP("TestServer")
+
+        @mcp.tool
+        async def my_tool(context: Context) -> str:
+            result = await context.elicit(message="", response_type=str)
+            return result.data  # type: ignore[attr-defined]
+
+        async def elicitation_handler(message, response_type, params, ctx):
+            return ElicitResult(action="accept", content={"value": "hello"})
+
+        async with Client(mcp, elicitation_handler=elicitation_handler) as client:
+            result = await client.call_tool("my_tool", {})
+            assert result.data == "hello"
+
+    async def test_elicitation_int_response(self):
+        """Test elicitation with number schema."""
+        mcp = FastMCP("TestServer")
+
+        @mcp.tool
+        async def my_tool(context: Context) -> int:
+            result = await context.elicit(message="", response_type=int)
+            return result.data  # type: ignore[attr-defined]
+
+        async def elicitation_handler(message, response_type, params, ctx):
+            return ElicitResult(action="accept", content={"value": 42})
+
+        async with Client(mcp, elicitation_handler=elicitation_handler) as client:
+            result = await client.call_tool("my_tool", {})
+            assert result.data == 42
+
+    async def test_elicitation_float_response(self):
+        """Test elicitation with number schema."""
+        mcp = FastMCP("TestServer")
+
+        @mcp.tool
+        async def my_tool(context: Context) -> float:
+            result = await context.elicit(message="", response_type=float)
+            return result.data  # type: ignore[attr-defined]
+
+        async def elicitation_handler(message, response_type, params, ctx):
+            return ElicitResult(action="accept", content={"value": 3.14})
+
+        async with Client(mcp, elicitation_handler=elicitation_handler) as client:
+            result = await client.call_tool("my_tool", {})
+            assert result.data == 3.14
+
+    async def test_elicitation_bool_response(self):
+        """Test elicitation with boolean schema."""
+        mcp = FastMCP("TestServer")
+
+        @mcp.tool
+        async def my_tool(context: Context) -> bool:
+            result = await context.elicit(message="", response_type=bool)
+            return result.data  # type: ignore[attr-defined]
+
+        async def elicitation_handler(message, response_type, params, ctx):
+            return ElicitResult(action="accept", content={"value": True})
+
+        async with Client(mcp, elicitation_handler=elicitation_handler) as client:
+            result = await client.call_tool("my_tool", {})
+            assert result.data is True
+
+    async def test_elicitation_literal_response(self):
+        """Test elicitation with literal schema."""
+        mcp = FastMCP("TestServer")
+
+        @mcp.tool
+        async def my_tool(context: Context) -> Literal["x", "y"]:
+            result = await context.elicit(message="", response_type=Literal["x", "y"])  # type: ignore
+            return result.data  # type: ignore[attr-defined]
+
+        async def elicitation_handler(message, response_type, params, ctx):
+            return ElicitResult(action="accept", content={"value": "x"})
+
+        async with Client(mcp, elicitation_handler=elicitation_handler) as client:
+            result = await client.call_tool("my_tool", {})
+            assert result.data == "x"
+
+    async def test_elicitation_enum_response(self):
+        """Test elicitation with enum schema."""
+        mcp = FastMCP("TestServer")
+
+        class ResponseEnum(Enum):
+            X = "x"
+            Y = "y"
+
+        @mcp.tool
+        async def my_tool(context: Context) -> ResponseEnum:
+            result = await context.elicit(message="", response_type=ResponseEnum)
+            return result.data  # type: ignore[attr-defined]
+
+        async def elicitation_handler(message, response_type, params, ctx):
+            return ElicitResult(action="accept", content={"value": "x"})
+
+        async with Client(mcp, elicitation_handler=elicitation_handler) as client:
+            result = await client.call_tool("my_tool", {})
+            assert result.data == "x"
+
+    async def test_elicitation_list_response(self):
+        """Test elicitation with list schema."""
+        mcp = FastMCP("TestServer")
+
+        @mcp.tool
+        async def my_tool(context: Context) -> str:
+            result = await context.elicit(message="", response_type=["x", "y"])
+            return result.data  # type: ignore[attr-defined]
+
+        async def elicitation_handler(message, response_type, params, ctx):
+            return ElicitResult(action="accept", content={"value": "x"})
+
+        async with Client(mcp, elicitation_handler=elicitation_handler) as client:
+            result = await client.call_tool("my_tool", {})
+            assert result.data == "x"
+
+
+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, response_type, params, 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.data
+
+
+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, response_type, params, ctx):
+        nonlocal call_count
+        call_count += 1
+        if call_count == 1:
+            return ElicitResult(action="accept", content={"value": "Bob"})
+        elif call_count == 2:
+            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.data == "Hello Bob, you are 25 years old"
+        assert call_count == 2
+
+
+@dataclass
+class UserInfo:
+    name: str
+    age: int
+
+
+class UserInfoTypedDict(TypedDict):
+    name: str
+    age: int
+
+
+class UserInfoPydantic(BaseModel):
+    name: str
+    age: int
+
+
+@pytest.mark.parametrize(
+    "structured_type", [UserInfo, UserInfoTypedDict, UserInfoPydantic]
+)
+async def test_structured_response_type(
+    structured_type: type[UserInfo | UserInfoTypedDict | UserInfoPydantic],
+):
+    """Test elicitation with dataclass response type."""
+    mcp = FastMCP("TestServer")
+
+    @mcp.tool
+    async def get_user_info(context: Context) -> str:
+        result = await context.elicit(
+            message="Please provide your information", response_type=structured_type
+        )
+        if result.action == "accept":
+            if isinstance(result.data, dict):
+                return f"User: {result.data['name']}, age: {result.data['age']}"
+            else:
+                return f"User: {result.data.name}, age: {result.data.age}"
+        return "No user info provided"
+
+    async def elicitation_handler(message, response_type, params, ctx):
+        # Verify we get the dataclass type
+        assert (
+            TypeAdapter(response_type).json_schema()
+            == TypeAdapter(structured_type).json_schema()
+        )
+
+        # Verify the schema has the dataclass fields (available in params)
+        schema = params.requestedSchema
+        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=UserInfo(name="Alice", age=30))
+
+    async with Client(mcp, elicitation_handler=elicitation_handler) as client:
+        result = await client.call_tool("get_user_info", {})
+        assert result.data == "User: Alice, age: 30"
+
+
+async def test_all_primitive_field_types():
+    class DataEnum(Enum):
+        X = "x"
+        Y = "y"
+
+    @dataclass
+    class Data:
+        integer: int
+        float_: float
+        number: int | float
+        boolean: bool
+        string: str
+        constant: Literal["x"]
+        union: Literal["x"] | Literal["y"]
+        choice: Literal["x", "y"]
+        enum: DataEnum
+
+    mcp = FastMCP("TestServer")
+
+    @mcp.tool
+    async def get_data(context: Context) -> Data:
+        result = await context.elicit(message="Enter data", response_type=Data)
+        return result.data  # type: ignore[attr-defined]
+
+    async def elicitation_handler(message, response_type, params, ctx):
+        return ElicitResult(
+            action="accept",
+            content=Data(
+                integer=1,
+                float_=1.0,
+                number=1.0,
+                boolean=True,
+                string="hello",
+                constant="x",
+                union="x",
+                choice="x",
+                enum=DataEnum.X,
+            ),
+        )
+
+    async with Client(mcp, elicitation_handler=elicitation_handler) as client:
+        result = await client.call_tool("get_data", {})
+
+        # Now all literal/enum fields should be preserved as strings
+        result_data = asdict(result.data)
+        result_data_enum = result_data.pop("enum")
+        assert result_data_enum == "x"  # Should be a string now, not an enum
+        assert result_data == {
+            "integer": 1,
+            "float_": 1.0,
+            "number": 1.0,
+            "boolean": True,
+            "string": "hello",
+            "constant": "x",
+            "union": "x",
+            "choice": "x",
+        }
+
+
+class TestValidation:
+    async def test_schema_validation_rejects_non_object(self):
+        """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(self):
+        """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(self):
+        """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(self):
+        """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"}}
+                    },
+                }
+            )
+
+
+class TestPatternMatching:
+    async def test_pattern_matching_accept(self):
+        """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, response_type, params, 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.data == "Hello Alice!"
+
+    async def test_pattern_matching_decline(self):
+        """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, response_type, params, 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.data == "You declined"
+
+    async def test_pattern_matching_cancel(self):
+        """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, response_type, params, 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.data == "Cancelled"

tests/utilities/test_json_schema_type.py

@@ -1,5 +1,7 @@
+from dataclasses import dataclass
 from datetime import datetime
-from typing import Any, Union
+from enum import Enum
+from typing import Any, Literal, Union
 
 import pytest
 from pydantic import AnyUrl, BaseModel, TypeAdapter, ValidationError
@@ -106,6 +108,65 @@ class TestSimpleTypes:
             validator.validate_python(False)
 
 
+class TestConstrainedTypes:
+    def test_constant(self):
+        validator = TypeAdapter(Literal["x"])
+        schema = validator.json_schema()
+        type_ = json_schema_to_type(schema)
+        assert type_ == Literal["x"]
+        assert TypeAdapter(type_).validate_python("x") == "x"
+        with pytest.raises(ValidationError):
+            TypeAdapter(type_).validate_python("y")
+
+    def test_union_constants(self):
+        validator = TypeAdapter(Literal["x"] | Literal["y"])
+        schema = validator.json_schema()
+        type_ = json_schema_to_type(schema)
+        assert type_ == Literal["x"] | Literal["y"]
+        assert TypeAdapter(type_).validate_python("x") == "x"
+        assert TypeAdapter(type_).validate_python("y") == "y"
+        with pytest.raises(ValidationError):
+            TypeAdapter(type_).validate_python("z")
+
+    def test_enum_str(self):
+        class MyEnum(Enum):
+            X = "x"
+            Y = "y"
+
+        validator = TypeAdapter(MyEnum)
+        schema = validator.json_schema()
+        type_ = json_schema_to_type(schema)
+        assert type_ == Literal["x", "y"]
+        assert TypeAdapter(type_).validate_python("x") == "x"
+        assert TypeAdapter(type_).validate_python("y") == "y"
+        with pytest.raises(ValidationError):
+            TypeAdapter(type_).validate_python("z")
+
+    def test_enum_int(self):
+        class MyEnum(Enum):
+            X = 1
+            Y = 2
+
+        validator = TypeAdapter(MyEnum)
+        schema = validator.json_schema()
+        type_ = json_schema_to_type(schema)
+        assert type_ == Literal[1, 2]
+        assert TypeAdapter(type_).validate_python(1) == 1
+        assert TypeAdapter(type_).validate_python(2) == 2
+        with pytest.raises(ValidationError):
+            TypeAdapter(type_).validate_python(3)
+
+    def test_choice(self):
+        validator = TypeAdapter(Literal["x", "y"])
+        schema = validator.json_schema()
+        type_ = json_schema_to_type(schema)
+        assert type_ == Literal["x", "y"]
+        assert TypeAdapter(type_).validate_python("x") == "x"
+        assert TypeAdapter(type_).validate_python("y") == "y"
+        with pytest.raises(ValidationError):
+            TypeAdapter(type_).validate_python("z")
+
+
 class TestStringConstraints:
     """Test suite for string constraint validation."""
 
@@ -386,6 +447,29 @@ class TestObjectTypes:
         with pytest.raises(ValidationError):
             validator.validate_python({"user": {"age": 30}})
 
+    def test_object_with_underscore_names(self):
+        @dataclass
+        class Data:
+            x: int
+            x_: int
+            _x: int
+
+        schema = TypeAdapter(Data).json_schema()
+        assert schema == {
+            "title": "Data",
+            "type": "object",
+            "properties": {
+                "x": {"type": "integer", "title": "X"},
+                "x_": {"type": "integer", "title": "X"},
+                "_x": {"type": "integer", "title": "X"},
+            },
+            "required": ["x", "x_", "_x"],
+        }
+
+        object = json_schema_to_type(schema)
+        object_schema = TypeAdapter(object).json_schema()
+        assert object_schema == schema
+
 
 class TestDefaultValues:
     """Test suite for default value handling."""