""" MCP Server for Collaborative Creative Projects Enables characters to autonomously manage creative collaborations """ import asyncio import json import logging from typing import Dict, List, Any, Optional, Sequence from datetime import datetime, timedelta from mcp.server import Server from mcp.server.models import InitializationOptions from mcp.server.stdio import stdio_server from mcp.types import ( CallToolRequestParams, GetToolRequestParams, ListToolsRequestParams, TextContent, Tool, INVALID_PARAMS, INTERNAL_ERROR ) from ..collaboration.creative_projects import ( CollaborativeCreativeManager, ProjectType, ContributionType, ProjectStatus ) from ..utils.logging import log_character_action, log_error_with_context logger = logging.getLogger(__name__) class CreativeProjectsMCPServer: """MCP server for autonomous creative project management""" def __init__(self, creative_manager: CollaborativeCreativeManager): self.creative_manager = creative_manager self.server = Server("creative-projects") self.current_character: Optional[str] = None # Register MCP tools self._register_tools() def _register_tools(self): """Register all creative project tools""" @self.server.list_tools() async def handle_list_tools(request: ListToolsRequestParams) -> list[Tool]: """List available creative project tools""" return [ Tool( name="propose_creative_project", description="Propose a new collaborative creative project", inputSchema={ "type": "object", "properties": { "title": {"type": "string", "description": "Project title"}, "description": {"type": "string", "description": "Project description and vision"}, "project_type": { "type": "string", "enum": ["story", "poem", "philosophy", "worldbuilding", "music", "art_concept", "dialogue", "research", "manifesto", "mythology"], "description": "Type of creative project" }, "target_collaborators": { "type": "array", "items": {"type": "string"}, "description": "List of characters to invite" }, "goals": { "type": "array", "items": {"type": "string"}, "description": "Project goals and objectives" }, "role_descriptions": { "type": "object", "description": "Specific roles for each collaborator" }, "estimated_duration": {"type": "string", "description": "Estimated project duration"} }, "required": ["title", "description", "project_type", "target_collaborators"] } ), Tool( name="respond_to_project_invitation", description="Respond to a creative project invitation", inputSchema={ "type": "object", "properties": { "invitation_id": {"type": "string", "description": "Invitation ID"}, "accept": {"type": "boolean", "description": "Whether to accept the invitation"}, "response_message": {"type": "string", "description": "Response message"} }, "required": ["invitation_id", "accept"] } ), Tool( name="contribute_to_project", description="Add a contribution to a creative project", inputSchema={ "type": "object", "properties": { "project_id": {"type": "string", "description": "Project ID"}, "content": {"type": "string", "description": "Contribution content"}, "contribution_type": { "type": "string", "enum": ["idea", "content", "revision", "feedback", "inspiration", "structure", "polish"], "description": "Type of contribution" }, "build_on_contribution_id": {"type": "string", "description": "ID of contribution to build upon"}, "feedback_for_contribution_id": {"type": "string", "description": "ID of contribution being reviewed"}, "metadata": {"type": "object", "description": "Additional contribution metadata"} }, "required": ["project_id", "content", "contribution_type"] } ), Tool( name="get_project_suggestions", description="Get personalized project suggestions based on interests and relationships", inputSchema={"type": "object", "properties": {}} ), Tool( name="get_active_projects", description="Get currently active projects for the character", inputSchema={"type": "object", "properties": {}} ), Tool( name="get_project_analytics", description="Get detailed analytics for a specific project", inputSchema={ "type": "object", "properties": { "project_id": {"type": "string", "description": "Project ID"} }, "required": ["project_id"] } ), Tool( name="get_pending_invitations", description="Get pending project invitations for the character", inputSchema={"type": "object", "properties": {}} ), Tool( name="search_projects", description="Search for projects by topic, type, or collaborators", inputSchema={ "type": "object", "properties": { "query": {"type": "string", "description": "Search query"}, "project_type": {"type": "string", "description": "Filter by project type"}, "status": {"type": "string", "description": "Filter by project status"}, "collaborator": {"type": "string", "description": "Filter by specific collaborator"} } } ) ] @self.server.call_tool() async def handle_call_tool(name: str, arguments: Dict[str, Any]) -> Sequence[TextContent]: """Handle tool calls for creative projects""" if not self.current_character: return [TextContent(type="text", text="Error: No character context set")] try: if name == "propose_creative_project": return await self._propose_creative_project(arguments) elif name == "respond_to_project_invitation": return await self._respond_to_project_invitation(arguments) elif name == "contribute_to_project": return await self._contribute_to_project(arguments) elif name == "get_project_suggestions": return await self._get_project_suggestions(arguments) elif name == "get_active_projects": return await self._get_active_projects(arguments) elif name == "get_project_analytics": return await self._get_project_analytics(arguments) elif name == "get_pending_invitations": return await self._get_pending_invitations(arguments) elif name == "search_projects": return await self._search_projects(arguments) else: return [TextContent(type="text", text=f"Unknown tool: {name}")] except Exception as e: log_error_with_context(e, {"tool": name, "character": self.current_character}) return [TextContent(type="text", text=f"Error executing {name}: {str(e)}")] async def set_character_context(self, character_name: str): """Set the current character context for tool calls""" self.current_character = character_name logger.info(f"Creative projects MCP context set to {character_name}") async def _propose_creative_project(self, args: Dict[str, Any]) -> Sequence[TextContent]: """Handle project proposal tool call""" try: success, message = await self.creative_manager.propose_project( initiator=self.current_character, project_idea=args ) if success: log_character_action(self.current_character, "mcp_proposed_project", { "project_title": args.get("title"), "project_type": args.get("project_type"), "collaborator_count": len(args.get("target_collaborators", [])) }) return [TextContent( type="text", text=f"✨ Successfully proposed creative project '{args['title']}'! {message}" )] else: return [TextContent( type="text", text=f"❌ Failed to propose project: {message}" )] except Exception as e: log_error_with_context(e, {"tool": "propose_creative_project", "character": self.current_character}) return [TextContent(type="text", text=f"Error proposing project: {str(e)}")] async def _respond_to_project_invitation(self, args: Dict[str, Any]) -> Sequence[TextContent]: """Handle invitation response tool call""" try: invitation_id = args["invitation_id"] accepted = args["accept"] response_message = args.get("response_message", "") success, message = await self.creative_manager.respond_to_invitation( invitee=self.current_character, invitation_id=invitation_id, accepted=accepted, response_message=response_message ) if success: action = "accepted" if accepted else "declined" log_character_action(self.current_character, f"mcp_{action}_project_invitation", { "invitation_id": invitation_id, "response": response_message }) emoji = "✅" if accepted else "❌" return [TextContent( type="text", text=f"{emoji} {message}" )] else: return [TextContent( type="text", text=f"❌ Failed to respond to invitation: {message}" )] except Exception as e: log_error_with_context(e, {"tool": "respond_to_project_invitation", "character": self.current_character}) return [TextContent(type="text", text=f"Error responding to invitation: {str(e)}")] async def _contribute_to_project(self, args: Dict[str, Any]) -> Sequence[TextContent]: """Handle project contribution tool call""" try: project_id = args["project_id"] contribution_data = { "content": args["content"], "contribution_type": args["contribution_type"], "build_on_contribution_id": args.get("build_on_contribution_id"), "feedback_for_contribution_id": args.get("feedback_for_contribution_id"), "metadata": args.get("metadata", {}) } success, message = await self.creative_manager.contribute_to_project( contributor=self.current_character, project_id=project_id, contribution=contribution_data ) if success: log_character_action(self.current_character, "mcp_contributed_to_project", { "project_id": project_id, "contribution_type": args["contribution_type"], "content_length": len(args["content"]) }) return [TextContent( type="text", text=f"🎨 Successfully added {args['contribution_type']} contribution! {message}" )] else: return [TextContent( type="text", text=f"❌ Failed to add contribution: {message}" )] except Exception as e: log_error_with_context(e, {"tool": "contribute_to_project", "character": self.current_character}) return [TextContent(type="text", text=f"Error adding contribution: {str(e)}")] async def _get_project_suggestions(self, args: Dict[str, Any]) -> Sequence[TextContent]: """Handle project suggestions tool call""" try: suggestions = await self.creative_manager.get_project_suggestions(self.current_character) if not suggestions: return [TextContent( type="text", text="💡 No specific project suggestions at the moment. Consider exploring new creative areas or connecting with other characters!" )] response = "🎯 Personalized Creative Project Suggestions:\n\n" for i, suggestion in enumerate(suggestions, 1): response += f"{i}. **{suggestion['title']}**\n" response += f" Type: {suggestion['project_type']}\n" response += f" Description: {suggestion['description']}\n" response += f" Suggested collaborators: {', '.join(suggestion.get('suggested_collaborators', []))}\n" response += f" Inspiration: {suggestion.get('inspiration', 'Unknown')}\n\n" response += "💭 These suggestions are based on your interests, creative history, and relationships with other characters." return [TextContent(type="text", text=response)] except Exception as e: log_error_with_context(e, {"tool": "get_project_suggestions", "character": self.current_character}) return [TextContent(type="text", text=f"Error getting suggestions: {str(e)}")] async def _get_active_projects(self, args: Dict[str, Any]) -> Sequence[TextContent]: """Handle active projects tool call""" try: active_projects = await self.creative_manager.get_active_projects(self.current_character) if not active_projects: return [TextContent( type="text", text="📂 You're not currently involved in any active creative projects. Why not propose something new or join an existing collaboration?" )] response = f"🎨 Your Active Creative Projects ({len(active_projects)}):\n\n" for project in active_projects: response += f"**{project.title}** ({project.project_type.value})\n" response += f"Status: {project.status.value} | Collaborators: {len(project.collaborators)}\n" response += f"Description: {project.description[:100]}...\n" response += f"Contributions: {len(project.contributions)} | Created: {project.created_at.strftime('%m/%d/%Y')}\n\n" return [TextContent(type="text", text=response)] except Exception as e: log_error_with_context(e, {"tool": "get_active_projects", "character": self.current_character}) return [TextContent(type="text", text=f"Error getting active projects: {str(e)}")] async def _get_project_analytics(self, args: Dict[str, Any]) -> Sequence[TextContent]: """Handle project analytics tool call""" try: project_id = args["project_id"] analytics = await self.creative_manager.get_project_analytics(project_id) if "error" in analytics: return [TextContent( type="text", text=f"❌ Error getting analytics: {analytics['error']}" )] project_info = analytics["project_info"] contrib_stats = analytics["contribution_stats"] health_metrics = analytics["health_metrics"] response = f"📊 Analytics for '{project_info['title']}':\n\n" response += f"**Project Status:**\n" response += f"• Status: {project_info['status']}\n" response += f"• Days active: {project_info['days_active']}\n" response += f"• Collaborators: {project_info['collaborators']}\n" response += f"• Content length: {project_info['current_content_length']} characters\n\n" response += f"**Contribution Statistics:**\n" response += f"• Total contributions: {contrib_stats['total_contributions']}\n" if contrib_stats['by_type']: response += f"• By type: {', '.join([f'{k}({v})' for k, v in contrib_stats['by_type'].items()])}\n" if contrib_stats['by_contributor']: response += f"• By contributor: {', '.join([f'{k}({v})' for k, v in contrib_stats['by_contributor'].items()])}\n" response += f"\n**Health Metrics:**\n" response += f"• Avg contributions/day: {health_metrics['avg_contributions_per_day']}\n" response += f"• Unique contributors: {health_metrics['unique_contributors']}\n" response += f"• Collaboration balance: {health_metrics['collaboration_balance']:.2f}\n" response += f"• Estimated completion: {health_metrics['completion_estimate']}\n" return [TextContent(type="text", text=response)] except Exception as e: log_error_with_context(e, {"tool": "get_project_analytics", "character": self.current_character}) return [TextContent(type="text", text=f"Error getting analytics: {str(e)}")] async def _get_pending_invitations(self, args: Dict[str, Any]) -> Sequence[TextContent]: """Handle pending invitations tool call""" try: # Get pending invitations for this character pending_invitations = [] for invitation in self.creative_manager.pending_invitations.values(): if invitation.invitee == self.current_character and invitation.status == "pending": if datetime.utcnow() <= invitation.expires_at: pending_invitations.append(invitation) if not pending_invitations: return [TextContent( type="text", text="📫 No pending project invitations at the moment." )] response = f"📮 You have {len(pending_invitations)} pending project invitation(s):\n\n" for invitation in pending_invitations: project = self.creative_manager.active_projects.get(invitation.project_id) project_title = project.title if project else "Unknown Project" response += f"**Invitation from {invitation.inviter}**\n" response += f"Project: {project_title}\n" response += f"Role: {invitation.role_description}\n" response += f"Message: {invitation.invitation_message}\n" response += f"Expires: {invitation.expires_at.strftime('%m/%d/%Y %H:%M')}\n" response += f"Invitation ID: {invitation.id}\n\n" response += "💡 Use 'respond_to_project_invitation' to accept or decline these invitations." return [TextContent(type="text", text=response)] except Exception as e: log_error_with_context(e, {"tool": "get_pending_invitations", "character": self.current_character}) return [TextContent(type="text", text=f"Error getting invitations: {str(e)}")] async def _search_projects(self, args: Dict[str, Any]) -> Sequence[TextContent]: """Handle project search tool call""" try: query = args.get("query", "") project_type_filter = args.get("project_type") status_filter = args.get("status") collaborator_filter = args.get("collaborator") # Simple search implementation - could be enhanced with vector search matching_projects = [] for project in self.creative_manager.active_projects.values(): # Check if character has access to this project if self.current_character not in project.collaborators: continue # Apply filters if project_type_filter and project.project_type.value != project_type_filter: continue if status_filter and project.status.value != status_filter: continue if collaborator_filter and collaborator_filter not in project.collaborators: continue # Text search in title and description if query: search_text = f"{project.title} {project.description}".lower() if query.lower() not in search_text: continue matching_projects.append(project) if not matching_projects: return [TextContent( type="text", text="🔍 No projects found matching your search criteria." )] response = f"🔍 Found {len(matching_projects)} matching project(s):\n\n" for project in matching_projects[:5]: # Limit to 5 results response += f"**{project.title}** ({project.project_type.value})\n" response += f"Status: {project.status.value} | Collaborators: {', '.join(project.collaborators)}\n" response += f"Description: {project.description[:100]}...\n" response += f"ID: {project.id}\n\n" if len(matching_projects) > 5: response += f"... and {len(matching_projects) - 5} more projects.\n" return [TextContent(type="text", text=response)] except Exception as e: log_error_with_context(e, {"tool": "search_projects", "character": self.current_character}) return [TextContent(type="text", text=f"Error searching projects: {str(e)}")] # Global creative projects MCP server creative_projects_mcp_server = None def get_creative_projects_mcp_server() -> CreativeProjectsMCPServer: global creative_projects_mcp_server return creative_projects_mcp_server def initialize_creative_projects_mcp_server(creative_manager: CollaborativeCreativeManager) -> CreativeProjectsMCPServer: global creative_projects_mcp_server creative_projects_mcp_server = CreativeProjectsMCPServer(creative_manager) return creative_projects_mcp_server