MCP Atlassian
by sooperset
- mcp-atlassian
- src
- mcp_atlassian
- confluence
"""Base client module for Confluence API interactions."""
import logging
from typing import Any
from atlassian import Confluence
from ..utils import configure_ssl_verification
from .config import ConfluenceConfig
# Configure logging
logger = logging.getLogger("mcp-atlassian")
class ConfluenceClient:
"""Base client for Confluence API interactions."""
def __init__(self, config: ConfluenceConfig | None = None) -> None:
"""Initialize the Confluence client with given or environment config.
Args:
config: Configuration for Confluence client. If None, will load from
environment.
Raises:
ValueError: If configuration is invalid or environment variables are missing
"""
self.config = config or ConfluenceConfig.from_env()
# Initialize the Confluence client based on auth type
if self.config.auth_type == "token":
self.confluence = Confluence(
url=self.config.url,
token=self.config.personal_token,
cloud=self.config.is_cloud,
verify_ssl=self.config.ssl_verify,
)
else: # basic auth
self.confluence = Confluence(
url=self.config.url,
username=self.config.username,
password=self.config.api_token, # API token is used as password
cloud=self.config.is_cloud,
)
# Configure SSL verification using the shared utility
configure_ssl_verification(
service_name="Confluence",
url=self.config.url,
session=self.confluence._session,
ssl_verify=self.config.ssl_verify,
)
# Import here to avoid circular imports
from ..preprocessing.confluence import ConfluencePreprocessor
self.preprocessor = ConfluencePreprocessor(
base_url=self.config.url, confluence_client=self.confluence
)
def get_user_details_by_accountid(
self, account_id: str, expand: str = None
) -> dict[str, Any]:
"""Get user details by account ID.
Args:
account_id: The account ID of the user
expand: OPTIONAL expand for get status of user.
Possible param is "status". Results are "Active, Deactivated"
Returns:
User details as a dictionary
Raises:
Various exceptions from the Atlassian API if user doesn't exist or
if there are permission issues
"""
return self.confluence.get_user_details_by_accountid(account_id, expand)
def _process_html_content(
self, html_content: str, space_key: str
) -> tuple[str, str]:
"""Process HTML content into both HTML and markdown formats.
Args:
html_content: Raw HTML content from Confluence
space_key: The key of the space containing the content
Returns:
Tuple of (processed_html, processed_markdown)
"""
return self.preprocessor.process_html_content(html_content, space_key)