# atlassian

Adopt this skill: https://proskills.md/skills/other/sanjay3290-ai-skills-skills-atlassian

Page: https://proskills.md/skills/other/sanjay3290-ai-skills-skills-atlassian
This document is the public skill bundle for that page. It contains SKILL.md and every helper file the catalog carries. No login is required.

## File: skills/atlassian/SKILL.md
size: 6334
sha: 8b1f82146d64c2c8b39e529f29a5c4721ccbdb8a
upstream: https://github.com/sanjay3290/ai-skills/blob/main/skills/atlassian/SKILL.md

````text
---
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 "user@example.com" --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 "user@example.com"
```

### 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).
````

## File: skills/atlassian/scripts/api_client.py
size: 5537
sha: 7638f8f7939e0c5ade2228428a765e6be8d6e964
upstream: https://github.com/sanjay3290/ai-skills/blob/main/skills/atlassian/scripts/api_client.py

```text
#!/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)
```

## File: skills/atlassian/scripts/auth.py
size: 23747
sha: ecbb3bd62778a927e60c533b6f3dff106d1ae14e
upstream: https://github.com/sanjay3290/ai-skills/blob/main/skills/atlassian/scripts/auth.py

```text
#!/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 ATLASSIAN_EMAIL=your-email@example.com", 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()
```

## File: skills/atlassian/scripts/confluence.py
size: 27538
sha: 269f2abf7a5285f8d4ed4547f30b022fbc8bc226
upstream: https://github.com/sanjay3290/ai-skills/blob/main/skills/atlassian/scripts/confluence.py

```text
#!/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()
```

## File: skills/atlassian/scripts/jira.py
size: 32867
sha: 58ce4420385bd8ee8ef68f33b712996cc2a55436
upstream: https://github.com/sanjay3290/ai-skills/blob/main/skills/atlassian/scripts/jira.py

Content is not inlined in the catalog. Read the upstream file: https://github.com/sanjay3290/ai-skills/blob/main/skills/atlassian/scripts/jira.py

## File: skills/atlassian/scripts/mcp_client.py
size: 10501
sha: 2a01c9b4d4fb1c83719497bf9db674dcb91efff8
upstream: https://github.com/sanjay3290/ai-skills/blob/main/skills/atlassian/scripts/mcp_client.py

```text
#!/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)
```

## File: skills/atlassian/.env.example
size: 688
sha: 45e84cb12aafd40aae5c85c9f63fd41fd6135c60
upstream: https://github.com/sanjay3290/ai-skills/blob/main/skills/atlassian/.env.example

```text
# 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
ATLASSIAN_EMAIL=your-email@example.com

# Required: Your Atlassian API token
ATLASSIAN_API_TOKEN=your-api-token-here

# Optional: Request timeout in seconds (default: 30)
# ATLASSIAN_TIMEOUT=30
```

## File: skills/atlassian/requirements.txt
size: 51
sha: 17c9fc684b2d28707834d4e87d57bf07f98893c7
upstream: https://github.com/sanjay3290/ai-skills/blob/main/skills/atlassian/requirements.txt

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

## File: README.md
upstream: https://github.com/sanjay3290/ai-skills/blob/main/README.md

````text
# AI Agent Skills

[![Ask DeepWiki](https://deepwiki.com/badge.svg)](https://deepwiki.com/sanjay3290/ai-skills)

A collection of portable skills for AI coding assistants. Works with all major AI clients that support the [Agent Skills Standard](https://agentskills.io).

## Supported AI Clients

<p align="center">
  <a href="#claude-code"><img src="https://img.shields.io/badge/Claude_Code-D97757?style=for-the-badge&logo=anthropic&logoColor=white" alt="Claude Code" /></a>
  <a href="#gemini-cli"><img src="https://img.shields.io/badge/Gemini_CLI-8E75B2?style=for-the-badge&logo=google&logoColor=white" alt="Gemini CLI" /></a>
  <a href="#google-antigravity"><img src="https://img.shields.io/badge/Antigravity-4285F4?style=for-the-badge&logo=google&logoColor=white" alt="Google Antigravity" /></a>
  <a href="#cursor"><img src="https://img.shields.io/badge/Cursor-000000?style=for-the-badge&logo=cursor&logoColor=white" alt="Cursor" /></a>
  <a href="#openai-codex-cli"><img src="https://img.shields.io/badge/OpenAI_Codex-412991?style=for-the-badge&logo=openai&logoColor=white" alt="OpenAI Codex" /></a>
  <a href="#goose"><img src="https://img.shields.io/badge/Goose-FF6B35?style=for-the-badge&logo=go&logoColor=white" alt="Goose" /></a>
</p>

| Client | Skills Directory | Documentation |
|--------|-----------------|---------------|
| **Claude Code** | `~/.claude/skills/` or `.claude/skills/` | [docs](https://docs.anthropic.com/en/docs/claude-code/skills) |
| **Gemini CLI** | `~/.gemini/skills/` or `.gemini/skills/` | [docs](https://geminicli.com/docs/cli/skills/) |
| **Google Antigravity** | `~/.gemini/antigravity/skills/` or `.agent/skills/` | [docs](https://antigravity.google/docs/skills) |
| **Cursor** | `~/.cursor/skills/` or `.cursor/skills/` | [docs](https://cursor.com/docs/context/skills) |
| **OpenAI Codex CLI** | `~/.codex/skills/` or `.codex/skills/` | [docs](https://developers.openai.com/codex/skills/) |
| **Goose** | `~/.config/goose/skills/` or `.goose/skills/` | [docs](https://block.github.io/goose/docs/guides/context-engineering/using-skills/) |

## Available Skills

| Skill | Description |
|-------|-------------|
| [postgres](skills/postgres/) | Read-only PostgreSQL queries with defense-in-depth security |
| [mysql](skills/mysql/) | Read-only MySQL queries with session-level write protection |
| [mssql](skills/mssql/) | Read-only Microsoft SQL Server queries with query validation security |
| [gemini-api](skills/gemini-api/) | Call the Gemini API from the shell with Google's `gemini-api` CLI — text, image generation and editing, TTS, music, video, document/audio/video analysis, transcription, and background managed agents (Deep Research) |
| [imagen](skills/imagen/) | AI image generation using Google Gemini (cross-platform) |
| [deep-research](skills/deep-research/) | Autonomous multi-step research using Gemini Deep Research Agent |
| [outline](skills/outline/) | Search, read, and manage Outline wiki documents |
| [jules](skills/jules/) | Delegate coding tasks to Google Jules AI agent (async bug fixes, docs, tests, features) |
| [manus](skills/manus/) | Delegate complex tasks to Manus AI agent (deep research, market analysis, reports) |
| [grok-build](skills/grok-build/) | Orchestrate coding by delegating well-specified tasks to xAI's Grok Build CLI headlessly — the assistant plans, specs, dispatches, and reviews every diff |
| [notebooklm](skills/notebooklm/) | Query and manage Google NotebookLM notebooks with persistent profile auth, source sync, batch/multi queries, and structured exports |
| [elevenlabs](skills/elevenlabs/) | Text-to-speech narration and two-host podcast generation from documents (PDF, DOCX, MD, TXT) using ElevenLabs API |
| [google-tts](skills/google-tts/) | Text-to-speech narration and podcast generation using Google Cloud TTS (Neural2, WaveNet, Studio voices, 40+ languages) |
| [atlassian](skills/atlassian/) | Manage Jira issues and Confluence wiki pages in Atlassian Cloud (OAuth 2.1 via MCP server or API token fallback) |
| [azure-devops](skills/azure-devops/) | Manage Azure DevOps projects, work items, repos, PRs, pipelines, wikis, test plans, security alerts, variable groups, environments/approvals, branch policies, and attachments (99 tools, 13 domains) |
| [apple-container](skills/apple-container/) | Build, run, and manage OCI/Linux containers as lightweight VMs on Apple-silicon macOS with Apple's `container` CLI — no Docker daemon |
| [telegram](skills/telegram/) | Send Telegram messages, files, and alerts via bot; ask-and-wait approvals with inline buttons; multi-bot and multi-target routing; Claude Code hook integration |
| [whatsapp](skills/whatsapp/) | Send and receive WhatsApp messages as a linked companion device via pywhats — QR pairing, text and images, group chat, read receipts, and a JSON event stream (unofficial client, not the Business/Cloud API) |

### Google Workspace Skills

Lightweight alternatives to the full [Google Workspace MCP server](https://github.com/gemini-cli-extensions/workspace). Each skill has standalone OAuth authentication with cross-platform token storage via keyring.

> **⚠️ Requires Google Workspace account.** Personal Gmail accounts are not supported. These skills use the same OAuth infrastructure as the official Google Workspace MCP.

| Skill | Description |
|-------|-------------|
| [google-chat](skills/google-chat/) | List spaces, send messages, DMs, create spaces |
| [google-docs](skills/google-docs/) | Create, read, edit Google Docs |
| [google-sheets](skills/google-sheets/) | Read spreadsheets, get ranges, find sheets |
| [google-slides](skills/google-slides/) | Read presentations, get text and metadata |
| [google-drive](skills/google-drive/) | Search files, list folders, download files |
| [google-calendar](skills/google-calendar/) | Events, scheduling, free time lookup |
| [gmail](skills/gmail/) | Search, read, send emails, manage labels |

## Installation

### Claude Code — Plugin Marketplace

The fastest path for Claude Code users, no extra tooling required:

```
/plugin marketplace add sanjay3290/ai-skills
/plugin install ai-skills@ai-skills
```

All 25 skills install as a single plugin; update later with `/plugin marketplace update ai-skills`.

### Any Agent — `skills` CLI

> **[`npx skills`](https://github.com/vercel-labs/skills)** — The package manager for the open [Agent Skills](https://agentskills.io) ecosystem. One command to install skills into any AI coding agent.

### Install Skills

```bash
# Browse all 25 available skills
npx skills add sanjay3290/ai-skills --list

# Install a single skill (auto-detects your agent)
npx skills add sanjay3290/ai-skills --skill postgres

# Install multiple skills at once
npx skills add sanjay3290/ai-skills --skill postgres --skill mysql --skill mssql

# Install all skills
npx skills add sanjay3290/ai-skills --all
```

### Target Specific Agents

Use `-a` to install into a specific agent's skills directory:

```bash
# Install for Claude Code
npx skills add sanjay3290/ai-skills --skill postgres -a claude-code

# Install for multiple agents at once
npx skills add sanjay3290/ai-skills --skill postgres -a claude-code -a gemini-cli -a cursor

# Install all skills into all supported agents
npx skills add sanjay3290/ai-skills --all -a '*'
```

### Global vs Project Install

By default, skills install to the current project directory. Use `-g` for global (user-level) installation:

```bash
# Global install — available in all projects
npx skills add sanjay3290/ai-skills --skill imagen -g

# Project install (default) — scoped to current repo
npx skills add sanjay3290/ai-skills --skill imagen
```

### Supported Agents

The skills CLI supports 40+ agents. Here are the most common:

| Agent | `-a` Flag | Project Directory | Global Directory |
|-------|-----------|-------------------|------------------|
| Claude Code | `claude-code` | `.claude/skills/` | `~/.claude/skills/` |
| Gemini CLI | `gemini-cli` | `.gemini/skills/` | `~/.gemini/skills/` |
````