""" title: Coolify API author: User description: | Manage your Coolify instance with an Open WebUI Agent. **Very useful for getting AI help with debugging.** Read-only, but able start/stop/restart and update services and applications. Once I get more experience using it, I will add write options. *AS ALWAYS - USE AT YOUR OWN RISK!* ## Example: Understand the Server > Familiarize yourself with my Coolify instance and give me an overview of all systems. The agent will use the following tools to explore and orient inside the instance. - list_servers: List all servers. - list_projects: List all projects. - list_applications: List all applications. - list_services: List all services. ## Example: Debug A Problem > Solve < problem > with < application > The agent will then gather additional information and debug using the following tools: - get_application: Get full application details. - get_service: Get full service details. - get_application_logs: Get the logs for an application. ## Example: Manage Lifecycle > Restart < application > The agent can also manage the lifecycle of applications and services: - start_application: Start an application. - stop_application: Stop an application. - restart_application: Restart an application. - deploy_application: Deploy an application (pulls latest image and restarts). - start_service: Start a service. - stop_service: Stop a service. - restart_service: Restart a service (optionally pulls latest image). required_open_webui_version: 0.6.0 requirements: - httpx - pydantic version: 1.0.0 license: MIT """ import asyncio import json import os from abc import ABC, abstractmethod from typing import Any, Literal, Annotated from datetime import datetime, timezone import httpx from pydantic import BaseModel, Field, ConfigDict class CoolifyAPI: """Base class that must be subclassed for specific Task API integrations.""" def __init__(self, api_url: str, api_key: str): self.api_url = api_url self.api_key = api_key self.client = httpx.AsyncClient( base_url=api_url, headers={"Authorization": f"Bearer {api_key}"} ) async def healthcheck(self, timeout: int = 10) -> str: """Checks the health of the Coolify instance.""" response = await self.client.get("/health", timeout=timeout) return response.text async def get_current_team(self) -> dict: """Gets the current team information.""" response = await self.client.get("v1/teams/current") return response.json() async def list_servers(self) -> list[dict]: """Lists all servers.""" response = await self.client.get("v1/servers") return response.json() async def list_projects(self) -> list[dict]: """Lists all projects.""" response = await self.client.get("v1/projects") return response.json() async def list_applications(self, raw: bool = False) -> list[dict]: """Lists all applications. Pass raw=True to get the full response, otherwise some fields are omitted for readability and context size.""" response = await self.client.get("v1/applications") # filter out deleted applications response = [app for app in response.json() if app.get("deleted_at") is None] if raw: return response else: return _whitelist_keys(response, self.LIST_APPLICATIONS_RETURN_KEYS) async def list_services(self, raw: bool = False) -> list[dict]: """Lists all services. Pass raw=True to get the full response, otherwise some fields are omitted for readability and context size.""" response = await self.client.get("v1/services") response = [ service for service in response.json() if service.get("deleted_at") is None ] if raw: return response else: return _whitelist_keys(response, self.LIST_SERVICES_RETURN_KEYS) async def get_application(self, uuid: str) -> dict: """Gets an Application by ID. Returns the full details - more information than list_applications does by default.""" response = await self.client.get(f"v1/applications/{uuid}") return response.json() async def get_service(self, uuid: str) -> dict: """Gets a Service by ID. Returns the full details - more information than list_services does by default.""" response = await self.client.get(f"v1/services/{uuid}") return response.json() async def get_application_logs(self, uuid: str, lines: int = 100) -> str: """Gets the logs for an Application by ID. Useful for debugging application issues.""" response = await self.client.get( f"v1/applications/{uuid}/logs", params={"lines": lines} ) return response.json().get("logs", response.text) async def start_service(self, uuid: str) -> dict: """Starts a service.""" response = await self.client.post(f"v1/services/{uuid}/start") return response.json() async def stop_service(self, uuid: str) -> dict: """Stops a service.""" response = await self.client.post(f"v1/services/{uuid}/stop") return response.json() async def restart_service(self, uuid: str, latest: bool = False) -> dict: """Restarts a service.""" params = {} if latest: params["latest"] = "true" response = await self.client.post(f"v1/services/{uuid}/restart", params=params) return response.json() async def start_application(self, uuid: str) -> dict: """Starts an application.""" response = await self.client.post(f"v1/applications/{uuid}/start") return response.json() async def stop_application(self, uuid: str) -> dict: """Stops an application.""" response = await self.client.post(f"v1/applications/{uuid}/stop") return response.json() async def restart_application(self, uuid: str) -> dict: """Restarts an application.""" response = await self.client.post(f"v1/applications/{uuid}/restart") return response.json() async def deploy_application(self, uuid: str, force: bool = False) -> dict: """ Deploys an application, pulling the latest image and restarting it if it is newer, or if force is specified. """ params = {} if force: params["force"] = "true" response = await self.client.post( f"v1/applications/{uuid}/deploy", params=params ) return response.json() async def list_service_environments(self, uuid: str) -> list[dict]: """Lists environments for a project (which contains services).""" response = await self.client.get(f"v1/projects/{uuid}/environments") return response.json() async def list_application_environments(self, uuid: str) -> list[dict]: """Lists environments for an application.""" response = await self.client.get(f"v1/applications/{uuid}/envs") return response.json() LIST_APPLICATIONS_RETURN_KEYS = [ "name", "id", "uuid", "status", "description", "git_repository", "fqdn", "server_id", ] LIST_SERVICES_RETURN_KEYS = [ "name", "id", "uuid", "status", "description", "service_type", "fqdn", "environment_id", "server_id", # Server "server.id", "server.uuid", "server.name" "server.description" # Applications "applications.id", "applications.uuid", "applications.name", "applications.human_name", "applications.status", "applications.description", "applications.git_repository", "applications.fqdn", "applications.image", "applications.ports", "applications.exposes", "applications.last_online_at", # Databases "databases.id", "databases.uuid", "databases.name", "databases.human_name", "databases.status", "databases.description", "databases.ports", "databases.exposes", "databases.last_online_at", ] class Tools: class Valves(BaseModel): model_config = ConfigDict(extra="forbid") default_coolify_api_url: str = Field( "", description="Default Coolify API URL, e.g. http://<ip or url>:8000/api. Leave blank to require users set their own.", ) default_coolify_api_key: str = Field( "", description="Default Coolify API Key. Leave blank to require users set their own.", ) class UserValves(BaseModel): model_config = ConfigDict(extra="forbid") coolify_api_url: str = Field( "", description="Your Coolify API URL, e.g. http://<ip or url>:8000/api" ) coolify_api_key: str = Field("", description="Your Coolify API Key") def __init__(self): self.valves = self.Valves() self.user_valves = self.UserValves() async def list_servers(self, __user__: dict = {}) -> list[dict]: """Lists all Servers. Useful when planning where to deploy a new application or service.""" return await _get_api(self, __user__).list_servers() async def list_projects(self, __user__: dict = {}) -> list[dict]: """Lists all Projects. Useful when planning where to deploy a new application or service.""" return await _get_api(self, __user__).list_projects() async def list_applications( self, __user__: dict = {}, ) -> list[dict]: """Lists all Applications. Useful for getting an overview.""" return await _get_api(self, __user__).list_applications() async def list_services( self, __user__: dict = {}, ) -> list[dict]: """Lists all Services. Useful for getting an overview.""" return await _get_api(self, __user__).list_services() async def get_application( self, uuid: Annotated[str, Field(...)], __user__: dict = {}, ) -> dict: """Gets full Application details. Useful for understanding an application.""" return await _get_api(self, __user__).get_application(uuid) async def get_service( self, uuid: Annotated[str, Field(...)], __user__: dict = {}, ) -> dict: """Gets full Service details. Useful for understanding a service.""" return await _get_api(self, __user__).get_service(uuid) async def get_application_logs( self, uuid: Annotated[str, Field(...)], lines: Annotated[ int, Field(default=100), ], __user__: dict = {}, ) -> str: """Gets recent loglines for an Application. Useful for debugging issues.""" return await _get_api(self, __user__).get_application_logs(uuid, lines) async def start_service( self, uuid: Annotated[str, Field(description="The UUID of the Service.")], __user__: dict = {}, ) -> dict: """Starts a Service.""" return await _get_api(self, __user__).start_service(uuid) async def stop_service( self, uuid: Annotated[str, Field(description="The UUID of the Service.")], __user__: dict = {}, ) -> dict: """Stops a Service.""" return await _get_api(self, __user__).stop_service(uuid) async def restart_service( self, uuid: Annotated[str, Field(description="The UUID of the Service.")], latest: Annotated[ bool, Field( description="Pull the latest image before restarting.", default=False ), ] = False, __user__: dict = {}, ) -> dict: """Restarts a Service, optionally updating it to the latest image.""" return await _get_api(self, __user__).restart_service(uuid, latest) async def start_application( self, uuid: Annotated[str, Field(description="The UUID of the Application.")], __user__: dict = {}, ) -> dict: """Starts an Application.""" return await _get_api(self, __user__).start_application(uuid) async def stop_application( self, uuid: Annotated[str, Field(description="The UUID of the Application.")], __user__: dict = {}, ) -> dict: """Stops an Application.""" return await _get_api(self, __user__).stop_application(uuid) async def restart_application( self, uuid: Annotated[str, Field(description="The UUID of the Application.")], __user__: dict = {}, ) -> dict: """Restarts an Application.""" return await _get_api(self, __user__).restart_application(uuid) async def deploy_application( self, uuid: Annotated[str, Field(description="The UUID of the Application.")], force: Annotated[ bool, Field( description="Restarts the application even if there are no changes.", default=False, ), ] = False, __user__: dict = {}, ) -> dict: """Deploys an Application, pulling the latest image and restarting it there are changes.""" return await _get_api(self, __user__).deploy_application(uuid, force) async def list_service_environments( self, uuid: Annotated[str, Field(description="The UUID of the Project.")], __user__: dict = {}, ) -> list[dict]: """Lists environments for a Project.""" return await _get_api(self, __user__).list_service_environments(uuid) async def list_application_environments( self, uuid: Annotated[str, Field(description="The UUID of the Application.")], __user__: dict = {}, ) -> list[dict]: """Lists environments for an Application.""" return await _get_api(self, __user__).list_application_environments(uuid) class Tester: """ Can be run non-destructively against a production implementation. """ def __init__(self, api: CoolifyAPI): self.api = api self.debug_prints = True self.applications = [] self.services = [] async def test_all(self): """ Tests all methods that can be done without interrupting the production implementation. """ if self.debug_prints: print( f"Testing Coolify API against {self.api.api_url} as {self.api.api_key[:4]}..." ) try: for name in dir(self): if name.startswith("test_") and name != "test_all": method = getattr(self, name) try: if self.debug_prints: print(f"- {name}..", end="") await method() if self.debug_prints: print(" PASS") except Exception as e: if self.debug_prints: print(" FAILED\n") raise Exception(f"Test {name} failed: {e}") from e finally: await self._cleanup_test_data() async def _cleanup_test_data(self): """Clean up all created test data with best-effort approach.""" async def test_0_healthcheck(self): """Tests the healthcheck method.""" healthcheck = await self.api.healthcheck() print(healthcheck) async def test_1_access(self): """Tests the check_access method.""" access = await self.api.get_current_team() print(json.dumps(access, indent=4)) async def test_2_list_servers(self): """Tests the list_servers method.""" servers = await self.api.list_servers() print(json.dumps(servers, indent=4)) async def test_3_list_projects(self): """Tests the list_projects method.""" projects = await self.api.list_projects() print(json.dumps(projects, indent=4)) async def test_4_list_applications(self): """Tests the list_applications method.""" self.applications = await self.api.list_applications() print(json.dumps(self.applications, indent=4)) async def test_5_list_services(self): """Tests the list_services method.""" self.services = await self.api.list_services() print(json.dumps(self.services, indent=4)) async def test_6_get_application(self): """Tests the get_application method.""" if not self.applications: print(" SKIPPED (No applications found)") return app_uuid = self.applications[0].get("uuid") if not app_uuid: print(" SKIPPED (No application UUID found)") return application = await self.api.get_application(app_uuid) print(json.dumps(application, indent=4)) async def test_7_get_service(self): """Tests the get_service method.""" if not self.services: print(" SKIPPED (No services found)") return service_uuid = self.services[0].get("uuid") if not service_uuid: print(" SKIPPED (No service UUID found)") return service = await self.api.get_service(service_uuid) print(json.dumps(service, indent=4)) async def test_8_get_application_logs(self): """Tests the get_application_logs method.""" if not self.applications: print(" SKIPPED (No applications found)") return app_uuid = self.applications[0].get("uuid") if not app_uuid: print(" SKIPPED (No application UUID found)") return logs = await self.api.get_application_logs(app_uuid) print(f"Received {len(logs)} log lines.") if logs: print(logs[:100]) async def test_9_list_service_environments(self): """Tests the list_service_environments method (using project UUID from list_projects).""" projects = await self.api.list_projects() if not projects: print(" SKIPPED (No projects found)") return project_uuid = projects[0].get("uuid") if not project_uuid: print(" SKIPPED (No project UUID found)") return envs = await self.api.list_service_environments(project_uuid) print(f"Received {len(envs)} environments.") if envs: print(json.dumps(envs[0], indent=4)) async def test_10_list_application_environments(self): """Tests the list_application_environments method.""" if not self.applications: print(" SKIPPED (No applications found)") return app_uuid = self.applications[0].get("uuid") if not app_uuid: print(" SKIPPED (No application UUID found)") return try: # Note: The endpoint /applications/{uuid}/envs implies getting environments for an application # However, applications are usually inside an environment. This might be listing variables or similar context. # We will proceed with the requested implementation. envs = await self.api.list_application_environments(app_uuid) print(f"Received {len(envs)} items.") if envs: print(json.dumps(envs[0], indent=4)) except Exception as e: print(f" FAILED with error: {e}") # Don't re-raise to allow other tests to continue if this specific endpoint is tricky pass def _get_api(self: Tools, user: dict = {}) -> "CoolifyAPI": # Prioritize UserValves from the injected __user__ dictionary user_valves = user.get("valves") if user else None url = ( (user_valves.coolify_api_url if user_valves else None) or self.user_valves.coolify_api_url or self.valves.default_coolify_api_url or os.getenv("COOLIFY_API_URL", "") ) key = ( (user_valves.coolify_api_key if user_valves else None) or self.user_valves.coolify_api_key or self.valves.default_coolify_api_key or os.getenv("COOLIFY_API_KEY", "") ) if not url: raise ValueError( "Coolify API URL is missing. Please configure it in User Valves or Global Valves." ) if not key: raise ValueError( "Coolify API Key is missing. Please configure it in User Valves or Global Valves." ) if not url.endswith("/"): url += "/" return CoolifyAPI(url, key) def _now_utc(): """Return current time in UTC.""" return datetime.now(timezone.utc) def _whitelist_keys( input: dict | list[dict], whitelist: list[str] = None, key_separator: str = ".", exclude_blank: bool = True, ) -> dict | list[dict]: """ Returns a copy of the input filtered by the whitelist, in the whitelist order. Keys can be separataed by a key_separator, e.g. "a.b" whitelists key A in the top level dict, and key B in the next level dict. Dicts can be in or separated by lists, every dict in a list of dicts is filtered by the whitelist. Keys can contain up to 1 * wildcard, allowing whitelisting all keys with a common prefix/suffix. If exclude_blank is True, removes keys with None or "" values (preserves 0 and False). """ if not whitelist: return input if isinstance(input, list): return [ _whitelist_keys(item, whitelist, key_separator, exclude_blank) for item in input ] if not isinstance(input, dict): return input # Group sub-keys by matching head keys to avoid complex merging later. # Using a dict preserves insertion order (important for the 'whitelist order' requirement). keys_to_keep = {} for key_path in whitelist: if key_separator in key_path: head, tail = key_path.split(key_separator, 1) else: head, tail = key_path, None # Find matching keys in the input matching_keys = [] if "*" in head: prefix, suffix = head.split("*", 1) matching_keys = [ k for k in input if k.startswith(prefix) and k.endswith(suffix) ] elif head in input: matching_keys = [head] for key in matching_keys: if key not in keys_to_keep: keys_to_keep[key] = [] # If we have a tail, add it. If tail is None, it means "keep everything" for this key. # We store None to indicate full retention. keys_to_keep[key].append(tail) result = {} for key, tails in keys_to_keep.items(): value = None # If any tail is None, it implies we selected the whole key at some point. if None in tails: value = input[key] else: # Recurse with collected tails value = _whitelist_keys(input[key], tails, key_separator, exclude_blank) if exclude_blank and ( value is None or (isinstance(value, str) and value == "") ): continue result[key] = value return result def _repr_llm( obj_or_list, hide_none_fields: bool = True, insert_classname: bool = False, public_only: bool = False, ) -> dict | list[dict]: """ Gets a LLM-friendly dict describing the object or list of dicts describing the objects. If insert_classname is True, the key "_classname" is inserted at the start of the dict. If public_only is True, only public fields are included (besides optional _classname key). """ if isinstance(obj_or_list, list): return [ _repr_llm( item, hide_none_fields=hide_none_fields, insert_classname=insert_classname, public_only=public_only, ) for item in obj_or_list ] contents = {} classname = None if isinstance(obj_or_list, dict): contents = obj_or_list elif isinstance(obj_or_list, BaseModel): contents = obj_or_list.model_dump(mode="json", exclude_none=hide_none_fields) classname = obj_or_list.__class__.__name__ if public_only: contents = {k: v for k, v in contents.items() if not k.startswith("_")} if hide_none_fields: contents = {k: v for k, v in contents.items() if v != ""} if insert_classname and classname: contents = {"_classname": classname, **contents} return contents if __name__ == "__main__": import os from dotenv import load_dotenv from pathlib import Path load_dotenv(Path(__file__).resolve().parent.parent.parent / ".env") api_url = os.getenv("COOLIFY_API_URL") api_key = os.getenv("COOLIFY_API_KEY") if api_url is not None and api_key is not None: api = CoolifyAPI(api_url, api_key) tester = Tester(api) asyncio.run(tester.test_all())