Shopify Liquid MCP Server

"""Shopify Liquid Documentation MCP Server."""

import logging
from typing import List, Optional

from fastmcp import FastMCP

from .ingest import (
    index_documentation,
    search_documentation,
    get_by_category,
    get_document,
)

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

# Initialize FastMCP server
mcp = FastMCP("Shopify Liquid Docs")


@mcp.tool()
def search_liquid_docs(queries: List[str]) -> str:
    """Search Shopify Liquid documentation using full-text search.

    Args:
        queries: List of search terms (maximum 3)

    Returns:
        Formatted search results with snippets
    """
    if not queries:
        return "Error: Please provide at least one search query"

    # Limit to 3 queries
    queries = queries[:3]

    logger.info(f"Searching for: {queries}")
    results = search_documentation(queries, limit=10)

    if not results:
        return f"No results found for: {', '.join(queries)}"

    # Format results
    output = [f"Found {len(results)} results for: {', '.join(queries)}\n"]

    for i, doc in enumerate(results, 1):
        output.append(f"{i}. **{doc['title']}** ({doc['category']})")
        output.append(f"   Path: {doc['path']}")
        if doc.get("snippet"):
            output.append(f"   {doc['snippet']}")
        output.append("")

    return "\n".join(output)


@mcp.tool()
def get_liquid_tag(tag_name: str) -> str:
    """Get documentation for a specific Shopify Liquid tag.

    Args:
        tag_name: Name of the tag (e.g., 'if', 'for', 'assign')

    Returns:
        Complete tag documentation
    """
    doc = get_document("tags", tag_name)

    if not doc:
        return (
            f"Tag '{tag_name}' not found. Use list_liquid_tags() to see available tags."
        )

    return doc["content"]


@mcp.tool()
def get_liquid_filter(filter_name: str) -> str:
    """Get documentation for a specific Shopify Liquid filter.

    Args:
        filter_name: Name of the filter (e.g., 'upcase', 'date', 'money')

    Returns:
        Complete filter documentation
    """
    doc = get_document("filters", filter_name)

    if not doc:
        return f"Filter '{filter_name}' not found. Use list_liquid_filters() to see available filters."

    return doc["content"]


@mcp.tool()
def get_liquid_object(object_name: str) -> str:
    """Get documentation for a specific Shopify Liquid object.

    Args:
        object_name: Name of the object (e.g., 'product', 'cart', 'shop')

    Returns:
        Complete object documentation
    """
    doc = get_document("objects", object_name)

    if not doc:
        return f"Object '{object_name}' not found. Use list_liquid_objects() to see available objects."

    return doc["content"]


@mcp.tool()
def list_liquid_tags() -> str:
    """List all available Shopify Liquid tags.

    Returns:
        List of all tag names with titles
    """
    docs = get_by_category("tags")

    if not docs:
        return "No tags found in database"

    output = [f"Available Liquid Tags ({len(docs)} total):\n"]

    # Group by common categories
    control_flow = []
    iteration = []
    variables = []
    theme = []
    other = []

    for doc in docs:
        name = doc["name"]
        title = doc["title"]
        item = f"- **{name}**: {title}"

        if name in ["if", "unless", "case", "conditional-else"]:
            control_flow.append(item)
        elif name in [
            "for",
            "break",
            "continue",
            "cycle",
            "tablerow",
            "iteration-else",
            "paginate",
        ]:
            iteration.append(item)
        elif name in ["assign", "capture", "increment", "decrement", "echo"]:
            variables.append(item)
        elif name in [
            "layout",
            "section",
            "sections",
            "render",
            "include",
            "content_for",
        ]:
            theme.append(item)
        else:
            other.append(item)

    if control_flow:
        output.append("**Control Flow:**")
        output.extend(control_flow)
        output.append("")

    if iteration:
        output.append("**Iteration:**")
        output.extend(iteration)
        output.append("")

    if variables:
        output.append("**Variables:**")
        output.extend(variables)
        output.append("")

    if theme:
        output.append("**Theme:**")
        output.extend(theme)
        output.append("")

    if other:
        output.append("**Other:**")
        output.extend(other)

    return "\n".join(output)


@mcp.tool()
def list_liquid_filters() -> str:
    """List all available Shopify Liquid filters.

    Returns:
        List of all filter names with titles organized by category
    """
    docs = get_by_category("filters")

    if not docs:
        return "No filters found in database"

    output = [f"Available Liquid Filters ({len(docs)} total):\n"]

    # Group by category
    string_filters = []
    array_filters = []
    math_filters = []
    money_filters = []
    other_filters = []

    string_keywords = [
        "case",
        "capitalize",
        "escape",
        "strip",
        "truncate",
        "replace",
        "remove",
        "append",
        "prepend",
        "split",
        "lstrip",
        "rstrip",
    ]
    array_keywords = [
        "concat",
        "compact",
        "first",
        "last",
        "join",
        "map",
        "reverse",
        "size",
        "sort",
        "uniq",
        "where",
    ]
    math_keywords = [
        "abs",
        "at_least",
        "at_most",
        "ceil",
        "floor",
        "minus",
        "plus",
        "times",
        "divided_by",
        "modulo",
        "round",
    ]
    money_keywords = ["money"]

    for doc in docs:
        name = doc["name"]
        title = doc["title"]
        item = f"- **{name}**: {title}"

        if any(kw in name for kw in string_keywords):
            string_filters.append(item)
        elif any(kw in name for kw in array_keywords):
            array_filters.append(item)
        elif any(kw in name for kw in math_keywords):
            math_filters.append(item)
        elif any(kw in name for kw in money_keywords):
            money_filters.append(item)
        else:
            other_filters.append(item)

    if string_filters:
        output.append("**String Filters:**")
        output.extend(string_filters[:15])  # Show first 15
        if len(string_filters) > 15:
            output.append(f"  ... and {len(string_filters) - 15} more")
        output.append("")

    if array_filters:
        output.append("**Array Filters:**")
        output.extend(array_filters)
        output.append("")

    if math_filters:
        output.append("**Math Filters:**")
        output.extend(math_filters)
        output.append("")

    if money_filters:
        output.append("**Money Filters:**")
        output.extend(money_filters)
        output.append("")

    if other_filters:
        output.append("**Other Filters:**")
        output.extend(other_filters[:15])
        if len(other_filters) > 15:
            output.append(f"  ... and {len(other_filters) - 15} more")

    return "\n".join(output)


@mcp.tool()
def list_liquid_objects() -> str:
    """List all available Shopify Liquid objects.

    Returns:
        List of all object names with titles organized by category
    """
    docs = get_by_category("objects")

    if not docs:
        return "No objects found in database"

    output = [f"Available Liquid Objects ({len(docs)} total):\n"]

    # Group by category
    core = []
    product_related = []
    cart_related = []
    customer_related = []
    content = []
    other_objects = []

    for doc in docs:
        name = doc["name"]
        title = doc["title"]
        item = f"- **{name}**: {title}"

        if name in ["shop", "settings", "theme", "request", "routes"]:
            core.append(item)
        elif "product" in name or "variant" in name or "collection" in name:
            product_related.append(item)
        elif "cart" in name or "checkout" in name or "line_item" in name:
            cart_related.append(item)
        elif "customer" in name or "company" in name:
            customer_related.append(item)
        elif name in ["page", "blog", "article", "articles", "comment"]:
            content.append(item)
        else:
            other_objects.append(item)

    if core:
        output.append("**Core Objects:**")
        output.extend(core)
        output.append("")

    if product_related:
        output.append("**Product Related:**")
        output.extend(product_related)
        output.append("")

    if cart_related:
        output.append("**Cart & Checkout:**")
        output.extend(cart_related)
        output.append("")

    if customer_related:
        output.append("**Customer Related:**")
        output.extend(customer_related)
        output.append("")

    if content:
        output.append("**Content:**")
        output.extend(content)
        output.append("")

    if other_objects:
        output.append("**Other Objects:**")
        output.extend(other_objects[:15])
        if len(other_objects) > 15:
            output.append(f"  ... and {len(other_objects) - 15} more")

    return "\n".join(output)


def main():
    """Main entry point for the MCP server."""
    # Index documentation on startup if not already indexed
    logger.info("Initializing Shopify Liquid Documentation MCP Server...")

    try:
        count = index_documentation(force=False)
        logger.info(f"Documentation indexed: {count} documents")
    except Exception as e:
        logger.error(f"Error indexing documentation: {e}")
        logger.info("Server will start but search may not work properly")

    # Run the MCP server
    mcp.run()


if __name__ == "__main__":
    main()