MCP Project Orchestrator
by sparesparrow
- src
- mcp_project_orchestrator
from mcp.server.fastmcp import FastMCP
import os
import json
from typing import List, Dict, Optional
# Load MCP configuration from JSON file
CONFIG_FILE = 'project_orchestration.json'
with open(CONFIG_FILE, 'r') as config_file:
MCP_CONFIG = json.load(config_file)
# MCP configuration details (e.g., communication_protocol, mcp_compliance) are now available in MCP_CONFIG
# Directory for projects
PROJECTS_DIR = './projects'
os.makedirs(PROJECTS_DIR, exist_ok=True)
# Load project templates from JSON file
with open('project_templates.json', 'r') as f:
PROJECT_TEMPLATES = json.load(f)
# Comprehensive README template aligned with JSON requirements
README_TEMPLATE = """
# {{project_name}}
## Overview
{{project_name}} is designed to {{primary_purpose}} using {{design}} patterns, adhering to systematic approaches for maintainability and scalability.
## Architecture
### Design Patterns
{{design_patterns}}
### Software Architecture
{{software_architecture}}
### Components and Modules
{{components_section}}
### Relationships
{{relationships}}
### Interfaces
{{interfaces_section}}
### Communication Protocols
{{communication_protocols}}
### Technologies
{{technologies}}
### Dependencies
{{dependencies}}
### Commands
- **Installation**: `{{install_command}}`
- **Build**: `{{build_command}}`
- **Run**: `{{run_command}}`
- **Test**: `{{test_command}}`
## File Structure
{{project_structure}}
## Implementation Strategy
{{implementation_strategy}}
## Mermaid Diagrams
{{mermaid_diagrams}}
## Instructions for Composer Implementor Agent
{{instructions}}
"""
# Initialize MCP server
mcp = FastMCP("ProjectOrchestrator")
mcp.config = MCP_CONFIG # attach configuration to MCP server instance
'''
MCP Project Orchestrator Server
-------------------------------
This MCP server orchestrates the creation and configuration of new software projects.
It performs the following steps:
1. Extracts key design patterns and architecture concepts from user input.
2. Selects an appropriate project template from a standardized catalogue.
3. Applies the template by creating well-structured directories and placeholder files.
4. Generates comprehensive documentation including software architecture, components, process flows, and file structures.
The server configuration is loaded from 'project_orchestration.json', which defines overall settings such as communication protocols and compliance standards.
Developers can extend or modify this orchestration process by updating the template definitions or the configuration JSON.
'''
# Tool: Analyze design patterns and architecture
@mcp.tool()
def analyze_design_patterns(idea: str) -> Dict[str, List[str]]:
"""Analyze the user's idea to identify design patterns and architecture concepts."""
idea_lower = idea.lower()
patterns = []
architectures = []
keyword_map = {
"microservices": ("Microservices Architecture", "Distributed System"),
"event": ("Event-Driven Architecture", "Asynchronous Processing"),
"async": ("Event-Driven Architecture", "Asynchronous Processing"),
"data": ("Repository Pattern", "Layered Architecture"),
"repository": ("Repository Pattern", "Layered Architecture"),
"cqrs": ("CQRS", "Event Sourcing"),
"client": ("Client-Server", "Request-Response"),
"server": ("Client-Server", "Request-Response"),
"modular": ("Modular Monolith", "Monolithic Architecture"),
"serverless": ("Serverless Architecture", "Function-as-a-Service"),
"bridge": ("Bridge Pattern", "Abstraction Separation"),
"composite": ("Composite Pattern", "Tree Structure"),
"flyweight": ("Flyweight Pattern", "Memory Optimization"),
"strategy": ("Strategy Pattern", "Behavioral Flexibility"),
"template": ("Template Method Pattern", "Algorithm Skeleton"),
"visitor": ("Visitor Pattern", "Operation Separation")
}
for keyword, (pattern, arch) in keyword_map.items():
if keyword in idea_lower:
if pattern not in patterns:
patterns.append(pattern)
if arch not in architectures:
architectures.append(arch)
if not patterns:
patterns.append("Modular Monolith")
architectures.append("Monolithic Architecture")
return {"design_patterns": patterns, "architectures": architectures}
# Tool: Generate Mermaid diagrams (aligned with JSON's MermaidTool)
@mcp.tool()
def mermaid_tool(diagram_planning: str, template_name: Optional[str] = None) -> str:
"""Generate Mermaid diagrams for visualization based on planning."""
planning_lower = diagram_planning.lower()
if "architecture" in planning_lower:
if template_name and "Microservices" in template_name:
return (
"```mermaid\n"
"graph TD\n"
" A[API Gateway] --> B[UserService]\n"
" A --> C[OrderService]\n"
" B --> D[UserDB]\n"
" C --> E[OrderDB]\n"
" B --> F[MessageQueue]\n"
" C --> F\n"
"```\n"
)
elif template_name and "EventDriven" in template_name:
return (
"```mermaid\n"
"graph TD\n"
" A[EventProducer] --> B[EventBus]\n"
" B --> C[EventConsumer]\n"
" C --> D[EventStore]\n"
"```\n"
)
return (
"```mermaid\n"
"graph TD\n"
" A[CoreModule] --> B[Services]\n"
" B --> C[Utilities]\n"
" A --> D[Database]\n"
"```\n"
)
elif "file structure" in planning_lower:
if template_name:
template = next((t for t in PROJECT_TEMPLATES if t["project_name"] == template_name), None)
if template:
components = "\n".join([f" E --> F{i+1}[{c['name']}]" for i, c in enumerate(template["components"])])
return (
"```mermaid\n"
"graph TD\n"
" A[ProjectRoot] --> B[src]\n"
" A --> C[tests]\n"
" A --> D[docs]\n"
" B --> E[components]\n"
f"{components}\n"
" B --> G[interfaces]\n"
" B --> H[services]\n"
" B --> I[utils]\n"
"```\n"
)
return (
"```mermaid\n"
"graph TD\n"
" A[ProjectRoot] --> B[src]\n"
" A --> C[tests]\n"
" A --> D[docs]\n"
" B --> E[components]\n"
" B --> F[interfaces]\n"
" B --> G[services]\n"
" B --> H[utils]\n"
"```\n"
)
elif "process flow" in planning_lower:
return (
"```mermaid\n"
"sequenceDiagram\n"
" participant U as User\n"
" participant S as System\n"
" U->>S: Initiate Action\n"
" S-->>U: Process Result\n"
"```\n"
)
return "```mermaid\n%% Placeholder diagram\n```"
# Tool: Apply project template
@mcp.tool()
def apply_project_template(template_name: str, project_name: str, user_idea: str, design_info: Dict[str, List[str]]) -> str:
"""Apply a template and create comprehensive documentation."""
template = next((t for t in PROJECT_TEMPLATES if t["project_name"] == template_name), None)
if not template:
return f"Error: Template '{template_name}' not found."
project_path = os.path.join(PROJECTS_DIR, project_name)
if os.path.exists(project_path):
return f"Error: Project '{project_name}' already exists."
# Step 5: Prepare detailed file structure
os.makedirs(os.path.join(project_path, "src", "components"), exist_ok=True)
os.makedirs(os.path.join(project_path, "src", "interfaces"), exist_ok=True)
os.makedirs(os.path.join(project_path, "src", "services"), exist_ok=True)
os.makedirs(os.path.join(project_path, "src", "utils"), exist_ok=True)
os.makedirs(os.path.join(project_path, "tests"), exist_ok=True)
os.makedirs(os.path.join(project_path, "docs"), exist_ok=True)
# Generate component files with consistent names and TODOs
components_section = ""
interfaces_section = ""
relationships = ""
communication_protocols = "REST API, Message Queues" if "Microservices" in template_name else "Internal Function Calls"
for i, component in enumerate(template["components"]):
name = component["name"]
# Interface
interface_file = f"i_{name.lower()}.py"
with open(os.path.join(project_path, "src", "interfaces", interface_file), "w") as f:
f.write(f"# TODO: Define interface methods for {name}\nclass I{name}:\n pass\n")
# Implementation
impl_file = f"{name.lower()}.py"
with open(os.path.join(project_path, "src", "components", impl_file), "w") as f:
f.write(f"# TODO: Implement {name} logic\nclass {name}:\n pass\n")
# Service (if applicable)
service_file = f"{name.lower()}_service.py"
with open(os.path.join(project_path, "src", "services", service_file), "w") as f:
f.write(f"# TODO: Implement service logic for {name}\n")
# Test
test_file = f"test_{name.lower()}.py"
with open(os.path.join(project_path, "tests", test_file), "w") as f:
f.write(f"# TODO: Write unit tests for {name}\n")
components_section += (
f"- **{name}**: {component.get('description', 'TBD')}\n"
f" - Interface: [{interface_file}](./src/interfaces/{interface_file})\n"
f" - Implementation: [{impl_file}](./src/components/{impl_file})\n"
f" - Service: [{service_file}](./src/services/{service_file})\n"
f" - Tests: [{test_file}](./tests/{test_file})\n"
)
interfaces_section += f"class I{name}:\n # TODO: Define {name} methods\n pass\n\n"
if i > 0:
relationships += f"- {template['components'][i-1]['name']} interacts with {name} via {communication_protocols}\n"
# Step 4: Comprehensive documentation
design_patterns = "- " + "\n- ".join(design_info["design_patterns"])
software_architecture = "- " + "\n- ".join(design_info["architectures"])
technologies = "Python, Flask, Docker, Kafka" if "Microservices" in template_name else "Python, Django"
dependencies = "requests, pytest, docker, confluent-kafka" if "Microservices" in template_name else "django, pytest"
install_command = "pip install -r requirements.txt"
build_command = "docker build ." if "Microservices" in template_name else "python manage.py migrate"
run_command = "docker-compose up" if "Microservices" in template_name else "python manage.py runserver"
test_command = "pytest"
# File structure visualization
project_structure = mermaid_tool("file structure", template_name)
# Step 6: Implementation strategy
impl_order = "\n".join([f"{i+1}. src/components/{c['name'].lower()}.py" for i, c in enumerate(template["components"])])
implementation_strategy = (
f"### File Implementation Order\n{impl_order}\n"
"### Testing Strategies\n- Unit Tests: Use pytest for component-level testing.\n- Integration Tests: Verify inter-component interactions.\n"
f"### Build and Deployment\n- Build: `{build_command}`\n- Deploy: Use Docker containers or a cloud platform like AWS.\n"
)
# Mermaid diagrams
mermaid_diagrams = (
f"### Architecture Diagram\n{mermaid_tool('architecture', template_name)}\n"
f"### File Structure\n{project_structure}\n"
f"### Process Flow\n{mermaid_tool('process flow', template_name)}"
)
# Instructions for the composer implementor agent
instructions = (
"1. Refine the generated documentation in README.md.\n"
"2. Implement components starting with core logic in src/components/.\n"
"3. Use mermaid_tool for additional visualizations (e.g., `mermaid_tool 'detailed process flow'`).\n"
"4. Follow the implementation strategy and test using provided commands."
)
# Substitutions for README
substitutions = {
"project_name": project_name,
"design": ", ".join(design_info["design_patterns"]),
"primary_purpose": template["description"].split(".")[0],
"design_patterns": design_patterns,
"software_architecture": software_architecture,
"components_section": components_section,
"relationships": relationships if relationships else "TBD - Define inter-component relationships",
"interfaces_section": interfaces_section,
"communication_protocols": communication_protocols,
"technologies": technologies,
"dependencies": dependencies,
"install_command": install_command,
"build_command": build_command,
"run_command": run_command,
"test_command": test_command,
"project_structure": project_structure,
"implementation_strategy": implementation_strategy,
"mermaid_diagrams": mermaid_diagrams,
"instructions": instructions
}
# Generate README
readme_content = README_TEMPLATE
for key, value in substitutions.items():
readme_content = readme_content.replace("{{" + key + "}}", value)
with open(os.path.join(project_path, "README.md"), "w") as f:
f.write(readme_content)
return f"Project '{project_name}' created successfully at '{project_path}'."
# Helper: Select template
def select_template(idea: str, design_info: Dict[str, List[str]]) -> str:
"""Select a project template based on design patterns and architectures."""
idea_lower = idea.lower()
patterns = design_info["design_patterns"]
template_map = {
"Microservices Architecture": "MicroservicesArchitectureProject",
"Event-Driven Architecture": "EventDrivenArchitectureProject",
"Repository Pattern": "RepositoryPatternProject",
"CQRS": "CQRSProject",
"Client-Server": "ClientServerProject",
"Modular Monolith": "ModularMonolithProject",
"Serverless Architecture": "ServerlessFunctionProject",
"Bridge Pattern": "BridgeProject",
"Composite Pattern": "CompositeProject",
"Flyweight Pattern": "FlyweightProject",
"Strategy Pattern": "StrategyProject",
"Template Method Pattern": "TemplateMethodProject",
"Visitor Pattern": "VisitorProject"
}
for pattern in patterns:
if pattern in template_map:
return template_map[pattern]
return "ModularMonolithProject" # Default
# Tool: Orchestrate project setup
@mcp.tool()
def orchestrate_new_project(user_idea: str) -> str:
"""Orchestrate the setup of a new software project from the user's idea."""
# Step 1: Information Extraction
design_info = analyze_design_patterns(user_idea)
# Step 2: Design Patterns & Architecture Identification (handled by analyze_design_patterns)
# Step 3: Project Template Application
template_name = select_template(user_idea, design_info)
project_name = user_idea.lower().replace(" ", "_")[:20]
# Steps 4-6: Apply template, generate documentation, prepare file structure, and define strategy
result = apply_project_template(template_name, project_name, user_idea, design_info)
if "Error" in result:
return result
return (
f"Project '{project_name}' has been initialized with template '{template_name}'.\n"
f"Design Patterns Identified: {', '.join(design_info['design_patterns'])}\n"
f"Architecture Concepts: {', '.join(design_info['architectures'])}\n"
"Next Steps: Review the generated README.md at '{project_path}/README.md' for detailed documentation and instructions."
)
# Run the server
if __name__ == "__main__":
mcp.run()