Skip to main content

Usage

Create the client​

from conduit import Conduit

conduit = Conduit(api_key="sk_...")
receipt = conduit.reports.create(
    source={"url": "https://storage.example.com/call.wav"},
    output={"template": "general_report"},
    target={"strategy": "dominant"},
    webhook={"url": "https://your-app.com/webhooks/conduit"},
    idempotency_key="report:user-42:call-1",
)

print(receipt.job_id, receipt.status, receipt.estimated_wait_sec)

reports.create(...) is receipt-first and asynchronous. It resolves the source, uploads when needed, creates the job, and returns immediately.

Source variants​

conduit.reports.create(
    source={"media_id": "med_123"},
    output={"template": "general_report"},
    target={"strategy": "dominant"},
    webhook={"url": "https://your-app.com/webhooks/conduit"},
)

conduit.reports.create(
    source={"path": "recordings/demo-call.wav", "label": "demo-call"},
    output={"template": "sales_playbook"},
    target={"strategy": "dominant"},
    webhook={"url": "https://your-app.com/webhooks/conduit"},
)
  • source accepts exactly one variant: media_id, file, url, or path.
  • source.url fetches remotely before upload.
  • source.path resolves from the current working directory.

Receipt helpers​

job = receipt.handle.job()
print(job.status, job.stage)

for event in receipt.handle.stream(timeout_ms=300_000):
    print(event.type, event.job.status, event.stage, event.progress)

report = receipt.handle.wait(timeout_ms=300_000)
print(report.output.markdown)
  • Prefer webhook delivery in production.
  • Use handle.stream() and handle.wait() for local scripts, tests, or controlled synchronous tooling.
  • handle.wait() raises typed errors like JobFailedError, JobCanceledError, and TimeoutError.

Upload once, then reuse media​

media = conduit.primitives.media.upload(path="recordings/demo-call.wav")

receipt = conduit.reports.create(
    source={"media_id": media.media_id},
    output={"template": "sales_playbook"},
    target={"strategy": "entity_id", "entity_id": "ent_123"},
    webhook={"url": "https://your-app.com/webhooks/conduit"},
)

Matching​

matching_receipt = conduit.matching.create(
    context="behavioral_compatibility",
    target={"entity_id": "ent_candidate"},
    group=[
        {"entity_id": "ent_manager"},
        {
            "media_id": "med_panel",
            "selector": {"strategy": "dominant"},
        },
    ],
    webhook={"url": "https://your-app.com/webhooks/conduit"},
)

print(matching_receipt.job_id)
  • context is currently the closed enum behavioral_compatibility.
  • target must be exactly one matching subject.
  • group must contain at least one matching subject.

Common completion flow​

report = conduit.reports.get("rpt_123")
print(report.output.template)
print(report.output.markdown)

Fetch the completed report_id or matching_id from a verified webhook event, then use reports.get(...) or matching.get(...) to read the finished resource.

Next steps​