ai-skills-api/mcp/skills.py

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

MCP_INSTRUCTIONS = """You have access to a shared knowledge base (skills) and project memory via these tools:

- get_context(project): Fetch relevant skills and project conventions for the given project identifier. Always call this before answering to load relevant context.
- search_skills(query): Find skills by content/tags.
- get_memory(project): Retrieve past decisions/learnings for a project.
- add_memory(project, key, content): Store a decision or learning. ASK permission first.
- create_skill(id, name, content, ...): Create a reusable skill. ASK permission first.

CRITICAL: When you identify an opportunity to store something (a decision, pattern, convention), you MUST propose it and get explicit user confirmation before calling add_memory or create_skill. Examples:
- "I decided to use PostgreSQL. Want me to save that to memory?"
- "That's a useful pattern. I could create a reusable skill for it. Should I?"

Project conventions are scoped to a project identifier (recommended: git remote origin URL like https://github.com/username/repo). This ensures your knowledge follows you across machines. Use your shell tool to detect the git remote locally: `git remote get-url origin`.

Always pass the same project identifier consistently across sessions."""

# Configure transport security to avoid 421 errors (DNS rebinding protection)
# Set MCP_ENABLE_DNS_PROTECTION=true to enable with custom hosts
enable_protection = os.getenv("MCP_ENABLE_DNS_PROTECTION", "false").lower() == "true"
if enable_protection:
    allowed_hosts = os.getenv("MCP_ALLOWED_HOSTS", "localhost:*,127.0.0.1:*,0.0.0.0:*").split(",")
    allowed_origins = os.getenv("MCP_ALLOWED_ORIGINS", "http://localhost:*,http://127.0.0.1:*,http://0.0.0.0:*").split(",")
    transport_security = TransportSecuritySettings(
        enable_dns_rebinding_protection=True,
        allowed_hosts=allowed_hosts,
        allowed_origins=allowed_origins,
    )
else:
    transport_security = TransportSecuritySettings(
        enable_dns_rebinding_protection=False,
    )

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

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()