"""
TTG Scratchpad MCP Server
Production-ready implementation with MongoDB persistence and user isolation.
"""
import os
import json
import logging
from typing import Optional, Dict, Any, List
from datetime import datetime
# Load environment variables from .env file FIRST
from dotenv import load_dotenv
load_dotenv()
from fastmcp import FastMCP
from fastmcp.server.context import Context
from database import (
database_lifespan,
create_workspace,
get_active_workspace,
get_workspace_by_id,
update_workspace,
complete_workspace,
get_user_workspaces,
create_or_update_file,
get_file,
list_workspace_files,
delete_file,
log_activity,
get_recent_activities,
format_file_for_response,
format_activity_for_response,
format_workspace_for_list,
clear_file_updated_flags,
WORKSPACE_TTL_DAYS,
)
from auth import get_current_user_id, get_current_user_id_optional
from models import (
validate_task_input,
validate_update_input,
validate_complete_input,
validate_file_path,
validate_write_file_input,
MAX_TASK_LENGTH,
MAX_FILE_CONTENT_SIZE,
)
# Configure logging
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)
# ============================================================
# SERVER SETUP WITH INSTRUCTIONS
# ============================================================
SERVER_INSTRUCTIONS = """
## Scratchpad MCP Server Instructions
Use the scratchpad tools to manage working memory during complex, multi-step tasks.
### When to Use Scratchpad
Use scratchpad tools when:
- Working on tasks with 3+ steps that need tracking
- Analyzing multiple tickers or data points
- Building watchlists, journals, or notes
- Tasks that may be interrupted and resumed
- Storing intermediate results that exceed context limits
- Complex research requiring organized notes
Do NOT use scratchpad for:
- Simple, single-step queries
- Information that should go to user memory (preferences, facts about user)
- Temporary calculations you can do inline
### Tool Usage Guide
1. **scratchpad_start** - Begin a new task
- Call at the start of multi-step work
- Provide clear task description
- Creates a new workspace for tracking
2. **scratchpad_update** - Track progress
- Update after completing each step
- Include current step number and total steps
- Describe what you're currently working on
- User sees real-time progress in the widget
3. **scratchpad_write_file** - Save data
- Store analysis results, notes, watchlists
- Use descriptive file names (e.g., "AAPL_analysis.md")
- Organize in folders for complex tasks
- Content persists across sessions
4. **scratchpad_read_file** - Retrieve saved data
- Read previously saved files
- Use to resume interrupted work
- Access saved analysis or notes
5. **scratchpad_list_files** - See workspace contents
- View all files in current workspace
- Helpful for orienting after resuming work
6. **scratchpad_complete** - Finish task
- Call when task is done
- Provide summary of accomplishments
- Workspace persists for 7 days (user can review)
7. **scratchpad_list_workspaces** - View workspace history
- Lists all workspaces (active and completed)
- Shows task summary, file count, TTL countdown
- Useful for finding previous work
8. **scratchpad_open_workspace** - Resume previous workspace
- Switch to a specific workspace by ID
- Access files and continue work
### Best Practices
- Always start with scratchpad_start for complex tasks
- Update progress regularly so user sees activity
- Use files to store data that exceeds context limits
- Complete tasks properly for clean state
- Use meaningful file names and organize with folders
- When resuming, check scratchpad_list_files first
### Example Workflow
User: "Analyze these 10 tickers and tell me which is best"
1. scratchpad_start("Analyzing 10 tickers for best opportunity")
2. For each ticker:
- scratchpad_update(step=N, working_on="Analyzing AAPL...")
- Get data, analyze
- scratchpad_write_file("analysis/AAPL.md", analysis_content)
3. scratchpad_write_file("summary.md", final_comparison)
4. scratchpad_complete("Analyzed 10 tickers, NVDA shows best setup")
"""
# Check if MongoDB is configured
MONGODB_URI = os.environ.get("MONGODB_URI")
USE_DATABASE = bool(MONGODB_URI)
if USE_DATABASE:
logger.info("MongoDB configured - using persistent storage")
mcp = FastMCP(
"scratchpad",
instructions=SERVER_INSTRUCTIONS,
lifespan=database_lifespan
)
else:
logger.warning("MONGODB_URI not set - running in stateless demo mode")
mcp = FastMCP(
"scratchpad",
instructions=SERVER_INSTRUCTIONS
)
# ============================================================
# HELPER FUNCTIONS
# ============================================================
def scratchpad_response(
task: str,
status: str,
files: List[Dict[str, Any]],
activity: List[Dict[str, Any]],
progress: Optional[Dict[str, int]] = None,
working_on: Optional[str] = None
) -> str:
"""Build standardized scratchpad response for widget."""
data = {
"task": task,
"status": status,
"files": files,
"activity": activity
}
if progress:
data["progress"] = progress
if working_on:
data["workingOn"] = working_on
return json.dumps(data)
def error_response(message: str) -> str:
"""Build error response."""
return json.dumps({
"error": message,
"status": "error",
"task": "Error",
"files": [],
"activity": [{"action": message, "icon": "delete"}]
})
# ============================================================
# SCRATCHPAD TOOLS - WORKSPACE MANAGEMENT
# ============================================================
@mcp.tool()
async def scratchpad_demo() -> str:
"""
Demo the scratchpad widget with sample data.
Use this to test the scratchpad UI rendering.
"""
return scratchpad_response(
task="Demo: Analyzing Market Data",
status="active",
progress={"current": 7, "total": 10},
working_on="Processing AAPL technical indicators...",
files=[
{"name": "watchlist.json", "type": "file", "updated": True},
{"name": "analysis/", "type": "folder", "children": 3},
{"name": "notes.md", "type": "file"},
{"name": "backtest_results.csv", "type": "file", "updated": True}
],
activity=[
{"action": "Fetched AAPL snapshot", "icon": "complete"},
{"action": "Calculated RSI indicators", "icon": "complete"},
{"action": "Writing analysis notes", "icon": "write"},
{"action": "Updated watchlist", "icon": "write"},
{"action": "Running backtest simulation", "icon": "read"}
]
)
@mcp.tool()
async def scratchpad_start(task_description: str, ctx: Context) -> str:
"""
Start a new scratchpad task/workspace.
Args:
task_description: Description of the task to begin (e.g., "Analyzing 10 tickers")
Returns:
JSON response with initialized workspace state
"""
# Validate and sanitize input
try:
validated = validate_task_input(task_description)
task_description = validated.task_description
except Exception as e:
return error_response(f"Invalid input: {str(e)}")
if not USE_DATABASE:
return scratchpad_response(
task=task_description,
status="active",
progress={"current": 0, "total": 1},
working_on="Initializing workspace...",
files=[{"name": "workspace/", "type": "folder", "children": 0}],
activity=[{"action": f"Started task: {task_description[:50]}", "icon": "write"}]
)
try:
user_id = get_current_user_id()
db = ctx.request_context.lifespan_context["db"]
# Create workspace in database
workspace = await create_workspace(db, user_id, task_description)
workspace_id = str(workspace["_id"])
# Log activity
await log_activity(db, user_id, workspace_id, f"Started task: {task_description[:50]}", "write")
logger.info(f"Created workspace {workspace_id} for user {user_id}")
return scratchpad_response(
task=task_description,
status="active",
progress={"current": 0, "total": 1},
working_on="Initializing workspace...",
files=[{"name": "workspace/", "type": "folder", "children": 0}],
activity=[{"action": f"Started task: {task_description[:50]}", "icon": "write"}]
)
except ValueError as e:
return error_response(str(e))
except Exception as e:
logger.error(f"scratchpad_start error: {e}")
return error_response(f"Failed to start workspace: {str(e)}")
@mcp.tool()
async def scratchpad_update(
task_description: str,
current_step: int,
total_steps: int,
working_on: str,
ctx: Context,
files: Optional[str] = None,
activity_message: Optional[str] = None
) -> str:
"""
Update scratchpad progress during a task.
Args:
task_description: Current task description
current_step: Current step number (e.g., 3)
total_steps: Total number of steps (e.g., 10)
working_on: What is currently being processed (e.g., "Analyzing AAPL...")
files: Optional JSON string of files array
activity_message: Optional new activity to log
Returns:
JSON response with updated progress state
"""
# Validate and sanitize input
try:
validated = validate_update_input(
task_description=task_description,
current_step=current_step,
total_steps=total_steps,
working_on=working_on,
files=files,
activity_message=activity_message
)
task_description = validated.task_description
current_step = validated.current_step
total_steps = validated.total_steps
working_on = validated.working_on
except Exception as e:
return error_response(f"Invalid input: {str(e)}")
# Parse files if provided
file_list = []
if files:
try:
file_list = json.loads(files)
except json.JSONDecodeError:
file_list = [{"name": "workspace/", "type": "folder", "children": current_step}]
else:
file_list = [{"name": "workspace/", "type": "folder", "children": current_step}]
# Build activity
activity = []
if activity_message:
activity.append({"action": activity_message, "icon": "write"})
activity.append({"action": f"Step {current_step}/{total_steps}", "icon": "read"})
if not USE_DATABASE:
return scratchpad_response(
task=task_description,
status="active",
progress={"current": current_step, "total": total_steps},
working_on=working_on,
files=file_list,
activity=activity
)
try:
user_id = get_current_user_id()
db = ctx.request_context.lifespan_context["db"]
# Get or create active workspace
workspace = await get_active_workspace(db, user_id)
if not workspace:
workspace = await create_workspace(db, user_id, task_description)
workspace_id = str(workspace["_id"])
# Update workspace
await update_workspace(db, user_id, workspace_id, {
"task": task_description,
"progress": {"current": current_step, "total": total_steps},
"working_on": working_on
})
# Log activity
if activity_message:
await log_activity(db, user_id, workspace_id, activity_message, "write")
await log_activity(db, user_id, workspace_id, f"Step {current_step}/{total_steps}", "read")
# Get files from database
db_files = await list_workspace_files(db, user_id, workspace_id)
if db_files:
file_list = [format_file_for_response(f) for f in db_files]
# Get recent activities
db_activities = await get_recent_activities(db, user_id, workspace_id, 5)
if db_activities:
activity = [format_activity_for_response(a) for a in db_activities]
return scratchpad_response(
task=task_description,
status="active",
progress={"current": current_step, "total": total_steps},
working_on=working_on,
files=file_list,
activity=activity
)
except ValueError as e:
return error_response(str(e))
except Exception as e:
logger.error(f"scratchpad_update error: {e}")
return error_response(f"Failed to update workspace: {str(e)}")
@mcp.tool()
async def scratchpad_complete(
task_description: str,
summary: str,
ctx: Context,
files: Optional[str] = None
) -> str:
"""
Mark a scratchpad task as complete.
Args:
task_description: The completed task description
summary: Summary of what was accomplished
files: Optional JSON string of final files array
Returns:
JSON response with completion state
"""
# Validate and sanitize input
try:
validated = validate_complete_input(
task_description=task_description,
summary=summary,
files=files
)
task_description = validated.task_description
summary = validated.summary
except Exception as e:
return error_response(f"Invalid input: {str(e)}")
# Parse files if provided
file_list = []
if files:
try:
file_list = json.loads(files)
except json.JSONDecodeError:
file_list = [{"name": "results/", "type": "folder", "children": 1}]
else:
file_list = [
{"name": "results/", "type": "folder", "children": 1},
{"name": "summary.md", "type": "file", "updated": True}
]
if not USE_DATABASE:
return scratchpad_response(
task=task_description,
status="complete",
files=file_list,
activity=[
{"action": summary[:100], "icon": "complete"},
{"action": "Task completed successfully", "icon": "complete"}
]
)
try:
user_id = get_current_user_id()
db = ctx.request_context.lifespan_context["db"]
# Get active workspace
workspace = await get_active_workspace(db, user_id)
if not workspace:
return error_response("No active workspace to complete")
workspace_id = str(workspace["_id"])
# Mark complete (cleanup=False - let TTL handle deletion after 7 days)
await complete_workspace(db, user_id, workspace_id, cleanup=False)
# Log completion
await log_activity(db, user_id, workspace_id, summary[:100], "complete")
await log_activity(db, user_id, workspace_id, "Task completed successfully", "complete")
# Get final files
db_files = await list_workspace_files(db, user_id, workspace_id)
if db_files:
file_list = [format_file_for_response(f) for f in db_files]
# Get activities
db_activities = await get_recent_activities(db, user_id, workspace_id, 5)
activity = [format_activity_for_response(a) for a in db_activities] if db_activities else [
{"action": summary[:100], "icon": "complete"},
{"action": "Task completed successfully", "icon": "complete"}
]
logger.info(f"Completed workspace {workspace_id} for user {user_id}")
return scratchpad_response(
task=task_description,
status="complete",
files=file_list,
activity=activity
)
except ValueError as e:
return error_response(str(e))
except Exception as e:
logger.error(f"scratchpad_complete error: {e}")
return error_response(f"Failed to complete workspace: {str(e)}")
# ============================================================
# SCRATCHPAD TOOLS - WORKSPACE LISTING & NAVIGATION
# ============================================================
@mcp.tool()
async def scratchpad_list_workspaces(
ctx: Context,
include_completed: bool = True,
limit: int = 20
) -> str:
"""
List all workspaces for the current user with TTL countdown.
Args:
include_completed: Include completed workspaces (default: True)
limit: Maximum number of workspaces to return (default: 20)
Returns:
JSON with list of workspaces including:
- id: Workspace ID (use with scratchpad_open_workspace)
- task: Task description summary
- status: "active" or "complete"
- files_count: Number of files in workspace
- created_at: When workspace was created
- updated_at: Last update time
- ttl_remaining_seconds: Seconds until auto-delete (completed only)
- ttl_display: Human-readable TTL (e.g., "5d 12h remaining")
"""
if not USE_DATABASE:
return json.dumps({
"workspaces": [],
"message": "Database not configured - workspace history unavailable in demo mode",
"ttl_days": WORKSPACE_TTL_DAYS
})
try:
user_id = get_current_user_id()
db = ctx.request_context.lifespan_context["db"]
# Get workspaces
workspaces = await get_user_workspaces(
db, user_id,
limit=min(limit, 50), # Cap at 50
include_completed=include_completed
)
# Format for response
formatted = [format_workspace_for_list(ws) for ws in workspaces]
# Count active vs completed
active_count = sum(1 for ws in formatted if ws["status"] == "active")
completed_count = sum(1 for ws in formatted if ws["status"] == "complete")
return json.dumps({
"workspaces": formatted,
"summary": {
"total": len(formatted),
"active": active_count,
"completed": completed_count
},
"ttl_days": WORKSPACE_TTL_DAYS,
"message": f"Found {len(formatted)} workspace(s). Completed workspaces auto-delete after {WORKSPACE_TTL_DAYS} days."
})
except ValueError as e:
return json.dumps({"error": str(e), "workspaces": []})
except Exception as e:
logger.error(f"scratchpad_list_workspaces error: {e}")
return json.dumps({"error": f"Failed to list workspaces: {str(e)}", "workspaces": []})
@mcp.tool()
async def scratchpad_open_workspace(
workspace_id: str,
ctx: Context
) -> str:
"""
Open/switch to a specific workspace by ID.
Use scratchpad_list_workspaces to find workspace IDs.
Args:
workspace_id: The workspace ID to open
Returns:
JSON response with workspace state (files, activity, etc.)
"""
if not USE_DATABASE:
return error_response("Database not configured - cannot open workspaces in demo mode")
try:
user_id = get_current_user_id()
db = ctx.request_context.lifespan_context["db"]
# Get workspace
workspace = await get_workspace_by_id(db, user_id, workspace_id)
if not workspace:
return error_response(f"Workspace not found: {workspace_id}")
workspace_id_str = str(workspace["_id"])
# Get files
db_files = await list_workspace_files(db, user_id, workspace_id_str)
file_list = [format_file_for_response(f) for f in db_files]
# Get activities
db_activities = await get_recent_activities(db, user_id, workspace_id_str, 10)
activity = [format_activity_for_response(a) for a in db_activities]
# Log that we opened it
await log_activity(db, user_id, workspace_id_str, "Opened workspace", "read")
return scratchpad_response(
task=workspace.get("task", "Workspace"),
status=workspace.get("status", "active"),
progress=workspace.get("progress"),
working_on=workspace.get("working_on"),
files=file_list,
activity=activity
)
except ValueError as e:
return error_response(str(e))
except Exception as e:
logger.error(f"scratchpad_open_workspace error: {e}")
return error_response(f"Failed to open workspace: {str(e)}")
# ============================================================
# SCRATCHPAD TOOLS - FILE MANAGEMENT
# ============================================================
@mcp.tool()
async def scratchpad_write_file(
path: str,
content: str,
ctx: Context
) -> str:
"""
Write content to a file in the scratchpad workspace.
Args:
path: File path (e.g., "analysis/AAPL.md" or "watchlist.json")
content: File content to write
Returns:
JSON response with updated workspace state
"""
# Validate and sanitize input
try:
validated = validate_write_file_input(path, content)
path = validated.path
content = validated.content
except Exception as e:
return error_response(f"Invalid input: {str(e)}")
if not USE_DATABASE:
# Extract filename from path
name = path.split("/")[-1] if "/" in path else path
return scratchpad_response(
task="File Operation",
status="active",
files=[{"name": name, "type": "file", "path": path, "updated": True}],
activity=[{"action": f"Wrote file: {path}", "icon": "write"}]
)
try:
user_id = get_current_user_id()
db = ctx.request_context.lifespan_context["db"]
# Get active workspace
workspace = await get_active_workspace(db, user_id)
if not workspace:
return error_response("No active workspace. Call scratchpad_start first.")
workspace_id = str(workspace["_id"])
# Extract filename from path
name = path.split("/")[-1] if "/" in path else path
# Write file
await create_or_update_file(db, user_id, workspace_id, name, path, content)
# Log activity
await log_activity(db, user_id, workspace_id, f"Wrote file: {path}", "write")
# Get updated files list
db_files = await list_workspace_files(db, user_id, workspace_id)
file_list = [format_file_for_response(f) for f in db_files]
# Get activities
db_activities = await get_recent_activities(db, user_id, workspace_id, 5)
activity = [format_activity_for_response(a) for a in db_activities]
return scratchpad_response(
task=workspace.get("task", "File Operation"),
status="active",
progress=workspace.get("progress"),
working_on=workspace.get("working_on"),
files=file_list,
activity=activity
)
except ValueError as e:
return error_response(str(e))
except Exception as e:
logger.error(f"scratchpad_write_file error: {e}")
return error_response(f"Failed to write file: {str(e)}")
@mcp.tool()
async def scratchpad_read_file(
path: str,
ctx: Context
) -> str:
"""
Read content from a file in the scratchpad workspace.
Args:
path: File path to read (e.g., "analysis/AAPL.md")
Returns:
JSON with file content or error
"""
# Validate file path
try:
path = validate_file_path(path)
except Exception as e:
return json.dumps({"error": f"Invalid path: {str(e)}", "path": path})
if not USE_DATABASE:
return json.dumps({
"error": "Database not configured - file operations unavailable in demo mode",
"path": path
})
try:
user_id = get_current_user_id()
db = ctx.request_context.lifespan_context["db"]
# Get active workspace
workspace = await get_active_workspace(db, user_id)
if not workspace:
return json.dumps({"error": "No active workspace", "path": path})
workspace_id = str(workspace["_id"])
# Read file
file_doc = await get_file(db, user_id, workspace_id, path)
if not file_doc:
return json.dumps({"error": f"File not found: {path}", "path": path})
# Log activity
await log_activity(db, user_id, workspace_id, f"Read file: {path}", "read")
return json.dumps({
"path": path,
"name": file_doc.get("name", ""),
"content": file_doc.get("content", ""),
"updated_at": file_doc.get("updated_at", "").isoformat() if file_doc.get("updated_at") else None
})
except ValueError as e:
return json.dumps({"error": str(e), "path": path})
except Exception as e:
logger.error(f"scratchpad_read_file error: {e}")
return json.dumps({"error": f"Failed to read file: {str(e)}", "path": path})
@mcp.tool()
async def scratchpad_list_files(ctx: Context) -> str:
"""
List all files in the current scratchpad workspace.
Returns:
JSON response with file list and workspace state
"""
if not USE_DATABASE:
return scratchpad_response(
task="File Listing",
status="idle",
files=[],
activity=[{"action": "Listed files (demo mode - no files)", "icon": "read"}]
)
try:
user_id = get_current_user_id()
db = ctx.request_context.lifespan_context["db"]
# Get active workspace
workspace = await get_active_workspace(db, user_id)
if not workspace:
return scratchpad_response(
task="No Active Workspace",
status="idle",
files=[],
activity=[{"action": "No active workspace found", "icon": "read"}]
)
workspace_id = str(workspace["_id"])
# Get files
db_files = await list_workspace_files(db, user_id, workspace_id)
file_list = [format_file_for_response(f) for f in db_files]
# Log activity
await log_activity(db, user_id, workspace_id, f"Listed {len(file_list)} files", "read")
# Get activities
db_activities = await get_recent_activities(db, user_id, workspace_id, 5)
activity = [format_activity_for_response(a) for a in db_activities]
return scratchpad_response(
task=workspace.get("task", "Workspace"),
status=workspace.get("status", "active"),
progress=workspace.get("progress"),
working_on=workspace.get("working_on"),
files=file_list,
activity=activity
)
except ValueError as e:
return error_response(str(e))
except Exception as e:
logger.error(f"scratchpad_list_files error: {e}")
return error_response(f"Failed to list files: {str(e)}")
@mcp.tool()
async def scratchpad_delete_file(
path: str,
ctx: Context
) -> str:
"""
Delete a file from the scratchpad workspace.
Args:
path: File path to delete (e.g., "analysis/AAPL.md")
Returns:
JSON response with updated workspace state
"""
# Validate file path
try:
path = validate_file_path(path)
except Exception as e:
return error_response(f"Invalid path: {str(e)}")
if not USE_DATABASE:
return error_response("Database not configured - file operations unavailable in demo mode")
try:
user_id = get_current_user_id()
db = ctx.request_context.lifespan_context["db"]
# Get active workspace
workspace = await get_active_workspace(db, user_id)
if not workspace:
return error_response("No active workspace")
workspace_id = str(workspace["_id"])
# Delete file
deleted = await delete_file(db, user_id, workspace_id, path)
if not deleted:
return error_response(f"File not found: {path}")
# Log activity
await log_activity(db, user_id, workspace_id, f"Deleted file: {path}", "delete")
# Get updated files list
db_files = await list_workspace_files(db, user_id, workspace_id)
file_list = [format_file_for_response(f) for f in db_files]
# Get activities
db_activities = await get_recent_activities(db, user_id, workspace_id, 5)
activity = [format_activity_for_response(a) for a in db_activities]
return scratchpad_response(
task=workspace.get("task", "File Operation"),
status="active",
progress=workspace.get("progress"),
working_on=workspace.get("working_on"),
files=file_list,
activity=activity
)
except ValueError as e:
return error_response(str(e))
except Exception as e:
logger.error(f"scratchpad_delete_file error: {e}")
return error_response(f"Failed to delete file: {str(e)}")
# ============================================================
# RUN SERVER
# ============================================================
if __name__ == "__main__":
mcp.settings.host = os.environ.get("HOST", "0.0.0.0")
mcp.settings.port = int(os.environ.get("PORT", 8001))
print(f"Starting Scratchpad MCP server on {mcp.settings.host}:{mcp.settings.port}")
print(f"Database mode: {'PERSISTENT' if USE_DATABASE else 'DEMO (stateless)'}")
mcp.run(transport="streamable-http")
