File size: 4,960 Bytes
35765b5 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 |
"""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()
|