Amal Nimmy Lal
feat : Project Memory
35765b5
"""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()