We provide all the information about MCP servers via our MCP API.
curl -X GET 'https://glama.ai/api/mcp/v1/servers/sooperset/mcp-atlassian'
If you have feedback or need assistance with the MCP directory API, please join our Discord server
"""Module for Confluence page operations."""
import logging
import requests
from requests.exceptions import HTTPError
from ..exceptions import MCPAtlassianAuthenticationError
from ..models.confluence import ConfluencePage
from .client import ConfluenceClient
from .utils import emoji_to_hex_id, extract_emoji_from_property
from .v2_adapter import ConfluenceV2Adapter
logger = logging.getLogger("mcp-atlassian")
class PagesMixin(ConfluenceClient):
"""Mixin for Confluence page operations."""
@property
def _v2_adapter(self) -> ConfluenceV2Adapter | None:
"""Get v2 API adapter for OAuth authentication.
Returns:
ConfluenceV2Adapter instance if OAuth is configured, None otherwise
"""
if self.config.auth_type == "oauth" and self.config.is_cloud:
return ConfluenceV2Adapter(
session=self.confluence._session, base_url=self.confluence.url
)
return None
def get_page_content(
self, page_id: str, *, convert_to_markdown: bool = True
) -> ConfluencePage:
"""
Get content of a specific page.
Args:
page_id: The ID of the page to retrieve
convert_to_markdown: When True, returns content in markdown format,
otherwise returns raw HTML (keyword-only)
Returns:
ConfluencePage model containing the page content and metadata
Raises:
MCPAtlassianAuthenticationError: If authentication fails with the Confluence API (401/403)
Exception: If there is an error retrieving the page
"""
try:
# Use v2 API for OAuth authentication, v1 API for token/basic auth
v2_adapter = self._v2_adapter
if v2_adapter:
logger.debug(
f"Using v2 API for OAuth authentication to get page '{page_id}'"
)
page = v2_adapter.get_page(
page_id=page_id,
expand="body.storage,version,space,children.attachment",
)
else:
logger.debug(
f"Using v1 API for token/basic authentication to get page '{page_id}'"
)
page = self.confluence.get_page_by_id(
page_id=page_id,
expand="body.storage,version,space,children.attachment",
)
# Check if API returned an error string instead of a dict
if isinstance(page, str):
error_msg = f"API returned error response: {page[:500]}"
raise Exception(error_msg)
space_key = page.get("space", {}).get("key", "")
try:
content = page["body"]["storage"]["value"]
except (KeyError, TypeError) as e:
logger.warning(
f"Page {page.get('id', 'unknown')} missing body.storage.value: {e}"
)
content = ""
page_id_str = str(page.get("id", ""))
page_attachments = (
page.get("children", {}).get("attachment", {}).get("results", [])
)
processed_html, processed_markdown = self.preprocessor.process_html_content(
content,
space_key=space_key,
confluence_client=self.confluence,
content_id=page_id_str,
attachments=page_attachments,
)
# Use the appropriate content format based on the convert_to_markdown flag
page_content = processed_markdown if convert_to_markdown else processed_html
# Fetch page emoji from content properties
emoji = self._get_page_emoji(page_id)
# Create and return the ConfluencePage model
return ConfluencePage.from_api_response(
page,
base_url=self.config.url,
include_body=True,
# Override content with our processed version
content_override=page_content,
content_format="storage" if not convert_to_markdown else "markdown",
is_cloud=self.config.is_cloud,
emoji=emoji,
)
except HTTPError as http_err:
if http_err.response is not None and http_err.response.status_code in [
401,
403,
]:
error_msg = (
f"Authentication failed for Confluence API ({http_err.response.status_code}). "
"Token may be expired or invalid. Please verify credentials."
)
logger.error(error_msg)
raise MCPAtlassianAuthenticationError(error_msg) from http_err
else:
logger.error(f"HTTP error during API call: {http_err}", exc_info=False)
raise http_err
except Exception as e:
logger.error(
f"Error retrieving page content for page ID {page_id}: {str(e)}"
)
raise Exception(f"Error retrieving page content: {str(e)}") from e
def get_page_ancestors(self, page_id: str) -> list[ConfluencePage]:
"""
Get ancestors (parent pages) of a specific page.
Args:
page_id: The ID of the page to get ancestors for
Returns:
List of ConfluencePage models representing the ancestors in hierarchical order
(immediate parent first, root ancestor last)
Raises:
MCPAtlassianAuthenticationError: If authentication fails with the Confluence API (401/403)
"""
try:
# Use the Atlassian Python API to get ancestors
ancestors = self.confluence.get_page_ancestors(page_id)
# Process each ancestor
ancestor_models = []
for ancestor in ancestors:
# Create the page model without fetching content
page_model = ConfluencePage.from_api_response(
ancestor,
base_url=self.config.url,
include_body=False,
)
ancestor_models.append(page_model)
return ancestor_models
except HTTPError as http_err:
if http_err.response is not None and http_err.response.status_code in [
401,
403,
]:
error_msg = (
f"Authentication failed for Confluence API ({http_err.response.status_code}). "
"Token may be expired or invalid. Please verify credentials."
)
logger.error(error_msg)
raise MCPAtlassianAuthenticationError(error_msg) from http_err
else:
logger.error(f"HTTP error during API call: {http_err}", exc_info=False)
raise http_err
except Exception as e:
logger.error(f"Error fetching ancestors for page {page_id}: {str(e)}")
logger.debug("Full exception details:", exc_info=True)
return []
def _get_page_emoji(self, page_id: str) -> str | None:
"""Get the page title emoji from content properties.
The page emoji (icon shown in navigation) is stored as a content property
with key 'emoji-title-published' or 'emoji-title-draft'.
Args:
page_id: The ID of the page
Returns:
The emoji character if set, None otherwise
"""
try:
# Use v2 API for OAuth authentication
v2_adapter = self._v2_adapter
if v2_adapter:
return v2_adapter.get_page_emoji(page_id)
# For token/basic auth, use v1 API via atlassian library
properties = self.confluence.get_page_properties(page_id)
if not properties:
return None
results = properties.get("results", [])
for prop in results:
key = prop.get("key", "")
if key in ("emoji-title-published", "emoji-title-draft"):
value = prop.get("value", {})
return extract_emoji_from_property(value)
return None
except Exception as e:
logger.debug(f"Error fetching emoji for page {page_id}: {str(e)}")
return None
def _set_single_property(
self, page_id: str, property_key: str, value: str | None
) -> bool:
"""Set or remove a single page property via v1 API.
Args:
page_id: The ID of the page
property_key: The property key to set
value: The value to set, or None to delete the property
Returns:
True if the operation succeeded, False otherwise
"""
try:
if value is None:
# Delete the property
try:
self.confluence.delete_page_property(page_id, property_key)
except Exception as e:
# Property might not exist, which is fine
logger.debug(f"Could not delete property '{property_key}': {e}")
return True
# Set/update the property
property_data = {
"key": property_key,
"value": value,
}
self.confluence.set_page_property(page_id, property_data)
return True
except Exception as e:
logger.debug(
f"Error setting property '{property_key}' for page {page_id}: {str(e)}"
)
return False
def _set_page_emoji(self, page_id: str, emoji: str | None) -> bool:
"""Set or remove the page title emoji.
The page emoji (icon shown in navigation) is stored as content properties.
Both 'emoji-title-published' and 'emoji-title-draft' are set to ensure
the emoji appears in both view and edit modes.
Args:
page_id: The ID of the page
emoji: The emoji character to set, or None to remove the emoji
Returns:
True if the operation succeeded, False otherwise
"""
try:
# Use v2 API for OAuth authentication
v2_adapter = self._v2_adapter
if v2_adapter:
return v2_adapter.set_page_emoji(page_id, emoji)
# For token/basic auth, use v1 API via atlassian library
# Convert emoji to hex code, or None to delete
emoji_value = emoji_to_hex_id(emoji) if emoji else None
# Set both published and draft properties
published_ok = self._set_single_property(
page_id, "emoji-title-published", emoji_value
)
draft_ok = self._set_single_property(
page_id, "emoji-title-draft", emoji_value
)
if not published_ok:
logger.warning(
f"Failed to set emoji-title-published for page {page_id}"
)
if not draft_ok:
logger.warning(f"Failed to set emoji-title-draft for page {page_id}")
return published_ok and draft_ok
except Exception as e:
logger.warning(f"Error setting emoji for page {page_id}: {str(e)}")
return False
def get_page_by_title(
self, space_key: str, title: str, *, convert_to_markdown: bool = True
) -> ConfluencePage | None:
"""
Get a specific page by its title from a Confluence space.
Args:
space_key: The key of the space containing the page
title: The title of the page to retrieve
convert_to_markdown: When True, returns content in markdown format,
otherwise returns raw HTML (keyword-only)
Returns:
ConfluencePage model containing the page content and metadata, or None if not found
"""
try:
# Directly try to find the page by title
page = self.confluence.get_page_by_title(
space=space_key, title=title, expand="body.storage,version"
)
if not page:
logger.warning(
f"Page '{title}' not found in space '{space_key}'. "
f"The space may be invalid, the page may not exist, or permissions may be insufficient."
)
return None
try:
content = page["body"]["storage"]["value"]
except (KeyError, TypeError) as e:
logger.warning(
f"Page {page.get('id', 'unknown')} missing body.storage.value: {e}"
)
content = ""
processed_html, processed_markdown = self.preprocessor.process_html_content(
content,
space_key=space_key,
confluence_client=self.confluence,
content_id=str(page.get("id", "")),
)
# Use the appropriate content format based on the convert_to_markdown flag
page_content = processed_markdown if convert_to_markdown else processed_html
# Fetch page emoji from content properties
emoji = self._get_page_emoji(str(page.get("id", "")))
# Create and return the ConfluencePage model
return ConfluencePage.from_api_response(
page,
base_url=self.config.url,
include_body=True,
# Override content with our processed version
content_override=page_content,
content_format="storage" if not convert_to_markdown else "markdown",
is_cloud=self.config.is_cloud,
emoji=emoji,
)
except KeyError as e:
logger.error(f"Missing key in page data: {str(e)}")
return None
except requests.RequestException as e:
logger.error(f"Network error when fetching page: {str(e)}")
return None
except (ValueError, TypeError) as e:
logger.error(f"Error processing page data: {str(e)}")
return None
except Exception as e: # noqa: BLE001 - Intentional fallback with full logging
logger.error(f"Unexpected error fetching page: {str(e)}")
# Log the full traceback at debug level for troubleshooting
logger.debug("Full exception details:", exc_info=True)
return None
def get_space_pages(
self,
space_key: str,
start: int = 0,
limit: int = 10,
*,
convert_to_markdown: bool = True,
) -> list[ConfluencePage]:
"""
Get all pages from a specific space.
Args:
space_key: The key of the space to get pages from
start: The starting index for pagination
limit: Maximum number of pages to return
convert_to_markdown: When True, returns content in markdown format,
otherwise returns raw HTML (keyword-only)
Returns:
List of ConfluencePage models containing page content and metadata
"""
pages = self.confluence.get_all_pages_from_space(
space=space_key, start=start, limit=limit, expand="body.storage"
)
page_models = []
for page in pages:
try:
content = page["body"]["storage"]["value"]
except (KeyError, TypeError) as e:
logger.warning(
f"Page {page.get('id', 'unknown')} missing body.storage.value: {e}"
)
content = ""
processed_html, processed_markdown = self.preprocessor.process_html_content(
content,
space_key=space_key,
confluence_client=self.confluence,
content_id=str(page.get("id", "")),
)
# Use the appropriate content format based on the convert_to_markdown flag
page_content = processed_markdown if convert_to_markdown else processed_html
# Ensure space information is included
if "space" not in page:
page["space"] = {
"key": space_key,
"name": space_key, # Use space_key as name if not available
}
# Create the ConfluencePage model
page_model = ConfluencePage.from_api_response(
page,
base_url=self.config.url,
include_body=True,
# Override content with our processed version
content_override=page_content,
content_format="storage" if not convert_to_markdown else "markdown",
is_cloud=self.config.is_cloud,
)
page_models.append(page_model)
return page_models
def create_page(
self,
space_key: str,
title: str,
body: str,
parent_id: str | None = None,
*,
is_markdown: bool = True,
enable_heading_anchors: bool = False,
content_representation: str | None = None,
emoji: str | None = None,
) -> ConfluencePage:
"""
Create a new page in a Confluence space.
Args:
space_key: The key of the space to create the page in
title: The title of the new page
body: The content of the page (markdown, wiki markup, or storage format)
parent_id: Optional ID of a parent page
is_markdown: Whether the body content is in markdown format (default: True, keyword-only)
enable_heading_anchors: Whether to enable automatic heading anchor generation (default: False, keyword-only)
content_representation: Content format when is_markdown=False ('wiki' or 'storage', keyword-only)
emoji: Optional emoji character for the page title icon (keyword-only)
Returns:
ConfluencePage model containing the new page's data
Raises:
Exception: If there is an error creating the page
"""
try:
# Determine body and representation based on content type
if is_markdown:
# Convert markdown to Confluence storage format
final_body = self.preprocessor.markdown_to_confluence_storage(
body, enable_heading_anchors=enable_heading_anchors
)
representation = "storage"
else:
# Use body as-is with specified representation
final_body = body
representation = content_representation or "storage"
# Use v2 API for OAuth authentication, v1 API for token/basic auth
v2_adapter = self._v2_adapter
if v2_adapter:
logger.debug(
f"Using v2 API for OAuth authentication to create page '{title}'"
)
result = v2_adapter.create_page(
space_key=space_key,
title=title,
body=final_body,
parent_id=parent_id,
representation=representation,
)
else:
logger.debug(
f"Using v1 API for token/basic authentication to create page '{title}'"
)
result = self.confluence.create_page(
space=space_key,
title=title,
body=final_body,
parent_id=parent_id,
representation=representation,
)
# Get the new page content
page_id = result.get("id")
if not page_id:
raise ValueError("Create page response did not contain an ID")
# Set the page emoji if provided
if emoji:
self._set_page_emoji(page_id, emoji)
return self.get_page_content(page_id)
except Exception as e:
logger.error(
f"Error creating page '{title}' in space {space_key}: {str(e)}"
)
raise Exception(
f"Failed to create page '{title}' in space {space_key}: {str(e)}"
) from e
def update_page(
self,
page_id: str,
title: str,
body: str,
*,
is_minor_edit: bool = False,
version_comment: str = "",
is_markdown: bool = True,
parent_id: str | None = None,
enable_heading_anchors: bool = False,
content_representation: str | None = None,
emoji: str | None = None,
) -> ConfluencePage:
"""
Update an existing page in Confluence.
Args:
page_id: The ID of the page to update
title: The new title of the page
body: The new content of the page (markdown, wiki markup, or storage format)
is_minor_edit: Whether this is a minor edit (keyword-only)
version_comment: Optional comment for this version (keyword-only)
is_markdown: Whether the body content is in markdown format (default: True, keyword-only)
parent_id: Optional new parent page ID (keyword-only)
enable_heading_anchors: Whether to enable automatic heading anchor generation (default: False, keyword-only)
content_representation: Content format when is_markdown=False ('wiki' or 'storage', keyword-only)
emoji: Optional emoji character for the page title icon (keyword-only). Pass empty string to remove emoji.
Returns:
ConfluencePage model containing the updated page's data
Raises:
Exception: If there is an error updating the page
"""
try:
# Determine body and representation based on content type
if is_markdown:
# Convert markdown to Confluence storage format
final_body = self.preprocessor.markdown_to_confluence_storage(
body, enable_heading_anchors=enable_heading_anchors
)
representation = "storage"
else:
# Use body as-is with specified representation
final_body = body
representation = content_representation or "storage"
logger.debug(f"Updating page {page_id} with title '{title}'")
# Use v2 API for OAuth authentication, v1 API for token/basic auth
v2_adapter = self._v2_adapter
if v2_adapter:
logger.debug(
f"Using v2 API for OAuth authentication to update page '{page_id}'"
)
response = v2_adapter.update_page(
page_id=page_id,
title=title,
body=final_body,
representation=representation,
version_comment=version_comment,
)
else:
logger.debug(
f"Using v1 API for token/basic authentication to update page '{page_id}'"
)
update_kwargs = {
"page_id": page_id,
"title": title,
"body": final_body,
"type": "page",
"representation": representation,
"minor_edit": is_minor_edit,
"version_comment": version_comment,
"always_update": True,
}
if parent_id:
update_kwargs["parent_id"] = parent_id
self.confluence.update_page(**update_kwargs)
# Set or remove the page emoji if provided
if emoji is not None:
# Empty string means remove emoji, otherwise set it
emoji_to_set = emoji if emoji else None
self._set_page_emoji(page_id, emoji_to_set)
# After update, refresh the page data
return self.get_page_content(page_id)
except Exception as e:
logger.error(f"Error updating page {page_id}: {str(e)}")
raise Exception(f"Failed to update page {page_id}: {str(e)}") from e
def get_page_children(
self,
page_id: str,
start: int = 0,
limit: int = 25,
expand: str = "version",
*,
convert_to_markdown: bool = True,
include_folders: bool = True,
) -> list[ConfluencePage]:
"""
Get child pages and folders of a specific Confluence page.
Args:
page_id: The ID of the parent page
start: The starting index for pagination
limit: Maximum number of child items to return
expand: Fields to expand in the response
convert_to_markdown: When True, returns content in markdown format,
otherwise returns raw HTML (keyword-only)
include_folders: When True, also returns child folders (keyword-only)
Returns:
List of ConfluencePage models containing the child pages and folders
"""
try:
# Use the Atlassian Python API's get_page_child_by_type method
# First, get child pages
page_results = self.confluence.get_page_child_by_type(
page_id=page_id, type="page", start=start, limit=limit, expand=expand
)
# Handle both pagination modes for pages
if isinstance(page_results, dict) and "results" in page_results:
child_items = page_results.get("results", [])
else:
child_items = page_results or []
# Also get child folders if requested
if include_folders:
try:
folder_results = self.confluence.get_page_child_by_type(
page_id=page_id,
type="folder",
start=start,
limit=limit,
expand=expand,
)
# Handle both pagination modes for folders
if isinstance(folder_results, dict) and "results" in folder_results:
child_folders = folder_results.get("results", [])
else:
child_folders = folder_results or []
# Combine pages and folders
child_items = child_items + child_folders
except Exception as folder_err:
# Log but don't fail if folder fetching fails
# (e.g., older Confluence versions might not support folders)
logger.debug(
f"Could not fetch child folders for page {page_id}: {folder_err}"
)
# Process results
page_models = []
space_key = ""
# Get space key from the first result if available
if child_items and "space" in child_items[0]:
space_key = child_items[0].get("space", {}).get("key", "")
# Process each child item (page or folder)
for item in child_items:
# Only process content if we have "body" expanded
content_override = None
if "body" in item and convert_to_markdown:
content = item.get("body", {}).get("storage", {}).get("value", "")
if content:
_, processed_markdown = self.preprocessor.process_html_content(
content,
space_key=space_key,
confluence_client=self.confluence,
content_id=str(item.get("id", "")),
)
content_override = processed_markdown
# Create the page model (works for both pages and folders)
page_model = ConfluencePage.from_api_response(
item,
base_url=self.config.url,
include_body=True,
content_override=content_override,
content_format="markdown" if convert_to_markdown else "storage",
)
page_models.append(page_model)
return page_models
except Exception as e:
logger.error(f"Error fetching child pages for page {page_id}: {str(e)}")
logger.debug("Full exception details:", exc_info=True)
return []
def delete_page(self, page_id: str) -> bool:
"""
Delete a Confluence page by its ID.
Args:
page_id: The ID of the page to delete
Returns:
Boolean indicating success (True) or failure (False)
Raises:
Exception: If there is an error deleting the page
"""
try:
logger.debug(f"Deleting page {page_id}")
# Use v2 API for OAuth authentication, v1 API for token/basic auth
v2_adapter = self._v2_adapter
if v2_adapter:
logger.debug(
f"Using v2 API for OAuth authentication to delete page '{page_id}'"
)
return v2_adapter.delete_page(page_id=page_id)
else:
logger.debug(
f"Using v1 API for token/basic authentication to delete page '{page_id}'"
)
response = self.confluence.remove_page(page_id=page_id)
# The Atlassian library's remove_page returns the raw response from
# the REST API call. For a successful deletion, we should get a
# response object, but it might be empty (HTTP 204 No Content).
# For REST DELETE operations, a success typically returns 204 or 200
# Check if we got a response object
if isinstance(response, requests.Response):
# Check if status code indicates success (2xx)
success = 200 <= response.status_code < 300
logger.debug(
f"Delete page {page_id} returned status code {response.status_code}"
)
return success
# If it's not a response object but truthy (like True), consider it a success
elif response:
return True
# Default to true since no exception was raised
# This is safer than returning false when we don't know what happened
return True
except Exception as e:
logger.error(f"Error deleting page {page_id}: {str(e)}")
raise Exception(f"Failed to delete page {page_id}: {str(e)}") from e
def get_page_history(
self, page_id: str, version: int, convert_to_markdown: bool = True
) -> ConfluencePage:
"""
Get the history of a specific page.
Args:
page_id: The ID of the page to get history for
version: The version to get history for
Returns:
ConfluencePage model containing the page history
"""
try:
# Use v2 API for OAuth authentication, v1 API for token/basic auth
v2_adapter = self._v2_adapter
if v2_adapter:
logger.debug(
f"Using v2 API for OAuth authentication to get page history for '{page_id}' version {version}"
)
page = v2_adapter.get_page_by_version(
page_id=page_id,
version=version,
expand="body.storage,version,space,children.attachment",
)
else:
logger.debug(
f"Using v1 API for token/basic authentication to get page history for '{page_id}'"
)
page = self.confluence.get_page_by_id(
page_id=page_id,
status="historical",
version=version,
expand="body.storage,version,space,children.attachment",
)
if isinstance(page, str):
error_msg = f"API returned error response: {page[:500]}"
raise Exception(error_msg)
try:
content = page["body"]["storage"]["value"]
except (KeyError, TypeError) as e:
logger.warning(
f"Page {page.get('id', 'unknown')} missing body.storage.value: {e}"
)
content = ""
space_key = page.get("space", {}).get("key", "")
page_attachments = (
page.get("children", {}).get("attachment", {}).get("results", [])
)
processed_html, processed_markdown = self.preprocessor.process_html_content(
content,
space_key=space_key,
confluence_client=self.confluence,
content_id=str(page.get("id", "")),
attachments=page_attachments,
)
page_content = processed_markdown if convert_to_markdown else processed_html
emoji = self._get_page_emoji(page_id)
return ConfluencePage.from_api_response(
page,
base_url=self.config.url,
include_body=True,
content_override=page_content,
content_format="markdown" if convert_to_markdown else "storage",
is_cloud=self.config.is_cloud,
emoji=emoji,
)
except HTTPError as http_err:
if http_err.response is not None and http_err.response.status_code in [
401,
403,
]:
error_msg = (
f"Authentication failed for Confluence API ({http_err.response.status_code}). "
"Token may be expired or invalid. Please verify credentials."
)
logger.error(error_msg)
raise MCPAtlassianAuthenticationError(error_msg) from http_err
else:
logger.error(f"HTTP error during API call: {http_err}", exc_info=False)
raise http_err
except Exception as e:
logger.error(f"Error getting page history for page {page_id}: {str(e)}")
raise Exception(f"Error getting page history: {str(e)}") from e