ai-skills-api/mcp/skills.py

from mcp.server.fastmcp import FastMCP
import httpx
import os
import uvicorn

MCP_INSTRUCTIONS = """You are a coding assistant with a shared knowledge base (skills) and project memory.

Your capabilities:
- Fetch relevant skills and project conventions using get_context(project) before answering
- Search for skills using search_skills(query)
- Store decisions and learnings using add_memory(project, key, content)
- Create reusable patterns using create_skill(id, name, content, ...)

CRITICAL: You get better over time. When you:
- Make an architecture decision → add_memory()
- Discover a best practice → create_skill()
- Start a new task → get_context() to retrieve past learnings

This compounds: future queries will benefit from what you store now.

Project conventions are scoped to a project path. Always pass the project path to get_context() and add_memory().

You are operating across a team/home lab. Use conventions to encode project standards. Use skills for reusable technical patterns.

Start every coding session by calling get_context(project) to load what you already know about this project."""

mcp = FastMCP("skills", instructions=MCP_INSTRUCTIONS)

SKILLS_API_URL = os.getenv("SKILLS_API_URL", "http://helm:8675")


@mcp.tool()
def get_skill(skill_id: str) -> dict:
    """Get a skill by ID from the skills database"""
    try:
        with httpx.Client() as client:
            response = client.get(f"{SKILLS_API_URL}/skills/{skill_id}")
            response.raise_for_status()
            return response.json()
    except httpx.HTTPError as e:
        return {"error": f"Failed to fetch skill: {e}"}


@mcp.tool()
def search_skills(query: str, category: str | None = None) -> list[dict]:
    """Search skills by query"""
    try:
        with httpx.Client() as client:
            params = {"q": query}
            if category:
                params["category"] = category
            response = client.get(f"{SKILLS_API_URL}/skills/search", params=params)
            response.raise_for_status()
            return response.json()
    except httpx.HTTPError as e:
        return [{"error": f"Failed to search skills: {e}"}]


@mcp.tool()
def list_skills(category: str | None = None) -> list[dict]:
    """List all skills, optionally filtered by category"""
    try:
        with httpx.Client() as client:
            params = {}
            if category:
                params["category"] = category
            response = client.get(f"{SKILLS_API_URL}/skills", params=params)
            response.raise_for_status()
            return response.json()
    except httpx.HTTPError as e:
        return [{"error": f"Failed to list skills: {e}"}]


@mcp.tool()
def get_context(project: str | None = None, skills: list[str] | None = None) -> dict:
    """Get context bundle for a project"""
    try:
        with httpx.Client() as client:
            params = {}
            if project:
                params["project"] = project
            if skills:
                params["skills"] = ",".join(skills)
            response = client.get(f"{SKILLS_API_URL}/context", params=params)
            response.raise_for_status()
            return response.json()
    except httpx.HTTPError as e:
        return {"error": f"Failed to fetch context: {e}"}


@mcp.tool()
def get_conventions(project: str | None = None) -> list[dict]:
    """Get conventions for a project"""
    try:
        with httpx.Client() as client:
            params = {}
            if project:
                params["project"] = project
            response = client.get(f"{SKILLS_API_URL}/conventions", params=params)
            response.raise_for_status()
            return response.json()
    except httpx.HTTPError as e:
        return [{"error": f"Failed to fetch conventions: {e}"}]


@mcp.tool()
def get_snippets(category: str | None = None, language: str | None = None) -> list[dict]:
    """Get code snippets"""
    try:
        with httpx.Client() as client:
            params = {}
            if category:
                params["category"] = category
            if language:
                params["language"] = language
            response = client.get(f"{SKILLS_API_URL}/snippets", params=params)
            response.raise_for_status()
            return response.json()
    except httpx.HTTPError as e:
        return [{"error": f"Failed to fetch snippets: {e}"}]


@mcp.tool()
def get_memory(project: str) -> list[dict]:
    """Get memory entries for a project"""
    try:
        with httpx.Client() as client:
            params = {"project": project}
            response = client.get(f"{SKILLS_API_URL}/memory", params=params)
            response.raise_for_status()
            return response.json()
    except httpx.HTTPError as e:
        return [{"error": f"Failed to fetch memory: {e}"}]


@mcp.tool()
def add_memory(project: str, key: str, content: str) -> dict:
    """Add a memory entry for a project"""
    import uuid
    try:
        with httpx.Client() as client:
            response = client.post(
                f"{SKILLS_API_URL}/memory",
                json={
                    "id": str(uuid.uuid4())[:8],
                    "project": project,
                    "key": key,
                    "content": content
                }
            )
            response.raise_for_status()
            return response.json()
    except httpx.HTTPError as e:
        return {"error": f"Failed to add memory: {e}"}


@mcp.tool()
def create_skill(
    id: str,
    name: str,
    content: str,
    category: str | None = None,
    description: str | None = None,
    tags: list[str] | None = None
) -> dict:
    """Create a new skill"""
    try:
        with httpx.Client() as client:
            response = client.post(
                f"{SKILLS_API_URL}/skills",
                json={
                    "id": id,
                    "name": name,
                    "content": content,
                    "category": category,
                    "description": description,
                    "tags": tags
                }
            )
            response.raise_for_status()
            return response.json()
    except httpx.HTTPError as e:
        return {"error": f"Failed to create skill: {e}"}


if __name__ == "__main__":
    transport = os.getenv("MCP_TRANSPORT", "stdio")
    
    if transport == "sse":
        host = os.getenv("MCP_HOST", "0.0.0.0")
        port = int(os.getenv("MCP_PORT", "3000"))
        app = mcp.sse_app()
        uvicorn.run(app, host=host, port=port)
    else:
        mcp.run()