2571 Context Features 5d1bde1e

FastMCP Context Features Reference

Complete reference for FastMCP's advanced context features: elicitation, progress tracking, and sampling.

Context Injection

To use context features, inject Context into your tool:

from fastmcp import Context

@mcp.tool()
async def tool_with_context(param: str, context: Context) -> dict:
    """Tool that uses context features."""
    # Access context features here
    pass

Important: Context parameter MUST have type hint Context for injection to work.

Feature 1: Elicitation (User Input)

Request user input during tool execution.

Basic Usage

from fastmcp import Context

@mcp.tool()
async def confirm_action(action: str, context: Context) -> dict:
    """Request user confirmation."""
    # Request user input
    user_response = await context.request_elicitation(
        prompt=f"Confirm {action}? (yes/no)",
        response_type=str
    )

    if user_response.lower() == "yes":
        result = await perform_action(action)
        return {"status": "completed", "action": action}
    else:
        return {"status": "cancelled", "action": action}

Type-Based Elicitation

@mcp.tool()
async def collect_user_info(context: Context) -> dict:
    """Collect information from user."""
    # String input
    name = await context.request_elicitation(
        prompt="What is your name?",
        response_type=str
    )

    # Boolean input
    confirmed = await context.request_elicitation(
        prompt="Do you want to continue?",
        response_type=bool
    )

    # Numeric input
    count = await context.request_elicitation(
        prompt="How many items?",
        response_type=int
    )

    return {
        "name": name,
        "confirmed": confirmed,
        "count": count
    }

Custom Type Elicitation

from dataclasses import dataclass

@dataclass
class UserChoice:
    option: str
    reason: str

@mcp.tool()
async def get_user_choice(options: list[str], context: Context) -> dict:
    """Get user choice with reasoning."""
    choice = await context.request_elicitation(
        prompt=f"Choose from: {', '.join(options)}",
        response_type=UserChoice
    )

    return {
        "selected": choice.option,
        "reason": choice.reason
    }

Client Handler for Elicitation

Client must provide handler:

from fastmcp import Client

async def elicitation_handler(message: str, response_type: type, context: dict):
    """Handle elicitation requests."""
    if response_type == str:
        return input(f"{message}: ")
    elif response_type == bool:
        response = input(f"{message} (y/n): ")
        return response.lower() == 'y'
    elif response_type == int:
        return int(input(f"{message}: "))
    else:
        return input(f"{message}: ")

async with Client(
    "server.py",
    elicitation_handler=elicitation_handler
) as client:
    result = await client.call_tool("collect_user_info", {})

Feature 2: Progress Tracking

Report progress for long-running operations.

Basic Progress

@mcp.tool()
async def long_operation(count: int, context: Context) -> dict:
    """Operation with progress tracking."""
    for i in range(count):
        # Report progress
        await context.report_progress(
            progress=i + 1,
            total=count,
            message=f"Processing item {i + 1}/{count}"
        )

        # Do work
        await asyncio.sleep(0.1)

    return {"status": "completed", "processed": count}

Multi-Phase Progress

@mcp.tool()
async def multi_phase_operation(data: list, context: Context) -> dict:
    """Operation with multiple phases."""
    # Phase 1: Loading
    await context.report_progress(0, 3, "Phase 1: Loading data")
    loaded = await load_data(data)

    # Phase 2: Processing
    await context.report_progress(1, 3, "Phase 2: Processing")
    for i, item in enumerate(loaded):
        await context.report_progress(
            progress=i,
            total=len(loaded),
            message=f"Processing {i + 1}/{len(loaded)}"
        )
        await process_item(item)

    # Phase 3: Saving
    await context.report_progress(2, 3, "Phase 3: Saving results")
    await save_results()

    await context.report_progress(3, 3, "Complete!")

    return {"status": "completed", "items": len(loaded)}

Indeterminate Progress

For operations where total is unknown:

@mcp.tool()
async def indeterminate_operation(context: Context) -> dict:
    """Operation with unknown duration."""
    stages = [
        "Initializing",
        "Loading data",
        "Processing",
        "Finalizing"
    ]

    for stage in stages:
        # No total - shows as spinner/indeterminate
        await context.report_progress(
            progress=stages.index(stage),
            total=None,
            message=stage
        )
        await perform_stage(stage)

    return {"status": "completed"}

Client Handler for Progress

async def progress_handler(progress: float, total: float | None, message: str | None):
    """Handle progress updates."""
    if total:
        pct = (progress / total) * 100
        # Use \r for same-line update
        print(f"\r[{pct:.1f}%] {message}", end="", flush=True)
    else:
        # Indeterminate progress
        print(f"\n[PROGRESS] {message}")

async with Client(
    "server.py",
    progress_handler=progress_handler
) as client:
    result = await client.call_tool("long_operation", {"count": 100})

Feature 3: Sampling (LLM Integration)

Request LLM completions from within tools.

Basic Sampling

@mcp.tool()
async def enhance_text(text: str, context: Context) -> str:
    """Enhance text using LLM."""
    response = await context.request_sampling(
        messages=[{
            "role": "system",
            "content": "You are a professional copywriter."
        }, {
            "role": "user",
            "content": f"Enhance this text: {text}"
        }],
        temperature=0.7,
        max_tokens=500
    )

    return response["content"]

Structured Output with Sampling

@mcp.tool()
async def classify_text(text: str, context: Context) -> dict:
    """Classify text using LLM."""
    prompt = f"""
    Classify this text: {text}

    Return JSON with:
    - category: one of [news, blog, academic, social]
    - sentiment: one of [positive, negative, neutral]
    - topics: list of main topics

    Return as JSON object.
    """

    response = await context.request_sampling(
        messages=[{"role": "user", "content": prompt}],
        temperature=0.3,  # Lower for consistency
        response_format="json"
    )

    import json
    return json.loads(response["content"])

Multi-Turn Sampling

@mcp.tool()
async def interactive_analysis(topic: str, context: Context) -> dict:
    """Multi-turn analysis with LLM."""
    # First turn: Generate questions
    questions_response = await context.request_sampling(
        messages=[{
            "role": "user",
            "content": f"Generate 3 key questions about: {topic}"
        }],
        max_tokens=200
    )

    # Second turn: Answer questions
    analysis_response = await context.request_sampling(
        messages=[{
            "role": "user",
            "content": f"Answer these questions about {topic}:\n{questions_response['content']}"
        }],
        max_tokens=500
    )

    return {
        "topic": topic,
        "questions": questions_response["content"],
        "analysis": analysis_response["content"]
    }

Client Handler for Sampling

Client provides LLM access:

async def sampling_handler(messages, params, context):
    """Handle LLM sampling requests."""
    # Call your LLM API
    from openai import AsyncOpenAI

    client = AsyncOpenAI()
    response = await client.chat.completions.create(
        model=params.get("model", "gpt-4"),
        messages=messages,
        temperature=params.get("temperature", 0.7),
        max_tokens=params.get("max_tokens", 1000)
    )

    return {
        "content": response.choices[0].message.content,
        "model": response.model,
        "usage": {
            "prompt_tokens": response.usage.prompt_tokens,
            "completion_tokens": response.usage.completion_tokens,
            "total_tokens": response.usage.total_tokens
        }
    }

async with Client(
    "server.py",
    sampling_handler=sampling_handler
) as client:
    result = await client.call_tool("enhance_text", {"text": "Hello world"})

Combined Example

All context features together:

@mcp.tool()
async def comprehensive_task(data: list, context: Context) -> dict:
    """Task using all context features."""
    # 1. Elicitation: Confirm operation
    confirmed = await context.request_elicitation(
        prompt="Start processing?",
        response_type=bool
    )

    if not confirmed:
        return {"status": "cancelled"}

    # 2. Progress: Track processing
    results = []
    for i, item in enumerate(data):
        await context.report_progress(
            progress=i + 1,
            total=len(data),
            message=f"Processing {i + 1}/{len(data)}"
        )

        # 3. Sampling: Use LLM for processing
        enhanced = await context.request_sampling(
            messages=[{
                "role": "user",
                "content": f"Analyze this item: {item}"
            }],
            temperature=0.5
        )

        results.append({
            "item": item,
            "analysis": enhanced["content"]
        })

    return {
        "status": "completed",
        "total": len(data),
        "results": results
    }

Best Practices

Elicitation

• Clear prompts: Be specific about what you're asking
• Type validation: Use appropriate response_type
• Handle cancellation: Allow users to cancel operations
• Provide context: Explain why input is needed

Progress Tracking

• Regular updates: Report every 5-10% or every item
• Meaningful messages: Describe what's happening
• Phase indicators: Show which phase of operation
• Final confirmation: Report 100% completion

Sampling

• System prompts: Set clear instructions
• Temperature control: Lower for factual, higher for creative
• Token limits: Set reasonable max_tokens
• Error handling: Handle API failures gracefully
• Cost awareness: Sampling uses LLM API (costs money)

Error Handling

Context Not Available

@mcp.tool()
async def safe_context_usage(context: Context) -> dict:
    """Safely use context features."""
    # Check if feature is available
    if hasattr(context, 'report_progress'):
        await context.report_progress(0, 100, "Starting")

    if hasattr(context, 'request_elicitation'):
        response = await context.request_elicitation(
            prompt="Continue?",
            response_type=bool
        )
    else:
        # Fallback behavior
        response = True

    return {"status": "completed"}

Timeout Handling

import asyncio

@mcp.tool()
async def elicitation_with_timeout(context: Context) -> dict:
    """Elicitation with timeout."""
    try:
        response = await asyncio.wait_for(
            context.request_elicitation(
                prompt="Your input (30 seconds):",
                response_type=str
            ),
            timeout=30.0
        )
        return {"status": "completed", "input": response}
    except asyncio.TimeoutError:
        return {"status": "timeout", "message": "No input received"}

Context Feature Availability

⚠️ = Feature availability depends on client implementation

Resources

• Context API: See SKILL.md for full Context API reference
• Client Handlers: See client-example.py template
• MCP Protocol: https://modelcontextprotocol.io