graphn — Python SDK

The official Python SDK for Graphn. Author workflows, agents, functions, and knowledge bases, then run them — or import a custom model and chat through the OpenAI-compatible inference endpoint — from Python.

v0.2.x covers the full CLI-reachable Graphn API (control plane, inference, gateway, and storage). Python is the only first-party wrapped package; other languages generate from the public OpenAPI spec. See Scope below.

import graphn

with graphn.Client() as c:
    wf = c.workflows.create(name="qa", dsl="name: qa\nsteps: []")
    c.workflows.publish(wf.id)
    run = c.workflows.run(wf.id, input={"q": "..."})
    result = c.executions.wait(run.execution_id or "")
    hits = c.knowledgebases.search("kb_...", query="...")

Chat and TTS still go through the official openai package:

with graphn.Client() as c:
    resp = c.chat.completions.create(
        model="Qwen/Qwen3-0.6B",
        messages=[{"role": "user", "content": "Hello!"}],
    )
    print(resp.choices[0].message.content)

Install

pip install graphn

Requires Python 3.10+. Tested on 3.10, 3.11, 3.12, 3.13.

Authentication

The SDK reads credentials from the environment by default:

export GRAPHN_API_KEY=gn_...           # required
export GRAPHN_WORKSPACE_ID=ws_...      # required
export GRAPHN_BASE_URL=https://cp.graphn.ai          # optional
export GRAPHN_INFERENCE_URL=https://model.graphn.ai  # optional
export GRAPHN_GATEWAY_URL=https://gateway.graphn.ai  # optional
export GRAPHN_STORAGE_URL=https://storage.graphn.ai  # optional

Or pass them explicitly:

client = graphn.Client(api_key="gn_...", workspace_id="ws_...")

Get an API key from the Graphn dashboard.

Scope

What's in the box (v0.2.x)

Module	What it does
client.workflows	CRUD, publish, bundle, versions, run, test, dry-run. create/update auto-link DSL agents/functions/mcp_servers via save_bundle (same as graphn wf create)
client.agents	CRUD, publish, archive, dry-run, run
client.functions	CRUD, builtins, publish, test, dry-run
client.mcp_servers	CRUD, publish, start/stop/status, tools, refresh
client.executions	list/get; wait polls until terminal; UUID ids go to the gateway
client.triggers	Workspace-scoped cron/webhook trigger CRUD
client.knowledgebases	CRUD, documents, search, ingest; wait_ingest
client.imported_models	Full BYO CRUD plus discover/test on the inference host
client.organizations / client.workspaces / client.api_keys	Org, workspace, and API-key administration
client.blueprints	Public catalog list/get and workspace deploy
client.storages	REST object-store overlay plus S3-host get/put/delete
client.batch	Gateway batch submit, poll, items, JSONL output, cancel
client.custom_models	Import from HuggingFace / S3; wait_until_ready, validate; LoRA adapters on a shared base (.adapters)
client.secrets	CRUD for workspace secrets
client.chat.completions	OpenAI-compatible chat, streaming + non-streaming, with auto-wake on cold start
client.models	List every callable model: CP catalog plus imported/custom from inference
client.tts	Text-to-speech: list voices, synthesize

Both graphn.Client and graphn.AsyncClient exist with identical APIs.

Not wrapped (generate from the spec, or wait)

These exist on the platform but are not first-class SDK resources in v0.2.x — file an issue on the SDK repo to vote on what you need next:

• Evals & datasets
• Guardrails (policy authoring)
• Voice agents / conversations
• Usage & billing beyond client.custom_models.gpu_hours()

Those endpoints can still be hit via raw HTTP using your gn_... API key. The OpenAPI 3.1 spec is mirrored at voltagepark/graphn-openapi and rendered at graphn.ai/api. Other languages generate from that spec (npx @hey-api/openapi-ts, oapi-codegen, openapi-generator-cli -g java).

The 80% recipe: import a model and chat with it

import graphn

with graphn.Client() as c:
    # 1. Import the model. Use a workspace secret for gated HF repos.
    model = c.custom_models.create(
        name="my-llama",
        huggingface_model_id="meta-llama/Llama-3.1-8B-Instruct",
        weight_source="huggingface",
        hf_token_secret_id="sec_...",  # optional, only for gated models
    )

    # 2. Wait for the deployment to be live.
    c.custom_models.wait_until_ready(model.id, timeout=1800)

    # 3. Chat. The first call will cold-start the model — the SDK
    #    transparently calls wake() and retries until it serves.
    resp = c.chat.completions.create(
        model=model.id,
        messages=[{"role": "user", "content": "Tell me a joke."}],
        wake_timeout=600,             # max time to wait for cold start
    )
    print(resp.choices[0].message.content)

Importing from S3

If your weights aren't on HuggingFace — fine-tunes, internal models, licensed checkpoints — import them straight from S3. Two flavors, both of which still require huggingface_model_id (see callout below).

huggingface_model_id is required for S3 imports too. It's the canonical identifier for the model — the name the inference endpoint advertises and the value you pass in model for chat completions. Use the upstream org/model-name your weights are based on (e.g. Qwen/Qwen3-0.6B, meta-llama/Llama-3.1-8B-Instruct). This is the same "Model ID" field the web UI requires for S3 imports. Omitting it raises graphn.ValidationError client-side; passing it but having a mismatched archive will surface as a deploy failure on the model record.

Presigned URL (no AWS credentials shared with Graphn):

model = c.custom_models.create(
    name="my-finetune",
    weight_source="s3_presigned",
    huggingface_model_id="meta-llama/Llama-3.1-8B-Instruct",
    s3_url="https://my-bucket.s3.amazonaws.com/llama-3.1-8b.tar.gz?X-Amz-Algorithm=...",
    gpu_count=1,
)

Package the weights as a single .tar.gz archive whose top level is the model directory (the same layout huggingface-cli download produces). Generate the URL with aws s3 presign s3://my-bucket/llama-3.1-8b.tar.gz or the AWS SDK; Graphn pulls weights through the URL on import. The URL only needs to be live for the import window (allow at least a few minutes for the download), not for the model's lifetime.

IAM role assumption (for buckets you control, longer-lived credentials):

model = c.custom_models.create(
    name="my-finetune",
    weight_source="s3_assume_role",
    huggingface_model_id="meta-llama/Llama-3.1-8B-Instruct",
    s3_url="s3://my-bucket/llama-3.1-8b.tar.gz",
    s3_role_arn="arn:aws:iam::123456789012:role/GraphnImport",
    gpu_count=1,
)

The role's trust policy must allow Graphn's importer principal to sts:AssumeRole; ask support for the principal ARN to put in your trust policy. Graphn re-assumes on every import / refresh, so rotating credentials underneath is safe.

Everything past the create call — wait_until_ready, chat completions, auto-wake, addressing by model.id — is identical regardless of weight source. See examples/import_from_s3.py for an end-to-end runnable script.

LoRA adapters on a shared base

Deploy one base with lora_enabled=True, attach up to eight LoRA adapters to it, and call each adapter as its own model. Adapters run on the base's replicas, so they add no GPUs. LoRA adapters are in public beta; see the guide for the adapter format and limits.

base = c.custom_models.create(
    name="qwen3-8b-base",
    huggingface_model_id="Qwen/Qwen3-8B",
    lora_enabled=True,
    max_lora_rank=64,  # 16 or 64, fixed once created
)
c.custom_models.wait_until_ready(base.id)

adapter = c.custom_models.adapters.attach(
    base.id,
    huggingface_model_id="my-org/qwen3-8b-support-lora",
    name="support",
)
adapter = c.custom_models.adapters.wait_until_ready(adapter.id)

resp = c.chat.completions.create(
    model=adapter.id,
    messages=[{"role": "user", "content": "Hello"}],
)

Adapters come from Hugging Face or S3, like a base: pass s3_url="https://...presigned..." for a .tar.gz of the adapter folder, or s3_url="s3://bucket/prefix/" with s3_role_arn. Attaching takes a few minutes, and the base and its other adapters keep serving while it does. A failed adapter raises graphn.APIError whose message starts with the reason, such as ADAPTER_RANK_EXCEEDS_BASE.

c.custom_models.adapters.list(base.id)      # oldest first
c.custom_models.list(include_adapters=True)  # bases and adapters together

c.custom_models.adapters.delete(adapter.id)
c.custom_models.adapters.wait_until_deleted(adapter.id)
c.custom_models.delete(base.id)  # ConflictError (HAS_ADAPTERS) until every adapter is gone

See examples/lora_adapters.py for a runnable script.

Streaming

stream = c.chat.completions.create(
    model=model.id,
    messages=[{"role": "user", "content": "Count to ten."}],
    stream=True,
)
for chunk in stream:
    delta = chunk.choices[0].delta.content
    if delta:
        print(delta, end="", flush=True)

Async

import asyncio
import graphn

async def main() -> None:
    async with graphn.AsyncClient() as c:
        async for m in c.custom_models.list():
            print(m.id, m.name, m.status)

        resp = await c.chat.completions.create(
            model="cm_abc123",
            messages=[{"role": "user", "content": "Hi!"}],
        )
        print(resp.choices[0].message.content)

asyncio.run(main())

Cold starts and auto-wake

Graphn custom models default to scale-to-zero: a model with no traffic for cooldown_seconds is descheduled, and the first request afterwards has to wait for the gateway to spin up a fresh replica (typically 60–600 seconds depending on weight size).

Without help, the first chat request after a cold period returns:

503 Service Unavailable: Model is scaled to zero and is now warming up.

The SDK detects this, calls POST /custom-models/{id}/wake to nudge the autoscaler, and retries with exponential backoff until the model serves or wake_timeout (default 180s) elapses. You don't have to do anything, but the knobs are there if you want them:

# Disable auto-wake — you handle the 503 yourself.
c.chat.completions.create(model=..., messages=[...], auto_wake=False)

# Give the warm-up more headroom (e.g. for large models).
c.chat.completions.create(model=..., messages=[...], wake_timeout=900)

See docs/cold-starts.md for the full story.

Drop-in for openai

The chat path is OpenAI-compatible all the way down — under the hood we delegate to the official openai Python SDK, configured against the Graphn gateway. So tools, structured outputs, multi-modal inputs, function calling, etc. all work out of the box.

If you already have OpenAI-shaped code and just want to point it at a Graphn model:

from openai import OpenAI

client = OpenAI(
    api_key="gn_...",
    base_url="https://model.graphn.ai/v1",
    default_headers={"X-Workspace-Id": "ws_..."},
)
resp = client.chat.completions.create(
    model="custom:cm_...",  # raw openai client => you type the prefix
    messages=[{"role": "user", "content": "Hello!"}],
)

The reason to use graphn.Client instead is everything around the chat call: lifecycle management, secrets, auto-wake, bare-cm_ addressing without the wire prefix, typed responses, and a stable URL contract.

More examples

See examples/ for runnable end-to-end scripts:

• examples/import_and_chat.py — full lifecycle (HuggingFace)
• examples/import_from_s3.py — S3 presigned + assume-role import
• examples/streaming.py — streaming chat
• examples/async_client.py — async usage
• examples/openai_compat.py — drop-in from openai

Configuration reference

Argument	Default	Notes
api_key	$GRAPHN_API_KEY	Bearer token starting with gn_. Required.
workspace_id	$GRAPHN_WORKSPACE_ID	Path parameter + X-Workspace-Id header. Required.
base_url	https://cp.graphn.ai	Control plane host.
inference_url	https://model.graphn.ai	Inference / OpenAI-compatible host.
timeout	60.0	Per-request HTTPX timeout (seconds).
max_retries	2	Retries on connect failures, 429, and 5xx.
default_headers	{}	Extra headers added to every request.

Generating clients in other languages

The OpenAPI 3.1 spec is the source of truth. It's published at:

• GitHub — voltagepark/graphn-openapi
• Live HTML reference — graphn.ai/docs/api
• Direct download — https://cp.graphn.ai/openapi.yaml

Point your favorite generator at any of these. We test against openapi-generator 6.0+, openapi-python-client 0.21+, and oapi-codegen 2.0+.

Contributing

git clone https://github.com/voltagepark/graphn-sdk-python
cd graphn-sdk-python
python -m venv .venv && source .venv/bin/activate
pip install -e '.[dev]'

ruff check src tests
pytest

Regenerate the typed transport from the upstream spec after a spec change:

./scripts/regenerate.sh

See CHANGELOG.md for release notes.

License

Apache 2.0 — see LICENSE.