Shopify Python MCP Server
- shopify-py-mcp
- 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={},
),
),
)