"""MCP Server for XER file analysis.""" from mcp.server import Server from mcp.server.stdio import stdio_server from mcp.types import TextContent, Tool from xer_mcp.db import db # Create MCP server instance server = Server("xer-mcp") # Server state _file_loaded: bool = False _current_project_id: str | None = None def is_file_loaded() -> bool: """Check if an XER file has been loaded.""" return _file_loaded def get_current_project_id() -> str | None: """Get the currently selected project ID.""" return _current_project_id def set_file_loaded(loaded: bool, project_id: str | None = None) -> None: """Set the file loaded state.""" global _file_loaded, _current_project_id _file_loaded = loaded _current_project_id = project_id @server.list_tools() async def list_tools() -> list[Tool]: """List available MCP tools.""" return [ Tool( name="load_xer", description="Load a Primavera P6 XER file and parse its schedule data. " "For multi-project files, specify project_id to select a project.", inputSchema={ "type": "object", "properties": { "file_path": { "type": "string", "description": "Absolute path to the XER file", }, "project_id": { "type": "string", "description": "Project ID to select (required for multi-project files)", }, "db_path": { "type": "string", "description": "Path for persistent SQLite database file. " "If omitted, uses in-memory database. " "If empty string, auto-generates path from XER filename (same directory, .sqlite extension).", }, }, "required": ["file_path"], }, ), Tool( name="list_activities", description="List activities from the loaded XER file with optional filtering and pagination.", inputSchema={ "type": "object", "properties": { "start_date": { "type": "string", "format": "date", "description": "Filter activities starting on or after this date (YYYY-MM-DD)", }, "end_date": { "type": "string", "format": "date", "description": "Filter activities ending on or before this date (YYYY-MM-DD)", }, "wbs_id": { "type": "string", "description": "Filter by WBS element ID", }, "activity_type": { "type": "string", "enum": ["TT_Task", "TT_Mile", "TT_LOE", "TT_WBS", "TT_Rsrc"], "description": "Filter by activity type", }, "limit": { "type": "integer", "default": 100, "minimum": 1, "maximum": 1000, "description": "Maximum number of activities to return", }, "offset": { "type": "integer", "default": 0, "minimum": 0, "description": "Number of activities to skip", }, }, }, ), Tool( name="get_activity", description="Get detailed information for a specific activity by ID.", inputSchema={ "type": "object", "properties": { "activity_id": { "type": "string", "description": "The task_id of the activity", }, }, "required": ["activity_id"], }, ), Tool( name="list_relationships", description="List all activity relationships (dependencies) with pagination.", inputSchema={ "type": "object", "properties": { "limit": { "type": "integer", "default": 100, "minimum": 1, "maximum": 1000, "description": "Maximum number of relationships to return", }, "offset": { "type": "integer", "default": 0, "minimum": 0, "description": "Number of relationships to skip", }, }, }, ), Tool( name="get_predecessors", description="Get all predecessor activities for a given activity.", inputSchema={ "type": "object", "properties": { "activity_id": { "type": "string", "description": "The task_id of the activity", }, }, "required": ["activity_id"], }, ), Tool( name="get_successors", description="Get all successor activities for a given activity.", inputSchema={ "type": "object", "properties": { "activity_id": { "type": "string", "description": "The task_id of the activity", }, }, "required": ["activity_id"], }, ), Tool( name="get_project_summary", description="Get a summary of the loaded project including dates and activity counts.", inputSchema={ "type": "object", "properties": {}, }, ), Tool( name="list_milestones", description="List all milestone activities in the loaded project.", inputSchema={ "type": "object", "properties": {}, }, ), Tool( name="get_critical_path", description="Get all activities on the critical path that determine project duration.", inputSchema={ "type": "object", "properties": {}, }, ), Tool( name="get_database_info", description="Get information about the currently loaded database including file path and schema. " "Use this to get connection details for direct SQL access.", inputSchema={ "type": "object", "properties": {}, }, ), ] @server.call_tool() async def call_tool(name: str, arguments: dict) -> list[TextContent]: """Handle MCP tool calls.""" import json if name == "load_xer": from xer_mcp.tools.load_xer import load_xer result = await load_xer( file_path=arguments["file_path"], project_id=arguments.get("project_id"), db_path=arguments.get("db_path"), ) return [TextContent(type="text", text=json.dumps(result, indent=2))] if name == "list_activities": from xer_mcp.tools.list_activities import list_activities result = await list_activities( start_date=arguments.get("start_date"), end_date=arguments.get("end_date"), wbs_id=arguments.get("wbs_id"), activity_type=arguments.get("activity_type"), limit=arguments.get("limit", 100), offset=arguments.get("offset", 0), ) return [TextContent(type="text", text=json.dumps(result, indent=2))] if name == "get_activity": from xer_mcp.tools.get_activity import get_activity result = await get_activity(activity_id=arguments["activity_id"]) return [TextContent(type="text", text=json.dumps(result, indent=2))] if name == "list_relationships": from xer_mcp.tools.list_relationships import list_relationships result = await list_relationships( limit=arguments.get("limit", 100), offset=arguments.get("offset", 0), ) return [TextContent(type="text", text=json.dumps(result, indent=2))] if name == "get_predecessors": from xer_mcp.tools.get_predecessors import get_predecessors result = await get_predecessors(activity_id=arguments["activity_id"]) return [TextContent(type="text", text=json.dumps(result, indent=2))] if name == "get_successors": from xer_mcp.tools.get_successors import get_successors result = await get_successors(activity_id=arguments["activity_id"]) return [TextContent(type="text", text=json.dumps(result, indent=2))] if name == "get_project_summary": from xer_mcp.tools.get_project_summary import get_project_summary result = await get_project_summary() return [TextContent(type="text", text=json.dumps(result, indent=2))] if name == "list_milestones": from xer_mcp.tools.list_milestones import list_milestones result = await list_milestones() return [TextContent(type="text", text=json.dumps(result, indent=2))] if name == "get_critical_path": from xer_mcp.tools.get_critical_path import get_critical_path result = await get_critical_path() return [TextContent(type="text", text=json.dumps(result, indent=2))] if name == "get_database_info": from xer_mcp.tools.get_database_info import get_database_info result = await get_database_info() return [TextContent(type="text", text=json.dumps(result, indent=2))] raise ValueError(f"Unknown tool: {name}") async def run_server() -> None: """Run the MCP server with stdio transport.""" db.initialize() async with stdio_server() as (read_stream, write_stream): await server.run( read_stream, write_stream, server.create_initialization_options(), )