server.py•6.04 kB
"""Jekyll Blog MCP Server."""
import asyncio
import os
import sys
from pathlib import Path
from mcp.server import Server
from mcp.server.stdio import stdio_server
from mcp.types import Tool, TextContent
from jekyll_mcp.indexer import PostIndexer
from jekyll_mcp.tools import JekyllTools
def get_blog_paths():
"""
Get blog paths from environment variables or command-line arguments.
Environment variables:
JEKYLL_POSTS_DIR: Path to _posts directory (required)
JEKYLL_DRAFTS_DIR: Path to _drafts directory (optional)
Returns:
Tuple of (posts_dir, drafts_dir)
"""
# Try environment variables first
posts_dir = os.getenv('JEKYLL_POSTS_DIR')
drafts_dir = os.getenv('JEKYLL_DRAFTS_DIR')
# If not set, try to infer from current directory
if not posts_dir:
cwd = Path.cwd()
# Check if we're in a Jekyll project
if (cwd / '_posts').exists():
posts_dir = str(cwd / '_posts')
if (cwd / '_drafts').exists():
drafts_dir = str(cwd / '_drafts')
if not posts_dir:
print("Error: JEKYLL_POSTS_DIR not set and no _posts directory found", file=sys.stderr)
print("Set JEKYLL_POSTS_DIR environment variable or run from Jekyll project root", file=sys.stderr)
sys.exit(1)
return Path(posts_dir), Path(drafts_dir) if drafts_dir else None
# Initialize server
app = Server("jekyll-blog-server")
# Get paths and initialize indexer
POSTS_DIR, DRAFTS_DIR = get_blog_paths()
indexer = PostIndexer(POSTS_DIR, DRAFTS_DIR)
indexer.index_all()
jekyll_tools = JekyllTools(indexer)
print(f"Indexed {len(indexer.posts)} posts from {POSTS_DIR}", file=sys.stderr)
if DRAFTS_DIR:
print(f"Including drafts from {DRAFTS_DIR}", file=sys.stderr)
print(f"Found {len(indexer.categories)} categories and {len(indexer.tags)} tags", file=sys.stderr)
@app.list_tools()
async def list_tools() -> list[Tool]:
"""List available tools."""
return [
Tool(
name="search_posts",
description="Search blog posts by query, category, or tags",
inputSchema={
"type": "object",
"properties": {
"query": {
"type": "string",
"description": "Search term to find in title, content, or slug"
},
"category": {
"type": "string",
"description": "Filter by category"
},
"tags": {
"type": "string",
"description": "Comma-separated list of tags to filter by"
},
"limit": {
"type": "number",
"description": "Maximum number of results (default: 10)",
"default": 10
}
}
}
),
Tool(
name="get_post",
description="Get full content of a specific post by slug",
inputSchema={
"type": "object",
"properties": {
"slug": {
"type": "string",
"description": "The post slug"
}
},
"required": ["slug"]
}
),
Tool(
name="list_categories",
description="List all blog categories with post counts",
inputSchema={
"type": "object",
"properties": {}
}
),
Tool(
name="list_tags",
description="List all blog tags with post counts",
inputSchema={
"type": "object",
"properties": {}
}
),
Tool(
name="compare_draft",
description="Compare draft content against published posts to find similar content",
inputSchema={
"type": "object",
"properties": {
"draft_content": {
"type": "string",
"description": "The draft text to compare against published posts"
},
"limit": {
"type": "number",
"description": "Maximum number of similar posts to return (default: 5)",
"default": 5
}
},
"required": ["draft_content"]
}
)
]
@app.call_tool()
async def call_tool(name: str, arguments: dict) -> list[TextContent]:
"""Handle tool calls."""
try:
if name == "search_posts":
result = jekyll_tools.search_posts(
query=arguments.get("query"),
category=arguments.get("category"),
tags=arguments.get("tags"),
limit=arguments.get("limit", 10)
)
elif name == "get_post":
result = jekyll_tools.get_post(arguments["slug"])
elif name == "list_categories":
result = jekyll_tools.list_categories()
elif name == "list_tags":
result = jekyll_tools.list_tags()
elif name == "compare_draft":
result = jekyll_tools.compare_draft(
draft_content=arguments["draft_content"],
limit=arguments.get("limit", 5)
)
else:
result = {"error": f"Unknown tool: {name}"}
return [TextContent(type="text", text=str(result))]
except Exception as e:
return [TextContent(type="text", text=f"Error: {str(e)}")]
async def main():
"""Run the MCP server."""
async with stdio_server() as (read_stream, write_stream):
await app.run(
read_stream,
write_stream,
app.create_initialization_options()
)
if __name__ == "__main__":
asyncio.run(main())