Storyblok MCP Server

from typing import Any, Optional
from httpx import AsyncClient
from mcp.server.fastmcp import FastMCP
from utils.api import build_management_url, get_management_headers, _handle_response, APIError

def register_branches(mcp: FastMCP, client: AsyncClient) -> None:

    @mcp.tool()
    async def retrieve_multiple_branches(
        by_ids: Optional[str] = None,
        search: Optional[str] = None
    ) -> Any:
        """
        Retrieves multiple branches (pipelines) in a Storyblok space via the Management API.

        - by_ids: Optional comma-separated list of branch IDs to filter.
        - search: Optional filter term for branch names.
        """
        try:
            params: dict[str, Any] = {}
            if by_ids is not None:
                params["by_ids"] = by_ids
            if search is not None:
                params["search"] = search

            url = build_management_url(f"/branches/")
            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 retrieve_single_branch(
        branch_id: int
    ) -> Any:
        """
        Retrieves a single branch (pipeline) by its ID via the Storyblok Management API.
        - branch_id: Numeric ID of the branch to retrieve.
        """
        try:
            url = build_management_url(f"/branches/{branch_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 create_branch(
        name: str,
        source_id: Optional[int] = None,
        url: Optional[str] = None,
        position: Optional[int] = None
    ) -> Any:
        """
        Creates a new branch (pipeline) in a Storyblok space via the Management API.

        - name: Required name for the new branch.
        - source_id: Optional ID of an existing branch to clone.
        - url: Optional preview URL for the branch.
        - position: Optional numeric position for ordering.
        """
        try:
            branch_data: dict[str, Any] = {"name": name}
            if source_id is not None:
                branch_data["source_id"] = source_id
            if url is not None:
                branch_data["url"] = url
            if position is not None:
                branch_data["position"] = position

            payload = {"branch": branch_data}
            url_path = build_management_url(f"/branches/")
            resp = await client.post(url_path, json=payload, headers=get_management_headers())
            return _handle_response(resp, url_path)
        except APIError as e:
            return {"isError": True, "content": [{"type": "text", "text": str(e)}]}
        
    @mcp.tool()
    async def update_branch(
        branch_id: int,
        name: Optional[str] = None,
        source_id: Optional[int] = None,
        url: Optional[str] = None,
        position: Optional[int] = None
    ) -> Any:
        """
        Updates an existing branch (pipeline) in a Storyblok space via the Management API.

        - branch_id: Numeric ID of the branch to update.
        Optional fields:
          - name: New branch name
          - source_id: Set/clear source branch (clone origin)
          - url: Preview URL
          - position: Position ordering number
        """
        try:
            branch_data: dict[str, Any] = {}
            if name is not None:
                branch_data["name"] = name
            if source_id is not None:
                branch_data["source_id"] = source_id
            if url is not None:
                branch_data["url"] = url
            if position is not None:
                branch_data["position"] = position

            payload = {"branch": branch_data}
            url_path = build_management_url(f"/branches/{branch_id}")
            resp = await client.put(url_path, json=payload, headers=get_management_headers())
            return _handle_response(resp, url_path)

        except APIError as e:
            return {"isError": True, "content": [{"type": "text", "text": str(e)}]}
        
    @mcp.tool()
    async def delete_branch(
        branch_id: int
    ) -> Any:
        """
        Deletes a branch (pipeline) by its ID in a Storyblok space.
        - branch_id: Numeric ID of the branch to delete.
        """
        try:
            url = build_management_url(f"/branches/{branch_id}")
            resp = await client.delete(url, headers=get_management_headers())
            return _handle_response(resp, url)
        except APIError as e:
            return {"isError": True, "content": [{"type": "text", "text": str(e)}]}