Shopify Python MCP Server

  • src
  • shopify_py_mcp
import asyncio import os import json import shopify import time from mcp.server.models import InitializationOptions import mcp.types as types from mcp.server import NotificationOptions, Server from pydantic import AnyUrl import mcp.server.stdio # Shopify API設定 SHOP_URL = os.environ.get("SHOPIFY_SHOP_URL", "") API_KEY = os.environ.get("SHOPIFY_API_KEY", "") API_PASSWORD = os.environ.get("SHOPIFY_API_PASSWORD", "") API_VERSION = os.environ.get("SHOPIFY_API_VERSION", "2025-01") # Shopify APIの初期化 def initialize_shopify_api(): shop_url = f"https://{API_KEY}:{API_PASSWORD}@{SHOP_URL}/admin/api/{API_VERSION}" shopify.ShopifyResource.set_site(shop_url) server = Server("shopify-py-mcp") def get_all_shopify_products(total_limit=None, per_page_limit=250): """ ShopifyAPIライブラリを使用して複数ページにわたる商品一覧を取得する関数 Parameters: total_limit (int): 取得する総商品数(Noneの場合はすべての商品を取得) per_page_limit (int): 1回のリクエストあたりの商品数(最大250) Returns: list: 商品のリスト """ # 1ページあたりの上限を250に制限 per_page_limit = min(per_page_limit, 250) all_products = [] next_page_url = None try: while True: # 既に十分な商品が取得されているか確認 if total_limit is not None and len(all_products) >= total_limit: break # 残りの取得数を計算 current_limit = per_page_limit if total_limit is not None: current_limit = min(per_page_limit, total_limit - len(all_products)) if current_limit <= 0: break # 商品一覧の取得 if next_page_url: # next_page_urlからpage_infoを抽出 page_info = extract_page_info(next_page_url) products = shopify.Product.find( limit=current_limit, page_info=page_info ) else: products = shopify.Product.find(limit=current_limit) # 結果が空の場合は終了 if not products: break # 取得した商品を追加 all_products.extend(products) # レスポンスヘッダーからページネーション情報を取得 response_headers = shopify.ShopifyResource.connection.response.headers link_header = response_headers.get("Link", "") # 次のページURLを抽出 next_page_url = extract_next_page_url(link_header) if not next_page_url: break # レート制限を避けるために少し待機 time.sleep(0.5) except Exception as e: print(f"エラーが発生しました: {e}") # total_limitが指定されている場合、指定した数だけ返す if total_limit is not None: return all_products[:total_limit] return all_products def extract_next_page_url(link_header): """ Linkヘッダーから次のページのURLを抽出する Parameters: link_header (str): レスポンスのLinkヘッダー Returns: str: 次のページのURL(存在しない場合はNone) """ if not link_header: return None links = link_header.split(",") for link in links: parts = link.split(";") if len(parts) != 2: continue url = parts[0].strip().strip("<>") rel = parts[1].strip() if 'rel="next"' in rel: return url return None def extract_page_info(next_page_url): """ URLからpage_infoパラメータを抽出する Parameters: next_page_url (str): 次のページのURL Returns: str: page_infoパラメータ """ import urllib.parse parsed_url = urllib.parse.urlparse(next_page_url) query_params = urllib.parse.parse_qs(parsed_url.query) if "page_info" in query_params: return query_params["page_info"][0] return None @server.list_tools() async def handle_list_tools() -> list[types.Tool]: """ 利用可能なツールのリストを返します。 各ツールはJSON Schemaを使用して引数を指定します。 """ return [ types.Tool( name="list_products", description="商品一覧を取得する", inputSchema={ "type": "object", "properties": { "limit": { "type": "number", "description": "取得する商品数(最大250)", "minimum": 1, "maximum": 250, "default": 50, }, }, }, ), types.Tool( name="get_product", description="商品の詳細情報を取得する", inputSchema={ "type": "object", "properties": { "product_id": {"type": "number", "description": "商品ID"} }, "required": ["product_id"], }, ), types.Tool( name="create_product", description="新しい商品を作成する", inputSchema={ "type": "object", "properties": { "title": {"type": "string", "description": "商品名"}, "body_html": { "type": "string", "description": "商品の説明(HTML形式)", }, "vendor": {"type": "string", "description": "ベンダー名"}, "product_type": {"type": "string", "description": "商品タイプ"}, "tags": {"type": "string", "description": "タグ(カンマ区切り)"}, "status": { "type": "string", "description": "ステータス", "enum": ["active", "draft", "archived"], "default": "active", }, "variants": { "type": "array", "description": "バリエーション", "items": { "type": "object", "properties": { "price": {"type": "string", "description": "価格"}, "sku": {"type": "string", "description": "SKU"}, "inventory_quantity": { "type": "number", "description": "在庫数", }, "option1": { "type": "string", "description": "オプション1の値", }, "option2": { "type": "string", "description": "オプション2の値", }, "option3": { "type": "string", "description": "オプション3の値", }, }, "required": ["price"], }, }, "options": { "type": "array", "description": "オプション", "items": { "type": "object", "properties": { "name": { "type": "string", "description": "オプション名", }, "position": { "position": "number", "description": "オプション順番", }, "values": { "type": "array", "description": "オプション値", "items": {"type": "string"}, }, }, "required": ["name", "position", "values"], }, }, "images": { "type": "array", "description": "画像", "items": { "type": "object", "properties": { "src": {"type": "string", "description": "画像URL"}, "alt": { "type": "string", "description": "代替テキスト", }, }, "required": ["src"], }, }, }, "required": ["title"], }, ), types.Tool( name="update_product", description="商品を更新する", inputSchema={ "type": "object", "properties": { "product_id": {"type": "number", "description": "商品ID"}, "title": {"type": "string", "description": "商品名"}, "body_html": { "type": "string", "description": "商品の説明(HTML形式)", }, "vendor": {"type": "string", "description": "ベンダー名"}, "product_type": {"type": "string", "description": "商品タイプ"}, "tags": {"type": "string", "description": "タグ(カンマ区切り)"}, "status": { "type": "string", "description": "ステータス", "enum": ["active", "draft", "archived"], }, "variants": { "type": "array", "description": "バリエーション", "items": { "type": "object", "properties": { "id": { "type": "number", "description": "バリエーションID", }, "price": {"type": "string", "description": "価格"}, "sku": {"type": "string", "description": "SKU"}, "inventory_quantity": { "type": "number", "description": "在庫数", }, "option1": { "type": "string", "description": "オプション1の値", }, "option2": { "type": "string", "description": "オプション2の値", }, "option3": { "type": "string", "description": "オプション3の値", }, }, }, }, "options": { "type": "array", "description": "オプション", "items": { "type": "object", "properties": { "id": {"type": "number", "description": "オプションID"}, "name": { "type": "string", "description": "オプション名", }, "position": { "position": "number", "description": "オプション順番", }, "values": { "type": "array", "description": "オプション値", "items": {"type": "string"}, }, }, "required": ["name", "values"], }, }, "images": { "type": "array", "description": "画像", "items": { "type": "object", "properties": { "id": {"type": "number", "description": "画像ID"}, "src": {"type": "string", "description": "画像URL"}, "alt": { "type": "string", "description": "代替テキスト", }, }, "required": ["src"], }, }, }, "required": ["product_id"], }, ), types.Tool( name="delete_product", description="商品を削除する", inputSchema={ "type": "object", "properties": { "product_id": {"type": "number", "description": "商品ID"} }, "required": ["product_id"], }, ), ] @server.call_tool() async def handle_call_tool( name: str, arguments: dict | None ) -> list[types.TextContent | types.ImageContent | types.EmbeddedResource]: """ ツール実行リクエストを処理します。 """ try: initialize_shopify_api() if name == "list_products": return await handle_list_products(arguments or {}) elif name == "get_product": return await handle_get_product(arguments or {}) elif name == "create_product": return await handle_create_product(arguments or {}) elif name == "update_product": return await handle_update_product(arguments or {}) elif name == "delete_product": return await handle_delete_product(arguments or {}) else: raise ValueError(f"Unknown tool: {name}") except Exception as e: return [ types.TextContent( type="text", text=f"エラーが発生しました: {str(e)}", ) ] async def handle_list_products(arguments: dict) -> list[types.TextContent]: """商品一覧を取得する""" limit = int(arguments.get("limit", 50)) products = get_all_shopify_products( total_limit=limit, per_page_limit=250 # 1回のリクエストで最大250件取得 ) result = [] for product in products: result.append( { "id": product.id, "title": product.title, "vendor": product.vendor, "product_type": product.product_type, "created_at": product.created_at, "updated_at": product.updated_at, "status": product.status, "variants_count": len(product.variants), "images_count": len(product.images), } ) return [ types.TextContent( type="text", text=json.dumps(result, indent=2, ensure_ascii=False), ) ] async def handle_get_product(arguments: dict) -> list[types.TextContent]: """商品の詳細情報を取得する""" product_id = arguments.get("product_id") if not product_id: raise ValueError("product_id is required") product = shopify.Product.find(product_id) # 商品情報を整形 result = { "id": product.id, "title": product.title, "body_html": product.body_html, "vendor": product.vendor, "product_type": product.product_type, "created_at": product.created_at, "updated_at": product.updated_at, "status": product.status, "tags": product.tags, "variants": [], "options": [], "images": [], } # バリエーション情報 for variant in product.variants: result["variants"].append( { "id": variant.id, "title": variant.title, "price": variant.price, "sku": variant.sku, "inventory_quantity": variant.inventory_quantity, "option1": variant.option1, "option2": variant.option2, "option3": variant.option3, } ) # オプション情報 for option in product.options: result["options"].append( {"id": option.id, "name": option.name, "values": option.values} ) # 画像情報 for image in product.images: result["images"].append({"id": image.id, "src": image.src, "alt": image.alt}) return [ types.TextContent( type="text", text=json.dumps(result, indent=2, ensure_ascii=False), ) ] async def handle_create_product(arguments: dict) -> list[types.TextContent]: """新しい商品を作成する""" # 必須パラメータのチェック title = arguments.get("title") if not title: raise ValueError("title is required") # 商品オブジェクトの作成 product = shopify.Product() product.title = title # オプションパラメータの設定 if "body_html" in arguments: product.body_html = arguments["body_html"] if "vendor" in arguments: product.vendor = arguments["vendor"] if "product_type" in arguments: product.product_type = arguments["product_type"] if "tags" in arguments: product.tags = arguments["tags"] if "status" in arguments: product.status = arguments["status"] # オプションの設定 if "options" in arguments and arguments["options"]: options = [] for option_data in arguments["options"]: option = shopify.Option() option.name = option_data["name"] option.position = option_data["position"] option.values = option_data["values"] options.append(option) product.options = options # バリエーションの設定 if "variants" in arguments and arguments["variants"]: variants = [] for variant_data in arguments["variants"]: variant = shopify.Variant() if "price" in variant_data: variant.price = variant_data["price"] if "sku" in variant_data: variant.sku = variant_data["sku"] if "inventory_quantity" in variant_data: variant.inventory_quantity = variant_data["inventory_quantity"] if "option1" in variant_data: variant.option1 = variant_data["option1"] if "option2" in variant_data: variant.option2 = variant_data["option2"] if "option3" in variant_data: variant.option3 = variant_data["option3"] variants.append(variant) product.variants = variants # 画像の設定 if "images" in arguments and arguments["images"]: for image_data in arguments["images"]: image = shopify.Image() image.src = image_data["src"] if "alt" in image_data: image.alt = image_data["alt"] product.images.append(image) # 商品の保存 product.save() return [ types.TextContent( type="text", text=json.dumps( { "success": True, "product_id": product.id, "message": f"商品「{product.title}」が作成されました", }, indent=2, ensure_ascii=False, ), ) ] async def handle_update_product(arguments: dict) -> list[types.TextContent]: """商品を更新する""" # 必須パラメータのチェック product_id = arguments.get("product_id") if not product_id: raise ValueError("product_id is required") # 商品の取得 product = shopify.Product.find(product_id) # 商品情報の更新 if "title" in arguments: product.title = arguments["title"] if "body_html" in arguments: product.body_html = arguments["body_html"] if "vendor" in arguments: product.vendor = arguments["vendor"] if "product_type" in arguments: product.product_type = arguments["product_type"] if "tags" in arguments: product.tags = arguments["tags"] if "status" in arguments: product.status = arguments["status"] # バリエーションの更新 if "variants" in arguments and arguments["variants"]: for variant_data in arguments["variants"]: # バリエーションIDがある場合は既存のバリエーションを更新 if "id" in variant_data: for variant in product.variants: if variant.id == variant_data["id"]: if "price" in variant_data: variant.price = variant_data["price"] if "sku" in variant_data: variant.sku = variant_data["sku"] if "inventory_quantity" in variant_data: variant.inventory_quantity = variant_data[ "inventory_quantity" ] if "option1" in variant_data: variant.option1 = variant_data["option1"] if "option2" in variant_data: variant.option2 = variant_data["option2"] if "option3" in variant_data: variant.option3 = variant_data["option3"] # バリエーションIDがない場合は新しいバリエーションを追加 else: variant = shopify.Variant() variant.product_id = product.id if "price" in variant_data: variant.price = variant_data["price"] if "sku" in variant_data: variant.sku = variant_data["sku"] if "inventory_quantity" in variant_data: variant.inventory_quantity = variant_data["inventory_quantity"] if "option1" in variant_data: variant.option1 = variant_data["option1"] if "option2" in variant_data: variant.option2 = variant_data["option2"] if "option3" in variant_data: variant.option3 = variant_data["option3"] product.variants.append(variant) # オプションの更新 if "options" in arguments and arguments["options"]: for option_data in arguments["options"]: # オプションIDがある場合は既存のオプションを更新 if "id" in option_data: for option in product.options: if option.id == option_data["id"]: option.name = option_data["name"] option.values = option_data["values"] # オプションIDがない場合は新しいオプションを追加 else: option = shopify.Option() option.product_id = product.id option.name = option_data["name"] option.values = option_data["values"] product.options.append(option) # 画像の更新 if "images" in arguments and arguments["images"]: for image_data in arguments["images"]: # 画像IDがある場合は既存の画像を更新 if "id" in image_data: for image in product.images: if image.id == image_data["id"]: image.src = image_data["src"] if "alt" in image_data: image.alt = image_data["alt"] # 画像IDがない場合は新しい画像を追加 else: image = shopify.Image() image.product_id = product.id image.src = image_data["src"] if "alt" in image_data: image.alt = image_data["alt"] product.images.append(image) # 商品の保存 product.save() return [ types.TextContent( type="text", text=json.dumps( { "success": True, "product_id": product.id, "message": f"商品「{product.title}」が更新されました", }, indent=2, ensure_ascii=False, ), ) ] async def handle_delete_product(arguments: dict) -> list[types.TextContent]: """商品を削除する""" # 必須パラメータのチェック product_id = arguments.get("product_id") if not product_id: raise ValueError("product_id is required") # 商品の取得 product = shopify.Product.find(product_id) # 商品名を保存 product_title = product.title # 商品の削除 product.destroy() return [ types.TextContent( type="text", text=json.dumps( { "success": True, "message": f"商品「{product_title}」が削除されました", }, indent=2, ensure_ascii=False, ), ) ] async def main(): # Run the server using stdin/stdout streams async with mcp.server.stdio.stdio_server() as (read_stream, write_stream): await server.run( read_stream, write_stream, InitializationOptions( server_name="shopify-py-mcp", server_version="0.1.0", capabilities=server.get_capabilities( notification_options=NotificationOptions(), experimental_capabilities={}, ), ), )