import os
import sys
import argparse
from typing import Dict, Any
from dotenv import load_dotenv
import html2text
from atlassian import Confluence
from fastmcp import FastMCP
load_dotenv()
# --- Environment and API Configuration ---
CONFLUENCE_URL = os.environ.get("CONFLUENCE_URL")
CONFLUENCE_USERNAME = os.environ.get("CONFLUENCE_USERNAME")
CONFLUENCE_API_TOKEN = os.environ.get("CONFLUENCE_API_TOKEN")
if not all([CONFLUENCE_URL, CONFLUENCE_USERNAME, CONFLUENCE_API_TOKEN]):
raise ValueError("Missing Confluence environment variables (URL, USERNAME, API_TOKEN)")
confluence = Confluence(
url=CONFLUENCE_URL,
username=CONFLUENCE_USERNAME,
password=CONFLUENCE_API_TOKEN,
cloud=True
)
# Initialize HTML to Markdown converter
h2t = html2text.HTML2Text()
h2t.ignore_links = False
h2t.ignore_images = False
h2t.body_width = 0 # Don't wrap lines
# --- Initialize MCP Server ---
mcp = FastMCP("Confluence MCP Server")
# --- Tool Functions ---
@mcp.tool()
def search_confluence(query: str) -> Dict[str, Any]:
"""Search for pages, blog posts, or attachments in Confluence.
Args:
query: Search query string to find content in Confluence
Returns:
Dictionary containing search results with title, type, and id
"""
try:
results = confluence.cql(f'siteSearch ~ "{query}"', limit=10)
return {
"results": [
{
"title": result["title"],
"type": result["content"]["type"],
"id": result["content"]["id"],
}
for result in results.get("results", [])
]
}
except confluence.exceptions.ApiError as e:
if e.status_code == 401:
return {"error": "Unauthorized: Check your Confluence credentials"}
elif e.status_code == 404:
return {"error": "Not Found: Unable to access Confluence API"}
else:
return {"error": f"API Error: {str(e)}"}
except Exception as e:
return {"error": f"Unexpected error: {str(e)}"}
@mcp.tool()
def read_page(page_id: str) -> Dict[str, Any]:
"""Fetch the full content of a specific page by its ID.
Args:
page_id: The Confluence page ID
Returns:
Dictionary containing page title and content in Markdown format
"""
try:
page = confluence.get_page_by_id(page_id, expand="body.storage")
html_content = page["body"]["storage"]["value"]
# Convert HTML to Markdown for better LLM context efficiency
markdown_content = h2t.handle(html_content)
return {
"title": page["title"],
"content": markdown_content,
"page_id": page_id
}
except confluence.exceptions.ApiError as e:
if e.status_code == 401:
return {"error": "Unauthorized: Check your Confluence credentials"}
elif e.status_code == 404:
return {"error": f"Not Found: Page with ID '{page_id}' does not exist"}
else:
return {"error": f"API Error: {str(e)}"}
except Exception as e:
return {"error": f"Unexpected error: {str(e)}"}
@mcp.tool()
def list_space_content(space_key: str) -> Dict[str, Any]:
"""List all pages within a specific Confluence workspace.
Args:
space_key: The Confluence space key (e.g., 'TEAM', 'DOCS')
Returns:
Dictionary containing list of pages with title and id
"""
try:
pages = confluence.get_all_pages_from_space(space_key, limit=100)
return {
"space_key": space_key,
"pages": [
{
"title": page["title"],
"id": page["id"]
}
for page in pages
],
"total": len(pages)
}
except confluence.exceptions.ApiError as e:
if e.status_code == 401:
return {"error": "Unauthorized: Check your Confluence credentials"}
elif e.status_code == 404:
return {"error": f"Not Found: Space with key '{space_key}' does not exist"}
else:
return {"error": f"API Error: {str(e)}"}
except Exception as e:
return {"error": f"Unexpected error: {str(e)}"}
if __name__ == "__main__":
# Parse command-line arguments
parser = argparse.ArgumentParser(description="Confluence MCP Server")
parser.add_argument(
"--transport",
choices=["stdio", "sse"],
default="stdio",
help="Transport protocol to use (stdio for local, sse for HTTP)"
)
parser.add_argument(
"--host",
default="0.0.0.0",
help="Host to bind to when using SSE transport (default: 0.0.0.0)"
)
parser.add_argument(
"--port",
type=int,
default=8000,
help="Port to bind to when using SSE transport (default: 8000)"
)
args = parser.parse_args()
# Run the MCP server with specified transport
if args.transport == "sse":
print(f"Starting Confluence MCP Server on {args.host}:{args.port}")
mcp.run(transport="sse", host=args.host, port=args.port)
else:
# Default to stdio for local development
mcp.run(transport="stdio")
