northflank 1.0.0

Python SDK for the Northflank platform API

Northflank Python SDK

The official Python client for the Northflank API. It provides typed endpoint methods for every Northflank resource — services, sandboxes, GPU workloads, jobs, addons, secrets, volumes, domains, metrics, and more — plus exec and logs (range fetch and live tailing).

Installation

pip install northflank

Requires Python 3.9–3.13.

Quick start

from northflank import ApiClient

client = ApiClient(api_token="your-api-token")

# List the services in a project
services = client.list.services(project_id="my-project")
for service in services.data["services"]:
    print(service["name"])

Every call returns an ApiCallResponse with .data, .status, and .error. Method arguments are keyword-only and fully typed — request bodies and response data are described by generated TypedDicts, so editors and type checkers can validate payloads.

Error handling

By default the client raises ApiCallError on any non-2xx response — this matches the convention of requests, httpx, boto3, and most Python SDKs:

from northflank import ApiCallError

try:
    client.get.service(project_id="my-project", service_id="does-not-exist")
except ApiCallError as exc:
    print(exc.status, exc.message)

Pass throw_on_http_error=False to instead surface the error on the response object and never raise:

client = ApiClient(api_token="...", throw_on_http_error=False)

resp = client.get.service(project_id="my-project", service_id="does-not-exist")
if resp.error is not None:
    print(resp.error.status, resp.error.message)

Working with services

Create a service

client.create.service.deployment(
    project_id="my-project",
    data={
        "name": "my-service",
        "billing": {"deploymentPlan": "nf-compute-20"},
        "deployment": {
            "instances": 1,
            "external": {"imagePath": "nginx:latest"},
            "docker": {"configType": "default"},
        },
        "ports": [
            {"name": "http", "internalPort": 80, "public": True, "protocol": "HTTP"},
        ],
    },
)

Wait for it to be running

import time

while True:
    service = client.get.service(project_id="my-project", service_id="my-service")
    status = service.data["status"]["deployment"]["status"]
    if status == "COMPLETED":
        break
    if status == "FAILED":
        raise RuntimeError("deployment failed")
    time.sleep(3)

Pause, resume, and delete

client.pause.service(project_id="my-project", service_id="my-service")
client.resume.service(project_id="my-project", service_id="my-service", data={})
client.delete.service(project_id="my-project", service_id="my-service")

Exec

One-shot commands

run_service_command runs a command, waits for it to finish, and returns an ExecResult (exit_code, stdout, stderr, status):

result = client.exec.run_service_command(
    project_id="my-project",
    service_id="my-service",
    command="cat /etc/os-release",
)
print(result.exit_code, result.stdout)

# Commands that need a shell (pipes, &&, redirection) — pass `shell`:
client.exec.run_service_command(
    project_id="my-project",
    service_id="my-service",
    command="echo hello && uname -a",
    shell="bash -c",
)

The same one-shot API works for jobs and addons:

client.exec.run_job_command(project_id="my-project", job_id="my-job", command="ls")
client.exec.run_addon_command(
    project_id="my-project", addon_id="my-redis", command="redis-cli ping", shell="bash -c"
)

Interactive sessions

open_service_session opens a persistent connection: write to the process's stdin, iterate output as it streams, resize the terminal. Use it as a context manager so the WebSocket is always closed. Send input, read the output it produces, then send_eof — closing stdin immediately after a send can race the proxy.

with client.exec.open_service_session(
    project_id="my-project", service_id="my-service", command="cat"
) as session:
    session.resize(rows=40, columns=120)
    session.send("hello\n")
    for chunk in session:            # chunk.stream is "stdout" / "stderr"
        print(chunk.data, end="")
        if "hello" in chunk.data:
            break
    session.send_eof()               # close stdin
    result = session.wait()          # ExecResult once the process exits

open_job_session and open_addon_session cover jobs and addons. The async client exposes aopen_service_session / aopen_job_session / aopen_addon_session, returning an AsyncExecSession driven with async with and async for.

Logs

Fetch a recent range

lines = client.logs.fetch_service_logs(
    project_id="my-project",
    service_id="my-service",
    line_limit=100,
    text_includes="error",
)
for line in lines:
    print(line.ts, line.log)

Live tail

tail_service_logs opens a WebSocket and returns a LogTail you iterate for LogLine items as they arrive:

with client.logs.tail_service_logs(
    project_id="my-project", service_id="my-service", recv_timeout=30.0
) as tail:
    for line in tail:
        print(line.ts, line.log)

tail_job_logs and tail_addon_logs cover jobs and addons; the async client provides atail_service_logs / atail_job_logs / atail_addon_logs (AsyncLogTail, driven with async for).

Metrics

client.get.{service,job,addon}.metrics fetches resource metrics. query_type selects a time range or a single snapshot, metric_types takes a list, and time-range arguments accept native datetime objects:

from datetime import datetime, timedelta, timezone
from northflank import METRIC_TYPES

metrics = client.get.service.metrics(
    project_id="my-project",
    service_id="my-service",
    query_type="range",
    metric_types=["cpu", "memory"],
    start_time=datetime.now(timezone.utc) - timedelta(hours=1),
)
print(metrics.data)

MetricType (a Literal of every valid metric) and METRIC_TYPES (all of them, as a tuple) are exported from northflank for annotation and reuse.

Async usage

AsyncApiClient mirrors ApiClient with awaitable calls:

import asyncio
from northflank import AsyncApiClient

async def main():
    client = AsyncApiClient(api_token="your-api-token")
    services = await client.list.services(project_id="my-project")
    print(services.data)

asyncio.run(main())

Pagination

Paginated list endpoints return one page by default. Use .all() to drain every page into a single response, or .pagination.get_next_page() to step through them:

# All pages at once
every_service = client.list.services.all(project_id="my-project")

# One page at a time
page = client.list.services(project_id="my-project")
while page.pagination and page.pagination.has_next_page:
    page = page.pagination.get_next_page()

Configuration

ApiClient reads configuration from arguments or environment variables:

Environment variable	Purpose
NF_API_TOKEN	API token (also NORTHFLANK_API_TOKEN)
NF_API_HOST / NF_HOST	API base URL (default https://api.northflank.com)

# Explicit
client = ApiClient(api_token="...", base_url="https://api.northflank.com")

# From the environment
client = ApiClient()

More examples

The examples/ directory has runnable scripts covering sandboxes, interactive exec sessions, log fetching and tailing, GPU workloads, networking, persistent volumes, and parallel sandbox pools. Each one reads NF_API_TOKEN and NF_PROJECT_ID from the environment.