Chat
Val Town is a collaborative website to build and scale JavaScript apps.
Deploy APIs, crons, & store data – all from the browser, and deployed in milliseconds.
Viewing readonly version of main branch: v1269View latest version
This document provides practical examples of how to implement NextSteps in your MCP server tools.
def analyze_file(file_path): """Analyze a file and suggest next steps""" # Perform analysis analysis_result = perform_analysis(file_path) # Create NextSteps for automatic continuation next_steps = { "nextSteps": { "type": "auto", "countdown": 8, "proposals": [ { "id": "analyze_next_section", "title": "Analyze Next Section", "description": "Continue with the next data segment", "action": { "type": "send_message", "content": f"Please analyze the next section of {file_path} using the same methodology." } } ] } } return { "content": [ { "type": "text", "text": json.dumps({ "analysis": analysis_result, "output": next_steps }) } ] }
def process_data(dataset_name): """Process data and offer multiple next steps""" processing_result = process_dataset(dataset_name) next_steps = { "nextSteps": { "type": "manual", "priority": 80, "proposals": [ { "id": "save_results", "title": "Save Results", "description": "Save processed data to file", "action": { "type": "send_message", "content": f"Save the processed results from {dataset_name} to a CSV file named 'processed_{dataset_name}.csv'" } }, { "id": "create_visualization", "title": "Create Chart", "description": "Generate data visualization", "action": { "type": "send_message", "content": f"Create a bar chart visualization of the processed {dataset_name} data showing the top 10 categories" } }, { "id": "statistical_analysis", "title": "Run Statistics", "description": "Perform statistical analysis", "action": { "type": "send_message", "content": f"Perform statistical analysis on the processed {dataset_name} data including mean, median, standard deviation, and correlation matrix" } } ] } } return { "content": [ { "type": "text", "text": json.dumps({ "processing_summary": processing_result, "output": next_steps }) } ] }
def upload_file(file_path, destination): """Upload file with automatic retry on failure""" try: result = perform_upload(file_path, destination) return success_response(result) except UploadError as e: # Offer automatic retry next_steps = { "nextSteps": { "type": "auto", "countdown": 5, "priority": 95, "proposals": [ { "id": "retry_upload", "title": "Retry Upload", "description": "Automatically retry the failed upload", "action": { "type": "send_message", "content": f"Retry uploading {file_path} to {destination} with increased timeout settings" } } ] } } return { "content": [ { "type": "text", "text": json.dumps({ "error": str(e), "output": next_steps }) } ] }
def finalize_report(report_id): """Finalize report and offer publishing options""" report = generate_final_report(report_id) next_steps = { "nextSteps": { "type": "manual", "priority": 70, "proposals": [ { "id": "publish_report", "title": "Publish Report", "description": "Make report available to stakeholders", "action": { "type": "send_message", "content": f"Publish report {report_id} to the stakeholder portal and send notification emails" } }, { "id": "schedule_presentation", "title": "Schedule Presentation", "description": "Set up presentation meeting", "action": { "type": "send_message", "content": f"Schedule a presentation meeting for report {report_id} with the project team for next week" } }, { "id": "archive_data", "title": "Archive Source Data", "description": "Move source data to archive", "action": { "type": "send_message", "content": f"Archive the source data used for report {report_id} to long-term storage" } } ] } } return { "content": [ { "type": "text", "text": json.dumps({ "report": report, "output": next_steps }) } ] }
def analyze_system_health(): """Analyze system health and suggest appropriate actions""" health_status = check_system_health() if health_status["critical_issues"]: # Critical issues - auto-execute immediate fixes next_steps = { "nextSteps": { "type": "auto", "countdown": 3, "priority": 100, "proposals": [ { "id": "emergency_restart", "title": "Emergency Restart", "description": "Restart critical services immediately", "action": { "type": "send_message", "content": "Execute emergency restart of critical services: database, web server, and cache" } } ] } } elif health_status["warnings"]: # Warnings - offer manual intervention options next_steps = { "nextSteps": { "type": "manual", "priority": 60, "proposals": [ { "id": "investigate_warnings", "title": "Investigate Warnings", "description": "Deep dive into warning conditions", "action": { "type": "send_message", "content": "Investigate the warning conditions in detail and provide recommendations" } }, { "id": "schedule_maintenance", "title": "Schedule Maintenance", "description": "Plan maintenance window", "action": { "type": "send_message", "content": "Schedule a maintenance window to address the identified warning conditions" } } ] } } else: # All good - suggest optimization next_steps = { "nextSteps": { "type": "manual", "priority": 30, "proposals": [ { "id": "optimize_performance", "title": "Optimize Performance", "description": "Look for optimization opportunities", "action": { "type": "send_message", "content": "Analyze system performance metrics and suggest optimization opportunities" } } ] } } return { "content": [ { "type": "text", "text": json.dumps({ "health_status": health_status, "output": next_steps }) } ] }
def create_dashboard(): """Create dashboard with interactive NextSteps""" dashboard_html = generate_dashboard_html() next_steps = { "nextSteps": { "type": "manual", "priority": 50, "proposals": [ { "id": "refresh_data", "title": "Refresh Data", "description": "Update dashboard with latest data", "action": { "type": "send_message", "content": "Refresh the dashboard with the latest data from all connected sources" } }, { "id": "export_pdf", "title": "Export PDF", "description": "Generate PDF report", "action": { "type": "send_message", "content": "Export the current dashboard as a PDF report for offline viewing" } }, { "id": "share_dashboard", "title": "Share Dashboard", "description": "Generate sharing link", "action": { "type": "send_message", "content": "Create a secure sharing link for this dashboard that expires in 7 days" } } ] } } return { "content": [ { "type": "text", "text": json.dumps({ "output": { "html": dashboard_html, **next_steps } }) } ] }
def test_nextsteps(): """Simple test tool for NextSteps functionality""" next_steps = { "nextSteps": { "type": "auto", "countdown": 5, "proposals": [ { "id": "test_success", "title": "Test Successful", "description": "NextSteps implementation is working", "action": { "type": "send_message", "content": "🎉 NextSteps test completed successfully! The feature is working correctly." } } ] } } return { "content": [ { "type": "text", "text": json.dumps({ "message": "NextSteps test initiated. This should auto-execute in 5 seconds.", "output": next_steps }) } ] }
def validate_nextsteps(nextsteps_data): """Validate NextSteps data structure""" required_fields = ["type", "proposals"] optional_fields = ["priority", "countdown"] # Check required fields for field in required_fields: if field not in nextsteps_data: raise ValueError(f"Missing required field: {field}") # Validate type if nextsteps_data["type"] not in ["auto", "manual"]: raise ValueError("Type must be 'auto' or 'manual'") # Validate proposals if not isinstance(nextsteps_data["proposals"], list) or len(nextsteps_data["proposals"]) == 0: raise ValueError("Proposals must be a non-empty list") # Validate each proposal for proposal in nextsteps_data["proposals"]: required_proposal_fields = ["id", "title", "action"] for field in required_proposal_fields: if field not in proposal: raise ValueError(f"Proposal missing required field: {field}") if proposal["action"]["type"] != "send_message": raise ValueError("Only 'send_message' action type is currently supported") if "content" not in proposal["action"]: raise ValueError("Action must include content field") # Validate optional fields if "priority" in nextsteps_data: priority = nextsteps_data["priority"] if not isinstance(priority, (int, float)) or priority < 0 or priority > 100: raise ValueError("Priority must be a number between 0 and 100") if "countdown" in nextsteps_data: countdown = nextsteps_data["countdown"] if not isinstance(countdown, (int, float)) or countdown < 1 or countdown > 60: raise ValueError("Countdown must be a number between 1 and 60 seconds") return True
- Use auto-execution sparingly - Only for obvious next steps or error recovery
- Keep countdown times reasonable - 3-10 seconds for most use cases
- Provide clear, descriptive titles - Users should understand what will happen
- Include helpful descriptions - Additional context in tooltips
- Validate your data structure - Use the validation helper above
- Test both execution modes - Verify auto and manual modes work correctly
- Consider user context - Tailor proposals to the current workflow state
- Handle errors gracefully - Malformed NextSteps should not break the client
- Use appropriate priorities - Guide user attention to most important actions
- Combine with other outputs - NextSteps work well with HTML and text results