Python SDK

1
# Webhook apps (aiohttp, FastAPI, Flask, etc.)
2
from jambonz_sdk.webhook import WebhookResponse
3
4
# WebSocket apps
5
from jambonz_sdk.websocket import create_endpoint
6
7
# REST API client (mid-call control, outbound calls)
8
from jambonz_sdk.client import JambonzClient

1
from aiohttp import web
2
from jambonz_sdk.webhook import WebhookResponse
3
4
async def handle_incoming(request: web.Request) -> web.Response:
5
    jambonz = WebhookResponse()
6
    jambonz.say(text="Hello from jambonz!").gather(
7
        input=["speech", "digits"],
8
        actionHook="/handle-input",
9
        say={"text": "Press 1 for sales or 2 for support."},
10
    ).hangup()
11
    return web.json_response(jambonz.to_json())
12
13
async def handle_input(request: web.Request) -> web.Response:
14
    body = await request.json()
15
    speech = body.get("speech", {}).get("alternatives", [{}])[0].get("transcript", "")
16
    jambonz = WebhookResponse()
17
    jambonz.say(text=f"You said: {speech}").hangup()
18
    return web.json_response(jambonz.to_json())
19
20
app = web.Application()
21
app.router.add_post("/incoming", handle_incoming)
22
app.router.add_post("/handle-input", handle_input)
23
web.run_app(app, port=3000)

1
import asyncio
2
from jambonz_sdk.websocket import create_endpoint
3
4
async def main():
5
    make_service, runner = await create_endpoint(port=3000)
6
    svc = make_service(path="/")
7
8
    async def handle_session(session):
9
        async def on_gather_result(evt):
10
            transcript = (
11
                evt.get("speech", {})
12
                .get("alternatives", [{}])[0]
13
                .get("transcript", "")
14
            )
15
            session.say(text=f"You said: {transcript}").hangup()
16
            await session.reply()
17
18
        session.on("/gather-result", on_gather_result)
19
20
        session.say(text="Hello! Say something.").gather(
21
            input=["speech"],
22
            actionHook="/gather-result",
23
            timeout=10,
24
        ).hangup()
25
        await session.send()
26
27
    svc.on("session:new", handle_session)
28
    await asyncio.Future()
29
30
asyncio.run(main())

1
make_service, runner = await create_endpoint(
2
    port=3000,
3
    env_vars={
4
        "OPENAI_MODEL": {
5
            "type": "string",
6
            "description": "LLM model to use",
7
            "default": "gpt-4.1-mini",
8
        },
9
        "SYSTEM_PROMPT": {
10
            "type": "string",
11
            "description": "System prompt",
12
            "uiHint": "textarea",
13
            "default": "You are a helpful assistant.",
14
        },
15
    },
16
)
17
18
# Read values in session handler
19
async def handle_session(session):
20
    model = session.data.get("env_vars", {}).get("OPENAI_MODEL", "gpt-4.1-mini")
21
    # ...

1
svc = make_service(path="/")
2
audio_svc = make_service.audio(path="/audio-stream")
3
4
async def handle_session(session):
5
    session.say(text="Listening...").listen(
6
        url="/audio-stream",
7
        sampleRate=8000,
8
        bidirectionalAudio={"enabled": True, "streaming": True, "sampleRate": 8000},
9
    )
10
    await session.send()
11
12
svc.on("session:new", handle_session)
13
14
async def handle_audio(stream):
15
    async def on_audio(pcm):
16
        # Process audio — feed to STT, record, etc.
17
        pass
18
19
    stream.on("audio", on_audio)
20
21
    # Send audio back
22
    stream.send_audio(pcm_buffer)
23
24
audio_svc.on("connection", handle_audio)

1
from jambonz_sdk.client import JambonzClient
2
3
async with JambonzClient(
4
    base_url="https://api.jambonz.us",
5
    account_sid="your-account-sid",
6
    api_key="your-api-key",
7
) as client:
8
    # Create an outbound call
9
    call_sid = await client.calls.create({
10
        "from": "+15085551212",
11
        "to": {"type": "phone", "number": "+15085551213"},
12
        "call_hook": "/incoming",
13
    })
14
15
    # Mid-call control
16
    await client.calls.mute(call_sid, "mute")
17
    await client.calls.redirect(call_sid, "https://example.com/new-flow")

1
async def on_llm_tokens(evt):
2
    tokens = evt.get("tokens")
3
    done = evt.get("done")
4
5
    if tokens:
6
        await session.send_tts_tokens(tokens)
7
8
    if done:
9
        session.flush_tts_tokens()
10
11
session.on("/llm-tokens", on_llm_tokens)

1
async def on_tool_call(evt):
2
    tool_call_id = evt["tool_call_id"]
3
    name = evt["name"]
4
    args = evt["arguments"]
5
    result = await handle_tool(name, args)
6
    session.send_tool_output(tool_call_id, result)
7
8
session.on("/tool-call", on_tool_call)

1
# Change the system prompt
2
session.update_agent({
3
    "type": "update_instructions",
4
    "instructions": "You are now a billing agent.",
5
})
6
7
# Inject context
8
session.update_agent({
9
    "type": "inject_context",
10
    "messages": [{"role": "user", "content": "Customer: Sarah, Gold tier."}],
11
})
12
13
# Replace tools
14
session.update_agent({"type": "update_tools", "tools": [...]})
15
16
# Trigger a new response
17
session.update_agent({
18
    "type": "generate_reply",
19
    "interrupt": True,
20
    "user_input": "Tell the customer about the flash sale.",
21
})

1
session.update_llm({"instructions": "Switch to Spanish."})

1
# Mute/unmute
2
session.inject_mute("mute")
3
session.inject_mute("unmute")
4
5
# Whisper to one party
6
session.inject_whisper({"verb": "say", "text": "The customer is a VIP."}, agent_call_sid)
7
8
# Control noise isolation
9
session.inject_noise_isolation("enable", vendor="krisp", level=80)
10
session.inject_noise_isolation("disable")
11
12
# Control recording
13
session.inject_record("startCallRecording", siprec_server_url="sip:siprec@recorder.example.com")
14
session.inject_record("pauseCallRecording")
15
16
# Send DTMF
17
session.inject_dtmf("1234")
18
19
# Redirect call flow
20
session.inject_redirect("/new-webhook")
21
22
# Tag the call with metadata
23
session.inject_tag({"priority": "high", "department": "billing"})