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 each query
- 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, ...)
- Retrieve project memory using get_memory(project)

CRITICAL: You get better over time, but you must ASK before creating new entries.

When you identify an opportunity to learn:
1. **Recognize**: You've made a decision, discovered a pattern, or learned something project-specific
2. **Propose**: Say "I can create a skill/memory for this" and explain what it would contain
3. **Await confirmation**: Wait for user to say "yes" or "create it" before calling the tool
4. **Execute**: Only after explicit permission, call add_memory() or create_skill()

Examples:
- "I decided to use PostgreSQL for this project. Want me to save that to memory?"
- "That's a useful Docker Compose pattern. I could create a reusable skill for it. Should I?"
- "This seems like a project-specific convention. Want me to record it as a convention?"

This respects user control while still building the knowledge base.

**Project identification**: At the start of each session, ask the user: "What is your project identifier? Use the git remote origin URL if available (e.g., https://github.com/username/repo) for cross-machine consistency. If not a git repo, use a unique name." Store this identifier and use it for all subsequent `project` parameters in get_context, add_memory, create_skill, etc.

If the user doesn't know their git remote, instruct them to run locally: `git remote get-url origin`

If they need to initialize a git repo or set up a new project with AI Skills, guide them to the "Project Setup Guide" skill (search for it) or have them clone the agentic-templates repository.

Project conventions are scoped to this project identifier, not filesystem path. This ensures your knowledge follows you across machines.

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

Never create skills/memories silently. Always propose and get explicit confirmation."""

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