server.pyā¢6.61 kB
#!/usr/bin/env python3
"""
Skills Registry MCP Server
Provides intelligent discovery and management of Claude Skills
"""
import os
from typing import List, Optional, Dict, Any
from fastmcp import FastMCP
from .database import Database
from .search import SearchEngine
from .models import SkillMetadata, SkillSearchResult
# Initialize FastMCP server
mcp = FastMCP("Skills Registry")
# Initialize database and search engine
db = Database(os.getenv("DATABASE_URL"))
search = SearchEngine(db, os.getenv("OPENAI_API_KEY"), os.getenv("ANTHROPIC_API_KEY"))
@mcp.tool()
async def skill_search(
query: str,
tags: Optional[List[str]] = None,
category: Optional[str] = None,
min_rating: Optional[float] = None,
ai_generated: Optional[bool] = None,
limit: int = 10
) -> List[SkillSearchResult]:
"""
Search for skills using natural language or filters.
Args:
query: Natural language search query
tags: Filter by specific tags
category: Filter by category (e.g., "documents", "data")
min_rating: Minimum average rating (1-5)
ai_generated: Filter by AI-generated vs human-authored
limit: Maximum number of results to return
Returns:
List of matching skills with relevance scores
"""
results = await search.search(
query=query,
tags=tags,
category=category,
min_rating=min_rating,
ai_generated=ai_generated,
limit=limit
)
return results
@mcp.tool()
async def skill_get(skill_id: str, user_id: Optional[str] = None) -> Dict[str, Any]:
"""
Fetch a specific skill's full content and metadata.
Args:
skill_id: Unique identifier for the skill
user_id: Optional user ID to track usage
Returns:
Complete skill data including SKILL.md content
"""
skill = await db.get_skill(skill_id)
if not skill:
return {"error": f"Skill '{skill_id}' not found"}
# Track usage
if user_id:
await db.track_usage(skill_id, user_id)
return skill.dict()
@mcp.tool()
async def skill_list_categories() -> Dict[str, Any]:
"""
Browse skills organized by category hierarchy.
Returns:
Tree structure of categories with skill counts
"""
categories = await db.get_categories()
return {"categories": categories}
@mcp.tool()
async def skill_favorite_add(skill_id: str, user_id: str) -> Dict[str, str]:
"""
Add a skill to user's favorites.
Args:
skill_id: Skill to favorite
user_id: User identifier
Returns:
Confirmation message
"""
success = await db.add_favorite(skill_id, user_id)
if success:
return {"status": "success", "message": f"Added '{skill_id}' to favorites"}
else:
return {"status": "error", "message": f"Skill '{skill_id}' not found"}
@mcp.tool()
async def skill_favorite_list(user_id: str) -> List[SkillMetadata]:
"""
Get user's favorite skills.
Args:
user_id: User identifier
Returns:
List of favorited skills with metadata
"""
favorites = await db.get_favorites(user_id)
return favorites
@mcp.tool()
async def skill_favorite_remove(skill_id: str, user_id: str) -> Dict[str, str]:
"""
Remove a skill from user's favorites.
Args:
skill_id: Skill to unfavorite
user_id: User identifier
Returns:
Confirmation message
"""
success = await db.remove_favorite(skill_id, user_id)
if success:
return {"status": "success", "message": f"Removed '{skill_id}' from favorites"}
else:
return {"status": "error", "message": "Favorite not found"}
@mcp.tool()
async def skill_rate(
skill_id: str,
user_id: str,
rating: int,
review: Optional[str] = None
) -> Dict[str, str]:
"""
Rate a skill and optionally leave a review.
Args:
skill_id: Skill to rate
user_id: User identifier
rating: Rating from 1-5
review: Optional text review
Returns:
Confirmation message with updated average rating
"""
if rating < 1 or rating > 5:
return {"status": "error", "message": "Rating must be between 1 and 5"}
success, avg_rating = await db.rate_skill(skill_id, user_id, rating, review)
if success:
return {
"status": "success",
"message": f"Rated '{skill_id}' {rating}/5",
"average_rating": f"{avg_rating:.1f}"
}
else:
return {"status": "error", "message": f"Skill '{skill_id}' not found"}
@mcp.tool()
async def skill_trending(
limit: int = 10,
timeframe: str = "week"
) -> List[Dict[str, Any]]:
"""
Get currently trending skills.
Args:
limit: Number of skills to return
timeframe: Time period - "day", "week", or "month"
Returns:
List of trending skills with usage statistics
"""
if timeframe not in ["day", "week", "month"]:
return {"error": "Timeframe must be 'day', 'week', or 'month'"}
trending = await db.get_trending(limit, timeframe)
return trending
@mcp.tool()
async def skill_upload(
name: str,
description: str,
skill_md_content: str,
category: str,
tags: List[str],
author_id: str,
visibility: str = "private",
ai_generated: bool = False
) -> Dict[str, str]:
"""
Upload a new custom skill.
Args:
name: Skill name
description: Brief description
skill_md_content: Full SKILL.md content
category: Category (e.g., "documents/pdf")
tags: List of tags
author_id: Author identifier
visibility: "public" or "private"
ai_generated: Whether skill was AI-generated
Returns:
Skill ID and confirmation
"""
if visibility not in ["public", "private"]:
return {"error": "Visibility must be 'public' or 'private'"}
skill_id = await db.create_skill(
name=name,
description=description,
skill_md_content=skill_md_content,
category=category,
tags=tags,
author_id=author_id,
visibility=visibility,
ai_generated=ai_generated
)
# Generate embedding for the new skill
await search.generate_embedding(skill_id)
return {
"status": "success",
"skill_id": skill_id,
"message": f"Created skill '{name}' with ID: {skill_id}"
}
if __name__ == "__main__":
# Run the MCP server
mcp.run()