Basic WP MCP Server

wp_mcp_server.py

from mcp.server.fastmcp import FastMCP
import httpx
import base64
import asyncio
import os
from dotenv import load_dotenv

# Load env vars
load_dotenv()

# Create an MCP server
mcp = FastMCP("wp-mcp-server")

WP_BASE_URL = os.getenv("WP_BASE_URL")
APP_USERNAME = os.getenv("APP_USERNAME")
APP_PASSWORD = os.getenv("APP_PASSWORD")

# Authentication headers
credentials = f"{APP_USERNAME}:{APP_PASSWORD}"
encoded_credentials = base64.b64encode(credentials.encode()).decode()
auth_headers = {
    "Authorization": f"Basic {encoded_credentials}"
} 

@mcp.tool()
async def create_post(title: str, content: str, status: str = "draft") -> str:
    """
    Create a new WordPress post
    
    Args:
        title: The title of the post
        content: The content of the post
        status: Post status (draft, publish, private, pending) - defaults to draft
    """
    print(f"MCP: Creating post on {WP_BASE_URL}/posts")
    
    # Validate status
    valid_statuses = ["draft", "publish", "private", "pending"]
    if status not in valid_statuses:
        status = "draft"
    
    # Prepare post data
    post_data = {
        "title": title,
        "content": content,
        "status": status
    }
    
    # Send request to WordPress
    try:
        async with httpx.AsyncClient(verify=False, timeout=10.0) as client:
            response = await client.post(
                f"{WP_BASE_URL}/posts", 
                json=post_data, 
                headers=auth_headers
            )
            
            if response.status_code not in [200, 201]:
                return f"Error creating post: {response.status_code} {response.text}"
            
            post = response.json()
            return f"Post created successfully!\nID: {post['id']}\nTitle: {post['title']['rendered']}\nStatus: {post['status']}\nLink: {post['link']}"
    except Exception as e:
        return f"Error connecting to WordPress: {str(e)}"

if __name__ == "__main__":
    # Start the MCP server without inspector and host parameters
    mcp.run()