cheechandcharliesnortan8balloffadeadhookersassinthedesert/README.md

conduit-client

A typed Python client for the Conduit unified API gateway.

Features

• Synchronous and asynchronous clients (ConduitClient and AsyncConduitClient).
• Pydantic v2 request/response models.
• Covers every documented Conduit endpoint across gateway, SMS/MMS, media, torrents, wiki, compute, and DNS façades.
• Clean error handling with typed exceptions.

Installation

pip install conduit-client

Quick start

from conduit_client import ConduitClient

client = ConduitClient("mega_sk_live_<prefix>_<secret>")
me = client.gateway.me()
print(me.scopes)

Async usage

import asyncio
from conduit_client import AsyncConduitClient

async def main():
    async with AsyncConduitClient("mega_sk_live_<prefix>_<secret>") as client:
        me = await client.gateway.me()
        print(me.scopes)

asyncio.run(main())

Per-façade examples

Gateway

from conduit_client import ConduitClient
from conduit_client.models.common import CreateKeyRequest
from conduit_client.models.gateway import AuditLogQuery

client = ConduitClient("mega_sk_live_<prefix>_<secret>")

# API keys
keys = client.gateway.list_keys()
new_key = client.gateway.create_key(CreateKeyRequest(display_name="ci", scopes=["sms:send"]))
print(new_key.key)  # one-time secret

# Audit log
entries = client.gateway.admin_audit(AuditLogQuery(scope="sms:send", limit=50))

SMS / MMS

from conduit_client.models.sms import SendSmsRequest, SendMmsRequest

sms_id = client.sms.send_sms(SendSmsRequest(did="5550100", dst="5550200", message="hi"))
print(sms_id.id)

mms_id = client.sms.send_mms(
    SendMmsRequest(
        did="5550100",
        dst="5550200",
        message="pic",
        media1="https://example.com/photo.jpg",
    )
)

Media ingestion (MeTube)

from conduit_client.models.media_ingest import CreateIngestJobRequest

job = client.media_ingest.create_job(
    CreateIngestJobRequest(url="https://www.youtube.com/watch?v=abc", quality="best")
)
print(job.id, job.status)

Media library (Jellyfin)

from conduit_client.models.media_library import LibraryLoginRequest, LibrarySearchQuery

client.media_library.login(LibraryLoginRequest(username="user", password="pw"))
results = client.media_library.search(LibrarySearchQuery(term="Inception"))

Torrents (Transmission)

from conduit_client.models.torrents import AddTorrentRequest

client.torrents.add(AddTorrentRequest(url="magnet:?xt=urn:btih:...", paused=False))
client.torrents.list(fields=["id", "name"])

Wiki (DokuWiki)

from conduit_client.models.wiki import WriteWikiPageRequest

page = client.wiki.get_raw("namespace:page")
client.wiki.write("namespace:page", WriteWikiPageRequest(text="updated content"))

Compute (Proxmox)

from conduit_client.models.compute import CreateVmRequest, ComputeTicketRequest

ticket = client.compute.auth_ticket(ComputeTicketRequest(username="root", password="pw"))
print(ticket.ticket)

# Create a VM; extra Proxmox fields are forwarded as-is.
client.compute.create_vm(
    CreateVmRequest(node="pve", vmid=100, cores=2, memory=2048, storage="local-lvm")
)
client.compute.start_vm(100)

DNS (BIND9)

from conduit_client.models.dns import (
    AddZoneRequest,
    BatchRecordUpdate,
    BatchUpdateRecordsRequest,
)

client.dns.create_zone(AddZoneRequest(zone="example.com", klass="IN"))
client.dns.batch_update_records(
    "example.com",
    BatchUpdateRecordsRequest(
        updates=[
            BatchRecordUpdate(op="add", name="www", ttl=300, rtype="A", rdata="1.2.3.4")
        ]
    ),
)
client.dns.reload_server("default")

Error handling

from conduit_client import ConduitClient, AuthenticationError, NotFoundError

client = ConduitClient("mega_sk_live_<prefix>_<secret>")

try:
    client.sms.delete_sms(999999)
except NotFoundError as exc:
    print(exc.status_code, exc.response_body)
except AuthenticationError as exc:
    print("Invalid API key", exc.status_code)

Configuration

from conduit_client import ConduitClient

client = ConduitClient(
    "mega_sk_live_<prefix>_<secret>",
    base_url="https://api.cowtunnel.com",
    timeout=60.0,
    max_retries=2,
)

License

MIT