Storyblok MCP Server

import json from typing import Optional, Dict, Any, Literal, List from httpx import AsyncClient from mcp.server.fastmcp import FastMCP from utils.api import ( build_management_url, get_management_headers, _handle_response, create_pagination_params, add_optional_params, APIError, ) from datetime import datetime def register_assets(mcp: FastMCP, client: AsyncClient) -> None: @mcp.tool() async def fetch_assets( page: Optional[int] = 1, per_page: Optional[int] = 25, search: Optional[str] = None, folder_id: Optional[int] = None, sort_by: Optional[Literal[ "created_at:asc", "created_at:desc", "updated_at:asc", "updated_at:desc", "short_filename:asc", "short_filename:desc" ]] = None, is_private: Optional[bool] = None, by_alt: Optional[str] = None, by_title: Optional[str] = None, by_copyright: Optional[str] = None, with_tags: Optional[str] = None ) -> Any: """ Retrieve multiple assets from Storyblok Management API. """ try: params = create_pagination_params(page, per_page) add_optional_params(params, { "search": search, "in_folder": folder_id, "sort_by": sort_by, "is_private": "1" if is_private else None, "by_alt": by_alt, "by_title": by_title, "by_copyright": by_copyright, "with_tags": with_tags, }) url = build_management_url("/assets") resp = await client.get(url, params=params, headers=get_management_headers()) return _handle_response(resp, url) except APIError as e: return {"isError": True, "content": [{"type": "text", "text": str(e)}]} @mcp.tool() async def get_asset(id: str) -> Any: """Gets a specific asset by ID.""" try: url = build_management_url(f"/assets/{id}") resp = await client.get(url, headers=get_management_headers()) return _handle_response(resp, url) except APIError as e: return {"isError": True, "content": [{"type": "text", "text": str(e)}]} @mcp.tool() async def delete_asset(id: str) -> Any: """Deletes an asset from Storyblok.""" try: url = build_management_url(f"/assets/{id}") resp = await client.delete(url, headers=get_management_headers()) _handle_response(resp, url) return {"content": [{"type": "text", "text": f"Asset {id} has been successfully deleted."}]} except APIError as e: return {"isError": True, "content": [{"type": "text", "text": str(e)}]} @mcp.tool() async def update_asset( asset_id: int, asset_folder_id: Optional[int] = None, internal_tag_ids: Optional[List[int]] = None, locked: Optional[bool] = None, is_private: Optional[bool] = None, publish_at: Optional[datetime] = None, expire_at: Optional[datetime] = None, focus: Optional[str] = None, alt: Optional[str] = None, title: Optional[str] = None, source: Optional[str] = None, copyright: Optional[str] = None, meta_data: Optional[Dict[str, Any]] = None ) -> Any: """ Update an existing asset’s metadata or settings. """ try: payload: Dict[str, Any] = {} # Core fields for field, value in [ ("asset_folder_id", asset_folder_id), ("internal_tag_ids", internal_tag_ids), ("locked", locked), ("is_private", is_private), ("publish_at", publish_at.isoformat() if publish_at else None), ("expire_at", expire_at.isoformat() if expire_at else None), ("focus", focus), ]: if value is not None: payload[field] = value # Metadata fields md: Dict[str, Any] = meta_data or {} for field, value in [ ("alt", alt), ("title", title), ("source", source), ("copyright", copyright), ]: if value is not None and field not in md: md[field] = value if md: payload["meta_data"] = md url = build_management_url(f"/assets/{asset_id}") resp = await client.put(url, json=payload, headers=get_management_headers()) return _handle_response(resp, url) except APIError as e: return {"isError": True, "content": [{"type": "text", "text": str(e)}]} @mcp.tool() async def delete_multiple_assets( ids: List[int] ) -> Any: """ Deletes multiple assets by numeric IDs using the Storyblok Management API. """ if not ids: return {"isError": True, "content": [{"type": "text", "text": "ids list cannot be empty"}]} try: payload = {"ids": ids} url = build_management_url("/assets/bulk_destroy") resp = await client.post(url, json=payload, headers=get_management_headers()) return _handle_response(resp, url) except APIError as e: return {"isError": True, "content": [{"type": "text", "text": str(e)}]} @mcp.tool() async def bulk_move_assets( ids: List[int], asset_folder_id: int ) -> Any: """ Move multiple assets to a specified folder. """ if not ids: return {"isError": True, "content": [{"type": "text", "text": "ids list cannot be empty"}]} if not isinstance(asset_folder_id, int): return {"isError": True, "content": [{"type": "text", "text": "asset_folder_id must be an integer"}]} try: payload = { "ids": ids, "asset_folder_id": asset_folder_id } url = build_management_url("/assets/bulk_update") resp = await client.post(url, json=payload, headers=get_management_headers()) return _handle_response(resp, url) except APIError as e: return {"isError": True, "content": [{"type": "text", "text": str(e)}]} @mcp.tool() async def bulk_restore_assets( ids: List[int] ) -> Any: """ Restores multiple previously deleted assets """ if not ids: return {"isError": True, "content": [{"type": "text", "text": "ids list cannot be empty"}]} try: payload = {"ids": ids} url = build_management_url("/assets/bulk_restore") resp = await client.post(url, json=payload, headers=get_management_headers()) return _handle_response(resp, url) except APIError as e: return {"isError": True, "content": [{"type": "text", "text": str(e)}]} @mcp.tool() async def init_asset_upload(filename: str, size: int, content_type: str) -> Any: """Initializes asset upload and returns signed S3 upload URL.""" try: url = build_management_url("/assets") payload = {"filename": filename, "size": size, "content_type": content_type} resp = await client.post(url, json=payload, headers=get_management_headers()) return _handle_response(resp, url) except APIError as e: return {"isError": True, "content": [{"type": "text", "text": str(e)}]} @mcp.tool() async def complete_asset_upload(asset_id: str) -> Any: """Completes the asset upload process after S3 upload.""" try: url = build_management_url(f"/assets/{asset_id}/finish_upload") resp = await client.post(url, headers=get_management_headers()) return _handle_response(resp, url) except APIError as e: return {"isError": True, "content": [{"type": "text", "text": str(e)}]}