argocd-readonly: a minimal read-only ArgoCD MCP server (Python, FastMCP). Inspect apps, sync/health, resource trees, diffs, events, and pod logs from any MCP client. 1Password integration for secrets.

.env.example

# Example .env for the argocd-readonly MCP server.
#
# You have two options for what goes here:
#
#   1. 1Password references (recommended). Values like op://VAULT/ITEM/FIELD
#      are resolved by `op run --env-file=.env -- ...` at launch time, so
#      real secrets never sit on disk. Requires the 1Password CLI (`op`).
#
#      NOTE: op:// references reject parens and some other special characters
#            in vault/item names. If your item name has them, use the item's
#            UUID instead (run `op item get "<name>" --format json | jq -r .id`).
#
#   2. Plain values. Just put the secret directly. Works with `op run`
#      (passed through untouched) or with any other env loader, or by
#      `set -a; source .env; set +a` before launching the server.
#
# Pick whichever matches your threat model. The server doesn't care — it
# just reads env vars.
#
# AUTH MODES — pick one:
#
#   Mode A — bearer token (preferred). Set ARGOCD_AUTH_TOKEN.
#   Mode B — username + password. Set ARGOCD_USERNAME and ARGOCD_PASSWORD.
#            The server will exchange these for a session token at startup
#            and refresh on 401. Use this when your account doesn't have
#            apiKey capability and you can't easily change cluster config.

# --- Required ---
ARGOCD_SERVER=op://Vault/ArgoCD/website

# --- Mode A: bearer token ---
ARGOCD_AUTH_TOKEN=op://Vault/ArgoCD/token

# --- Mode B: username + password (use instead of AUTH_TOKEN, not in addition) ---
# ARGOCD_USERNAME=op://Vault/ArgoCD/username
# ARGOCD_PASSWORD=op://Vault/ArgoCD/password

# --- Plain-value examples (no 1Password) ---
# ARGOCD_SERVER=argocd.example.com
# ARGOCD_AUTH_TOKEN=eyJhbGciOi...your-token-here...

# Set to 1 to skip TLS verification (self-signed certs). Boolean, doesn't
# need to live in a vault.
ARGOCD_INSECURE=1

README.md

argocd-readonly MCP server

A minimal read-only Model Context Protocol server for ArgoCD. Lets an MCP client (Claude Code, Claude Desktop, etc.) inspect application sync/health, resource trees, live-vs-desired diffs, events, and pod logs — without exposing any mutating endpoints.

No third-party services. The server runs locally on your machine and talks directly to your ArgoCD API over HTTPS.

Tools exposed

All read-only — no sync, refresh, terminate-op, rollback, or delete endpoints are wired up.

Tool	What it does
list_applications(project?, selector?)	Apps with sync status, health, project, target revision
get_application(name)	Full app details: spec, status, conditions, current operation
get_application_resource_tree(name)	Every k8s resource managed by the app, with per-node health
get_application_managed_resources(name)	Live-vs-desired diff (great for OutOfSync drift)
get_application_events(name)	Recent k8s events for the app
get_application_manifests(name, revision?)	Rendered desired-state manifests
get_application_logs(name, pod, container?, tail_lines?, namespace?)	Pod logs via ArgoCD's proxy
list_projects()	ArgoCD projects + their source/destination allowlists
list_clusters()	Registered clusters (credentials stripped)
list_repositories()	Registered Git repos (credentials stripped)

Setup

1. Install dependencies

pip3 install mcp httpx

2. Pick an auth mode

The server supports two:

Mode A — bearer token (preferred). In the ArgoCD UI: Settings → Accounts → <your account> → Generate New Token (set "Expires In" to 0 for "never expires" if you want a long-lived token). The account needs the apiKey capability — by default the built-in admin user does not have this; an admin will need to add accounts.<name>: apiKey, login to the argocd-cm ConfigMap.

Mode B — username + password. Use this when you can't easily change cluster config. The server POSTs your credentials to /api/v1/session at startup, gets a session token, and refreshes it automatically when it expires (handles 401s transparently). Slightly less ideal — sessions are short-lived (~24h default) — but the refresh logic makes that invisible.

3. Set environment variables

Var	Required	Notes
ARGOCD_SERVER	yes	Hostname (full URLs like https://argocd.example.com/applications are stripped to host automatically)
ARGOCD_AUTH_TOKEN	Mode A	Bearer token from step 2
ARGOCD_USERNAME	Mode B	Username for session login
ARGOCD_PASSWORD	Mode B	Password for session login
ARGOCD_INSECURE	no	Set to 1 to skip TLS verification (self-signed certs)

You can export them however you like. The recommended pattern is 1Password CLI (op run) so secrets never touch disk in plaintext — see below.

4. Wire up your MCP client

For Claude Code, add to ~/.claude/settings.json:

{
  "mcpServers": {
    "argocd-readonly": {
      "command": "python3",
      "args": ["/absolute/path/to/server.py"],
      "env": {
        "ARGOCD_SERVER": "argocd.example.com",
        "ARGOCD_AUTH_TOKEN": "your-token-here",
        "ARGOCD_INSECURE": "1"
      }
    }
  }
}

Restart your client and the tools become available.

Using a .env file

If you'd rather not paste secrets into settings.json, drop them in a .env file next to server.py and load them at launch. You have two options:

Option A — Plain values

Simplest. Put the actual secrets in .env:

ARGOCD_SERVER=argocd.example.com
ARGOCD_AUTH_TOKEN=eyJhbGciOi...your-token-here...
ARGOCD_INSECURE=1

Load it however your shell or process manager prefers. With op run (works without 1Password too — it just passes plain values through):

{
  "mcpServers": {
    "argocd-readonly": {
      "command": "op",
      "args": ["run", "--env-file=/absolute/path/to/.env", "--", "python3", "/absolute/path/to/server.py"]
    }
  }
}

Or skip op entirely with a tiny shell wrapper:

{
  "mcpServers": {
    "argocd-readonly": {
      "command": "bash",
      "args": ["-c", "set -a; source /absolute/path/to/.env; set +a; exec python3 /absolute/path/to/server.py"]
    }
  }
}

This is fine for personal machines. Just remember .env now holds real secrets — gitignore it, and don't commit it to a dotfiles repo without a vault.

Option B — 1Password references (recommended for shared/synced machines)

Put secret references in .env instead of values. op run resolves them against your vault at launch, so real secrets never sit on disk:

ARGOCD_SERVER=op://Vault/ArgoCD/website
ARGOCD_AUTH_TOKEN=op://Vault/ArgoCD/token
ARGOCD_INSECURE=1

For Mode B (username/password), use:

ARGOCD_SERVER=op://Vault/ArgoCD/website
ARGOCD_USERNAME=op://Vault/ArgoCD/username
ARGOCD_PASSWORD=op://Vault/ArgoCD/password
ARGOCD_INSECURE=1

Wire your client to op run exactly as in Option A — same command, same args. The only difference is what's in the file. You can mix and match: ARGOCD_INSECURE=1 as a literal, secrets as op:// refs.

Heads up — special characters in item names. op:// references reject parentheses and some other special characters in vault/item names. If your item is called something like ArgoCD (prod), look up its UUID and use that instead:

op item get "ArgoCD (prod)" --vault "Vault" --format json | jq -r '.id'
# -> zmwp44ddjeta252qt5rttuk6ky

# then in .env:
ARGOCD_SERVER=op://Vault/zmwp44ddjeta252qt5rttuk6ky/website

Requires the 1Password CLI (op) signed in to your account.

Security notes

• Read-only by design. No mutating ArgoCD endpoints are exposed. If you want sync/rollback, fork it — but be aware an LLM with sync power can do real damage on a misread.
• Token scope. Use a dedicated ArgoCD account with the narrowest project RBAC you can get away with. If you only need to inspect one project, restrict the token to that project.
• ARGOCD_INSECURE. Only enable when you know you're hitting a cluster with a self-signed cert. It disables certificate verification entirely — it does not just "ignore hostname mismatch."
• Credential scrubbing. list_clusters and list_repositories drop credential fields (passwords, SSH keys, GitHub App keys, etc.) before returning. The full ArgoCD API does include these in responses for users with the right RBAC; the server filters them out so they don't end up in your LLM context.
• Logs leak content. get_application_logs returns raw pod log content into your LLM context. Skip that tool if your workloads log sensitive data.

License

MIT — do whatever you want with it.

server.py

"""
Minimal read-only ArgoCD MCP server. Runs locally, talks directly
to your ArgoCD API. No third-party services involved.

Setup:
1. pip install mcp httpx
2. Set env vars (one of two auth modes):
     ARGOCD_SERVER       hostname only, no scheme (e.g. argocd.example.com).
                         Paths/schemes are stripped automatically, so a full
                         URL like https://argocd.example.com/applications also works.
     ARGOCD_INSECURE     optional, "1"/"true" to skip TLS verification

   Mode A — bearer token (preferred):
     ARGOCD_AUTH_TOKEN   long-lived API token
                         (ArgoCD UI -> User Info -> Generate Token)

   Mode B — username + password (works against accounts without apiKey
            capability; the server exchanges credentials for a session
            token at startup and refreshes on 401):
     ARGOCD_USERNAME
     ARGOCD_PASSWORD

3. Add to your MCP client's settings (see README.md)

Read-only by design: no sync, refresh, terminate-op, rollback, delete,
or any other mutating endpoint is exposed.
"""

import os
import json
import asyncio
from urllib.parse import urlparse
import httpx
from mcp.server.fastmcp import FastMCP

ARGOCD_SERVER = os.environ.get("ARGOCD_SERVER")
ARGOCD_AUTH_TOKEN = os.environ.get("ARGOCD_AUTH_TOKEN")
ARGOCD_USERNAME = os.environ.get("ARGOCD_USERNAME")
ARGOCD_PASSWORD = os.environ.get("ARGOCD_PASSWORD")
ARGOCD_INSECURE = os.environ.get("ARGOCD_INSECURE", "").lower() in ("1", "true", "yes")

if not ARGOCD_SERVER:
    raise ValueError("ARGOCD_SERVER environment variable is required")
if not ARGOCD_AUTH_TOKEN and not (ARGOCD_USERNAME and ARGOCD_PASSWORD):
    raise ValueError(
        "Auth required: set ARGOCD_AUTH_TOKEN, or both ARGOCD_USERNAME and ARGOCD_PASSWORD"
    )


def _normalize_host(value: str) -> str:
    """Strip scheme, path, query, and trailing slashes — leave host[:port] only."""
    v = value.strip()
    if "://" not in v:
        v = "https://" + v
    parsed = urlparse(v)
    if not parsed.netloc:
        raise ValueError(f"Could not parse host from ARGOCD_SERVER={value!r}")
    return parsed.netloc


ARGOCD_HOST = _normalize_host(ARGOCD_SERVER)
BASE_URL = f"https://{ARGOCD_HOST}/api/v1"
VERIFY = not ARGOCD_INSECURE
USING_SESSION_AUTH = not ARGOCD_AUTH_TOKEN


class _AuthState:
    """Holds the current bearer token. Refreshes via /api/v1/session when in session mode."""

    def __init__(self):
        self.token: str | None = ARGOCD_AUTH_TOKEN
        self.lock = asyncio.Lock()

    async def _login(self) -> str:
        async with httpx.AsyncClient(verify=VERIFY, timeout=30.0) as client:
            r = await client.post(
                f"{BASE_URL}/session",
                json={"username": ARGOCD_USERNAME, "password": ARGOCD_PASSWORD},
            )
            r.raise_for_status()
            self.token = r.json()["token"]
            return self.token

    async def get_token(self) -> str:
        if self.token:
            return self.token
        async with self.lock:
            if self.token:
                return self.token
            return await self._login()

    async def refresh(self) -> str:
        async with self.lock:
            return await self._login()


_auth = _AuthState()

mcp = FastMCP(
    "argocd-readonly",
    instructions="Read-only access to an ArgoCD instance. Use this for inspecting application sync/health status, resource trees, live-vs-desired diffs, events, and pod logs.",
)


async def _request(method: str, path: str, **kwargs) -> httpx.Response:
    """Authenticated request. Auto-refreshes session token on 401 in session-auth mode."""
    token = await _auth.get_token()
    headers = kwargs.pop("headers", {}) or {}
    headers["Authorization"] = f"Bearer {token}"
    async with httpx.AsyncClient(verify=VERIFY, timeout=60.0) as client:
        r = await client.request(method, f"{BASE_URL}{path}", headers=headers, **kwargs)
        if r.status_code == 401 and USING_SESSION_AUTH:
            token = await _auth.refresh()
            headers["Authorization"] = f"Bearer {token}"
            r = await client.request(method, f"{BASE_URL}{path}", headers=headers, **kwargs)
        r.raise_for_status()
        return r


async def _get_json(path: str, params: dict | None = None) -> dict:
    r = await _request("GET", path, params=params)
    return r.json()


async def _get_text(path: str, params: dict | None = None) -> str:
    r = await _request("GET", path, params=params)
    return r.text


def _scrub_repo(repo: dict) -> dict:
    drop = {
        "password", "sshPrivateKey", "tlsClientCertKey", "tlsClientCertData",
        "githubAppPrivateKey", "githubAppEnterpriseBaseUrl", "githubAppId",
        "githubAppInstallationId", "gcpServiceAccountKey", "proxy",
        "username",
    }
    return {k: v for k, v in repo.items() if k not in drop}


def _scrub_cluster(cluster: dict) -> dict:
    cleaned = dict(cluster)
    cleaned.pop("config", None)
    return cleaned


@mcp.tool()
async def list_applications(project: str | None = None, selector: str | None = None) -> str:
    """List ArgoCD applications with sync status, health, project, and target revision.

    Args:
        project: Optional project name to filter by.
        selector: Optional label selector (e.g. "env=prod,team=platform").
    """
    params: dict = {}
    if project:
        params["projects"] = project
    if selector:
        params["selector"] = selector
    data = await _get_json("/applications", params=params or None)
    apps = []
    for item in data.get("items", []) or []:
        spec = item.get("spec", {})
        status = item.get("status", {})
        sync = status.get("sync", {})
        health = status.get("health", {})
        source = spec.get("source") or (spec.get("sources") or [{}])[0]
        apps.append({
            "name": item.get("metadata", {}).get("name"),
            "namespace": item.get("metadata", {}).get("namespace"),
            "project": spec.get("project"),
            "sync_status": sync.get("status"),
            "health_status": health.get("status"),
            "target_revision": source.get("targetRevision"),
            "repo_url": source.get("repoURL"),
            "path": source.get("path"),
            "destination": spec.get("destination"),
        })
    return json.dumps(apps, indent=2)


@mcp.tool()
async def get_application(name: str) -> str:
    """Get full details for a single application: spec, sync status, health, conditions, and current operation state."""
    data = await _get_json(f"/applications/{name}")
    return json.dumps(data, indent=2)


@mcp.tool()
async def get_application_resource_tree(name: str) -> str:
    """Get the resource tree for an application: every Kubernetes resource managed by the app, with per-node health and parent/child relationships."""
    data = await _get_json(f"/applications/{name}/resource-tree")
    return json.dumps(data, indent=2)


@mcp.tool()
async def get_application_managed_resources(name: str) -> str:
    """Get managed resources for an application with live-vs-desired diff. Useful for inspecting drift on OutOfSync apps."""
    data = await _get_json(f"/applications/{name}/managed-resources")
    return json.dumps(data, indent=2)


@mcp.tool()
async def get_application_events(name: str) -> str:
    """Get recent Kubernetes events for an application. Useful for debugging stuck syncs and failed resources."""
    data = await _get_json(f"/applications/{name}/events")
    return json.dumps(data, indent=2)


@mcp.tool()
async def get_application_manifests(name: str, revision: str | None = None) -> str:
    """Get the rendered desired-state manifests for an application at a given revision (defaults to current target revision)."""
    params = {"revision": revision} if revision else None
    data = await _get_json(f"/applications/{name}/manifests", params=params)
    return json.dumps(data, indent=2)


@mcp.tool()
async def get_application_logs(
    name: str,
    pod: str,
    container: str | None = None,
    tail_lines: int = 200,
    namespace: str | None = None,
) -> str:
    """Get logs from a pod managed by an ArgoCD application.

    Args:
        name: Application name.
        pod: Pod name (from get_application_resource_tree).
        container: Optional container name (defaults to first container).
        tail_lines: Number of lines from the end to return (default 200).
        namespace: Pod namespace (defaults to app destination namespace).
    """
    params: dict = {"podName": pod, "tailLines": str(tail_lines)}
    if container:
        params["container"] = container
    if namespace:
        params["namespace"] = namespace
    raw = await _get_text(f"/applications/{name}/logs", params=params)
    # The logs endpoint streams newline-delimited JSON: {"result":{"content":"..."}}.
    lines = []
    for line in raw.splitlines():
        if not line.strip():
            continue
        try:
            obj = json.loads(line)
            content = obj.get("result", {}).get("content", "")
            if content:
                lines.append(content)
        except json.JSONDecodeError:
            lines.append(line)
    return "\n".join(lines) or "(no log output)"


@mcp.tool()
async def list_projects() -> str:
    """List ArgoCD projects with their allowed source repos, destinations, and cluster-resource whitelists."""
    data = await _get_json("/projects")
    projects = []
    for item in data.get("items", []) or []:
        projects.append({
            "name": item.get("metadata", {}).get("name"),
            "description": item.get("spec", {}).get("description"),
            "source_repos": item.get("spec", {}).get("sourceRepos"),
            "destinations": item.get("spec", {}).get("destinations"),
        })
    return json.dumps(projects, indent=2)


@mcp.tool()
async def list_clusters() -> str:
    """List clusters registered with ArgoCD. Credentials are stripped from the output."""
    data = await _get_json("/clusters")
    clusters = [_scrub_cluster(c) for c in (data.get("items", []) or [])]
    return json.dumps(clusters, indent=2)


@mcp.tool()
async def list_repositories() -> str:
    """List Git repositories registered with ArgoCD. Credentials are stripped from the output."""
    data = await _get_json("/repositories")
    repos = [_scrub_repo(r) for r in (data.get("items", []) or [])]
    return json.dumps(repos, indent=2)


if __name__ == "__main__":
    mcp.run()