atlassian

9 files shipped with this skill. Inline text is the catalog copy. The upstream link is the source repository.

• skills/atlassian/SKILL.md6.2 KB8b1f82146d64upstream ↗

  ---
  name: atlassian
  description: |
    Manage Jira issues and Confluence wiki pages in Atlassian Cloud.
    Use when: (1) searching/creating/updating Jira issues with JQL, (2) searching/reading/creating Confluence pages
    with CQL, (3) managing Jira workflows, transitions, and comments, (4) browsing Confluence spaces and page
    hierarchies. Supports OAuth 2.1 via MCP server (recommended) or API token authentication (fallback).
  ---
  
  # Atlassian (Jira + Confluence)
  
  Full Jira and Confluence integration with two authentication methods:
  - **OAuth 2.1** via Atlassian MCP server — browser-based consent, auto-refresh tokens, calls MCP tools
  - **API token** — email + token stored in keyring, calls REST API directly
  
  ## First-Time Setup
  
  ### Option 1: OAuth 2.1 via MCP Server (Recommended)
  
  No API tokens or instance URLs needed. Uses dynamic client registration and PKCE.
  
  ```bash
  pip install -r requirements.txt
  python scripts/auth.py login --oauth
  ```
  
  A browser opens for Atlassian authorization. Select which products (Jira, Confluence, Compass) to grant access. Tokens are stored in the system keyring and auto-refresh when expired.
  
  Check status:
  ```bash
  python scripts/auth.py status
  ```
  
  ### Option 2: API Token (Fallback)
  
  For environments where browser-based OAuth isn't available.
  
  ```bash
  pip install -r requirements.txt
  python scripts/auth.py login
  ```
  
  Follow the prompts to enter your Atlassian URL, email, and API token. Credentials are stored securely in the system keyring.
  
  Create an API token at: https://id.atlassian.com/manage-profile/security/api-tokens
  
  Check authentication status:
  ```bash
  python scripts/auth.py status
  ```
  
  Logout (clears both OAuth and API token credentials):
  ```bash
  python scripts/auth.py logout
  ```
  
  ## Backend Selection
  
  The scripts automatically detect which backend to use based on your auth type:
  - **OAuth** → MCP backend (calls Atlassian MCP server tools)
  - **API token** → REST backend (calls Atlassian REST API directly)
  
  All commands work identically regardless of backend.
  
  ## Jira (scripts/jira.py)
  
  ### Search issues with JQL
  ```bash
  python scripts/jira.py search "project = DEV AND status = Open"
  python scripts/jira.py search "assignee = currentUser() ORDER BY updated DESC" --limit 10
  ```
  
  ### Get issue details
  ```bash
  python scripts/jira.py get DEV-123
  ```
  
  ### Create an issue
  ```bash
  python scripts/jira.py create --project DEV --summary "Fix login bug" --type Bug
  python scripts/jira.py create --project DEV --summary "New feature" --type Story \
    --description "Details here" --priority High --assignee "[email protected]" --labels "backend,urgent"
  ```
  
  ### Update an issue
  ```bash
  python scripts/jira.py update DEV-123 --summary "Updated summary" --priority High
  python scripts/jira.py update DEV-123 --assignee "[email protected]"
  ```
  
  ### Transition issue status
  ```bash
  python scripts/jira.py transition DEV-123 "In Progress"
  python scripts/jira.py transition DEV-123 "Done"
  ```
  
  ### Add and list comments
  ```bash
  python scripts/jira.py comment DEV-123 --add "This is a comment"
  python scripts/jira.py comment DEV-123 --list
  ```
  
  ### List projects and statuses
  ```bash
  python scripts/jira.py list-projects
  python scripts/jira.py list-statuses DEV
  ```
  
  ### Test authentication
  ```bash
  python scripts/jira.py auth-info
  ```
  
  ### List available MCP tools (OAuth only)
  ```bash
  python scripts/jira.py list-tools
  ```
  
  ## Confluence (scripts/confluence.py)
  
  ### Search pages
  ```bash
  python scripts/confluence.py search "deployment guide"
  python scripts/confluence.py search "type=page AND space=DEV AND text~\"deployment\""
  python scripts/confluence.py search "onboarding" --limit 10
  ```
  
  ### Read a page
  ```bash
  python scripts/confluence.py read <page-id>
  python scripts/confluence.py read <page-id> --json
  ```
  
  ### List spaces
  ```bash
  python scripts/confluence.py list-spaces
  python scripts/confluence.py list-spaces --limit 50
  ```
  
  ### Get space details
  ```bash
  python scripts/confluence.py get-space <space-id>
  ```
  
  ### List pages in a space
  ```bash
  python scripts/confluence.py list-pages --space-id <space-id>
  ```
  
  ### Create a page
  ```bash
  python scripts/confluence.py create --title "New Page" --space-id <space-id>
  python scripts/confluence.py create --title "Guide" --space-id <id> --body "<p>Content here</p>"
  python scripts/confluence.py create --title "Child" --space-id <id> --parent-id <parent-id>
  ```
  
  ### Update a page
  ```bash
  python scripts/confluence.py update <page-id> --title "Updated Title"
  python scripts/confluence.py update <page-id> --body "<p>New content</p>"
  ```
  
  ### Get child pages
  ```bash
  python scripts/confluence.py get-children <page-id>
  ```
  
  ### Test authentication
  ```bash
  python scripts/confluence.py auth-info
  ```
  
  ### List available MCP tools (OAuth only)
  ```bash
  python scripts/confluence.py list-tools
  ```
  
  ## Operations Reference
  
  ### Jira
  
  | Command | Description | Required Args |
  |---------|-------------|---------------|
  | search | Search issues with JQL | jql |
  | get | Get issue details | issue_key |
  | create | Create new issue | --project, --summary, --type |
  | update | Update existing issue | issue_key |
  | transition | Change issue status | issue_key, status |
  | comment | Add or list comments | issue_key |
  | list-projects | List accessible projects | - |
  | list-statuses | List statuses for project | project_key |
  | auth-info | Test API connection | - |
  | list-tools | List MCP tools (OAuth only) | - |
  
  ### Confluence
  
  | Command | Description | Required Args |
  |---------|-------------|---------------|
  | search | Search using CQL | query |
  | read | Get page content | page_id |
  | list-spaces | List all spaces | - |
  | get-space | Get space details | space_id |
  | list-pages | List pages in a space | --space-id |
  | create | Create new page | --title, --space-id |
  | update | Update existing page | page_id |
  | get-children | Get child pages | page_id |
  | auth-info | Test API connection | - |
  | list-tools | List MCP tools (OAuth only) | - |
  
  ## JSON Output
  
  Add `--json` flag to any script command for machine-readable output.
  
  ## Token Management
  
  Credentials stored securely using the system keyring:
  - **macOS**: Keychain
  - **Windows**: Windows Credential Locker
  - **Linux**: Secret Service API
  
  Service name: `atlassian-skill`
  
  OAuth tokens auto-refresh when expired (if refresh token is available).
  

• skills/atlassian/scripts/api_client.py5.4 KB7638f8f7939eupstream ↗

  #!/usr/bin/env python3
  """
  Shared client layer for Atlassian Cloud APIs (Jira + Confluence).
  
  Supports two backends:
    - MCP: OAuth-authenticated calls to the Atlassian MCP server (tools/call)
    - REST: API token-authenticated calls to the Atlassian REST API (httpx)
  
  Domain scripts (jira.py, confluence.py) use `get_backend()` to determine
  which backend is active, then call accordingly.
  """
  
  import json
  import sys
  from typing import Any, Dict, Optional
  
  from auth import get_auth_header, get_auth_type, require_config
  
  # httpx is deferred — only needed for REST backend (API token auth).
  # MCP-only users (OAuth) don't need httpx installed.
  httpx = None
  
  
  def _ensure_httpx():
      """Import httpx on demand so MCP-only users don't need it installed."""
      global httpx
      if httpx is None:
          try:
              import httpx as _httpx  # type: ignore[no-redef]
              httpx = _httpx
          except ImportError:
              print("Error: httpx is required for REST backend. Install with: pip install httpx", file=sys.stderr)
              sys.exit(1)
  
  
  class AtlassianAPIError(Exception):
      """Raised when an Atlassian API returns an error."""
      pass
  
  
  class AtlassianClient:
      """HTTP client for Atlassian REST APIs (API token auth)."""
  
      def __init__(self, base_url: str, auth_header: str, timeout: int = 30):
          _ensure_httpx()
          self.base_url = base_url.rstrip("/")
          self._auth_header = auth_header
          self.client = httpx.Client(
              timeout=timeout,
              headers={
                  "Accept": "application/json",
                  "Authorization": auth_header,
              },
          )
  
      def request(
          self,
          method: str,
          url: str,
          data: Optional[Dict[str, Any]] = None,
          params: Optional[Dict[str, Any]] = None,
      ) -> dict:
          """Make an authenticated HTTP request."""
          try:
              headers = {"Content-Type": "application/json"} if data is not None else None
              response = self.client.request(
                  method, url, params=params, json=data, headers=headers,
              )
  
              if response.status_code == 204:
                  return {}
  
              response.raise_for_status()
  
              if response.content:
                  return response.json()
              return {}
  
          except httpx.HTTPStatusError as e:
              error_body = e.response.text
              try:
                  error_json = e.response.json()
                  errors = error_json.get("errorMessages", [])
                  field_errors = error_json.get("errors", {})
                  message = error_json.get("message", "")
  
                  parts = []
                  if message:
                      parts.append(message)
                  if errors:
                      parts.extend(errors)
                  if field_errors and isinstance(field_errors, dict):
                      parts.extend(f"{k}: {v}" for k, v in field_errors.items())
                  error_msg = "; ".join(parts) if parts else error_body
              except Exception:
                  error_msg = error_body
  
              if self._auth_header:
                  token_part = self._auth_header.split(" ", 1)[-1]
                  error_msg = error_msg.replace(token_part, "***")
  
              raise AtlassianAPIError(f"HTTP {e.response.status_code}: {error_msg}")
          except httpx.RequestError as e:
              raise AtlassianAPIError(f"Request failed: {e}")
  
      def get(self, url: str, params: Optional[dict] = None) -> dict:
          """Make a GET request."""
          return self.request("GET", url, params=params)
  
      def post(self, url: str, data: dict) -> dict:
          """Make a POST request with JSON body."""
          return self.request("POST", url, data=data)
  
      def put(self, url: str, data: dict) -> dict:
          """Make a PUT request with JSON body."""
          return self.request("PUT", url, data=data)
  
      def close(self):
          """Close the HTTP client."""
          self.client.close()
  
  
  def get_backend() -> str:
      """
      Determine which backend to use based on auth type.
  
      Returns:
          "mcp" for OAuth authentication (uses MCP tools)
          "rest" for API token authentication (uses REST API)
      """
      auth_type = get_auth_type()
      if auth_type == "oauth":
          return "mcp"
      return "rest"
  
  
  def create_rest_client(timeout: int = 30) -> AtlassianClient:
      """
      Create an authenticated REST API client.
      Exits with helpful error if not authenticated with API token.
      """
      config = require_config()
      if config.get("auth_type") == "oauth":
          print("Error: REST client requires API token authentication.", file=sys.stderr)
          print("Current auth type is OAuth. Use MCP backend instead,", file=sys.stderr)
          print("or re-authenticate: python scripts/auth.py login", file=sys.stderr)
          sys.exit(1)
  
      auth_header = get_auth_header()
      if not auth_header:
          print("Error: Failed to build auth header.", file=sys.stderr)
          sys.exit(1)
      return AtlassianClient(
          base_url=config["base_url"],
          auth_header=auth_header,
          timeout=timeout,
      )
  
  
  # Keep backward compat alias
  create_client = create_rest_client
  
  
  def output_result(data: Any, as_json: bool = False):
      """Output result in requested format."""
      if as_json:
          if hasattr(data, "__dataclass_fields__"):
              data = {k: v for k, v in data.__dict__.items() if v is not None}
          print(json.dumps(data, indent=2, default=str))
      else:
          print(data)
  

• skills/atlassian/scripts/auth.py23 KBecbb3bd62778upstream ↗

  #!/usr/bin/env python3
  """
  Authentication management for Atlassian Cloud APIs (Jira + Confluence).
  
  Supports two authentication methods:
    1. OAuth 2.1 via Atlassian MCP server (recommended)
       - Dynamic client registration (RFC 7591)
       - Authorization Code + PKCE (S256)
       - Tokens stored in system keyring
    2. API token with Basic auth (fallback)
       - Email + API token stored in system keyring or env vars
  
  Credentials stored in system keyring:
    - macOS: Keychain
    - Windows: Windows Credential Locker
    - Linux: Secret Service API (GNOME Keyring, KDE Wallet)
  
  Usage:
      python3 auth.py login --oauth     # OAuth 2.1 via MCP server
      python3 auth.py login             # API token (email + token)
      python3 auth.py status
      python3 auth.py logout
  """
  
  import argparse
  import base64
  import hashlib
  import html
  import http.server
  import json
  import os
  import secrets
  import sys
  import time
  import urllib.error
  import urllib.parse
  import urllib.request
  import webbrowser
  from pathlib import Path
  
  KEYCHAIN_SERVICE = "atlassian-skill"
  KEYCHAIN_ACCOUNT = "default"
  
  # OAuth 2.1 endpoints (from Atlassian MCP server)
  OAUTH_METADATA_URL = "https://mcp.atlassian.com/.well-known/oauth-authorization-server"
  OAUTH_REGISTER_URL = "https://cf.mcp.atlassian.com/v1/register"
  OAUTH_AUTHORIZE_URL = "https://mcp.atlassian.com/v1/authorize"
  OAUTH_TOKEN_URL = "https://cf.mcp.atlassian.com/v1/token"
  MCP_ENDPOINT = "https://mcp.atlassian.com/v1/mcp"
  
  # Local callback server
  CALLBACK_HOST = "127.0.0.1"
  CALLBACK_PORT = 39827  # arbitrary high port
  
  # Try importing keyring
  try:
      import keyring
      HAS_KEYRING = True
  except ImportError:
      HAS_KEYRING = False
  
  # Try importing dotenv
  try:
      from dotenv import load_dotenv
  except ImportError:
      load_dotenv = None
  
  
  def _load_env():
      """Load environment variables from .env file if python-dotenv is available."""
      if load_dotenv is not None:
          skill_dir = Path(__file__).parent.parent
          env_file = skill_dir / ".env"
          if env_file.exists():
              load_dotenv(env_file)
          elif Path(".env").exists():
              load_dotenv()
  
  
  _load_env()
  
  
  # ─── Storage ────────────────────────────────────────────────────────────────
  
  def _get_keyring_config():
      """Get stored config from keyring."""
      if not HAS_KEYRING:
          return None
      try:
          data_str = keyring.get_password(KEYCHAIN_SERVICE, KEYCHAIN_ACCOUNT)
          if not data_str:
              return None
          return json.loads(data_str)
      except json.JSONDecodeError:
          return None
      except Exception as e:
          print(f"Warning: keyring read failed: {e}", file=sys.stderr)
          return None
  
  
  def _save_keyring_config(config):
      """Save config to keyring."""
      if not HAS_KEYRING:
          return False
      try:
          keyring.set_password(KEYCHAIN_SERVICE, KEYCHAIN_ACCOUNT, json.dumps(config))
          return True
      except Exception as e:
          print(f"Warning: keyring write failed: {e}", file=sys.stderr)
          return False
  
  
  def _clear_keyring_config():
      """Clear stored credentials from keyring."""
      if not HAS_KEYRING:
          return False
      try:
          keyring.delete_password(KEYCHAIN_SERVICE, KEYCHAIN_ACCOUNT)
          return True
      except Exception as e:
          print(f"Warning: keyring clear failed: {e}", file=sys.stderr)
          return False
  
  
  def _get_env_config():
      """Get API token config from environment variables."""
      base_url = os.environ.get("ATLASSIAN_URL", "")
      email = os.environ.get("ATLASSIAN_EMAIL", "")
      api_token = os.environ.get("ATLASSIAN_API_TOKEN", "")
  
      if base_url and email and api_token:
          return {
              "auth_type": "api_token",
              "base_url": base_url.rstrip("/"),
              "email": email,
              "api_token": api_token,
          }
      return None
  
  
  # ─── PKCE helpers ───────────────────────────────────────────────────────────
  
  def _generate_pkce():
      """Generate PKCE code_verifier and code_challenge (S256)."""
      code_verifier = secrets.token_urlsafe(64)[:128]
      digest = hashlib.sha256(code_verifier.encode("ascii")).digest()
      code_challenge = base64.urlsafe_b64encode(digest).rstrip(b"=").decode("ascii")
      return code_verifier, code_challenge
  
  
  # ─── OAuth callback server ──────────────────────────────────────────────────
  
  class _OAuthCallbackHandler(http.server.BaseHTTPRequestHandler):
      """HTTP handler that captures the OAuth authorization code callback."""
  
      auth_code = None
      auth_error = None
      state_received = None
  
      def do_GET(self):
          parsed = urllib.parse.urlparse(self.path)
          params = urllib.parse.parse_qs(parsed.query)
  
          if "code" in params:
              _OAuthCallbackHandler.auth_code = params["code"][0]
              _OAuthCallbackHandler.state_received = params.get("state", [None])[0]
              self.send_response(200)
              self.send_header("Content-Type", "text/html")
              self.end_headers()
              self.wfile.write(b"<html><body><h2>Authorization successful!</h2>"
                               b"<p>You can close this browser tab and return to the terminal.</p>"
                               b"</body></html>")
          elif "error" in params:
              _OAuthCallbackHandler.auth_error = params.get("error_description",
                                                             params.get("error", ["Unknown error"]))[0]
              self.send_response(400)
              self.send_header("Content-Type", "text/html")
              self.end_headers()
              error_msg = html.escape(_OAuthCallbackHandler.auth_error)
              self.wfile.write(f"<html><body><h2>Authorization failed</h2>"
                               f"<p>{error_msg}</p></body></html>".encode())
          else:
              self.send_response(404)
              self.end_headers()
  
      def log_message(self, format, *args):  # type: ignore[override]
          pass  # suppress HTTP logs
  
  
  def _wait_for_callback(server, timeout=120):
      """Run the callback server until we get a response or timeout."""
      server.timeout = 1
      deadline = time.time() + timeout
      while time.time() < deadline:
          server.handle_request()
          if _OAuthCallbackHandler.auth_code or _OAuthCallbackHandler.auth_error:
              break
  
  
  # ─── OAuth flow ─────────────────────────────────────────────────────────────
  
  def _http_json_request(url, data=None, method=None, headers=None):
      """Make an HTTP request and return parsed JSON response."""
      if data is not None:
          if isinstance(data, dict):
              body = json.dumps(data).encode("utf-8")
              content_type = "application/json"
          else:
              body = data.encode("utf-8") if isinstance(data, str) else data
              content_type = "application/x-www-form-urlencoded"
          if method is None:
              method = "POST"
      else:
          body = None
          content_type = None
          if method is None:
              method = "GET"
  
      req = urllib.request.Request(url, data=body, method=method)
      req.add_header("User-Agent", "atlassian-skill/3.0")
      req.add_header("Accept", "application/json")
      if content_type:
          req.add_header("Content-Type", content_type)
      if headers:
          for k, v in headers.items():
              req.add_header(k, v)
  
      try:
          with urllib.request.urlopen(req, timeout=30) as resp:
              return json.loads(resp.read().decode("utf-8"))
      except urllib.error.HTTPError as e:
          error_body = e.read().decode("utf-8", errors="replace")
          try:
              error_json = json.loads(error_body)
              msg = error_json.get("error_description", error_json.get("error", error_body))
          except Exception:
              msg = error_body
          raise RuntimeError(f"HTTP {e.code}: {msg}")
  
  
  def _register_oauth_client(redirect_uri):
      """Dynamically register a public OAuth client (RFC 7591)."""
      registration_data = {
          "client_name": "atlassian-skill-cli",
          "redirect_uris": [redirect_uri],
          "grant_types": ["authorization_code", "refresh_token"],
          "response_types": ["code"],
          "token_endpoint_auth_method": "none",
      }
      return _http_json_request(OAUTH_REGISTER_URL, data=registration_data)
  
  
  def _exchange_code_for_tokens(client_id, auth_code, redirect_uri, code_verifier):
      """Exchange authorization code for tokens at the token endpoint."""
      token_data = urllib.parse.urlencode({
          "grant_type": "authorization_code",
          "client_id": client_id,
          "code": auth_code,
          "redirect_uri": redirect_uri,
          "code_verifier": code_verifier,
      })
      return _http_json_request(OAUTH_TOKEN_URL, data=token_data)
  
  
  def _refresh_access_token(client_id, refresh_token):
      """Use refresh token to get a new access token."""
      token_data = urllib.parse.urlencode({
          "grant_type": "refresh_token",
          "client_id": client_id,
          "refresh_token": refresh_token,
      })
      return _http_json_request(OAUTH_TOKEN_URL, data=token_data)
  
  
  def oauth_login():
      """Run the full OAuth 2.1 + PKCE login flow."""
      redirect_uri = f"http://{CALLBACK_HOST}:{CALLBACK_PORT}/callback"
  
      # Step 1: Dynamic client registration
      print("Registering OAuth client...", file=sys.stderr)
      try:
          reg = _register_oauth_client(redirect_uri)
      except RuntimeError as e:
          print(f"Error: Client registration failed: {e}", file=sys.stderr)
          sys.exit(1)
  
      client_id = reg.get("client_id")
      if not client_id:
          print("Error: No client_id returned from registration.", file=sys.stderr)
          sys.exit(1)
  
      # Step 2: Generate PKCE values
      code_verifier, code_challenge = _generate_pkce()
      state = secrets.token_urlsafe(32)
  
      # Step 3: Build authorization URL
      auth_params = urllib.parse.urlencode({
          "response_type": "code",
          "client_id": client_id,
          "redirect_uri": redirect_uri,
          "code_challenge": code_challenge,
          "code_challenge_method": "S256",
          "state": state,
      })
      auth_url = f"{OAUTH_AUTHORIZE_URL}?{auth_params}"
  
      # Step 4: Start local callback server
      _OAuthCallbackHandler.auth_code = None
      _OAuthCallbackHandler.auth_error = None
      _OAuthCallbackHandler.state_received = None
  
      server = http.server.HTTPServer((CALLBACK_HOST, CALLBACK_PORT), _OAuthCallbackHandler)
  
      # Step 5: Open browser for authorization
      print(f"\nOpening browser for Atlassian authorization...", file=sys.stderr)
      print(f"If the browser doesn't open, visit this URL:\n{auth_url}\n", file=sys.stderr)
      webbrowser.open(auth_url)
  
      # Step 6: Wait for callback
      print("Waiting for authorization (2 minute timeout)...", file=sys.stderr)
      try:
          _wait_for_callback(server, timeout=120)
      finally:
          server.server_close()
  
      if _OAuthCallbackHandler.auth_error:
          print(f"Error: Authorization failed: {_OAuthCallbackHandler.auth_error}", file=sys.stderr)
          sys.exit(1)
  
      if not _OAuthCallbackHandler.auth_code:
          print("Error: No authorization code received (timed out).", file=sys.stderr)
          sys.exit(1)
  
      # Validate state
      if _OAuthCallbackHandler.state_received != state:
          print("Error: State mismatch — possible CSRF attack.", file=sys.stderr)
          sys.exit(1)
  
      # Step 7: Exchange code for tokens
      print("Exchanging code for tokens...", file=sys.stderr)
      try:
          tokens = _exchange_code_for_tokens(
              client_id,
              _OAuthCallbackHandler.auth_code,
              redirect_uri,
              code_verifier,
          )
      except RuntimeError as e:
          print(f"Error: Token exchange failed: {e}", file=sys.stderr)
          sys.exit(1)
  
      access_token = tokens.get("access_token")
      refresh_token = tokens.get("refresh_token")
      expires_in = tokens.get("expires_in", 3600)
  
      if not access_token:
          print("Error: No access_token in response.", file=sys.stderr)
          sys.exit(1)
  
      # Step 8: Store in keyring
      config = {
          "auth_type": "oauth",
          "client_id": client_id,
          "access_token": access_token,
          "refresh_token": refresh_token,
          "expires_at": int(time.time()) + expires_in,
          "mcp_endpoint": MCP_ENDPOINT,
      }
  
      if _save_keyring_config(config):
          print("OAuth login successful!")
          print(f"Token expires in {expires_in // 60} minutes.")
          if refresh_token:
              print("Refresh token stored — will auto-refresh when expired.")
          print("Credentials stored in system keyring.")
      else:
          print("Error: Failed to store credentials in keyring.", file=sys.stderr)
          sys.exit(1)
  
  
  def ensure_valid_oauth_token():
      """
      Check if the OAuth token is expired and refresh if needed.
      Returns the config with a valid access_token, or None if refresh fails.
      """
      config = _get_keyring_config()
      if not config or config.get("auth_type") != "oauth":
          return config
  
      expires_at = config.get("expires_at", 0)
      # Refresh if token expires within 60 seconds
      if time.time() < expires_at - 60:
          return config
  
      refresh_token = config.get("refresh_token")
      if not refresh_token:
          return None
  
      try:
          tokens = _refresh_access_token(config["client_id"], refresh_token)
      except RuntimeError:
          return None
  
      config["access_token"] = tokens.get("access_token", config["access_token"])
      if tokens.get("refresh_token"):
          config["refresh_token"] = tokens["refresh_token"]
      config["expires_at"] = int(time.time()) + tokens.get("expires_in", 3600)
  
      _save_keyring_config(config)
      return config
  
  
  # ─── Validation (API token) ─────────────────────────────────────────────────
  
  def _validate_credentials(base_url, email, api_token):
      """Validate API token credentials by calling Jira /rest/api/3/myself."""
      url = f"{base_url.rstrip('/')}/rest/api/3/myself"
      creds = base64.b64encode(f"{email}:{api_token}".encode()).decode()
      headers = {
          "Authorization": f"Basic {creds}",
          "Accept": "application/json",
      }
      try:
          req = urllib.request.Request(url, headers=headers)
          req.add_header("User-Agent", "atlassian-skill/3.0")
          with urllib.request.urlopen(req, timeout=15) as resp:
              return json.loads(resp.read().decode("utf-8"))
      except urllib.error.HTTPError as e:
          body = e.read().decode("utf-8", errors="replace")[:200]
          print(f"  Validation failed (HTTP {e.code}): {body}", file=sys.stderr)
          return None
      except (urllib.error.URLError, Exception) as e:
          print(f"  Validation failed: {e}", file=sys.stderr)
          return None
  
  
  # ─── Public API (used by api_client.py and domain scripts) ──────────────────
  
  def get_auth_type():
      """
      Get the current authentication type.
  
      Returns:
          "oauth" | "api_token" | None
      """
      config = get_config()
      if not config:
          return None
      return config.get("auth_type", "api_token")
  
  
  def get_config():
      """
      Get Atlassian config. Priority: keyring > environment variables.
  
      For OAuth: returns dict with auth_type, client_id, access_token, refresh_token, etc.
      For API token: returns dict with auth_type, base_url, email, api_token.
      Returns None if not configured.
      """
      config = _get_keyring_config()
      if config:
          # Ensure auth_type is set (backward compat with old configs)
          if "auth_type" not in config:
              config["auth_type"] = "api_token"
          # Auto-refresh OAuth tokens
          if config.get("auth_type") == "oauth":
              config = ensure_valid_oauth_token()
          return config
  
      config = _get_env_config()
      if config:
          return config
  
      return None
  
  
  def get_auth_header():
      """
      Get the auth header value.
  
      For OAuth: returns "Bearer <access_token>"
      For API token: returns "Basic <base64(email:token)>"
      """
      config = get_config()
      if not config:
          return None
  
      if config.get("auth_type") == "oauth":
          access_token = config.get("access_token")
          if access_token:
              return f"Bearer {access_token}"
          return None
  
      # API token path
      email = config.get("email", "")
      api_token = config.get("api_token", "")
      if email and api_token:
          creds = base64.b64encode(f"{email}:{api_token}".encode()).decode()
          return f"Basic {creds}"
      return None
  
  
  def get_base_url():
      """Get the Atlassian instance base URL (API token auth only)."""
      config = get_config()
      if not config:
          return None
      return config.get("base_url")
  
  
  def get_mcp_endpoint():
      """Get the MCP endpoint URL (OAuth auth only)."""
      config = get_config()
      if not config or config.get("auth_type") != "oauth":
          return None
      return config.get("mcp_endpoint", MCP_ENDPOINT)
  
  
  def require_config():
      """
      Get config or exit with helpful error message.
  
      Returns:
          dict with auth config
      """
      config = get_config()
      if config:
          return config
  
      print("Error: Not authenticated.", file=sys.stderr)
      print("\nTo authenticate, choose one of:", file=sys.stderr)
      if HAS_KEYRING:
          print("\n  Option 1 (recommended): OAuth via Atlassian MCP server", file=sys.stderr)
          print("  python scripts/auth.py login --oauth", file=sys.stderr)
          print("\n  Option 2: API token via keyring", file=sys.stderr)
          print("  python scripts/auth.py login", file=sys.stderr)
      print("\n  Option 3: Set environment variables", file=sys.stderr)
      print("  export ATLASSIAN_URL=https://your-domain.atlassian.net", file=sys.stderr)
      print("  export [email protected]", file=sys.stderr)
      print("  export ATLASSIAN_API_TOKEN=your-api-token", file=sys.stderr)
      print("\n  Get your API token at:", file=sys.stderr)
      print("  https://id.atlassian.com/manage-profile/security/api-tokens", file=sys.stderr)
      sys.exit(1)
  
  
  # ─── CLI ────────────────────────────────────────────────────────────────────
  
  def cmd_login(args):
      """Handle login command."""
      if args.oauth:
          if not HAS_KEYRING:
              print("Error: keyring package required for OAuth. Install with: pip install keyring",
                    file=sys.stderr)
              sys.exit(1)
          oauth_login()
          return
  
      # API token login
      if not HAS_KEYRING:
          print("Error: keyring package not installed.", file=sys.stderr)
          print("Install with: pip install keyring", file=sys.stderr)
          print("Or use environment variables instead (see .env.example).", file=sys.stderr)
          sys.exit(1)
  
      base_url = args.url
      if not base_url:
          base_url = input("Atlassian URL (e.g. https://your-domain.atlassian.net): ").strip()
      if not base_url:
          print("URL is required.", file=sys.stderr)
          sys.exit(1)
      base_url = base_url.rstrip("/")
  
      email = args.email
      if not email:
          email = input("Email: ").strip()
      if not email:
          print("Email is required.", file=sys.stderr)
          sys.exit(1)
  
      api_token = args.token
      if not api_token:
          import getpass
          api_token = getpass.getpass("API token: ").strip()
      if not api_token:
          print("API token is required.", file=sys.stderr)
          sys.exit(1)
  
      print("Validating credentials...", file=sys.stderr)
      user = _validate_credentials(base_url, email, api_token)
      if not user:
          print("Error: Invalid credentials. Check your URL, email, and API token.", file=sys.stderr)
          sys.exit(1)
  
      config = {
          "auth_type": "api_token",
          "base_url": base_url,
          "email": email,
          "api_token": api_token,
      }
      if _save_keyring_config(config):
          print(f"Login successful!")
          print(f"User: {user.get('displayName', 'Unknown')}")
          print(f"Instance: {base_url}")
          print(f"Credentials stored in system keyring.")
      else:
          print("Failed to store credentials in keyring.", file=sys.stderr)
          sys.exit(1)
  
  
  def cmd_logout(_args):
      """Handle logout command."""
      if _clear_keyring_config():
          print("Logged out successfully. Credentials removed from keyring.")
      else:
          print("No credentials to clear (or keyring not available).")
  
  
  def cmd_status(_args):
      """Handle status command."""
      config = get_config()
      if not config:
          print("Status: Not authenticated")
          print("\nRun: python scripts/auth.py login --oauth  (OAuth)")
          print("  or: python scripts/auth.py login           (API token)")
          sys.exit(1)
  
      auth_type = config.get("auth_type", "api_token")
      source = "keyring" if _get_keyring_config() else "environment variables"
  
      print(f"Status: Authenticated")
      print(f"Auth type: {auth_type}")
      print(f"Source: {source}")
  
      if auth_type == "oauth":
          expires_at = int(config.get("expires_at", 0))
          remaining = expires_at - int(time.time())
          if remaining > 0:
              print(f"Token expires in: {remaining // 60}m {remaining % 60}s")
          else:
              print(f"Token: expired (will auto-refresh)")
          has_refresh = "yes" if config.get("refresh_token") else "no"
          print(f"Refresh token: {has_refresh}")
          print(f"MCP endpoint: {config.get('mcp_endpoint', MCP_ENDPOINT)}")
      else:
          base_url = config.get("base_url", "unknown")
          email = config.get("email", "unknown")
          print(f"Instance: {base_url}")
          print(f"Email: {email}")
  
          # Validate connection
          print("\nTesting connection...", file=sys.stderr)
          user = _validate_credentials(base_url, email, config.get("api_token", ""))
          if user:
              print(f"Connection: OK")
              print(f"User: {user.get('displayName', 'Unknown')}")
          else:
              print(f"Connection: FAILED (credentials may be expired)")
  
  
  def main():
      parser = argparse.ArgumentParser(description="Atlassian authentication management")
      subparsers = parser.add_subparsers(dest="command")
  
      login_parser = subparsers.add_parser("login", help="Authenticate with Atlassian Cloud")
      login_parser.add_argument("--oauth", action="store_true",
                                help="Use OAuth 2.1 via Atlassian MCP server (recommended)")
      login_parser.add_argument("--url", help="Atlassian instance URL (API token mode)")
      login_parser.add_argument("--email", help="Account email (API token mode)")
      login_parser.add_argument("--token", help="API token (API token mode)")
  
      subparsers.add_parser("logout", help="Clear stored credentials")
      subparsers.add_parser("status", help="Check authentication status")
  
      args = parser.parse_args()
  
      if args.command == "login":
          cmd_login(args)
      elif args.command == "logout":
          cmd_logout(args)
      elif args.command == "status":
          cmd_status(args)
      else:
          parser.print_help()
  
  
  if __name__ == "__main__":
      main()
  

• skills/atlassian/scripts/confluence.py27 KB269f2abf7a52upstream ↗

  #!/usr/bin/env python3
  """
  Confluence Wiki CLI - Search, read, and manage Confluence wiki pages.
  
  Supports two backends:
    - MCP: OAuth-authenticated calls to Atlassian MCP server (tools/call)
    - REST: API token-authenticated calls to Confluence REST API v2
  
  Usage:
      python3 confluence.py search "query"
      python3 confluence.py read <page-id>
      python3 confluence.py list-spaces
      python3 confluence.py get-space <space-id>
      python3 confluence.py list-pages --space-id <space-id>
      python3 confluence.py create --title "Title" --space-id <id> [--body "<p>content</p>"]
      python3 confluence.py update <page-id> --title "New Title"
      python3 confluence.py get-children <page-id>
      python3 confluence.py auth-info
      python3 confluence.py list-tools   # MCP only: show available tools
  """
  
  import argparse
  import html
  import json
  import re
  import sys
  from dataclasses import dataclass
  from typing import Any, Dict, Optional
  
  from api_client import AtlassianAPIError, AtlassianClient, create_rest_client, get_backend, output_result
  
  
  # ─── Data Models ─────────────────────────────────────────────────────────────
  
  @dataclass
  class ConfluencePage:
      """Represents a Confluence page."""
      id: str
      title: str
      space_id: Optional[str] = None
      status: Optional[str] = None
      body: Optional[str] = None
      version_number: Optional[int] = None
      parent_id: Optional[str] = None
      created_at: Optional[str] = None
      updated_at: Optional[str] = None
      url: Optional[str] = None
  
      @classmethod
      def from_v2_dict(cls, data: dict, base_url: str = "") -> "ConfluencePage":
          """Create from v2 API response."""
          body_content = None
          if "body" in data:
              body_obj = data["body"]
              if "storage" in body_obj:
                  body_content = body_obj["storage"].get("value", "")
              elif "view" in body_obj:
                  body_content = body_obj["view"].get("value", "")
              elif "atlas_doc_format" in body_obj:
                  body_content = body_obj["atlas_doc_format"].get("value", "")
  
          version_num = None
          if "version" in data:
              version_num = data["version"].get("number")
  
          page_url = None
          if base_url and data.get("_links", {}).get("webui"):
              page_url = base_url.rstrip("/") + data["_links"]["webui"]
  
          return cls(
              id=str(data.get("id", "")),
              title=data.get("title", ""),
              space_id=data.get("spaceId"),
              status=data.get("status"),
              body=body_content,
              version_number=version_num,
              parent_id=data.get("parentId"),
              created_at=data.get("createdAt"),
              updated_at=data.get("version", {}).get("createdAt"),
              url=page_url,
          )
  
      @classmethod
      def from_v1_search(cls, data: dict, base_url: str = "") -> "ConfluencePage":
          """Create from v1 search API response."""
          body_content = None
          if "body" in data:
              body_obj = data["body"]
              if "storage" in body_obj:
                  body_content = body_obj["storage"].get("value", "")
              elif "view" in body_obj:
                  body_content = body_obj["view"].get("value", "")
  
          version_num = None
          if "version" in data:
              version_num = data["version"].get("number")
  
          page_url = None
          if base_url and data.get("_links", {}).get("webui"):
              page_url = base_url.rstrip("/") + data["_links"]["webui"]
  
          return cls(
              id=str(data.get("id", "")),
              title=data.get("title", ""),
              space_id=data.get("space", {}).get("id") if isinstance(data.get("space"), dict) else None,
              status=data.get("status"),
              body=body_content,
              version_number=version_num,
              parent_id=None,
              created_at=None,
              updated_at=data.get("version", {}).get("when"),
              url=page_url,
          )
  
  
  @dataclass
  class ConfluenceSpace:
      """Represents a Confluence space."""
      id: str
      key: str
      name: str
      description: Optional[str] = None
      type: Optional[str] = None
      status: Optional[str] = None
      homepage_id: Optional[str] = None
  
      @classmethod
      def from_v2_dict(cls, data: dict) -> "ConfluenceSpace":
          return cls(
              id=str(data.get("id", "")),
              key=data.get("key", ""),
              name=data.get("name", ""),
              description=data.get("description", {}).get("plain", {}).get("value") if isinstance(data.get("description"), dict) else data.get("description"),
              type=data.get("type"),
              status=data.get("status"),
              homepage_id=str(data.get("homepageId", "")) if data.get("homepageId") else None,
          )
  
  
  # ─── Helpers ─────────────────────────────────────────────────────────────────
  
  def strip_html_tags(text: str) -> str:
      """Strip HTML tags and decode entities for readable output."""
      if not text:
          return ""
      clean = re.sub(r"<[^>]+>", " ", text)
      clean = html.unescape(clean)
      clean = re.sub(r"\s+", " ", clean).strip()
      return clean
  
  
  # ─── REST Client ─────────────────────────────────────────────────────────────
  
  class ConfluenceClient:
      """Confluence API client wrapping shared AtlassianClient."""
  
      def __init__(self, client: AtlassianClient):
          self.client = client
          self.base_url = client.base_url
          wiki_base = self.base_url
          if not wiki_base.endswith("/wiki"):
              wiki_base = f"{wiki_base}/wiki"
          self.wiki_base = wiki_base
          self.api_url = f"{wiki_base}/api/v2"
          self.v1_api_url = f"{wiki_base}/rest/api"
  
      def get_current_user(self) -> dict:
          url = f"{self.wiki_base}/rest/api/user/current"
          return self.client.get(url)
  
      def search(self, cql: str, limit: int = 25, start: int = 0) -> dict:
          """Search content using CQL (v1 API)."""
          url = f"{self.v1_api_url}/content/search"
          params = {
              "cql": cql,
              "limit": min(limit, 100),
              "start": start,
              "expand": "version,space",
          }
          result = self.client.get(url, params=params)
          pages = [ConfluencePage.from_v1_search(r, self.wiki_base) for r in result.get("results", [])]
          return {
              "pages": pages,
              "total_size": result.get("totalSize", 0),
              "start": result.get("start", 0),
              "limit": result.get("limit", limit),
          }
  
      def get_page(self, page_id: str) -> ConfluencePage:
          url = f"{self.api_url}/pages/{page_id}"
          params = {"body-format": "storage"}
          result = self.client.get(url, params=params)
          return ConfluencePage.from_v2_dict(result, self.wiki_base)
  
      def list_spaces(self, limit: int = 25, cursor: Optional[str] = None) -> dict:
          url = f"{self.api_url}/spaces"
          params: dict = {"limit": min(limit, 250)}
          if cursor:
              params["cursor"] = cursor
          result = self.client.get(url, params=params)
          spaces = [ConfluenceSpace.from_v2_dict(s) for s in result.get("results", [])]
          return {
              "spaces": spaces,
              "next_cursor": result.get("_links", {}).get("next"),
          }
  
      def get_space(self, space_id: str) -> ConfluenceSpace:
          url = f"{self.api_url}/spaces/{space_id}"
          params = {"description-format": "plain"}
          result = self.client.get(url, params=params)
          return ConfluenceSpace.from_v2_dict(result)
  
      def list_pages(self, space_id: str, limit: int = 25, cursor: Optional[str] = None, status: str = "current") -> dict:
          url = f"{self.api_url}/spaces/{space_id}/pages"
          params: dict = {"limit": min(limit, 250), "status": status}
          if cursor:
              params["cursor"] = cursor
          result = self.client.get(url, params=params)
          pages = [ConfluencePage.from_v2_dict(p, self.wiki_base) for p in result.get("results", [])]
          return {
              "pages": pages,
              "next_cursor": result.get("_links", {}).get("next"),
          }
  
      def create_page(self, space_id: str, title: str, body: Optional[str] = None,
                      parent_id: Optional[str] = None, status: str = "current") -> ConfluencePage:
          url = f"{self.api_url}/pages"
          payload: dict = {
              "spaceId": space_id,
              "status": status,
              "title": title,
              "body": {
                  "representation": "storage",
                  "value": body or "",
              },
          }
          if parent_id:
              payload["parentId"] = parent_id
          result = self.client.post(url, payload)
          return ConfluencePage.from_v2_dict(result, self.wiki_base)
  
      def update_page(self, page_id: str, title: Optional[str] = None,
                      body: Optional[str] = None, status: str = "current") -> ConfluencePage:
          """Update a page. Automatically handles version increment."""
          current = self.get_page(page_id)
          new_version = (current.version_number or 0) + 1
  
          url = f"{self.api_url}/pages/{page_id}"
          payload = {
              "id": page_id,
              "status": status,
              "title": title if title is not None else current.title,
              "version": {
                  "number": new_version,
                  "message": "Updated via Confluence CLI",
              },
              "body": {
                  "representation": "storage",
                  "value": body if body is not None else (current.body or ""),
              },
          }
          result = self.client.put(url, payload)
          return ConfluencePage.from_v2_dict(result, self.wiki_base)
  
      def get_children(self, page_id: str, limit: int = 25, cursor: Optional[str] = None) -> dict:
          url = f"{self.api_url}/pages/{page_id}/children"
          params: dict = {"limit": min(limit, 250)}
          if cursor:
              params["cursor"] = cursor
          result = self.client.get(url, params=params)
          pages = [ConfluencePage.from_v2_dict(p, self.wiki_base) for p in result.get("results", [])]
          return {
              "pages": pages,
              "next_cursor": result.get("_links", {}).get("next"),
          }
  
  
  # ─── MCP Backend ─────────────────────────────────────────────────────────────
  
  from mcp_client import AtlassianMCPClient, MCPError, mcp_output as _mcp_output  # noqa: E402
  
  
  class ConfluenceMCPClient(AtlassianMCPClient):
      """Confluence operations via Atlassian MCP server tools."""
  
      def __init__(self):
          super().__init__(product_name="Confluence")
  
  
  def run_mcp_command(args):
      """Execute a Confluence command via MCP backend."""
      client = ConfluenceMCPClient()
      cmd = args.command
  
      if cmd == "list-tools":
          tools = client.list_tools()
          if getattr(args, "json", False):
              _mcp_output(tools, as_json=True)
          else:
              conf_tools = [t for t in tools if "confluence" in t.get("name", "").lower()
                            or t.get("name", "") in ("search", "fetch",
                                                      "atlassianUserInfo",
                                                      "getAccessibleAtlassianResources")]
              print(f"Available Confluence-related MCP tools ({len(conf_tools)}):\n")
              for t in conf_tools:
                  desc = t.get("description", "")[:80]
                  print(f"  {t.get('name', '?')}")
                  if desc:
                      print(f"    {desc}")
                  print()
          return
  
      if cmd == "auth-info":
          result = client.call("atlassianUserInfo")
          _mcp_output(result, getattr(args, "json", False))
          return
  
      if cmd == "search":
          cql = args.query
          cql_operators = ["=", "~", "AND", "OR", "NOT", "IN", "order by"]
          if not any(op in cql for op in cql_operators):
              cql = f'type=page AND text~"{cql}"'
  
          result = client.call("searchConfluenceUsingCql", {
              "cql": cql,
              "limit": args.limit,
          })
          _mcp_output(result, getattr(args, "json", False))
          return
  
      if cmd == "read":
          result = client.call("getConfluencePage", {
              "pageId": args.page_id,
          })
          _mcp_output(result, getattr(args, "json", False))
          return
  
      if cmd == "list-spaces":
          result = client.call("getConfluenceSpaces", {
              "limit": args.limit,
          })
          _mcp_output(result, getattr(args, "json", False))
          return
  
      if cmd == "get-space":
          # MCP doesn't have a direct get-space tool, use getConfluenceSpaces with filter
          try:
              space_id_int = int(args.space_id)
          except ValueError:
              print(f"Error: Space ID must be numeric, got '{args.space_id}'.", file=sys.stderr)
              sys.exit(1)
          result = client.call("getConfluenceSpaces", {
              "ids": [space_id_int],
          })
          _mcp_output(result, getattr(args, "json", False))
          return
  
      if cmd == "list-pages":
          result = client.call("getPagesInConfluenceSpace", {
              "spaceId": args.space_id,
              "limit": args.limit,
          })
          _mcp_output(result, getattr(args, "json", False))
          return
  
      if cmd == "create":
          tool_args: Dict[str, Any] = {
              "spaceId": args.space_id,
              "title": args.title,
              "body": args.body if args.body else "",  # MCP requires body
          }
          if args.parent_id:
              tool_args["parentId"] = args.parent_id
          result = client.call("createConfluencePage", tool_args)
          if getattr(args, "json", False):
              _mcp_output(result, as_json=True)
          else:
              print("Page created successfully!")
              if isinstance(result, dict):
                  print(f"Title: {result.get('title', '')}")
                  print(f"ID: {result.get('id', '')}")
              else:
                  print(result)
          return
  
      if cmd == "update":
          tool_args_update: Dict[str, Any] = {
              "pageId": args.page_id,
          }
          if args.title:
              tool_args_update["title"] = args.title
          if args.body:
              tool_args_update["body"] = args.body
          else:
              # MCP requires body — fetch current page body if not provided
              page = client.call("getConfluencePage", {"pageId": args.page_id})
              if isinstance(page, dict) and "body" in page:
                  body = page["body"]
                  adf = {
                      "type": body.get("type", "doc"),
                      "version": body.get("version", 1),
                      "content": body.get("content", []),
                  }
                  tool_args_update["body"] = json.dumps(adf)
                  tool_args_update["contentFormat"] = "adf"
              else:
                  print("Error: Could not fetch current page body. Provide --body.", file=sys.stderr)
                  sys.exit(1)
          result = client.call("updateConfluencePage", tool_args_update)
          if getattr(args, "json", False):
              _mcp_output(result, as_json=True)
          else:
              print("Page updated successfully!")
              if isinstance(result, dict):
                  print(f"Title: {result.get('title', '')}")
              else:
                  print(result)
          return
  
      if cmd == "get-children":
          result = client.call("getConfluencePageDescendants", {
              "pageId": args.page_id,
              "limit": args.limit,
          })
          _mcp_output(result, getattr(args, "json", False))
          return
  
      print(f"Error: Unknown command '{cmd}'.", file=sys.stderr)
      sys.exit(1)
  
  
  # ─── Formatting ──────────────────────────────────────────────────────────────
  
  def format_page(page: ConfluencePage, verbose: bool = False) -> str:
      lines = [
          f"Title: {page.title}",
          f"ID: {page.id}",
      ]
      if page.url:
          lines.append(f"URL: {page.url}")
      if page.space_id:
          lines.append(f"Space: {page.space_id}")
      if page.status:
          lines.append(f"Status: {page.status}")
      if page.version_number:
          lines.append(f"Version: {page.version_number}")
      if page.updated_at:
          lines.append(f"Updated: {page.updated_at}")
      if verbose and page.body:
          readable = strip_html_tags(page.body)
          lines.append(f"\n--- Content ---\n{readable}")
      return "\n".join(lines)
  
  
  def format_space(space: ConfluenceSpace) -> str:
      lines = [
          f"Name: {space.name}",
          f"Key: {space.key}",
          f"ID: {space.id}",
      ]
      if space.type:
          lines.append(f"Type: {space.type}")
      if space.description:
          lines.append(f"Description: {space.description}")
      if space.homepage_id:
          lines.append(f"Homepage: {space.homepage_id}")
      return "\n".join(lines)
  
  
  # ─── REST Command Handlers ──────────────────────────────────────────────────
  
  def cmd_search(args, client: ConfluenceClient):
      cql = args.query
      cql_operators = ["=", "~", "AND", "OR", "NOT", "IN", "order by"]
      if not any(op in cql for op in cql_operators):
          cql = f'type=page AND text~"{cql}"'
  
      result = client.search(cql=cql, limit=args.limit, start=args.offset)
  
      if args.json:
          output_result({
              "pages": [p.__dict__ for p in result["pages"]],
              "total_size": result["total_size"],
              "start": result["start"],
              "limit": result["limit"],
          }, as_json=True)
      else:
          pages = result["pages"]
          if not pages:
              print(f"No pages found for query: {args.query}")
              return
          print(f"Found {len(pages)} page(s) (total: {result['total_size']}):\n")
          for i, page in enumerate(pages, 1):
              print(f"{i}. {page.title}")
              print(f"   ID: {page.id}")
              if page.url:
                  print(f"   URL: {page.url}")
              if page.updated_at:
                  print(f"   Updated: {page.updated_at}")
              print()
  
  
  def cmd_read(args, client: ConfluenceClient):
      page = client.get_page(args.page_id)
      if args.json:
          output_result(page.__dict__, as_json=True)
      else:
          print(format_page(page, verbose=True))
  
  
  def cmd_list_spaces(args, client: ConfluenceClient):
      result = client.list_spaces(limit=args.limit)
      if args.json:
          output_result({"spaces": [s.__dict__ for s in result["spaces"]]}, as_json=True)
      else:
          spaces = result["spaces"]
          if not spaces:
              print("No spaces found.")
              return
          print(f"Found {len(spaces)} space(s):\n")
          for space in spaces:
              print(format_space(space))
              print()
  
  
  def cmd_get_space(args, client: ConfluenceClient):
      space = client.get_space(args.space_id)
      if args.json:
          output_result(space.__dict__, as_json=True)
      else:
          print(format_space(space))
  
  
  def cmd_list_pages(args, client: ConfluenceClient):
      result = client.list_pages(space_id=args.space_id, limit=args.limit)
      if args.json:
          output_result({"pages": [p.__dict__ for p in result["pages"]]}, as_json=True)
      else:
          pages = result["pages"]
          if not pages:
              print("No pages found.")
              return
          print(f"Found {len(pages)} page(s):\n")
          for page in pages:
              print(f"- {page.title}")
              print(f"  ID: {page.id}")
              if page.url:
                  print(f"  URL: {page.url}")
              print()
  
  
  def cmd_create(args, client: ConfluenceClient):
      status = "draft" if args.draft else "current"
      page = client.create_page(
          space_id=args.space_id,
          title=args.title,
          body=args.body,
          parent_id=args.parent_id,
          status=status,
      )
      if args.json:
          output_result(page.__dict__, as_json=True)
      else:
          print("Page created successfully!\n")
          print(format_page(page))
  
  
  def cmd_update(args, client: ConfluenceClient):
      page = client.update_page(
          page_id=args.page_id,
          title=args.title,
          body=args.body,
      )
      if args.json:
          output_result(page.__dict__, as_json=True)
      else:
          print("Page updated successfully!\n")
          print(format_page(page))
  
  
  def cmd_get_children(args, client: ConfluenceClient):
      result = client.get_children(page_id=args.page_id, limit=args.limit)
      if args.json:
          output_result({"pages": [p.__dict__ for p in result["pages"]]}, as_json=True)
      else:
          pages = result["pages"]
          if not pages:
              print("No child pages found.")
              return
          print(f"Found {len(pages)} child page(s):\n")
          for page in pages:
              print(f"- {page.title}")
              print(f"  ID: {page.id}")
              if page.url:
                  print(f"  URL: {page.url}")
              print()
  
  
  def cmd_auth_info(args, client: ConfluenceClient):
      try:
          user = client.get_current_user()
      except AtlassianAPIError as e:
          print(f"Error: Unable to connect to Confluence API. {e}", file=sys.stderr)
          sys.exit(1)
  
      if args.json:
          output_result(user, as_json=True)
      else:
          print("Authentication successful!\n")
          print(f"User: {user.get('displayName', 'Unknown')}")
          print(f"Email: {user.get('email', 'N/A')}")
          print(f"Type: {user.get('type', 'Unknown')}")
          print(f"URL: {client.wiki_base}")
  
  
  # ─── Main ────────────────────────────────────────────────────────────────────
  
  def main():
      parser = argparse.ArgumentParser(
          description="Confluence Wiki CLI - Search, read, and manage Confluence wiki pages.",
          formatter_class=argparse.RawDescriptionHelpFormatter,
      )
      parser.add_argument("--json", action="store_true", help="Output results as JSON")
  
      subparsers = parser.add_subparsers(dest="command", help="Available commands")
  
      def add_json_flag(subparser):
          subparser.add_argument("--json", action="store_true", help="Output as JSON")
  
      # search
      search_parser = subparsers.add_parser("search", help="Search pages using CQL")
      search_parser.add_argument("query", help="Search query or CQL expression")
      search_parser.add_argument("--limit", type=int, default=25, help="Max results (default: 25)")
      search_parser.add_argument("--offset", type=int, default=0, help="Pagination offset")
      add_json_flag(search_parser)
  
      # read
      read_parser = subparsers.add_parser("read", help="Read a page")
      read_parser.add_argument("page_id", help="Page ID")
      add_json_flag(read_parser)
  
      # list-spaces
      list_spaces_parser = subparsers.add_parser("list-spaces", help="List all spaces")
      list_spaces_parser.add_argument("--limit", type=int, default=25, help="Max results (default: 25)")
      add_json_flag(list_spaces_parser)
  
      # get-space
      get_space_parser = subparsers.add_parser("get-space", help="Get space details")
      get_space_parser.add_argument("space_id", help="Space ID")
      add_json_flag(get_space_parser)
  
      # list-pages
      list_pages_parser = subparsers.add_parser("list-pages", help="List pages in a space")
      list_pages_parser.add_argument("--space-id", required=True, help="Space ID")
      list_pages_parser.add_argument("--limit", type=int, default=25, help="Max results (default: 25)")
      add_json_flag(list_pages_parser)
  
      # create
      create_parser = subparsers.add_parser("create", help="Create a new page")
      create_parser.add_argument("--title", required=True, help="Page title")
      create_parser.add_argument("--space-id", required=True, help="Space ID")
      create_parser.add_argument("--body", help="Page content (Confluence storage format / HTML)")
      create_parser.add_argument("--parent-id", help="Parent page ID")
      create_parser.add_argument("--draft", action="store_true", help="Create as draft")
      add_json_flag(create_parser)
  
      # update
      update_parser = subparsers.add_parser("update", help="Update a page")
      update_parser.add_argument("page_id", help="Page ID")
      update_parser.add_argument("--title", help="New title")
      update_parser.add_argument("--body", help="New content (Confluence storage format / HTML)")
      add_json_flag(update_parser)
  
      # get-children
      children_parser = subparsers.add_parser("get-children", help="Get child pages")
      children_parser.add_argument("page_id", help="Parent page ID")
      children_parser.add_argument("--limit", type=int, default=25, help="Max results (default: 25)")
      add_json_flag(children_parser)
  
      # auth-info
      auth_parser = subparsers.add_parser("auth-info", help="Test authentication and show user info")
      add_json_flag(auth_parser)
  
      # list-tools (MCP only)
      tools_parser = subparsers.add_parser("list-tools", help="List available MCP tools (OAuth only)")
      add_json_flag(tools_parser)
  
      args = parser.parse_args()
  
      if not args.command:
          parser.print_help()
          sys.exit(1)
  
      try:
          backend = get_backend()
  
          if backend == "mcp":
              run_mcp_command(args)
          else:
              atlassian_client = create_rest_client()
              client = ConfluenceClient(atlassian_client)
              try:
                  commands = {
                      "search": cmd_search,
                      "read": cmd_read,
                      "list-spaces": cmd_list_spaces,
                      "get-space": cmd_get_space,
                      "list-pages": cmd_list_pages,
                      "create": cmd_create,
                      "update": cmd_update,
                      "get-children": cmd_get_children,
                      "auth-info": cmd_auth_info,
                  }
                  handler = commands.get(args.command)
                  if handler:
                      handler(args, client)
                  elif args.command == "list-tools":
                      print("Error: list-tools is only available with OAuth authentication.", file=sys.stderr)
                      print("Run: python scripts/auth.py login --oauth", file=sys.stderr)
                      sys.exit(1)
                  else:
                      parser.print_help()
                      sys.exit(1)
              finally:
                  atlassian_client.close()
  
      except AtlassianAPIError as e:
          print(f"API Error: {e}", file=sys.stderr)
          sys.exit(1)
      except MCPError as e:
          print(f"MCP Error: {e}", file=sys.stderr)
          sys.exit(1)
      except KeyboardInterrupt:
          print("\nCancelled.", file=sys.stderr)
          sys.exit(130)
      except Exception as e:
          print(f"Error: {e}", file=sys.stderr)
          sys.exit(1)
  
  
  if __name__ == "__main__":
      main()
  

• skills/atlassian/scripts/jira.py32 KB58ce4420385bupstream ↗

  Contents are not inlined in the catalog. Open the upstream file.

• skills/atlassian/scripts/mcp_client.py10 KB2a01c9b4d4fbupstream ↗

  #!/usr/bin/env python3
  """
  MCP (Model Context Protocol) client for Atlassian Cloud.
  
  Communicates with the Atlassian MCP server at https://mcp.atlassian.com/v1/mcp
  using JSON-RPC 2.0 over Streamable HTTP (POST with SSE or JSON responses).
  
  Requires OAuth 2.1 authentication — run `python auth.py login --oauth` first.
  
  Protocol sequence:
    1. POST initialize → server returns capabilities + Mcp-Session-Id
    2. POST notifications/initialized → server returns 202
    3. POST tools/list → server returns available tools
    4. POST tools/call → server returns tool results
  """
  
  import json
  import sys
  import urllib.error
  import urllib.request
  from typing import Any, Dict, List, Optional
  
  from auth import get_auth_header, get_auth_type, get_mcp_endpoint
  
  MCP_PROTOCOL_VERSION = "2025-03-26"
  
  
  class MCPError(Exception):
      """Raised when the MCP server returns an error."""
      pass
  
  
  class MCPClient:
      """Client for the Atlassian MCP server (JSON-RPC 2.0 over HTTP)."""
  
      def __init__(self, endpoint: str, auth_header: str):
          self.endpoint = endpoint
          self._auth_header = auth_header
          self._session_id: Optional[str] = None
          self._request_id = 0
          self._initialized = False
  
      def _next_id(self) -> int:
          self._request_id += 1
          return self._request_id
  
      def _make_request(self, method: str, params: Optional[dict] = None,
                        is_notification: bool = False) -> Any:
          """
          Send a JSON-RPC 2.0 request to the MCP server.
  
          Args:
              method: JSON-RPC method name (e.g. "initialize", "tools/list")
              params: Method parameters
              is_notification: If True, no id field (server returns 202)
  
          Returns:
              Parsed result from the JSON-RPC response, or None for notifications
          """
          payload: Dict[str, Any] = {
              "jsonrpc": "2.0",
              "method": method,
          }
          if params:
              payload["params"] = params
          if not is_notification:
              payload["id"] = self._next_id()
  
          body = json.dumps(payload).encode("utf-8")
  
          req = urllib.request.Request(self.endpoint, data=body, method="POST")
          req.add_header("User-Agent", "atlassian-skill/3.0")
          req.add_header("Content-Type", "application/json")
          req.add_header("Accept", "application/json, text/event-stream")
          req.add_header("Authorization", self._auth_header)
          if self._session_id:
              req.add_header("Mcp-Session-Id", self._session_id)
  
          try:
              with urllib.request.urlopen(req, timeout=60) as resp:
                  # Capture session ID from response headers
                  session_id = resp.headers.get("Mcp-Session-Id")
                  if session_id:
                      self._session_id = session_id
  
                  # Notifications return 202 with no body
                  if resp.status == 202:
                      return None
  
                  content_type = resp.headers.get("Content-Type", "")
                  raw = resp.read().decode("utf-8")
  
                  if "text/event-stream" in content_type:
                      return self._parse_sse(raw)
                  else:
                      return self._parse_json_rpc(raw)
  
          except urllib.error.HTTPError as e:
              error_body = e.read().decode("utf-8", errors="replace")
              # Sanitize auth header from error messages
              if self._auth_header:
                  token_part = self._auth_header.split(" ", 1)[-1]
                  error_body = error_body.replace(token_part, "***")
              raise MCPError(f"HTTP {e.code}: {error_body}")
          except urllib.error.URLError as e:
              raise MCPError(f"Connection failed: {e.reason}")
  
      def _parse_json_rpc(self, raw: str) -> Any:
          """Parse a JSON-RPC 2.0 response."""
          data = json.loads(raw)
          if "error" in data:
              err = data["error"]
              raise MCPError(f"MCP error {err.get('code', '?')}: {err.get('message', raw)}")
          return data.get("result")
  
      def _parse_sse(self, raw: str) -> Any:
          """Parse Server-Sent Events response, extracting the last JSON-RPC message."""
          last_result = None
          for line in raw.split("\n"):
              line = line.strip()
              if line.startswith("data:"):
                  data_str = line[5:].strip()
                  if not data_str:
                      continue
                  try:
                      data = json.loads(data_str)
                      if "error" in data:
                          err = data["error"]
                          raise MCPError(
                              f"MCP error {err.get('code', '?')}: {err.get('message', data_str)}")
                      if "result" in data:
                          last_result = data["result"]
                  except json.JSONDecodeError:
                      continue
          if last_result is None:
              raise MCPError("No valid JSON-RPC result in SSE response")
          return last_result
  
      def initialize(self) -> dict:
          """
          Initialize the MCP session.
  
          Returns:
              Server capabilities dict
          """
          result = self._make_request("initialize", {
              "protocolVersion": MCP_PROTOCOL_VERSION,
              "capabilities": {},
              "clientInfo": {
                  "name": "atlassian-skill",
                  "version": "3.0",
              },
          })
  
          # Send initialized notification
          self._make_request("notifications/initialized", is_notification=True)
          self._initialized = True
          return result
  
      def _ensure_initialized(self):
          """Initialize the session if not already done."""
          if not self._initialized:
              self.initialize()
  
      def list_tools(self) -> List[dict]:
          """List available MCP tools."""
          self._ensure_initialized()
          result = self._make_request("tools/list")
          return result.get("tools", []) if result else []
  
      def call_tool(self, tool_name: str, arguments: Optional[dict] = None) -> Any:
          """
          Call an MCP tool.
  
          Args:
              tool_name: Name of the tool (e.g. "jira_search_issues")
              arguments: Tool-specific arguments
  
          Returns:
              Tool result (parsed from content array)
          """
          self._ensure_initialized()
          params: Dict[str, Any] = {"name": tool_name}
          params["arguments"] = arguments if arguments is not None else {}
  
          result = self._make_request("tools/call", params)
          if not result:
              return None
  
          # MCP tools/call returns {content: [{type, text}], isError}
          if result.get("isError"):
              content = result.get("content", [])
              error_text = "\n".join(c.get("text", "") for c in content if c.get("type") == "text")
              raise MCPError(f"Tool error: {error_text or 'Unknown error'}")
  
          # Extract text content
          content = result.get("content", [])
          texts = [c.get("text", "") for c in content if c.get("type") == "text"]
          combined = "\n".join(texts)
  
          # Try to parse as JSON
          try:
              return json.loads(combined)
          except (json.JSONDecodeError, ValueError):
              return combined
  
  
  def create_mcp_client() -> MCPClient:
      """
      Create an authenticated MCP client.
      Exits with helpful error if OAuth is not configured.
      """
      auth_type = get_auth_type()
      if auth_type != "oauth":
          print("Error: MCP client requires OAuth authentication.", file=sys.stderr)
          print("Run: python scripts/auth.py login --oauth", file=sys.stderr)
          sys.exit(1)
  
      endpoint = get_mcp_endpoint()
      if not endpoint:
          print("Error: No MCP endpoint configured.", file=sys.stderr)
          sys.exit(1)
  
      auth_header = get_auth_header()
      if not auth_header:
          print("Error: Failed to get OAuth token. Re-authenticate with:", file=sys.stderr)
          print("  python scripts/auth.py login --oauth", file=sys.stderr)
          sys.exit(1)
  
      return MCPClient(endpoint=endpoint, auth_header=auth_header)
  
  
  # ─── Shared MCP wrapper with cloudId management ────────────────────────────
  
  # Tools that don't require cloudId
  _NO_CLOUD_ID = {"atlassianUserInfo", "getAccessibleAtlassianResources", "search", "fetch"}
  
  
  class AtlassianMCPClient:
      """Base MCP client with automatic cloudId injection.
  
      Subclass or use directly for Jira/Confluence MCP operations.
      """
  
      def __init__(self, product_name: str = "Atlassian"):
          self.mcp = create_mcp_client()
          self._cloud_id: Optional[str] = None
          self._product_name = product_name
  
      def _get_cloud_id(self) -> str:
          """Get the Atlassian Cloud ID (required for most MCP tools)."""
          if self._cloud_id:
              return self._cloud_id
          result = self.mcp.call_tool("getAccessibleAtlassianResources")
          if isinstance(result, list) and result:
              self._cloud_id = result[0].get("id", "")
          elif isinstance(result, dict):
              resources = result.get("resources", [result])
              if resources:
                  self._cloud_id = resources[0].get("id", "")
          if not self._cloud_id:
              raise RuntimeError(
                  f"No accessible Atlassian resources found. "
                  f"Ensure your OAuth consent includes {self._product_name} access.")
          return self._cloud_id
  
      def call(self, tool_name: str, arguments: Optional[dict] = None) -> Any:
          """Call an MCP tool with automatic cloudId injection."""
          args = dict(arguments) if arguments else {}
          if tool_name not in _NO_CLOUD_ID and "cloudId" not in args:
              args["cloudId"] = self._get_cloud_id()
          return self.mcp.call_tool(tool_name, args)
  
      def list_tools(self) -> List[dict]:
          """List all available MCP tools."""
          return self.mcp.list_tools()
  
  
  def mcp_output(result: Any, as_json: bool = False):
      """Output MCP tool result to stdout."""
      if as_json:
          if isinstance(result, (dict, list)):
              print(json.dumps(result, indent=2, default=str))
          else:
              print(json.dumps({"result": str(result)}, indent=2))
      else:
          if isinstance(result, str):
              print(result)
          elif isinstance(result, (dict, list)):
              print(json.dumps(result, indent=2, default=str))
          else:
              print(result)
  

• skills/atlassian/.env.example688 B45e84cb12aafupstream ↗

  # Atlassian API Configuration (fallback - keyring preferred)
  # ============================================================
  # These are only needed if you prefer .env over keyring storage.
  # For keyring-based auth, use: python scripts/auth.py login
  #
  # Get your API token from:
  # https://id.atlassian.com/manage-profile/security/api-tokens
  
  # Required: Your Atlassian instance URL (without /wiki suffix)
  ATLASSIAN_URL=https://your-domain.atlassian.net
  
  # Required: Your Atlassian account email
  [email protected]
  
  # Required: Your Atlassian API token
  ATLASSIAN_API_TOKEN=your-api-token-here
  
  # Optional: Request timeout in seconds (default: 30)
  # ATLASSIAN_TIMEOUT=30
  

• skills/atlassian/requirements.txt51 B17c9fc684b2dupstream ↗

  httpx>=0.25.0
  keyring>=24.0.0
  python-dotenv>=1.0.0
  

• README.mdupstream ↗

  Contents are not inlined in the catalog. Open the upstream file.

Plain-text copy, no login: /skill-bundles/sanjay3290-ai-skills-skills-atlassian.md