Storyblok MCP Server

import json
from typing import Any, Optional, List
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_activities(mcp: FastMCP, client: AsyncClient) -> None:

    @mcp.tool()
    async def retrieve_multiple_activities(
        created_at_gte: Optional[str] = None,
        created_at_lte: Optional[str] = None,
        by_owner_ids: Optional[List[int]] = None,
        types: Optional[List[str]] = None
    ) -> Any:
        """
        Retrieves activity logs for a specified Storyblok space.

        Optional filters:
        - created_at_gte / created_at_lte: 'YYYY-MM-DD' date strings
        - by_owner_ids: list of user IDs
        - types: list of activity types (e.g. 'Story', 'Component', 'Asset')
        """
        try:
            params: dict[str, Any] = {}
            if created_at_gte:
                params["created_at_gte"] = created_at_gte
            if created_at_lte:
                params["created_at_lte"] = created_at_lte
            if by_owner_ids:
                params["by_owner_ids"] = ",".join(map(str, by_owner_ids))
            if types:
                params["types"] = ",".join(types)

            url = build_management_url("/activities/")
            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_activity(
        activity_id: int
    ) -> Any:
        """
        Retrieves a single activity log by its ID from a specified Storyblok space.
        """
        try:
            url = build_management_url(f"/activities/{activity_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)}]}