"""Project-related MCP tools.""" from sqlalchemy.orm import Session from app.database import SessionLocal from app.models import Project, ProjectMembership, User def check_project_id_available(project_id: str) -> dict: """ Check if a project ID is available. Args: project_id: The project ID to check Returns: Dict with 'available' boolean """ db: Session = SessionLocal() try: existing = db.query(Project).filter(Project.id == project_id).first() return {"available": existing is None, "project_id": project_id} finally: db.close() def create_project(name: str, description: str, user_id: str) -> dict: """ Create a new project and add the creator as owner. The project ID will be the project name (used as-is). Args: name: Project name (will also be used as project ID) description: Project description user_id: ID of the user creating the project Returns: Created project data or error if project ID already exists """ db: Session = SessionLocal() try: # Verify user exists user = db.query(User).filter(User.id == user_id).first() if not user: return {"error": f"User {user_id} not found"} # Use name as project ID project_id = name # Check if project ID already exists existing = db.query(Project).filter(Project.id == project_id).first() if existing: return {"error": f"Project ID '{project_id}' already exists. Please choose a different name."} # Create project with name as ID project = Project( id=project_id, name=name, description=description, created_by=user_id ) db.add(project) db.flush() # Add creator as owner membership = ProjectMembership( project_id=project.id, user_id=user_id, role="owner" ) db.add(membership) db.commit() return { "id": project.id, "name": project.name, "description": project.description, "created_at": project.created_at.isoformat(), "created_by": project.created_by } finally: db.close() def list_projects(user_id: str) -> dict: """ List all projects for a user. Args: user_id: ID of the user Returns: List of projects the user is a member of """ db: Session = SessionLocal() try: # Get all project memberships for user memberships = ( db.query(ProjectMembership) .filter(ProjectMembership.user_id == user_id) .all() ) projects = [] for membership in memberships: project = membership.project projects.append({ "id": project.id, "name": project.name, "description": project.description, "created_at": project.created_at.isoformat(), "role": membership.role }) return {"projects": projects} finally: db.close() def join_project(project_id: str, user_id: str) -> dict: """ Add a user to an existing project. Args: project_id: ID of the project to join user_id: ID of the user joining Returns: Membership confirmation """ db: Session = SessionLocal() try: # Verify project exists project = db.query(Project).filter(Project.id == project_id).first() if not project: return {"error": f"Project {project_id} not found"} # Verify user exists user = db.query(User).filter(User.id == user_id).first() if not user: return {"error": f"User {user_id} not found"} # Check if already a member existing = ( db.query(ProjectMembership) .filter( ProjectMembership.project_id == project_id, ProjectMembership.user_id == user_id ) .first() ) if existing: return { "message": "Already a member", "project_id": project_id, "user_id": user_id, "role": existing.role } # Create membership membership = ProjectMembership( project_id=project_id, user_id=user_id, role="member" ) db.add(membership) db.commit() return { "message": "Joined project successfully", "project_id": project_id, "project_name": project.name, "user_id": user_id, "role": "member" } finally: db.close()