argocd-mcp

Instructions

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