Nash MCP Server
by nash-app
import json
import os
import subprocess
import sys
import traceback
import logging
import signal
import atexit
from datetime import datetime
# Store active subprocesses
active_python_processes = []
# Cleanup function
def cleanup_python_subprocesses():
for proc in active_python_processes:
try:
if proc.poll() is None: # If process is still running
os.killpg(os.getpgid(proc.pid), signal.SIGTERM)
except Exception:
pass
# Register cleanup on exit
atexit.register(cleanup_python_subprocesses)
# For signal handling
def python_signal_handler(sig, frame):
cleanup_python_subprocesses()
sys.exit(0)
# Register signal handlers
signal.signal(signal.SIGINT, python_signal_handler)
signal.signal(signal.SIGTERM, python_signal_handler)
from nash_mcp.constants import MAC_SECRETS_PATH, NASH_SESSION_DIR
def list_session_files() -> str:
"""
List all Python files in the current session directory.
This function is essential to check what files already exist before creating new ones.
ALWAYS use this function before creating a new file to avoid duplicating functionality.
USE CASES:
- Before creating a new file to check if something similar already exists
- When starting work on a new task to understand available resources
- To discover relevant code that could be modified instead of rewritten
- When fixing errors to find the file that needs editing
EXAMPLES:
```python
# List all existing Python files in the session
list_session_files()
# After seeing available files, check content of a specific file
get_file_content("data_processor.py")
```
WORKFLOW:
1. ALWAYS start by listing available files with list_session_files()
2. Check content of relevant files with get_file_content()
3. Edit existing files with edit_python_file() instead of creating new ones
4. Only create new files for entirely new functionality
Returns:
A formatted list of Python files in the current session directory
"""
try:
# Ensure session directory exists
if not NASH_SESSION_DIR.exists():
return "No session directory found."
# Find all Python files in the session directory
py_files = list(NASH_SESSION_DIR.glob("*.py"))
if not py_files:
return "No Python files found in the current session."
# Sort files by modification time (newest first)
py_files.sort(key=lambda x: x.stat().st_mtime, reverse=True)
# Format output
result = "Python files in current session:\n\n"
for file_path in py_files:
mod_time = datetime.fromtimestamp(file_path.stat().st_mtime).strftime("%Y-%m-%d %H:%M:%S")
result += f"- {file_path.name} (Modified: {mod_time})\n"
result += "\nTo view file content: get_file_content(\"filename.py\")"
return result
except Exception as e:
logging.error(f"Error listing session files: {str(e)}")
return f"Error listing files: {str(e)}"
def get_file_content(file_name: str) -> str:
"""
Retrieve the contents of a Python file from the session directory.
This function reads a file from the current session directory and returns its
contents. This is essential for viewing the current state of a file before
making edits with edit_python_file().
USE CASES:
- Before making edits to an existing file
- When checking the current implementation of a script
- To understand the structure of a previously saved script
- For reviewing code to identify parts that need modification
EXAMPLES:
```python
# View a file named "data_analysis.py"
get_file_content("data_analysis.py")
# View a file without .py extension (extension will be added automatically)
get_file_content("data_analysis")
```
WORKFLOW:
1. Use get_file_content() to check if a file exists and view its current content
2. Identify the exact content you want to modify
3. Use edit_python_file() to make targeted changes by replacing specific content
4. Use execute_python() with an empty code string to run the modified file
Args:
file_name: The name of the file to read from the session directory
Returns:
The file contents as a string, or an error message if the file doesn't exist
"""
try:
# Ensure file has .py extension
if not file_name.endswith('.py'):
file_name = f"{file_name}.py"
file_path = NASH_SESSION_DIR / file_name
if not file_path.exists():
return f"Error: File '{file_name}' not found in the current session."
with open(file_path, 'r') as f:
content = f.read()
return content
except Exception as e:
logging.error(f"Error reading file '{file_name}': {str(e)}")
return f"Error reading file: {str(e)}"
def edit_python_file(file_name: str, old_content: str, new_content: str) -> str:
"""
Edit a Python file by replacing specific content with new content.
ALWAYS PRIORITIZE EDITING EXISTING FILES RATHER THAN CREATING NEW ONES WHEN MAKING CHANGES.
This should be your first choice whenever modifying existing code - even for seemingly significant changes.
This function uses exact string matching to find and replace code snippets,
similar to how Claude edits files. This approach is more reliable for complex
changes and matches how LLMs naturally think about editing text.
USE CASES:
- Fix bugs or errors in existing code
- Refactor code to improve readability or maintainability
- Add new features to an existing script
- Update variable names, function signatures, or other identifiers
- Replace entire blocks of code with improved implementations
- Change algorithm implementations or logic flows
- Modify large portions of files (you can replace almost the entire content if needed)
ADVANTAGES:
- Uses exact pattern matching, similar to how Claude handles edits
- Avoids problems with line numbers shifting during edits
- Can replace multi-line content with precise context
- More reliable for complex edits than line-based approaches
- Preserves script history and context
WHEN TO EDIT vs. CREATE NEW:
EDIT when (almost always):
- Making any modification to existing functionality
- Fixing bugs or issues in existing code
- Adding new functions or classes to existing modules
- Changing logic, algorithms, or implementations
- Adjusting parameters or configuration values
- Updating imports or dependencies
- Improving error handling or adding validation
- Enhancing existing features in any way
- Refactoring or restructuring code
- Even for major changes that affect large portions of the file
CREATE NEW only when:
- Creating a completely separate utility with an entirely different purpose
- Explicitly asked by the user to create a new standalone file
- Testing isolated functionality that shouldn't affect existing code
- The existing file is explicitly described as a template or example
EXAMPLES:
```python
# Fix a calculation by replacing the specific function
edit_python_file(
"data_analysis.py",
"def calculate_average(values):\n return sum(values) / len(values)",
"def calculate_average(values):\n return np.mean(values) # Using numpy for better handling of edge cases"
)
# Fix a bug by replacing a specific line with its surrounding context
edit_python_file(
"processor.py",
" data = load_data()\n result = process(data)\n save_results(data) # Bug: saving wrong data",
" data = load_data()\n result = process(data)\n save_results(result) # Fixed: save processed results"
)
# Add a new import statement
edit_python_file(
"api_client.py",
"import requests\nimport json",
"import requests\nimport json\nimport logging"
)
# Adding error handling to a function (NOTICE INDENTATION IS PRESERVED)
edit_python_file(
"fetch_data.py",
"def fetch_user_data(user_id):\n url = f\"https://api.example.com/users/{user_id}\"\n response = requests.get(url)\n response.raise_for_status()\n return response.json()",
"def fetch_user_data(user_id):\n url = f\"https://api.example.com/users/{user_id}\"\n try:\n response = requests.get(url)\n response.raise_for_status()\n return response.json()\n except requests.RequestException as e:\n logging.error(f\"Failed to fetch user data: {e}\")\n return None"
)
# Major change: Replace an entire function with a completely new implementation
edit_python_file(
"processor.py",
"def process_data(data):\n # Old inefficient implementation\n result = []\n for item in data:\n if item['value'] > 0:\n result.append(item['value'] * 2)\n return result",
"def process_data(data):\n # New vectorized implementation\n import pandas as pd\n df = pd.DataFrame(data)\n return df[df['value'] > 0]['value'] * 2"
)
# Even major changes that add multiple functions should use edit_python_file
edit_python_file(
"utils.py",
"# Utility functions for data processing",
"# Utility functions for data processing\n\ndef validate_input(data):\n \"\"\"Validate input data format.\"\"\"\n if not isinstance(data, list):\n raise TypeError(\"Data must be a list\")\n return True\n\ndef normalize_data(data):\n \"\"\"Normalize values to 0-1 range.\"\"\"\n min_val = min(data)\n max_val = max(data)\n return [(x - min_val) / (max_val - min_val) for x in data]"
)
```
WORKFLOW:
1. Always check if a relevant file already exists with get_file_content()
2. When modifying any existing file, use edit_python_file()
3. Identify the exact content to replace (including enough context)
4. Create the new replacement content
5. Apply the change with edit_python_file()
6. Use execute_python() to run the modified file
7. Only create new files when specifically creating a new utility
BEST PRACTICES:
- Include sufficient context around the text to be replaced (3-5 lines before and after)
- For major rewrites, you can replace large chunks of the file or even nearly all content
- Ensure the old_content exactly matches text in the file, including spacing and indentation
- Make focused, targeted changes rather than multiple changes at once
- When a user asks to "fix", "update", "modify", or "change" something, they typically want edits to existing files
INDENTATION GUIDELINES (CRITICAL FOR PYTHON):
- Always preserve correct indentation in both old_content and new_content
- When adding control structures (if/else, try/except, loops), replace the entire block
- Never try to insert just the opening part of a control structure without its closing part
- For adding error handling, replace the entire function or block, not just parts of it
- Watch for common indentation errors, especially with nested structures
- When debugging indentation issues, view the entire file first with get_file_content()
- For complex control flow changes, prefer replacing larger blocks to ensure consistency
PATTERN RECOGNITION:
- When a user asks to "fix", "update", "modify", or "change" something, they typically want edits to existing files
- Use list_session_files() and get_file_content() first to check what files already exist
- Only create new files when the user explicitly requests a completely new utility
SAFETY FEATURES:
- Creates a backup of the original file (.py.bak extension)
- Returns a diff of changes made
- Will only replace exact matches, preventing unintended changes
Args:
file_name: The name of the file to edit in the session directory
old_content: The exact content to replace (must match exactly, including whitespace)
new_content: The new content to insert as a replacement
Returns:
Success message with diff of changes, or error message if the operation fails
"""
try:
# Ensure file has .py extension
if not file_name.endswith('.py'):
file_name = f"{file_name}.py"
file_path = NASH_SESSION_DIR / file_name
if not file_path.exists():
return f"Error: File '{file_name}' not found in the current session."
# Read the original file
with open(file_path, 'r') as f:
content = f.read()
# Check if the old content exists in the file
if old_content not in content:
return f"Error: The specified content was not found in '{file_name}'. Please check that the content matches exactly, including whitespace and indentation."
# Create a backup of the original file
backup_path = file_path.with_suffix('.py.bak')
with open(backup_path, 'w') as f:
f.write(content)
# Replace the content
new_file_content = content.replace(old_content, new_content)
# Write the modified content back to the file
with open(file_path, 'w') as f:
f.write(new_file_content)
# Generate a unified diff for the changes
from difflib import unified_diff
old_lines = content.splitlines()
new_lines = new_file_content.splitlines()
diff = list(unified_diff(
old_lines,
new_lines,
fromfile=f"{file_name} (original)",
tofile=f"{file_name} (modified)",
lineterm='',
n=3 # Context lines
))
if diff:
diff_result = '\n'.join(diff)
else:
diff_result = "No changes detected."
return f"Successfully edited '{file_name}'.\n\nChanges:\n{diff_result}"
except Exception as e:
logging.error(f"Error editing file '{file_name}': {str(e)}")
logging.error(traceback.format_exc())
return f"Error editing file: {str(e)}"
def execute_python(code: str, file_name: str) -> str:
"""Execute arbitrary Python code and return the result.
⚠️ MANDATORY PRE-CODING CHECKLIST - COMPLETE BEFORE WRITING ANY CODE: ⚠️
STOP! Before writing or executing ANY code, have you completed these REQUIRED checks?
1. Check available packages: list_installed_packages()
- Know what libraries you can use
- Avoid importing unavailable packages
2. Check available secrets: nash_secrets()
- See what API keys and credentials are available
- Don't write code requiring credentials you don't have
3. Check existing files: list_session_files()
- See what code already exists
- Avoid duplicating existing functionality
4. Review relevant file contents: get_file_content("filename.py")
- Understand existing implementations
- Decide whether to edit or create new
These steps are MANDATORY. Skipping them is the #1 cause of inefficient code development.
AFTER completing the checklist, consider output efficiency:
- If a similar file exists with MINOR or MODERATE changes needed:
- Use edit_python_file() to make targeted changes
- This is usually more efficient for small to medium changes
- When it's MORE EFFICIENT to create a new file:
- If changes would require replacing almost the entire file
- If explaining the edits would require more tokens than a new file
- If creating a brand new file results in a cleaner, smaller response
Remember: The goal is to minimize token usage while maintaining context.
Choose the approach that results in the smallest, most efficient output.
This function executes standard Python code with access to imported modules and packages.
The code is saved to a named file in the session directory and executed in a subprocess
using the same Python interpreter. All available secrets are automatically loaded as
environment variables.
If you provide an empty code string and the file already exists, it will execute the
existing file without overwriting it. This is useful for running previously edited files.
USAGE GUIDE:
1. Use list_installed_packages() first to check available packages
2. Provide a descriptive file_name that reflects the purpose of your code
3. Write standard Python code including imports, functions, and print statements
4. All output from print() statements will be returned to the conversation
5. Always include proper error handling with try/except blocks
6. Clean up any resources (files, connections) your code creates
FILE NAMING:
- Always provide a descriptive filename that reflects what the code does
- Examples: "data_analysis.py", "api_fetch.py", "report_generation.py"
- Files are saved in the session directory for later reference or modification
EXAMPLES:
```python
# Basic operations
import os
print(f"Current directory: {os.getcwd()}")
# Using subprocess
import subprocess
result = subprocess.run(["ls", "-la"], capture_output=True, text=True)
print(result.stdout)
# Data processing
import json
data = {"name": "John", "age": 30}
print(json.dumps(data, indent=2))
# Using temporary files
import tempfile
import os
with tempfile.NamedTemporaryFile(suffix='.txt', delete=False) as tmp:
tmp_path = tmp.name
tmp.write("Hello World".encode('utf-8'))
try:
with open(tmp_path, 'r') as f:
print(f.read())
finally:
os.unlink(tmp_path) # Always clean up
```
SECRET MANAGEMENT:
- Use nash_secrets() to see available API keys/credentials
- Access secrets in your code with: os.environ.get('SECRET_NAME')
- Secrets are loaded from: ~/Library/Application Support/Nash/secrets.json
IMPLEMENTATION DETAILS:
- Code is saved to a file in the session directory with the provided name
- File is executed as a subprocess with the system Python interpreter
- Files are preserved in the session directory for reference and reuse
- All secrets are passed as environment variables to the subprocess
- All stdout output is captured and returned
WEB AUTOMATION NOTE:
For interactive web automation, browser-based scraping of dynamic sites,
or any tasks requiring browser interactions (clicking, form filling, etc.),
use the operate_browser tool instead of writing custom automation code.
SECURITY CONSIDERATIONS:
- Never write code that could harm the user's system
- Avoid creating persistent files; use tempfile module when needed
- Don't leak or expose secret values in output
- Avoid making unauthorized network requests
- Do not attempt to bypass system security controls
- When scraping websites, respect robots.txt and rate limits
BEST PRACTICES:
- Keep Python code focused on data retrieval, computation, and transformations
- Write minimal code that extracts and formats data, letting the LLM analyze the results
- Avoid embedding complex analysis logic in Python when the LLM can do it better
- Return clean, structured data that the LLM can easily interpret
- For static web content, use fetch_webpage; for dynamic sites or interactive features, use operate_browser
Args:
code: Python code to execute (multi-line string)
file_name: Descriptive name for the Python file (will be saved in session directory)
Returns:
Captured stdout from code execution or detailed error message
"""
# Log the full code being executed
logging.info(f"Executing Python code in file '{file_name}':\n{code}")
try:
# Load secrets as environment variables
env_vars = os.environ.copy()
if MAC_SECRETS_PATH.exists():
try:
with open(MAC_SECRETS_PATH, 'r') as f:
secrets = json.load(f)
# Add secrets to environment variables for subprocess
for secret in secrets:
if 'key' in secret and 'value' in secret:
env_vars[secret['key']] = secret['value']
logging.info("Loaded secrets for Python execution")
except Exception as e:
# Log the error but continue execution
logging.warning(f"Error loading secrets: {str(e)}")
print(f"Warning: Error loading secrets: {str(e)}")
else:
logging.info("No secrets file found")
# Ensure file name has .py extension
if not file_name.endswith('.py'):
file_name = f"{file_name}.py"
# Create the full file path in the session directory
file_path = NASH_SESSION_DIR / file_name
# If code is empty and file exists, use existing file
if not code and file_path.exists():
logging.info(f"Using existing file: {file_path}")
else:
# Write the code to the file
with open(file_path, 'w') as f:
f.write(code)
logging.info(f"Saved Python code to: {file_path}")
try:
# Execute the file using the same Python interpreter
logging.info(f"Running Python code from: {file_path}")
try:
proc = subprocess.Popen(
[sys.executable, str(file_path)],
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
text=True,
env=env_vars,
preexec_fn=os.setpgrp # Make this process the group leader
)
# Add to active processes list for cleanup
active_python_processes.append(proc)
try:
stdout, stderr = proc.communicate()
result = subprocess.CompletedProcess(
[sys.executable, str(file_path)],
proc.returncode,
stdout,
stderr
)
finally:
# Remove from active processes list
if proc in active_python_processes:
active_python_processes.remove(proc)
# Return stdout if successful, or stderr if there was an error
if result.returncode == 0:
logging.info(f"Python code in {file_name} executed successfully")
return result.stdout if result.stdout else f"Code in {file_name} executed successfully (no output)"
else:
logging.warning(f"Python code execution of {file_name} failed with return code {result.returncode}")
# Save error information to companion file
try:
error_file = file_path.with_suffix('.error')
with open(error_file, 'w') as f:
f.write(result.stderr)
logging.info(f"Saved error output to: {error_file}")
except Exception as write_err:
logging.error(f"Failed to write error file: {str(write_err)}")
return f"Error in {file_name} (return code {result.returncode}):\n{result.stderr}"
except subprocess.SubprocessError as sub_err:
# Specifically handle subprocess errors
error_msg = f"Subprocess error executing {file_name}: {str(sub_err)}"
logging.error(error_msg)
try:
exception_file = file_path.with_suffix('.exception')
with open(exception_file, 'w') as f:
f.write(f"SubprocessError: {str(sub_err)}\nTraceback: {traceback.format_exc()}")
logging.info(f"Saved exception information to: {exception_file}")
except Exception:
logging.error("Failed to write exception file")
return error_msg
except Exception as exec_err:
# Handle other exceptions during execution
error_msg = f"Error executing {file_name}: {str(exec_err)}"
logging.error(error_msg)
logging.error(traceback.format_exc())
try:
exception_file = file_path.with_suffix('.exception')
with open(exception_file, 'w') as f:
f.write(f"Error: {str(exec_err)}\nTraceback: {traceback.format_exc()}")
logging.info(f"Saved exception information to: {exception_file}")
except Exception:
logging.error("Failed to write exception file")
return error_msg
except Exception as e:
# Catch-all for any other exceptions in the outer try block
logging.error(f"Unexpected error in execute_python for {file_name}: {str(e)}")
logging.error(traceback.format_exc())
try:
if 'file_path' in locals():
exception_file = file_path.with_suffix('.exception')
with open(exception_file, 'w') as f:
f.write(f"Unexpected error: {str(e)}\nTraceback: {traceback.format_exc()}")
logging.info(f"Saved exception information to: {exception_file}")
except Exception:
logging.error("Failed to write exception file")
# Return error message instead of raising to prevent MCP server crash
return f"Unexpected error in {file_name}: {str(e)}\nSee logs for details."
except Exception as e:
logging.error(f"Python execution error: {str(e)}")
logging.error(traceback.format_exc())
return f"Error in {file_name}: {str(e)}\nTraceback: {traceback.format_exc()}\n\n"