import os
import json
import logging
from typing import List, Dict, Any, Optional
from pydantic import BaseModel
from mcp.server.fastmcp import FastMCP
# Setup logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("mcp-backend-architect")
# Initialize FastMCP server
mcp = FastMCP("Backend Architect")
STATE_FILE = ".mcp_state.json"
class RoleSchema(BaseModel):
name: str
description: str = ""
permissions: List[str] = []
class ModelSchema(BaseModel):
name: str
fields: List[Dict[str, str]]
dependencies: List[str] = []
status: str = "pending"
class RouteSchema(BaseModel):
path: str
method: str
summary: str
name: str
request_model: Optional[str] = None # Name of the Pydantic model for input
response_model: Optional[str] = None # Name of the model for output
required_permissions: List[str] = [] # Link to your RoleSchema
logic: str = "" # Describe what the route actually does
status: str = "pending"
class TestSchema(BaseModel):
name: str
scenario: str
status: str = "pending"
class ProjectState(BaseModel):
project_root: str
organization_context: str = ""
roles: List[RoleSchema] = []
models: List[ModelSchema] = []
routes: List[RouteSchema] = []
tests: List[TestSchema] = []
def read_state() -> ProjectState:
if not os.path.exists(STATE_FILE):
return ProjectState(project_root="")
with open(STATE_FILE, "r") as f:
data = json.load(f)
# Migration: if roles are strings, convert to RoleSchema objects
if "roles" in data and data["roles"] and isinstance(data["roles"][0], str):
data["roles"] = [{"name": r} for r in data["roles"]]
return ProjectState(**data)
def write_state(state: ProjectState):
with open(STATE_FILE, "w") as f:
json.dump(state.model_dump(), f, indent=2)
def update_init_file(directory: str, name: str):
init_file = os.path.join(directory, "__init__.py")
if not os.path.exists(init_file):
with open(init_file, "w") as f:
f.write("")
import_line = f"from .{name} import *"
with open(init_file, "r") as f:
content = f.read()
if import_line not in content:
with open(init_file, "a") as f:
f.write(f"\n{import_line}")
@mcp.tool()
def initialize_project(root_path: str = ".") -> str:
"""Creates the folder structure and pyproject.toml configured for uv. Defaults to current directory."""
abs_root = os.path.abspath(root_path)
os.makedirs(abs_root, exist_ok=True)
folders = ["app/models", "app/routes", "app/core", "docs"]
for folder in folders:
os.makedirs(os.path.join(abs_root, folder), exist_ok=True)
# Touch __init__.py
with open(os.path.join(abs_root, folder, "__init__.py"), "w") as f:
pass
# Initialize uv if root_path is different from current or if pyproject.toml missing
pyproject_path = os.path.join(abs_root, "pyproject.toml")
if not os.path.exists(pyproject_path):
# We assume uv is installed and we're just creating a standard FastAPI structure
pyproject_content = f"""[project]
name = "{os.path.basename(abs_root)}"
version = "0.1.0"
description = "FastAPI project managed by Backend Architect"
readme = "README.md"
requires-python = ">=3.12"
dependencies = [
"fastapi",
"sqlmodel",
"pydantic-settings",
"python-multipart",
"passlib[bcrypt]",
"python-jose[cryptography]",
"alembic",
"uvicorn",
"supabase",
]
[tool.uv]
dev-dependencies = [
"pytest",
"httpx",
]
"""
with open(pyproject_path, "w") as f:
f.write(pyproject_content)
state = ProjectState(project_root=abs_root)
write_state(state)
return f"Project initialized at {abs_root}"
@mcp.tool()
def save_organization_context(context: str) -> str:
"""Saves the organizational context and complex operation requirements to state."""
state = read_state()
state.organization_context = context
write_state(state)
return "Organization context saved."
@mcp.tool()
def save_roles_plan(roles: List[Dict[str, Any]]) -> str:
"""Saves user roles/permissions to state. Each role should have 'name', 'description', and 'permissions'."""
state = read_state()
state.roles = [RoleSchema(**r) for r in roles]
write_state(state)
return f"Roles plan saved with {len(roles)} roles."
@mcp.tool()
def save_database_plan(models: List[Dict[str, Any]]) -> str:
"""Saves table schemas (fields, types, relationships) to state."""
state = read_state()
state.models = [ModelSchema(**m) for m in models]
write_state(state)
return f"Database plan saved with {len(models)} models."
@mcp.tool()
def save_route_plan(routes: List[Dict[str, Any]]) -> str:
"""Saves API endpoints (method, path, summary) to state."""
state = read_state()
state.routes = [RouteSchema(**r) for r in routes]
write_state(state)
return f"Route plan saved with {len(routes)} routes."
@mcp.tool()
def save_test_plan(tests: List[Dict[str, Any]]) -> str:
"""Saves simulation scenarios to state."""
state = read_state()
state.tests = [TestSchema(**t) for t in tests]
write_state(state)
return f"Test plan saved with {len(tests)} tests."
@mcp.tool()
def get_next_pending_task() -> Optional[Dict[str, str]]:
"""Returns the first unbuilt item. Priority: Models -> Routes -> Tests."""
state = read_state()
# Priority 1: Models
done_models = {m.name for m in state.models if m.status == "done"}
for m in state.models:
if m.status == "pending":
# Check dependencies
if all(dep in done_models for dep in m.dependencies):
return {"type": "model", "name": m.name, "description": f"Build Database Model: {m.name}"}
# Priority 2: Routes (Only if ALL models are done)
if all(m.status == "done" for m in state.models):
for r in state.routes:
if r.status == "pending":
return {"type": "route", "name": r.name, "description": f"Build Route: {r.method} {r.path}"}
# Priority 3: Tests (Only if ALL routes are done)
if all(m.status == "done" for m in state.models) and all(r.status == "done" for r in state.routes):
for t in state.tests:
if t.status == "pending":
return {"type": "test", "name": t.name, "description": f"Build Test: {t.scenario}"}
return None
@mcp.tool()
def get_file_instruction(task_type: str, task_name: str) -> str:
"""Returns a strict SYSTEM PROMPT for the agent to write the code."""
state = read_state()
if task_type == "model":
model = next((m for m in state.models if m.name == task_name), None)
if not model:
return f"Error: Model {task_name} not found in plan."
fields_str = "\n".join([f"- {f.get('name')}: {f.get('type')}" for f in model.fields])
prompt = f"""Write the SQLModel for '{task_name}'.
Organization Context: {state.organization_context}
Technical Requirements:
- Project root: {state.project_root}
- Fields:
{fields_str}
- Dependencies: {', '.join(model.dependencies) if model.dependencies else 'None'}
- Constraints:
- Use Python 3.12 type hints.
- Use SQLModel for database models.
- For collections, use `typing.List` or the built-in `list` (do NOT import from a 'list' module).
- Include relationships if applicable.
- File path: app/models/{task_name.lower()}.py
"""
return prompt
elif task_type == "route":
route = next((r for r in state.routes if r.name == task_name), None)
if not route:
return f"Error: Route {task_name} not found."
# Get names of existing models for better imports
available_models = [m.name for m in state.models]
prompt = f"""Write the FastAPI router for '{task_name}'.
Organization Context: {state.organization_context}
Technical Requirements:
- Path: {route.path}
- Method: {route.method}
- Summary: {route.summary}
- Logic Description: {route.logic}
- Request Model: {route.request_model or "None"}
- Response Model: {route.response_model or "None"}
- Permissions Required: {', '.join(route.required_permissions) if route.required_permissions else 'None'}
Constraints:
- Use FastAPI APIRouter.
- Available models to import from `app.models`: {', '.join(available_models)}
- If models are missing, define appropriate Pydantic schemas in the file.
- Implement error handling with HTTPException.
- File path: app/routes/{task_name.lower()}.py
"""
return prompt
elif task_type == "test":
test = next((t for t in state.tests if t.name == task_name), None)
if not test:
return f"Error: Test {task_name} not found in plan."
prompt = f"""Write the pytest for scenario: '{test.scenario}'.
Context:
- Name: {test.name}
- Constraints: Use pytest and httpx for async testing.
- File path: tests/{test.name.lower()}.py
"""
return prompt
return "Invalid task type."
@mcp.tool()
def write_component_file(type: str, name: str, content: str) -> str:
"""Writes the file and marks the task as done."""
state = read_state()
base_dir = state.project_root
if not base_dir:
return "Error: Project not initialized. Call initialize_project first."
subdir = f"{type}s" if type in ["model", "route"] else "tests"
file_path = os.path.join(base_dir, "app", subdir, f"{name.lower()}.py")
# Ensure directory exists
os.makedirs(os.path.dirname(file_path), exist_ok=True)
with open(file_path, "w") as f:
f.write(content)
# Atomic auto-import update for models and routes
if type in ["model", "route"]:
update_init_file(os.path.join(base_dir, "app", subdir), name.lower())
# Mark as done in state
if type == "model":
for m in state.models:
if m.name == name:
m.status = "done"
elif type == "route":
for r in state.routes:
if r.name == name:
r.status = "done"
elif type == "test":
for t in state.tests:
if t.name == name:
t.status = "done"
write_state(state)
return f"File written successfully to {file_path} and state updated."
if __name__ == "__main__":
mcp.run()
