MCP Integration

class MCPApp:
    def resource(self, uri: str, name: str, description: str):
        """
        Register an MCP resource that AI agents can access.
        
        Args:
            uri: Unique resource identifier (URI format)
            name: Human-readable name for the resource
            description: Description of what the resource provides
            
        Usage:
            @app.mcp.resource("users://all", "User List", "Complete list of system users")
            def get_all_users():
                return {"users": [...]}
        """
    
    def tool(self, name: str, description: str, input_schema: dict):
        """
        Register an MCP tool that AI agents can invoke.
        
        Args:
            name: Tool name (used by agents to invoke the tool)
            description: Description of what the tool does
            input_schema: JSON schema defining tool input parameters
            
        Usage:
            @app.mcp.tool(
                "create_user",
                "Create a new user account",
                {
                    "type": "object",
                    "properties": {
                        "username": {"type": "string"},
                        "email": {"type": "string", "format": "email"}
                    },
                    "required": ["username", "email"]
                }
            )
            def create_user_tool(username, email):
                return {"user_id": 123, "message": "User created"}
        """
    
    def prompt(self, name: str, description: str):
        """
        Register an MCP prompt template for AI agents.
        
        Args:
            name: Prompt template name
            description: Description of the prompt's purpose
            
        Usage:
            @app.mcp.prompt("user_summary", "Generate a summary of user activity")
            def user_summary_prompt(user_id):
                return f"Summarize activity for user {user_id}"
        """

from robyn import Robyn

app = Robyn(__file__)

# Sample data
users_db = [
    {"id": 1, "username": "alice", "email": "alice@example.com", "active": True},
    {"id": 2, "username": "bob", "email": "bob@example.com", "active": False},
    {"id": 3, "username": "charlie", "email": "charlie@example.com", "active": True}
]

# Register MCP resources that AI agents can access
@app.mcp.resource("users://all", "All Users", "Complete list of all users in the system")
def get_all_users():
    """Resource providing all users"""
    return {"users": users_db, "count": len(users_db)}

@app.mcp.resource("users://active", "Active Users", "List of currently active users")
def get_active_users():
    """Resource providing only active users"""
    active_users = [user for user in users_db if user["active"]]
    return {"users": active_users, "count": len(active_users)}

@app.mcp.resource("system://stats", "System Statistics", "Current system statistics and metrics")
def get_system_stats():
    """Resource providing system statistics"""
    return {
        "total_users": len(users_db),
        "active_users": len([u for u in users_db if u["active"]]),
        "server_uptime": "5 days",
        "memory_usage": "256 MB"
    }

# Regular HTTP endpoints
@app.get("/users")
def list_users(request):
    return {"users": users_db}

app.start()

from robyn import Robyn
import json
import uuid

app = Robyn(__file__)

# In-memory data store
users_db = {}
tasks_db = {}

# MCP Tools - AI agents can invoke these
@app.mcp.tool(
    "create_user",
    "Create a new user account with username and email",
    {
        "type": "object",
        "properties": {
            "username": {
                "type": "string",
                "description": "Unique username for the account",
                "minLength": 3,
                "maxLength": 50
            },
            "email": {
                "type": "string",
                "format": "email",
                "description": "User's email address"
            },
            "full_name": {
                "type": "string",
                "description": "User's full name"
            }
        },
        "required": ["username", "email"]
    }
)
def create_user_tool(username, email, full_name=None):
    """Tool for creating users that AI agents can invoke"""
    user_id = str(uuid.uuid4())
    
    # Check if username already exists
    if any(user["username"] == username for user in users_db.values()):
        return {
            "success": False,
            "error": "Username already exists",
            "user_id": None
        }
    
    new_user = {
        "id": user_id,
        "username": username,
        "email": email,
        "full_name": full_name,
        "created_at": "2023-12-01T12:00:00Z",
        "active": True
    }
    
    users_db[user_id] = new_user
    
    return {
        "success": True,
        "message": "User created successfully",
        "user_id": user_id,
        "user": new_user
    }

@app.mcp.tool(
    "create_task",
    "Create a new task for a user",
    {
        "type": "object",
        "properties": {
            "user_id": {
                "type": "string",
                "description": "ID of the user to assign the task to"
            },
            "title": {
                "type": "string",
                "description": "Task title",
                "maxLength": 200
            },
            "description": {
                "type": "string",
                "description": "Detailed task description"
            },
            "priority": {
                "type": "string",
                "enum": ["low", "medium", "high"],
                "description": "Task priority level"
            }
        },
        "required": ["user_id", "title"]
    }
)
def create_task_tool(user_id, title, description=None, priority="medium"):
    """Tool for creating tasks that AI agents can invoke"""
    if user_id not in users_db:
        return {
            "success": False,
            "error": "User not found",
            "task_id": None
        }
    
    task_id = str(uuid.uuid4())
    new_task = {
        "id": task_id,
        "user_id": user_id,
        "title": title,
        "description": description,
        "priority": priority,
        "status": "pending",
        "created_at": "2023-12-01T12:00:00Z"
    }
    
    tasks_db[task_id] = new_task
    
    return {
        "success": True,
        "message": "Task created successfully",
        "task_id": task_id,
        "task": new_task
    }

@app.mcp.tool(
    "search_users",
    "Search for users by username or email",
    {
        "type": "object",
        "properties": {
            "query": {
                "type": "string",
                "description": "Search query to match against username or email",
                "minLength": 1
            },
            "active_only": {
                "type": "boolean",
                "description": "Only return active users",
                "default": False
            }
        },
        "required": ["query"]
    }
)
def search_users_tool(query, active_only=False):
    """Tool for searching users that AI agents can invoke"""
    query_lower = query.lower()
    results = []
    
    for user in users_db.values():
        if active_only and not user["active"]:
            continue
            
        if (query_lower in user["username"].lower() or 
            query_lower in user["email"].lower() or
            (user["full_name"] and query_lower in user["full_name"].lower())):
            results.append(user)
    
    return {
        "query": query,
        "results": results,
        "count": len(results),
        "active_only": active_only
    }

# Regular HTTP endpoints for web interface
@app.get("/users")
def list_users(request):
    return {"users": list(users_db.values())}

@app.get("/tasks")
def list_tasks(request):
    return {"tasks": list(tasks_db.values())}

app.start()

from robyn import Robyn
import json

app = Robyn(__file__)

# Sample data
users_db = {
    "1": {"username": "alice", "email": "alice@example.com", "role": "admin"},
    "2": {"username": "bob", "email": "bob@example.com", "role": "user"}
}

projects_db = {
    "proj1": {"name": "Website Redesign", "status": "active", "assigned_to": "1"},
    "proj2": {"name": "Mobile App", "status": "completed", "assigned_to": "2"}
}

# MCP Prompt Templates - AI agents can use these for generating content
@app.mcp.prompt("user_report", "Generate a comprehensive report about a user")
def user_report_prompt(user_id):
    """Generate prompt for creating user reports"""
    user = users_db.get(user_id)
    if not user:
        return "User not found"
    
    user_projects = [p for p in projects_db.values() if p["assigned_to"] == user_id]
    
    prompt = f"""
    Generate a comprehensive report for the following user:
    
    User Information:
    - Username: {user['username']}
    - Email: {user['email']}
    - Role: {user['role']}
    
    Assigned Projects: {len(user_projects)}
    {json.dumps(user_projects, indent=2)}
    
    Please include:
    1. User activity summary
    2. Project involvement analysis
    3. Performance assessment
    4. Recommendations for future assignments
    """
    
    return prompt.strip()

@app.mcp.prompt("project_summary", "Generate a summary of project status and progress")
def project_summary_prompt(project_id=None):
    """Generate prompt for project summaries"""
    if project_id:
        project = projects_db.get(project_id)
        if not project:
            return "Project not found"
        
        assigned_user = users_db.get(project["assigned_to"], {})
        
        prompt = f"""
        Generate a detailed summary for the following project:
        
        Project: {project['name']}
        Status: {project['status']}
        Assigned to: {assigned_user.get('username', 'Unknown')} ({assigned_user.get('email', 'Unknown')})
        
        Please analyze:
        1. Current progress and milestones
        2. Risk assessment
        3. Resource allocation
        4. Timeline adherence
        5. Recommendations for improvement
        """
    else:
        prompt = f"""
        Generate an executive summary of all projects:
        
        Total Projects: {len(projects_db)}
        
        Project Details:
        {json.dumps(projects_db, indent=2)}
        
        User Assignments:
        {json.dumps(users_db, indent=2)}
        
        Please provide:
        1. Overall project portfolio health
        2. Resource utilization analysis
        3. Risk factors across projects
        4. Strategic recommendations
        """
    
    return prompt.strip()

@app.mcp.prompt("onboarding_guide", "Generate personalized onboarding guide for new users")
def onboarding_guide_prompt(user_role="user", department=None):
    """Generate onboarding prompts for new team members"""
    prompt = f"""
    Create a comprehensive onboarding guide for a new team member with the following profile:
    
    Role: {user_role}
    Department: {department or "General"}
    
    Available System Features:
    - User Management
    - Project Tracking  
    - MCP AI Integration
    - Real-time WebSocket Communication
    
    Include:
    1. Welcome message and company overview
    2. Role-specific responsibilities and expectations
    3. System access and navigation guide
    4. Key contacts and resources
    5. First-week checklist and goals
    6. Training schedule and materials
    """
    
    return prompt.strip()

# MCP Resource that provides prompt context
@app.mcp.resource("prompts://context", "Prompt Context Data", "Data context for prompt generation")
def get_prompt_context():
    """Provide context data for prompt generation"""
    return {
        "users": users_db,
        "projects": projects_db,
        "system_info": {
            "total_users": len(users_db),
            "active_projects": len([p for p in projects_db.values() if p["status"] == "active"]),
            "features": ["User Management", "Project Tracking", "MCP Integration", "WebSocket Support"]
        }
    }

# Regular endpoints
@app.get("/")
def home(request):
    return {
        "message": "MCP-enabled Robyn application",
        "features": {
            "mcp_resources": 1,
            "mcp_tools": 3,
            "mcp_prompts": 3
        }
    }

app.start()

from robyn import Robyn
import json
import uuid
from datetime import datetime, timedelta

app = Robyn(__file__)

# Data stores
users_db = {}
articles_db = {}
analytics_db = {"views": 0, "registrations": 0}

# === MCP RESOURCES ===

@app.mcp.resource("content://articles", "All Articles", "Complete article database")
def get_all_articles():
    return {
        "articles": list(articles_db.values()),
        "count": len(articles_db),
        "last_updated": datetime.now().isoformat()
    }

@app.mcp.resource("analytics://dashboard", "Analytics Data", "Site analytics and metrics")
def get_analytics():
    return {
        "total_users": len(users_db),
        "total_articles": len(articles_db),
        "page_views": analytics_db["views"],
        "registrations": analytics_db["registrations"],
        "active_authors": len(set(a["author_id"] for a in articles_db.values()))
    }

# === MCP TOOLS ===

@app.mcp.tool(
    "publish_article",
    "Publish a new article with title, content, and tags",
    {
        "type": "object",
        "properties": {
            "title": {"type": "string", "description": "Article title"},
            "content": {"type": "string", "description": "Article content (markdown supported)"},
            "author_id": {"type": "string", "description": "ID of the article author"},
            "tags": {
                "type": "array",
                "items": {"type": "string"},
                "description": "Article tags for categorization"
            },
            "publish_immediately": {
                "type": "boolean", 
                "description": "Whether to publish immediately or save as draft",
                "default": True
            }
        },
        "required": ["title", "content", "author_id"]
    }
)
def publish_article_tool(title, content, author_id, tags=None, publish_immediately=True):
    if author_id not in users_db:
        return {"success": False, "error": "Author not found"}
    
    article_id = str(uuid.uuid4())
    article = {
        "id": article_id,
        "title": title,
        "content": content,
        "author_id": author_id,
        "author_name": users_db[author_id]["username"],
        "tags": tags or [],
        "status": "published" if publish_immediately else "draft",
        "created_at": datetime.now().isoformat(),
        "views": 0
    }
    
    articles_db[article_id] = article
    
    return {
        "success": True,
        "message": f"Article {'published' if publish_immediately else 'saved as draft'} successfully",
        "article_id": article_id,
        "article": article
    }

@app.mcp.tool(
    "moderate_content",
    "Review and moderate article content for policy compliance",
    {
        "type": "object",
        "properties": {
            "article_id": {"type": "string", "description": "ID of article to moderate"},
            "action": {
                "type": "string", 
                "enum": ["approve", "reject", "flag"],
                "description": "Moderation action to take"
            },
            "reason": {"type": "string", "description": "Reason for the moderation action"}
        },
        "required": ["article_id", "action"]
    }
)
def moderate_content_tool(article_id, action, reason=None):
    if article_id not in articles_db:
        return {"success": False, "error": "Article not found"}
    
    article = articles_db[article_id]
    
    if action == "approve":
        article["status"] = "published"
        article["moderated_at"] = datetime.now().isoformat()
    elif action == "reject":
        article["status"] = "rejected"
        article["rejection_reason"] = reason
    elif action == "flag":
        article["status"] = "flagged"
        article["flag_reason"] = reason
    
    return {
        "success": True,
        "message": f"Article {action}ed successfully",
        "article_id": article_id,
        "new_status": article["status"]
    }

# === MCP PROMPTS ===

@app.mcp.prompt("content_review", "Generate content review and improvement suggestions")
def content_review_prompt(article_id):
    if article_id not in articles_db:
        return "Article not found"
    
    article = articles_db[article_id]
    
    return f"""
    Please review the following article and provide detailed feedback:
    
    Title: {article['title']}
    Author: {article['author_name']}
    Tags: {', '.join(article['tags'])}
    Word Count: {len(article['content'].split())}
    
    Content:
    {article['content']}
    
    Please analyze:
    1. Content quality and readability
    2. SEO optimization opportunities
    3. Factual accuracy and citations
    4. Engagement potential
    5. Specific improvement recommendations
    """

@app.mcp.prompt("author_performance", "Generate author performance analysis")
def author_performance_prompt(author_id):
    if author_id not in users_db:
        return "Author not found"
    
    author = users_db[author_id]
    author_articles = [a for a in articles_db.values() if a["author_id"] == author_id]
    total_views = sum(a["views"] for a in author_articles)
    
    return f"""
    Generate a comprehensive performance analysis for author: {author['username']}
    
    Author Statistics:
    - Total Articles: {len(author_articles)}
    - Total Views: {total_views}
    - Average Views per Article: {total_views / len(author_articles) if author_articles else 0:.1f}
    - Join Date: {author.get('created_at', 'Unknown')}
    
    Article Breakdown:
    {json.dumps([{
        'title': a['title'], 
        'status': a['status'],
        'views': a['views'],
        'tags': a['tags']
    } for a in author_articles], indent=2)}
    
    Please provide:
    1. Performance overview and trends
    2. Content quality assessment
    3. Engagement analysis
    4. Areas for improvement
    5. Recognition and achievements
    """

# Regular HTTP endpoints
@app.post("/register")
def register(request):
    data = request.json()
    user_id = str(uuid.uuid4())
    
    users_db[user_id] = {
        "id": user_id,
        "username": data["username"],
        "email": data["email"],
        "created_at": datetime.now().isoformat()
    }
    
    analytics_db["registrations"] += 1
    
    return {"message": "User registered", "user_id": user_id}

@app.get("/articles")
def list_articles(request):
    analytics_db["views"] += 1
    return {"articles": list(articles_db.values())}

@app.get("/articles/<article_id>")
def get_article(request):
    article_id = request.path_params["article_id"]
    if article_id in articles_db:
        articles_db[article_id]["views"] += 1
        analytics_db["views"] += 1
        return {"article": articles_db[article_id]}
    return Response(404, {}, {"error": "Article not found"})

app.start()