argocd-mcp

Instructions

Create a new application in ArgoCD

Args:
    name: The name of the application (required)
    project: The project name (required)
    repo_url: The Git repository URL (required)
    path: Path within the repository (required)
    dest_server: Destination K8s API server URL (required)
    dest_namespace: Destination namespace (required)
    revision: Git revision (default: HEAD)
    automated_sync: Enable automated sync (default: False)
    prune: Auto-prune resources (default: False)
    self_heal: Enable self-healing (default: False)
    namespace: Application namespace
    validate: Whether to validate the application before creation
    upsert: Whether to update the application if it already exists

Returns:
    The created application details

Implementation Reference

• tools/applications.py:101-177 (handler)

  The handler function that implements the create_application tool. It constructs an Application object from parameters, adds sync policy if needed, converts to API format, and POSTs to ArgoCD API.

  async def create_application(
      name: str,
      project: str,
      repo_url: str,
      path: str,
      dest_server: str,
      dest_namespace: str,
      revision: str = "HEAD",
      automated_sync: bool = False,
      prune: bool = False,
      self_heal: bool = False,
      namespace: str = "",
      validate: bool = True,
      upsert: bool = False,
  ) -> Dict[str, Any]:
      """
      Create a new application in ArgoCD
  
      Args:
          name: The name of the application (required)
          project: The project name (required)
          repo_url: The Git repository URL (required)
          path: Path within the repository (required)
          dest_server: Destination K8s API server URL (required)
          dest_namespace: Destination namespace (required)
          revision: Git revision (default: HEAD)
          automated_sync: Enable automated sync (default: False)
          prune: Auto-prune resources (default: False)
          self_heal: Enable self-healing (default: False)
          namespace: Application namespace
          validate: Whether to validate the application before creation
          upsert: Whether to update the application if it already exists
  
      Returns:
          The created application details
      """
      # Create application structure
      app = Application(
          name=name,
          project=project,
          source=ApplicationSource(
              repo_url=repo_url, path=path, target_revision=revision
          ),
          destination=ApplicationDestination(
              server=dest_server, namespace=dest_namespace
          ),
      )
  
      # Add sync policy if automated sync is enabled
      if automated_sync:
          app.sync_policy = ApplicationSyncPolicy(
              automated=True, prune=prune, self_heal=self_heal
          )
  
      if namespace:
          app.namespace = namespace
  
      # Convert to API format
      data = application_to_api_format(app)
  
      # Add query parameters
      params = {}
      if validate:
          params["validate"] = "true"
      if upsert:
          params["upsert"] = "true"
  
      success, response = await make_api_request(
          "applications", method="POST", data=data, params=params
      )
  
      if success:
          logger.info(f"Application '{name}' created successfully")
          return response
      else:
          logger.error(f"Failed to create application '{name}': {response.get('error')}")
          return {"error": response.get("error", "Failed to create application")}

• server.py:43-43 (registration)

  Registers the create_application function as an MCP tool using the FastMCP tool decorator.

  mcp.tool()(applications.create_application)

• models/applications.py:60-72 (schema)

  Dataclass defining the Application model used by the create_application handler for structuring input data and serialization to ArgoCD API format.

  class Application:
      """
      ArgoCD application model
      """
  
      name: str
      project: str
      source: ApplicationSource
      destination: ApplicationDestination
      sync_policy: Optional[ApplicationSyncPolicy] = None
      namespace: str = "argocd"
      status: Optional[ApplicationStatus] = None

• models/applications.py:74-134 (helper)

  Helper function to convert the Application dataclass to the exact dictionary format required by the ArgoCD API, used within the handler.

  def application_to_api_format(app: Application) -> Dict[str, Any]:
      """
      Convert an Application object to the format expected by ArgoCD API
  
      Args:
          app: Application object
  
      Returns:
          Dictionary in ArgoCD API format
      """
      source = {
          "repoURL": app.source.repo_url,
          "path": app.source.path,
          "targetRevision": app.source.target_revision,
      }
  
      # Add optional source fields if they exist
      if app.source.helm is not None:
          source["helm"] = app.source.helm  # type: ignore
      if app.source.kustomize is not None:
          source["kustomize"] = app.source.kustomize  # type: ignore
      if app.source.directory is not None:
          source["directory"] = app.source.directory  # type: ignore
  
      destination = {
          "server": app.destination.server,
          "namespace": app.destination.namespace,
      }
  
      if app.destination.name:
          destination["name"] = app.destination.name
  
      result = {
          "metadata": {"name": app.name, "namespace": app.namespace},
          "spec": {"project": app.project, "source": source, "destination": destination},
      }
  
      # Add sync policy if provided
      if app.sync_policy:
          sync_policy = {}
  
          if app.sync_policy.automated:
              sync_policy["automated"] = {
                  "prune": app.sync_policy.prune,
                  "selfHeal": app.sync_policy.self_heal,
                  "allowEmpty": app.sync_policy.allow_empty,
              }
  
          if app.sync_policy.sync_options:
              # Convert to list for compatibility with API
              sync_policy["syncOptions"] = list(app.sync_policy.sync_options)  # type: ignore
  
          if app.sync_policy.retry:
              sync_policy["retry"] = app.sync_policy.retry
  
          # Add the sync policy to the spec
          spec = result.get("spec")
          if isinstance(spec, dict):
              spec["syncPolicy"] = sync_policy
  
      return result