Block a user
conduit-client (0.2.4)
Published 2026-07-08 17:06:44 -05:00 by chelsea
Installation
pip install --index-url About this package
Typed Python client for the Conduit / Mega API gateway
conduit-client
A typed Python client for the Conduit / Mega API unified API gateway.
Features
- Synchronous and asynchronous clients (
ConduitClient/ClientandAsyncConduitClient/AsyncClient). - Pydantic v2 request/response models.
- Covers documented Conduit endpoints across gateway, SMS/MMS, media, torrents, United cache, wiki, compute, and DNS façades.
- Clean error handling with typed exceptions.
Installation
pip install conduit-client
The megaapi import alias is also supported for the Mega API naming used in the live docs.
Quick start
from conduit_client import ConduitClient
client = ConduitClient("mega_sk_live_<prefix>_<secret>")
me = client.gateway.me()
print(me.scopes)
The Mega API import alias is also supported:
from megaapi import Client
client = Client.from_env() # MEGA_API_KEY or CONDUIT_API_KEY
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"])
United cache
from conduit_client.models.united import UnitedVelocityQuery
records = client.united.list_torrents()
velocity = client.united.velocity(
UnitedVelocityQuery(window_minutes=60, sort="snatches_per_hour")
)
torrent_bytes = client.united.download_torrent(records[0].record_hash)
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,
)
Documentation site
A generated static API reference lives in site/index.html. It lists every façade namespace in a sidebar, shows each method's endpoint and signature, and includes copyable sync and async examples.
To regenerate the site after changing the client code:
python scripts/generate_docs.py
Then open site/index.html in a browser, or serve it locally:
python -m http.server 8000 --directory site
The generator (scripts/generate_docs.py) parses the client source with the AST and introspects Pydantic models to build the examples automatically.
License
MIT
Requirements
Requires Python: >=3.10
Details
Assets (2)
Versions (6)
View all