Ansible + Kubernetes MCP Server

How to use

You interact with your MCP servers through Claude Desktop. Start Claude Desktop and ensure the MCP hammer icon is visible, indicating your MCP tools are loaded. Click the hammer to view available tools, then run any tool to perform actions against your connected platforms. For example, you can list inventories, launch job templates, or fetch job statuses from Ansible Automation Platform, and you can access Kubernetes-related MCP capabilities if you have the Kubernetes MCP Server configured.

How to install

Prerequisites you need before starting: install an Ansible Automation Platform (AAP) environment, an OpenShift cluster (with OpenShift Virtualization if needed), Claude Desktop, and Python 3.10 or higher. Ensure you are authenticated with your OpenShift cluster (for example, export kubeconfig).

Install the required tooling and set up your Python project. Run the following commands in sequence:

curl -LsSf https://astral.sh/uv/install.sh | sh

# Optional: install jbang if you will use the Kubernetes MCP Server
# (jbang should be installed globally for Claude to access it)

# Restart terminal to ensure uv and jbang are available

Step 2: Create and Setup your Project

Create a new directory for your project, set up a virtual environment, and add MCP dependencies.

# Create a new directory for our project
uv init ansible
cd ansible

# Create virtual environment and activate it
uv venv
source .venv/bin/activate

# Install dependencies
uv add "mcp[cli]" httpx

# Create our server file
touch ansible.py

Step 3 Building your Ansible Automation Controller MCP Server

Create the Ansible MCP Server to interact with your Ansible Automation Platform. This example shows the complete Python server file you can place in ansible.py. If you connect to self-signed SSL, you may set verify=False in the HTTP client.

import os
import httpx
from mcp.server.fastmcp import FastMCP
from typing import Any

# Environment variables for authentication
AAP_URL = os.getenv("AAP_URL")
AAP_TOKEN = os.getenv("AAP_TOKEN")

if not AAP_TOKEN:
    raise ValueError("AAP_TOKEN is required")

# Headers for API authentication
HEADERS = {
    "Authorization": f"Bearer {AAP_TOKEN}",
    "Content-Type": "application/json"
}

# Initialize FastMCP
mcp = FastMCP("ansible")

async def make_request(url: str, method: str = "GET", json: dict = None) -> Any:
    """Helper function to make authenticated API requests to AAP."""
    async with httpx.AsyncClient() as client:
        response = await client.request(method, url, headers=HEADERS, json=json)
    if response.status_code not in [200, 201]:
        return f"Error {response.status_code}: {response.text}"
    return response.json() if "application/json" in response.headers.get("Content-Type", "") else response.text

@mcp.tool()
async def list_inventories() -> Any:
    """List all inventories in Ansible Automation Platform."""
    return await make_request(f"{AAP_URL}/inventories/")

@mcp.tool()
async def get_inventory(inventory_id: str) -> Any:
    """Get details of a specific inventory by ID."""
    return await make_request(f"{AAP_URL}/inventories/{inventory_id}/")

@mcp.tool()
async def run_job(template_id: int, extra_vars: dict = {}) -> Any:
    """Run a job template by ID, optionally with extra_vars."""
    return await make_request(f"{AAP_URL}/job_templates/{template_id}/launch/", method="POST", json={"extra_vars": extra_vars})

@mcp.tool()
async def job_status(job_id: int) -> Any:
    """Check the status of a job by ID."""
    return await make_request(f"{AAP_URL}/jobs/{job_id}/")

@mcp.tool()
async def job_logs(job_id: int) -> str:
    """Retrieve logs for a job."""
    return await make_request(f"{AAP_URL}/jobs/{job_id}/stdout/")

@mcp.tool()
async def create_project(
    name: str,
    organization_id: int,
    source_control_url: str,
    source_control_type: str = "git",
    description: str = "",
    execution_environment_id: int = None,
    content_signature_validation_credential_id: int = None,
    source_control_branch: str = "",
    source_control_refspec: str = "",
    source_control_credential_id: int = None,
    clean: bool = False,
    update_revision_on_launch: bool = False,
    delete: bool = False,
    allow_branch_override: bool = False,
    track_submodules: bool = False,
) -> Any:
    """Create a new project in Ansible Automation Platform."""

    payload = {
        "name": name,
        "description": description,
        "organization": organization_id,
        "scm_type": source_control_type.lower(),  # Git is default
        "scm_url": source_control_url,
        "scm_branch": source_control_branch,
        "scm_refspec": source_control_refspec,
        "scm_clean": clean,
        "scm_delete_on_update": delete,
        "scm_update_on_launch": update_revision_on_launch,
        "allow_override": allow_branch_override,
        "scm_track_submodules": track_submodules,
    }

    if execution_environment_id:
        payload["execution_environment"] = execution_environment_id
    if content_signature_validation_credential_id:
        payload["signature_validation_credential"] = content_signature_validation_credential_id
    if source_control_credential_id:
        payload["credential"] = source_control_credential_id

    return await make_request(f"{AAP_URL}/projects/", method="POST", json=payload)

@mcp.tool()
async def create_job_template(
    name: str,
    project_id: int,
    playbook: str,
    inventory_id: int,
    job_type: str = "run",
    description: str = "",
    credential_id: int = None,
    execution_environment_id: int = None,
    labels: list[str] = None,
    forks: int = 0,
    limit: str = "",
    verbosity: int = 0,
    timeout: int = 0,
    job_tags: list[str] = None,
    skip_tags: list[str] = None,
    extra_vars: dict = None,
    privilege_escalation: bool = False,
    concurrent_jobs: bool = False,
    provisioning_callback: bool = False,
    enable_webhook: bool = False,
    prevent_instance_group_fallback: bool = False,
) -> Any:
    """Create a new job template in Ansible Automation Platform."""

    payload = {
        "name": name,
        "description": description,
        "job_type": job_type,
        "project": project_id,
        "playbook": playbook,
        "inventory": inventory_id,
        "forks": forks,
        "limit": limit,
        "verbosity": verbosity,
        "timeout": timeout,
        "ask_variables_on_launch": bool(extra_vars),
        "ask_tags_on_launch": bool(job_tags),
        "ask_skip_tags_on_launch": bool(skip_tags),
        "ask_credential_on_launch": credential_id is None,
        "ask_execution_environment_on_launch": execution_environment_id is None,
        "ask_labels_on_launch": labels is None,
        "ask_inventory_on_launch": False,
        "ask_job_type_on_launch": False,
        "become_enabled": privilege_escalation,
        "allow_simultaneous": concurrent_jobs,
        "scm_branch": "",
        "webhook_service": "github" if enable_webhook else "",
        "prevent_instance_group_fallback": prevent_instance_group_fallback,
    }

    if credential_id:
        payload["credential"] = credential_id
    if execution_environment_id:
        payload["execution_environment"] = execution_environment_id
    if labels:
        payload["labels"] = labels
    if job_tags:
        payload["job_tags"] = job_tags
    if skip_tags:
        payload["skip_tags"] = skip_tags
    if extra_vars:
        payload["extra_vars"] = extra_vars

    return await make_request(f"{AAP_URL}/job_templates/", method="POST", json=payload)

@mcp.tool()
async def list_inventory_sources() -> Any:
    """List all inventory sources in Ansible Automation Platform."""
    return await make_request(f"{AAP_URL}/inventory_sources/")

@mcp.tool()
async def get_inventory_source(inventory_source_id: int) -> Any:
    """Get details of a specific inventory source."""
    return await make_request(f"{AAP_URL}/inventory_sources/{inventory_source_id}/")

@mcp.tool()
async def create_inventory_source(
    name: str,
    inventory_id: int,
    source: str,
    credential_id: int,
    source_vars: dict = None,
    update_on_launch: bool = True,
    timeout: int = 0,
) -> Any:
    """Create a dynamic inventory source. Claude will ask for the source type and credential before proceeding."""
    valid_sources = [
        "file", "constructed", "scm", "ec2", "gce", "azure_rm", "vmware", "satellite6", "openstack", 
        "rhv", "controller", "insights", "terraform", "openshift_virtualization"
    ]
    
    if source not in valid_sources:
        return f"Error: Invalid source type '{source}'. Please select from: {', '.join(valid_sources)}"
    
    if not credential_id:
        return "Error: Credential is required to create an inventory source."
    
    payload = {
        "name": name,
        "inventory": inventory_id,
        "source": source,
        "credential": credential_id,
        "source_vars": source_vars,
        "update_on_launch": update_on_launch,
        "timeout": timeout,
    }
    return await make_request(f"{AAP_URL}/inventory_sources/", method="POST", json=payload)

@mcp.tool()
async def update_inventory_source(inventory_source_id: int, update_data: dict) -> Any:
    """Update an existing inventory source."""
    return await make_request(f"{AAP_URL}/inventory_sources/{inventory_source_id}/", method="PATCH", json=update_data)

@mcp.tool()
async def delete_inventory_source(inventory_source_id: int) -> Any:
    """Delete an inventory source."""
    return await make_request(f"{AAP_URL}/inventory_sources/{inventory_source_id}/", method="DELETE")

@mcp.tool()
async def sync_inventory_source(inventory_source_id: int) -> Any:
    """Manually trigger a sync for an inventory source."""
    return await make_request(f"{AAP_URL}/inventory_sources/{inventory_source_id}/update/", method="POST")

@mcp.tool()
async def create_inventory(
    name: str,
    organization_id: int,
    description: str = "",
    kind: str = "",
    host_filter: str = "",
    variables: dict = None,
    prevent_instance_group_fallback: bool = False,
) -> Any:
    """Create an inventory in Ansible Automation Platform."""
    payload = {
        "name": name,
        "organization": organization_id,
        "description": description,
        "kind": kind,
        "host_filter": host_filter,
        "variables": variables,
        "prevent_instance_group_fallback": prevent_instance_group_fallback,
    }
    return await make_request(f"{AAP_URL}/inventories/", method="POST", json=payload)

@mcp.tool()
async def delete_inventory(inventory_id: int) -> Any:
    """Delete an inventory from Ansible Automation Platform."""
    return await make_request(f"{AAP_URL}/inventories/{inventory_id}/", method="DELETE")

@mcp.tool()
async def list_job_templates() -> Any:
    """List all job templates available in Ansible Automation Platform."""
    return await make_request(f"{AAP_URL}/job_templates/")

@mcp.tool()
async def get_job_template(template_id: int) -> Any:
    """Retrieve details of a specific job template."""
    return await make_request(f"{AAP_URL}/job_templates/{template_id}/")

@mcp.tool()
async def list_jobs() -> Any:
    """List all jobs available in Ansible Automation Platform."""
    return await make_request(f"{AAP_URL}/jobs/")

@mcp.tool()
async def list_recent_jobs(hours: int = 24) -> Any:
    """List all jobs executed in the last specified hours (default 24 hours)."""
    from datetime import datetime, timedelta
    
    time_filter = (datetime.utcnow() - timedelta(hours=hours)).isoformat() + "Z"
    return await make_request(f"{AAP_URL}/jobs/?created__gte={time_filter}")

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

Step 4 Configuring Claude Desktop to use your MCP Servers

Prepare Claude Desktop to connect to your MCP servers by editing the configuration file. You will define two servers: the Ansible MCP Server and the Kubernetes MCP Server. Provide the absolute path to your uv runtime and the correct project paths.

{
  "mcpServers": {
    "ansible": {
        "command": "/absolute/path/to/uv",
        "args": [
            "--directory",
            "/absolute/path/to/ansible_mcp",
            "run",
            "ansible.py"
        ],
        "env": {
            "AAP_TOKEN": "<aap-token>",
            "AAP_URL": "https://<aap-url>/api/controller/v2"
        }
    },
    "kubernetes": {
      "command": "jbang",
      "args": [
        "--quiet",
        "https://github.com/quarkiverse/quarkus-mcp-servers/blob/main/kubernetes/src/main/java/io/quarkiverse/mcp/servers/kubernetes/MCPServerKubernetes.java"
      ]
    }
  }
}

Step 5: Re-Launch Claude Desktop

If Claude Desktop is already open, restart it to pick up the new MCP servers. If you are launching it for the first time, ensure the hammer icon appears, indicating MCP tools are loaded. The number next to the hammer varies with the number of available MCP tools.

Step 6: Test your Environment

Ask Claude Desktop to perform practical checks against your environment, such as querying the number of Job Templates, listing inventories, or retrieving the status of a Kubernetes cluster-related task. This confirmation helps ensure your MCP servers are correctly integrated and responsive.

Notes and tips

The Ansible MCP Server configuration shown requires environment variables for authentication to AAP. Ensure you securely provide AAP_TOKEN and AAP_URL. For local development, you may run into SSL verification issues with self-signed certificates; adjust the client to skip verification if needed, but only in trusted environments.

Security considerations

Protect your AAP_TOKEN and ensure access to Claude Desktop is restricted to authorized users. Use HTTPS endpoints for MCP servers and rotate tokens regularly.

list_inventories

List all inventories in Ansible Automation Platform.

get_inventory

Get details of a specific inventory by ID.

run_job

Launch a job template by ID with optional extra variables.

job_status

Check the status of a specific job by ID.

job_logs

Retrieve the logs for a specific job.

create_project

Create a new project in Ansible Automation Platform.

create_job_template

Create a new job template in Ansible Automation Platform.

list_inventory_sources

List all inventory sources in Ansible Automation Platform.

get_inventory_source

Get details of a specific inventory source.

create_inventory_source

Create a dynamic inventory source with source type and credentials.

update_inventory_source

Update an existing inventory source.

delete_inventory_source

Delete an inventory source.

sync_inventory_source

Manually trigger a sync for an inventory source.

create_inventory

Create an inventory in Ansible Automation Platform.

delete_inventory

Delete an inventory from Ansible Automation Platform.

list_job_templates

List all job templates available in Ansible Automation Platform.

get_job_template

Retrieve details of a specific job template.

list_jobs

List all jobs available in Ansible Automation Platform.

list_recent_jobs

List jobs launched in the past N hours.