mcp_server.py

from typing import Any
import httpx
import jwt
import json
import re
from datetime import datetime
from mcp.server.fastmcp import FastMCP, Image
from PIL import Image as PILImage
from dotenv import load_dotenv
import os

load_dotenv()

# Constants
USER_AGENT = "ghost-app/1.0"
GHOST_API_KEY = os.getenv("GHOST_API_KEY")
GHOST_API_URL = os.getenv("GHOST_API_URL")
GHOST_API_VERSION = os.getenv("GHOST_API_VERSION", "v5.116.1")


# Helper functions for Ghost blog integration
def create_ghost_auth_token():
    """Create a JWT authentication token for Ghost Admin API."""
    id, secret = GHOST_API_KEY.split(":")
    iat = int(datetime.now().timestamp())

    header = {"alg": "HS256", "typ": "JWT", "kid": id}
    payload = {"iat": iat, "exp": iat + 5 * 60, "aud": "/admin/"}

    return jwt.encode(payload, bytes.fromhex(secret), algorithm="HS256", headers=header)


def preprocess_code_blocks(markdown_content):
    """
    Preprocess markdown content to ensure code blocks are properly formatted for Ghost.

    This function ensures that:
    1. Code blocks have proper language tags
    2. Code blocks are properly indented and formatted
    3. Inline code is properly formatted

    Args:
        markdown_content: Original markdown content

    Returns:
        Processed markdown with improved code block formatting
    """
    # Regular expression to match code blocks (```language ... ```)
    code_block_pattern = r"```([a-zA-Z0-9]*)\n([\s\S]*?)```"

    def format_code_block(match):
        language = match.group(1) or "plaintext"
        code = match.group(2).rstrip()

        # Clean up the code: normalize indentation and ensure proper line breaks
        lines = code.split("\n")
        # Find minimum indentation (excluding empty lines)
        non_empty_lines = [line for line in lines if line.strip()]
        if non_empty_lines:
            min_indent = min(len(line) - len(line.lstrip()) for line in non_empty_lines)
            # Remove common indentation
            code = "\n".join(
                line[min_indent:] if line.strip() else line for line in lines
            )

        # Ensure proper formatting with language tag
        return f"```{language}\n{code}\n```"

    # Replace all code blocks with properly formatted ones
    processed_content = re.sub(code_block_pattern, format_code_block, markdown_content)

    # Add custom styling for better code display (as HTML at the beginning of the post)
    code_style = """
<style>
pre {
  background-color: #282c34;
  border-radius: 5px;
  padding: 15px;
  overflow-x: auto;
}
code {
  font-family: 'Fira Code', Consolas, Monaco, 'Andale Mono', 'Ubuntu Mono', monospace;
  font-size: 14px;
  line-height: 1.5;
}
:not(pre) > code {
  background-color: #f1f1f1;
  border-radius: 3px;
  padding: 2px 4px;
  color: #e83e8c;
}
</style>

"""
    # Only add the style if there's at least one code block
    if "```" in processed_content:
        processed_content = code_style + processed_content

    return processed_content


def markdown_to_mobiledoc(markdown_content):
    """
    Convert markdown content to Ghost's mobiledoc format.

    This function properly handles code blocks with syntax highlighting.

    Args:
        markdown_content: Content in markdown format

    Returns:
        String with JSON representation of mobiledoc
    """
    # Preprocess the markdown to ensure code blocks are properly formatted
    processed_markdown = preprocess_code_blocks(markdown_content)

    # Basic mobiledoc structure
    mobiledoc = {
        "version": "0.3.1",
        "markups": [],
        "atoms": [],
        "cards": [
            [
                "markdown",
                {
                    "markdown": processed_markdown,
                    "cardName": "markdown",
                    "options": {
                        "syntaxTheme": "atom-one-dark"  # Add syntax highlighting theme
                    },
                },
            ]
        ],
        "sections": [[10, 0]],
    }

    return json.dumps(mobiledoc)


async def upload_to_ghost(
    title,
    content,
    author_id="1",
    tags=None,
    status="draft",
    feature_image: str = None,
):
    """Upload a blog post to Ghost CMS."""
    token = create_ghost_auth_token()

    headers = {
        "Authorization": f"Ghost {token}",
        "Accept-Version": GHOST_API_VERSION,
        "Content-Type": "application/json",
    }

    post_data = {
        "title": title,
        "mobiledoc": content,
        "status": status,
        "authors": [author_id],
    }

    if tags:
        post_data["tags"] = tags
    if feature_image:
        post_data["feature_image"] = feature_image

    async with httpx.AsyncClient() as client:
        response = await client.post(
            GHOST_API_URL, json={"posts": [post_data]}, headers=headers
        )

        if response.status_code != 201:
            return {"success": False, "error": response.text}

        return {"success": True, "data": response.json()}


# Initialize the MCP server
mcp = FastMCP("ghost-mcp")


# Register tools
@mcp.tool(name="ghost_post")
async def ghost_post(
    title: str,
    content: str,
    author_id: str = "1",
    tags: list = None,
    status: str = "draft",
    feature_image: str = None,
    code_language: str = "plaintext",
) -> dict:
    """
    Create a new blog post on Ghost CMS.

    Args:
        title: The title of the blog post
        content: The content of the blog post in markdown format (will be converted to mobiledoc)
        author_id: The ID of the author (default: "1" for admin)
        tags: list of tag names
        status: Post status (draft, published)
        feature_image: URL for the post's cover image (optional)
        code_language: Default language for unlabeled code blocks (default: "plaintext")

    Returns:
        Dict with success status and response data
    """
    # Add default language to unlabeled code blocks
    if code_language != "plaintext":
        content = content.replace("```\n", f"```{code_language}\n")

    mobiledoc_content = markdown_to_mobiledoc(content)
    result = await upload_to_ghost(
        title, mobiledoc_content, author_id, tags, status, feature_image
    )
    return result


if __name__ == "__main__":
    # Initialize and run the server
    mcp.run(transport="stdio")