Notifications

from mcp_use.server import Context, MCPServer

server = MCPServer(name="My Server")

@server.tool()
async def process_batch(items: list[str], ctx: Context) -> str:
    """Process items with progress reporting."""
    total = len(items)

    for i, item in enumerate(items):
        # Process the item
        await process_item(item)

        # Report progress
        await ctx.report_progress(
            progress=(i + 1) / total,
            total=1.0,
            message=f"Processed {i + 1}/{total} items"
        )

    return f"Processed {total} items successfully"

@server.tool()
async def add_new_tool(tool_name: str, ctx: Context) -> str:
    """Dynamically add a new tool and notify clients."""
    # Add the tool dynamically
    register_new_tool(tool_name)

    # Notify clients that the tool list has changed
    await ctx.send_tool_list_changed()

    return f"Added tool: {tool_name}"

@server.tool()
async def refresh_resources(ctx: Context) -> str:
    """Refresh available resources and notify clients."""
    # Update resources
    update_resource_cache()

    # Notify clients
    await ctx.send_resource_list_changed()

    return "Resources refreshed"

@server.tool()
async def analyze_dataset(file_path: str, ctx: Context) -> str:
    """Analyze a dataset with progress notifications."""
    await ctx.info(f"Starting analysis of {file_path}")

    steps = ["Loading data", "Cleaning", "Analyzing", "Generating report"]

    for i, step in enumerate(steps):
        await ctx.report_progress(
            progress=(i + 1) / len(steps),
            total=1.0,
            message=step
        )
        await perform_step(step, file_path)

    return "Analysis complete"