coolipy.async_client

Asynchronous coolipy client.

Classes

class AsyncCoolipy (coolify_api_key: str,
coolify_endpoint: str,
coolify_port: int = 8000,
omit_port: bool = False,
http_protocol: str = 'http',
timeout: float = 30.0,
transport: httpx.AsyncBaseTransport | None = None)

Expand source code

class AsyncCoolipy:
    """Asynchronous client for the Coolify API.

    Args:
        coolify_api_key: Bearer token used to authenticate with the Coolify API.
        coolify_endpoint: Hostname or IP of the Coolify instance.
        coolify_port: Port of the Coolify instance (ignored when ``omit_port``).
        omit_port: When ``True``, build the base URL without a port.
        http_protocol: ``"http"`` or ``"https"``.
        timeout: Request timeout in seconds.
        transport: Optional httpx async transport override (mainly for testing).
    """

    def __init__(
        self,
        coolify_api_key: str,
        coolify_endpoint: str,
        coolify_port: int = 8000,
        omit_port: bool = False,
        http_protocol: str = "http",
        timeout: float = 30.0,
        transport: httpx.AsyncBaseTransport | None = None,
    ) -> None:
        if not coolify_api_key:
            raise CoolipyConfigError("coolify_api_key must be a non-empty string.")
        base_url = build_base_url(http_protocol, coolify_endpoint, coolify_port, omit_port)
        self._transport = AsyncTransport(
            base_url, coolify_api_key, timeout=timeout, transport=transport
        )
        self.applications = AsyncApplications(self._transport)
        self.databases = AsyncDatabases(self._transport)
        self.services = AsyncServices(self._transport)
        self.servers = AsyncServers(self._transport)
        self.projects = AsyncProjects(self._transport)
        self.teams = AsyncTeams(self._transport)
        self.deployments = AsyncDeployments(self._transport)
        self.tags = AsyncTags(self._transport)
        self.s3_storages = AsyncS3Storages(self._transport)
        self.security = AsyncSecurity(self._transport)

    async def version(self) -> CoolipyAPIResponse[str]:
        """Get the Coolify version."""
        return parse_response(await self._transport.request("GET", "/version"), str)

    async def health(self) -> CoolipyAPIResponse[str]:
        """Run the Coolify healthcheck."""
        return parse_response(await self._transport.request("GET", "/health"), str)

    async def enable_api(self) -> CoolipyAPIResponse[SystemMessage]:
        """Enable the Coolify API (requires root permissions)."""
        return parse_response(await self._transport.request("POST", "/enable"), SystemMessage)

    async def disable_api(self) -> CoolipyAPIResponse[SystemMessage]:
        """Disable the Coolify API (requires root permissions)."""
        return parse_response(await self._transport.request("POST", "/disable"), SystemMessage)

    async def enable_mcp(self) -> CoolipyAPIResponse[SystemMessage]:
        """Enable the Coolify MCP server (requires root permissions)."""
        return parse_response(await self._transport.request("POST", "/mcp/enable"), SystemMessage)

    async def disable_mcp(self) -> CoolipyAPIResponse[SystemMessage]:
        """Disable the Coolify MCP server (requires root permissions)."""
        return parse_response(await self._transport.request("POST", "/mcp/disable"), SystemMessage)

    async def close(self) -> None:
        """Close the underlying HTTP client."""
        await self._transport.close()

    async def __aenter__(self) -> AsyncCoolipy:
        return self

    async def __aexit__(self, *exc: object) -> None:
        await self.close()

Asynchronous client for the Coolify API.

Args

-----=

coolify_api_key

Bearer token used to authenticate with the Coolify API.

coolify_endpoint

Hostname or IP of the Coolify instance.

coolify_port

Port of the Coolify instance (ignored when omit_port).

omit_port

When True, build the base URL without a port.

http_protocol

"http" or "https".

timeout

Request timeout in seconds.

transport

Optional httpx async transport override (mainly for testing).

Methods

async def close(self) ‑> None

Expand source code

async def close(self) -> None:
    """Close the underlying HTTP client."""
    await self._transport.close()

Close the underlying HTTP client.

async def disable_api(self) ‑> coolipy._response.CoolipyAPIResponse[SystemMessage]

Expand source code

async def disable_api(self) -> CoolipyAPIResponse[SystemMessage]:
    """Disable the Coolify API (requires root permissions)."""
    return parse_response(await self._transport.request("POST", "/disable"), SystemMessage)

Disable the Coolify API (requires root permissions).

async def disable_mcp(self) ‑> coolipy._response.CoolipyAPIResponse[SystemMessage]

Expand source code

async def disable_mcp(self) -> CoolipyAPIResponse[SystemMessage]:
    """Disable the Coolify MCP server (requires root permissions)."""
    return parse_response(await self._transport.request("POST", "/mcp/disable"), SystemMessage)

Disable the Coolify MCP server (requires root permissions).

async def enable_api(self) ‑> coolipy._response.CoolipyAPIResponse[SystemMessage]

Expand source code

async def enable_api(self) -> CoolipyAPIResponse[SystemMessage]:
    """Enable the Coolify API (requires root permissions)."""
    return parse_response(await self._transport.request("POST", "/enable"), SystemMessage)

Enable the Coolify API (requires root permissions).

async def enable_mcp(self) ‑> coolipy._response.CoolipyAPIResponse[SystemMessage]

Expand source code

async def enable_mcp(self) -> CoolipyAPIResponse[SystemMessage]:
    """Enable the Coolify MCP server (requires root permissions)."""
    return parse_response(await self._transport.request("POST", "/mcp/enable"), SystemMessage)

Enable the Coolify MCP server (requires root permissions).

async def health(self) ‑> coolipy._response.CoolipyAPIResponse[str]

Expand source code

async def health(self) -> CoolipyAPIResponse[str]:
    """Run the Coolify healthcheck."""
    return parse_response(await self._transport.request("GET", "/health"), str)

Run the Coolify healthcheck.

async def version(self) ‑> coolipy._response.CoolipyAPIResponse[str]

Expand source code

async def version(self) -> CoolipyAPIResponse[str]:
    """Get the Coolify version."""
    return parse_response(await self._transport.request("GET", "/version"), str)

Get the Coolify version.