ai-skills-api/mcp/skills.py

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

mcp = FastMCP("skills")

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 check_cache(prompt: str, model: str | None = None) -> dict | None:
    """Check if a response is cached for this prompt"""
    try:
        with httpx.Client() as client:
            response = client.post(
                f"{SKILLS_API_URL}/cache/lookup",
                json={"prompt": prompt, "model": model}
            )
            response.raise_for_status()
            return response.json()
    except httpx.HTTPError as e:
        return {"error": f"Failed to check cache: {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__":
    mcp.run()