> ## Documentation Index
> Fetch the complete documentation index at: https://docs.vatel.ai/llms.txt
> Use this file to discover all available pages before exploring further.

# Python SDK Example

> Run a full-duplex voice session from the command line with the Python SDK: microphone to agent, agent audio to speaker.

This example runs a **Session WebSocket** demo from the terminal: your microphone is streamed to Vatel, and the agent’s audio is played through your speaker. Transcripts and events are printed to the console. It uses the Vatel Python SDK with `numpy` and `sounddevice` for capture and playback.

<Card title="Session token" icon="key" href="/api-reference/session/session-token">
  REST endpoint used to obtain a short-lived JWT for the WebSocket connection.
</Card>

<Card title="Connection" icon="radio" href="/api-reference/session/connection">
  WebSocket channel, message types, and request/response flow.
</Card>

## Prerequisites

* **Python** 3.9+
* **Organization API key** and **agent ID** (agent UUID)
* **Dependencies**: Vatel Python SDK, `numpy`, `sounddevice` (for mic and speaker)

## Setup

<Steps>
  <Step title="Clone or copy the example">
    Use the example from the [vatel-nextjs-starters](https://github.com/Devpro-Software/vatel-nextjs-starters) repo, or copy `examples/python-cli/` (e.g. `run_session.py` and `requirements.txt`) into your project.
  </Step>

  <Step title="Install dependencies">
    From the directory that contains `requirements.txt`:

    ```bash theme={null}
    pip install -r requirements.txt
    ```

    This installs the Vatel SDK (from the repo), `numpy`, and `sounddevice`.
  </Step>

  <Step title="Run the session">
    Pass your API key and agent ID via arguments or environment variables:

    ```bash theme={null}
    python run_session.py --api-key YOUR_API_KEY --agent-id AGENT_UUID
    ```

    Or set `VATEL_API_KEY` and `VATEL_AGENT_ID`. Optional: `--base-url` or `VATEL_BASE_URL` to use a different API host.
  </Step>
</Steps>

## How it works

1. **Client and token** - A `Client` is created with the API key and base URL (default `https://api.vatel.ai`). The script gets a session token with `client.session.generate_token_async(agent_id)`.
2. **Connection** - `client.connect(token=token, url=base_url)` opens the WebSocket. A background thread captures microphone audio at 24 kHz mono 16-bit PCM, base64-encodes it, and feeds it into an async queue; a coroutine reads from the queue and sends chunks via `conn.send_input_audio(chunk)`.
3. **Playback** - A `sounddevice` output stream runs in a callback. When `response_audio` messages arrive, the base64 payload is decoded and pushed into a queue; the callback plays the PCM data.
4. **Events** - The script prints `session_started`, `response_text`, `input_audio_transcript`, and `session_ended` to the console. For `tool_call` it sends a simple `"ok"` reply so the agent can continue.

## Project structure

| Path               | Purpose                                                                                                       |
| ------------------ | ------------------------------------------------------------------------------------------------------------- |
| `run_session.py`   | CLI entrypoint: argparse, client, connection, mic thread, playback queue, event loop over `stream_messages()` |
| `requirements.txt` | Vatel SDK (git), `numpy`, `sounddevice`                                                                       |

## Code

<Tabs>
  <Tab title="run_session.py">
    Full-duplex session: mic → server, agent audio → speaker. Uses a thread for mic capture and a queue + sounddevice callback for playback.

    ```python theme={null}
    #!/usr/bin/env python3
    from __future__ import annotations

    import argparse
    import asyncio
    import base64
    import os
    import queue
    import sys
    import threading
    from pathlib import Path

    from vatel import Client
    from vatel.connection import Connection
    from vatel.models.ws import ToolCall

    SAMPLE_RATE = 24000
    BLOCKSIZE = 480

    try:
        import numpy as np
        import sounddevice as sd
    except ImportError:
        print("error: pip install numpy sounddevice", file=sys.stderr)
        sys.exit(1)


    def mic_thread(
        loop: asyncio.AbstractEventLoop,
        out_queue: asyncio.Queue[str | None],
        stop: threading.Event,
    ) -> None:
        with sd.InputStream(samplerate=SAMPLE_RATE, channels=1, dtype="int16", blocksize=BLOCKSIZE) as stream:
            while not stop.is_set():
                data, _ = stream.read(BLOCKSIZE)
                if data.size > 0:
                    loop.call_soon_threadsafe(out_queue.put_nowait, base64.b64encode(data.tobytes()).decode("ascii"))


    async def send_mic(conn: Connection, q: asyncio.Queue[str | None]) -> None:
        while True:
            chunk = await q.get()
            if chunk is None:
                break
            await conn.send_input_audio(chunk)


    def make_player():
        q: queue.Queue[bytes | None] = queue.Queue()
        buf = np.array([], dtype=np.int16)

        def callback(outdata: np.ndarray, frames: int, time: object, status: object) -> None:
            nonlocal buf
            need = frames * outdata.shape[1]
            if status:
                outdata.flat[:] = 0
                return
            while len(buf) < need:
                try:
                    chunk = q.get_nowait()
                    if chunk is None:
                        break
                    buf = np.concatenate([buf, np.frombuffer(chunk, dtype=np.int16)])
                except queue.Empty:
                    break
            n = min(need, len(buf))
            if n:
                outdata.flat[:n] = buf[:n]
                buf = buf[n:]
            outdata.flat[n:] = 0

        stream = sd.OutputStream(samplerate=SAMPLE_RATE, channels=1, dtype=np.int16, blocksize=BLOCKSIZE, callback=callback)
        stream.start()
        return q, stream


    async def run(api_key: str, agent_id: str, base_url: str | None) -> None:
        base_url = base_url or "https://api.vatel.ai"
        client = Client(api_key=api_key, base_url=base_url)
        audio_q, playback = make_player()
        try:
            token = (await client.session.generate_token_async(agent_id)).token
            conn = await client.connect(token=token, url=base_url)
            mic_stop = threading.Event()
            mic_q: asyncio.Queue[str | None] = asyncio.Queue()
            mic_t = threading.Thread(target=mic_thread, args=(asyncio.get_running_loop(), mic_q, mic_stop), daemon=True)
            mic_t.start()
            send_task = asyncio.create_task(send_mic(conn, mic_q))
            try:
                async for msg in conn.stream_messages():
                    t = getattr(msg, "type", None)
                    if t == "session_started":
                        print("Session started:", msg.data.id)
                    elif t == "response_text":
                        print("Agent:", msg.data.text)
                    elif t == "input_audio_transcript":
                        print("You:", msg.data.transcript)
                    elif t == "response_audio":
                        audio_q.put(base64.b64decode(msg.data.audio))
                    elif isinstance(msg, ToolCall):
                        print("Tool:", msg.data.toolName)
                        await conn.send_tool_call_output(msg.data.toolCallId, "ok")
                    elif t == "session_ended":
                        print("Session ended.")
                        break
            finally:
                mic_stop.set()
                mic_q.put_nowait(None)
                await send_task
                mic_t.join(timeout=2.0)
                await conn.close()
            audio_q.put(None)
            await asyncio.sleep(0.4)
            playback.stop()
            playback.close()
        finally:
            await client.aclose()


    def main() -> None:
        p = argparse.ArgumentParser(description="Run a call session (mic + speaker) using the Vatel SDK.")
        p.add_argument("--api-key", default=os.environ.get("VATEL_API_KEY"), help="API key")
        p.add_argument("--agent-id", default=os.environ.get("VATEL_AGENT_ID"), help="Agent UUID")
        p.add_argument("--base-url", default=os.environ.get("VATEL_BASE_URL"), help="Optional base URL")
        args = p.parse_args()
        if not args.api_key or not args.agent_id:
            print("error: set --api-key and --agent-id (or VATEL_API_KEY, VATEL_AGENT_ID)", file=sys.stderr)
            sys.exit(1)
        asyncio.run(run(args.api_key, args.agent_id, args.base_url))


    if __name__ == "__main__":
        main()
    ```
  </Tab>

  <Tab title="requirements.txt">
    Vatel SDK from the repo plus audio dependencies.

    ```
    numpy
    sounddevice
    vatel @ git+https://github.com/Devpro-Software/vatel-py-sdk.git
    ```
  </Tab>
</Tabs>