vxsdk · Python SDK

Provision infrastructure, deploy applications, run AI agents, and drive the SalesShift GTM stack on the vxcloud platform — straight from Python.

vxsdk is the canonical Python distribution of the vxcloud SDK. Same wire contract, same auth model (API key with automatic refresh on 401), and the same resource layout as the Go, TypeScript, C++ and Java SDKs. The sync client is stdlib-only — zero third-party dependencies; the async client is one extra away.

Prefer the brand name? pip install vxcloud installs the identical surface under import vxcloud.

Installation · Quick start · SalesShift · Resource map · Async · Errors · Docs

Two flavors

Same class hierarchy, same method signatures, same auth model. Switching is essentially vxsdk.Client → vxsdk_async.AsyncClient plus async/await.

Why a parallel Python implementation?

Python and Go don't share runtimes. The Go SDK is the right answer for Go services and customers; this Python file is the right answer for Python customers. Both are wrappers over the same FastAPI HTTP surface, so the JSON wire is identical and they can be regenerated from a single OpenAPI spec when the platform team enables one.

For now, both are hand-written and kept in sync by review.

Installation

# PyPI (canonical name)
pip install vxsdk
pip install vxsdk[async]      # adds httpx for vxsdk_async

# PyPI (brand-name alias — same code, just `import vxcloud`)
pip install vxcloud
pip install vxcloud[async]

# Drop-in (no install, stdlib-only)
cp services/sdk/python/vxsdk.py /path/to/your/project/

Stdlib only for the sync flavor — no extra deps. Tested on Python 3.9+.

Entry-point styles

All four below resolve to the same class object — pick the name you and your team prefer. There is no behavior difference:

import vxsdk

c = vxsdk.Client.load_from_vxcli()      # canonical
c = vxsdk.VxCloud.load_from_vxcli()     # PascalCase brand (matches TS SDK)
c = vxsdk.vxcloud.load_from_vxcli()     # lowercase brand

# Or via the alias package:
import vxcloud
c = vxcloud.Client.load_from_vxcli()
c = vxcloud.load_from_vxcli()            # module-level convenience

Quick start

import vxsdk

# Reads ~/.vxcloud/credentials.json (the file `vxcli auth login` writes)
c = vxsdk.Client.load_from_vxcli()

# Or with explicit credentials
# c = vxsdk.Client(api_key="xc_dev_...", username="alice")

# Provision a VM — `c.cloud.vm.provision(...)` mirrors the TypeScript SDK.
# (The legacy flat `c.cloud.create_vm(...)` also still works.)
vm = c.cloud.vm.provision(
    name="api-vm", cloud="aws", region="us-east-1",
    instance_type="t3.small", key_pair_name="AWSPRODKEY2",
)
print(vm["public_ip"])

# Deploy FastAPI onto the new VM
sess = c.deploy.fastapi(
    path="./", entry="app.app:app",
    requirements="requirements.txt",
    app_port=8000, http_port=80, app_name="studio-backend",
    host=vm["public_ip"], ssh_user="ubuntu",
    key_pair_name="AWSPRODKEY1.PEM",
)
print(sess["session_id"])

# Deploy a Docker container onto a VM
result = c.deploy.container(
    host="54.197.71.181", ssh_user="ubuntu", key_pair_name="AWSPRODKEY1.PEM",
    image="grafana/grafana:latest",
    name="grafana", ports=["3000:3000"],
    restart_policy="unless-stopped",
)
print(result["session_id"], result.get("status"))

# Deploy a language stack from a public git repo
result = c.deploy.stack(
    "golang",
    host="54.197.71.181", ssh_user="ubuntu", key_pair_name="AWSPRODKEY1.PEM",
    repo_url="https://github.com/joelwembo/va-sample-golang.git", branch="main",
    git_provider="github", app_name="va-sample-golang",
    http_port="80", app_port="8080", go_version="1.22",
)

# Run a custom shell script over SSH
result = c.install.script(
    host="54.197.71.181", ssh_user="ubuntu", key_pair_name="AWSPRODKEY1.PEM",
    script="#!/bin/bash\necho hello\n",
    script_name="hello.sh",
)

# CI/CD
for p in c.cicd.pipelines.list():
    print(p["id"], p["name"], p.get("repository_url"))

build = c.cicd.pipelines.trigger(pipeline_id="abc...", branch="main")

# Marketplace
agents = c.marketplace.agents.list()
result = c.marketplace.agents.deploy(
    "golang_url_status_agent",
    host="54.197.71.181", ssh_user="ubuntu", key_pair_name="AWSPRODKEY1.PEM",
    http_port="8094",
)

# Cloud provisioning (real AWS resources)
result = c.cloud.create_s3_bucket("my-bucket-001", region="us-east-1")
result = c.cloud.create_iam_policy("my-policy-001",
                                    policy_document={"Version": "2012-10-17", "Statement": [...]})

SalesShift — leads, CRM, campaigns and signals

SalesShift is the go-to-market layer of the platform, and all of it hangs off c.salesshift. It covers the global prospect pool, the CRM that pool feeds, tracked email and campaigns, the cross-tenant opportunity signal pool, tasks, social distribution, and the workspace's own billing.

The prospect pool

Search returns masked addresses (j•••@acme.com). A mask is not an address — revealing one spends quota, so price the batch before you spend it.

import vxsdk

c = vxsdk.Client.load_from_vxcli()
ss = c.salesshift

# Search the global pool. Paging is keyset-based; the cursor is opaque —
# pass back exactly what the server gave you.
page = ss.search_leads(
    filters={"seniority": ["c_level", "vp"], "employee_range": ["51-200"],
             "country": ["AU"], "department": ["engineering"]},
    limit=50,
)
for p in page["results"]:
    print(p["full_name"], p["title"], p["email"])   # email is MASKED here

# Walk every page without hand-rolling the cursor loop
for person in ss.search_all_leads(filters={"country": ["AU"]}):
    ...

# Where does this org stand on reveal quota?
q = ss.reveal_quota()
print(q["remaining"], "of", q["allowance"], "unlimited:", q["unlimited"])

# Price a batch BEFORE spending anything on it
ids = [p["pool_person_id"] for p in page["results"][:10]]
est = ss.preview_reveal_cost(ids)
print("would cost", est["cost"], "already revealed:", est["already_revealed"])

# Reveal one. Nothing is charged if this raises VxLeadQuotaExhaustedError.
try:
    person = ss.reveal_lead(ids[0])
    print(person["email"])                          # real address
except vxsdk.VxLeadQuotaExhaustedError:
    print("allowance spent — you were NOT charged for this attempt")
except vxsdk.VxLeadErasedError:
    print("erased at the person's request — terminal, never retry")

Pool → lead → contact

A pool row is not mailable. It becomes a lead when saved, and a mailable CRM contact only when converted.

ss.save_leads(ids)                       # pool → saved leads
leads = ss.list_leads(status="new", limit=100)

report = ss.convert_from_pool(ids, lifecycle_stage="lead")

# A convert splits into buckets. Reporting a partial success as success is
# how duplicate contacts get created — render every bucket.
print(vxsdk.describe_convert(report))

ss.request_erasure(email="someone@example.com")     # GDPR, global + terminal
ss.enrich_company(domain="acme.com")                # crawl + fill in firmographics

Tracked email and campaigns

ss.send_email(
    to_email="ada@acme.com",
    subject="Quick question about your deploy pipeline",
    body_html="<p>Hi Ada — noticed you are hiring platform engineers…</p>",
)

for e in ss.list_emails(status="delivered"):
    print(e["to_email"], e["opened_at"], e["clicked_at"])

print(ss.get_stats())                    # sends, opens, clicks, bounces
print(ss.get_worker_health())            # the Go email worker behind it

cid = ss.create_campaign("AU founders Q3", "Subject line", "<p>Body</p>")["id"]
ss.send_campaign(cid)
report = ss.wait_for_campaign(cid, timeout=120)

Opportunities, tasks and social

# Opportunities are a cross-tenant signal pool — save/dismiss are per-org
# side-table state, never a mutation of the shared signal.
opps = ss.list_opportunities(source="hn", min_score=70, saved_only=False)
ss.save_opportunity(opps["results"][0]["id"])
ss.push_opportunity_to_lead(opps["results"][0]["id"])

ss.create_task("Follow up with Ada", goal="Book a 20-min call", priority="high")
for t in ss.list_tasks(status="open")["results"]:
    print(t["title"], t["progress"], t["assignee_name"])

post = ss.create_social_post("Shipping vxcli 2026.8.13 today.", title="Release")
job = ss.distribute_post(post["id"])

# Fan-out is one goroutine per network. `speedup` is measured, not claimed.
print(job["job"]["speedup"], "x faster than sequential")

# A deployment without social API credentials still returns delivery records.
# Surface `simulated` — reporting a simulated post as published is the one
# unforgivable lie this SDK could tell.
for d in job["job"]["deliveries"]:
    print(d["channel"], "SIMULATED" if d["simulated"] else "published")

Billing (what the workspace pays for SalesShift)

plans = ss.billing_plans()
sub   = ss.billing_subscription()

# Quota fields are None when UNLIMITED. A plain 0 would read as "no
# allowance" — the exact opposite of what the API means.
for code, limit in sub["plan"]["quotas"].items():
    print(code, "unlimited" if limit is None else limit)

print(ss.billing_checkout("growth", seats=5)["url"])
for inv in ss.billing_invoices()["invoices"]:
    print(inv["number"], inv["amount_due"], inv["status"])

The same surface from vxcli

Every call above has a CLI equivalent. All honour --output json|yaml, and spending or destructive commands confirm first and take --yes.

vxcli salesshift leads search --seniority c_level --country AU --limit 25
vxcli salesshift leads quota
vxcli salesshift leads reveal <pool-id>
vxcli salesshift leads save <pool-id>…
vxcli salesshift leads convert-from-pool <pool-id>… --lifecycle-stage lead
vxcli salesshift leads enrich acme.com

vxcli salesshift email send --to ada@acme.com --subject "…" --html "<p>…</p>"
vxcli salesshift email stats
vxcli salesshift campaigns list
vxcli salesshift campaigns report <campaign-id>

vxcli salesshift contacts list
vxcli salesshift workflows test-run <id>
vxcli salesshift sequences list

vxcli salesshift opportunities list --source hn --min-score 70
vxcli salesshift opportunities push-to-lead <id>
vxcli salesshift tasks add --title "Follow up" --goal "Book a call"
vxcli salesshift social post --content "…" && vxcli salesshift social send <post-id>
vxcli salesshift webmaster inspect https://example.com

vxcli salesshift billing plans
vxcli salesshift billing invoices
vxcli salesshift billing cancel --yes

Resource map

Async parity: vxsdk_async.AsyncClient exposes the same modules, including c.vxcomputer, c.workflow, c.vxchrono, and the full c.salesshift surface.

Errors

try:
    c.cicd.pipelines.list()
except vxsdk.VxAuthError as e:        # 401/403
    ...
except vxsdk.VxValidationError as e:  # 400/422
    ...
except vxsdk.VxRateLimitError as e:   # 429 — e.retry_after
    ...
except vxsdk.VxNotFoundError as e:    # 404
    ...
except vxsdk.VxServerError as e:      # 5xx
    ...
except vxsdk.VxNetworkError as e:     # transport
    ...
except vxsdk.VxError as e:            # base, anything else
    ...

The SDK retries VxNetworkError, VxServerError, and VxRateLimitError up to 3 times with exponential backoff. Auth errors and validation errors are surfaced immediately — retrying them as-is would not succeed.

On 401, the SDK calls POST /api/v1/auth/developer/keys/login once with the configured API key, replays the original request, and only surfaces the error if the refresh itself fails. Application code should not see token expiration.

Run the sync deploy program

cd services/sdk/python
python3 deploy_app.py                                  # whoami container on inst3:8085
python3 deploy_app.py --image redis:7 --name r --ports 6380:6379
python3 deploy_app.py --mode stack --kind golang \
    --repo-url https://github.com/joelwembo/va-sample-golang.git

Default: deploys traefik/whoami:latest to inst3:8085 and polls until it returns HTTP 200. See python3 deploy_app.py --help for flags.

Run the async demo

pip install httpx
python3 deploy_async.py

Drops three redis containers onto a single host in parallel via asyncio.gather(). Verified live: 3 deploys in 22.8s wall clock vs. ~57s sequential — a 2.5× speedup, and a 3× win at higher fan-out.

import asyncio, vxsdk_async as vx

async def main():
    async with await vx.AsyncClient.load_from_vxcli() as c:
        results = await asyncio.gather(
            c.deploy.container(host=h1, ssh_user="ubuntu", key_pair_name=K, image="redis:7", ports=["6381:6379"], name="r1"),
            c.deploy.container(host=h2, ssh_user="ubuntu", key_pair_name=K, image="redis:7", ports=["6381:6379"], name="r2"),
            c.deploy.container(host=h3, ssh_user="ubuntu", key_pair_name=K, image="redis:7", ports=["6381:6379"], name="r3"),
        )

asyncio.run(main())

SDKs for every stack

Same JSON wire contract, same auth model, same error taxonomy in every language.

Links

• 📦 PyPI: pypi.org/project/vxcloud · pypi.org/project/vxsdk
• 📖 Documentation: vxcloud.io/docs/sdks
• 🛠️ Source & issues: github.com/prodxcloud/vxcloud
• 📝 Changelog: CHANGELOG.md

Author

Built and maintained by Joel O. Wembo — linkedin.com/in/joelwembo

License

Apache-2.0 © vxcloud / ProdXCloud