Spaces:
Build error
Build error
| """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() | |