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