server.py•6.12 kB
#!/usr/bin/env python3
"""Claude Continuity MCP Server - Project State Management"""
import os
import json
import logging
from pathlib import Path
from typing import Dict, Any, List
from datetime import datetime
from mcp.server import Server, NotificationOptions
from mcp.server.models import InitializationOptions
import mcp.server.stdio
import mcp.types as types
# Setup logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
# Configuration
STATE_DIR = Path.home() / "claude-continuity" / ".claude_states"
STATE_DIR.mkdir(parents=True, exist_ok=True)
server = Server("claude-continuity")
@server.list_tools()
async def handle_list_tools() -> List[types.Tool]:
"""List available tools"""
return [
types.Tool(
name="save_project_state",
description="Save the current project state",
inputSchema={
"type": "object",
"properties": {
"project_name": {"type": "string", "description": "Name of the project"},
"state": {"type": "object", "description": "Project state to save"},
"summary": {"type": "string", "description": "Summary of current progress"}
},
"required": ["project_name", "state"]
}
),
types.Tool(
name="load_project_state",
description="Load a saved project state",
inputSchema={
"type": "object",
"properties": {
"project_name": {"type": "string", "description": "Name of the project to load"}
},
"required": ["project_name"]
}
),
types.Tool(
name="list_active_projects",
description="List all active projects with saved states",
inputSchema={"type": "object", "properties": {}}
),
types.Tool(
name="get_project_summary",
description="Get a summary of a project's current state",
inputSchema={
"type": "object",
"properties": {
"project_name": {"type": "string", "description": "Name of the project"}
},
"required": ["project_name"]
}
)
]
@server.call_tool()
async def handle_call_tool(
name: str, arguments: Dict[str, Any]
) -> List[types.TextContent]:
"""Handle tool calls"""
if name == "save_project_state":
project_name = arguments["project_name"]
state = arguments["state"]
summary = arguments.get("summary", "")
project_dir = STATE_DIR / project_name
project_dir.mkdir(exist_ok=True)
state_file = project_dir / "state.json"
state_data = {
"state": state,
"summary": summary,
"updated_at": datetime.now().isoformat()
}
with open(state_file, 'w') as f:
json.dump(state_data, f, indent=2)
return [types.TextContent(
type="text",
text=f"✅ Saved state for project: {project_name}"
)]
elif name == "load_project_state":
project_name = arguments["project_name"]
state_file = STATE_DIR / project_name / "state.json"
if not state_file.exists():
return [types.TextContent(
type="text",
text=f"❌ No saved state found for project: {project_name}"
)]
with open(state_file, 'r') as f:
state_data = json.load(f)
return [types.TextContent(
type="text",
text=json.dumps(state_data, indent=2)
)]
elif name == "list_active_projects":
projects = []
for project_dir in STATE_DIR.iterdir():
if project_dir.is_dir():
state_file = project_dir / "state.json"
if state_file.exists():
with open(state_file, 'r') as f:
data = json.load(f)
projects.append({
"name": project_dir.name,
"updated_at": data.get("updated_at", "unknown"),
"summary": data.get("summary", "No summary")
})
return [types.TextContent(
type="text",
text=json.dumps(projects, indent=2)
)]
elif name == "get_project_summary":
project_name = arguments["project_name"]
state_file = STATE_DIR / project_name / "state.json"
if not state_file.exists():
return [types.TextContent(
type="text",
text=f"❌ No saved state found for project: {project_name}"
)]
with open(state_file, 'r') as f:
state_data = json.load(f)
summary = f"📁 Project: {project_name}\n"
summary += f"📅 Updated: {state_data.get('updated_at', 'unknown')}\n"
summary += f"📝 Summary: {state_data.get('summary', 'No summary')}\n"
return [types.TextContent(
type="text",
text=summary
)]
else:
return [types.TextContent(
type="text",
text=f"Unknown tool: {name}"
)]
async def main():
"""Main entry point"""
logger.info(f"Starting Claude Continuity MCP Server")
logger.info(f"State directory: {STATE_DIR}")
async with mcp.server.stdio.stdio_server() as (read_stream, write_stream):
await server.run(
read_stream,
write_stream,
InitializationOptions(
server_name="claude-continuity",
server_version="1.0.0",
capabilities=server.get_capabilities(
notification_options=NotificationOptions(),
experimental_capabilities={},
),
),
)
if __name__ == "__main__":
import asyncio
asyncio.run(main())