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_story_schedules(mcp: FastMCP, client: AsyncClient) -> None:

    @mcp.tool()
    async def retrieve_multiple_story_schedules(
        space_id: int,
        by_status: Optional[str] = None
    ) -> Any:
        """
        Retrieves multiple story scheduling entries in a Storyblok space via the Management API.

        - by_status: Optional status filter ("published_before_schedule" or "scheduled").
        """
        try:
            params: dict[str, Any] = {}
            if by_status is not None:
                params["by_status"] = by_status

            url = build_management_url(f"/story_schedulings/")
            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_one_story_schedule(
        story_scheduling_id: int
    ) -> Any:
        """
        Retrieves a single story schedule entry by its ID in a Storyblok space via the Management API.

        - story_scheduling_id: Numeric ID of the schedule to retrieve.
        """
        try:
            url = build_management_url(
                f"/story_schedulings/{story_scheduling_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_story_schedule(
        story_id: int,
        publish_at: str,
        language: Optional[str] = None
    ) -> Any:
        """
        Creates a new story schedule via the Storyblok Management API.

        - story_id: Numeric ID of the story to be scheduled.
        - publish_at: ISO‑8601 date/time string in UTC (e.g., "2025‑06‑20T15:30:00Z").
        - language: Optional language code (e.g., "en", "pt‑br").
        """
        try:
            payload: dict[str, Any] = {
                "story_scheduling": {
                    "story_id": story_id,
                    "publish_at": publish_at
                }
            }
            if language:
                payload["story_scheduling"]["language"] = language

            url = build_management_url(f"/story_schedulings")
            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 update_story_schedule(
        space_id: int,
        story_scheduling_id: int,
        publish_at: Optional[str] = None,
        language: Optional[str] = None
    ) -> Any:
        """
        Updates an existing story schedule via the Storyblok Management API.

        - space_id: Numeric ID of the Storyblok space.
        - story_scheduling_id: Numeric ID of the schedule to update.
        - publish_at: New ISO‑8601 UTC date/time string.
        - language: Optional new language code.
        """
        try:
            payload: dict[str, Any] = {"story_scheduling": {}}
            if publish_at is not None:
                payload["story_scheduling"]["publish_at"] = publish_at
            if language is not None:
                payload["story_scheduling"]["language"] = language

            url = build_management_url(
                f"/story_schedulings/{story_scheduling_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_story_schedule(
        story_scheduling_id: int
    ) -> Any:
        """
        Deletes a story schedule entry via the Storyblok Management API.
        """
        try:
            url = build_management_url(
                f"/story_schedulings/{story_scheduling_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)}]}