Build Form Filling Session Streaming Client

Tutorial: Build an end-to-end Form filling session client from login to structured form data

What you will build

20 min | Intermediate

• Login and create a Form filling session.
• Add form template context before streaming.
• Stream raw PCM audio over WebSocket.
• End the session and retrieve structured form data.
• Handle common API errors.

Using an AI coding tool?

Copy the prompt below to point your agent at the audio streaming skill and Documentation MCP. For every task skill, refer to AI coding tools.

Fetch the audio streaming skill and connect the documentation MCP.

CopiedCopy promptOpen in Cursor

In this tutorial, you build an end-to-end Form filling streaming client. You authenticate, create a Form filling session, add form template context, stream PCM audio over a WebSocket connection, end the session, and retrieve structured form data. Examples are provided in Python, TypeScript, and Go coding languages.

By the end of this tutorial, you’ll have a working client that connects to the production endpoints https://sdp.suki.ai and wss://sdp.suki.ai/ws/stream API, and handles common API errors.

Form filling uses the same /ws/stream wire format as ambient clinical notes. The response field ambient_session_id from Create Form filling session is your Form filling session ID. Do not use an ID from Create ambient session.

Prerequisites

Before you begin, make sure you have:

• Completed Partner onboarding and received your partner credentials.
• You have partner_id and partner_token.
• Optional provider_id or sdp_provider_id if your integration uses Bearer authentication or the Single Auth Token flow.
• At least one form_template_id from Suki Medical form templates (GET /api/v1/info/suki-medical-form-templates, field template_id). This tutorial seeds a Suki template. You can also send your own form templates. Refer to Dynamic Form filling.
• An audio file in LINEAR16 PCM format (16 kHz, mono, 16-bit little-endian). If you’re using a WAV file, remove the WAV header before streaming and send only the raw PCM audio.

Store your partner_token only on your backend. Never expose it in a client-side or public application.

Architecture

Project setup

1. Create a script or web page in the language used by this tutorial (Python, Go, or TypeScript).

2. Install the required dependencies:.

   • Python: pip install requests websocket-client.

   • Go: go get github.com/gorilla/websocket.

   • TypeScript (browser): No additional packages are required. The example uses the built-in fetch and WebSocket APIs.

3. Configure your credentials:.

   • Add your partner_id.
   • Store partner_token on your backend only. Never expose it in browser code.
   • Add provider_id or sdp_provider_id only if your integration requires them.
   • Set at least one real form in context before you stream. This tutorial uses a Suki form_template_id.

4. Prepare an audio file in LINEAR16 PCM format or a WAV file. If you use a WAV file, remove the 44-byte WAV header before streaming the PCM audio.

How WebSocket authentication works in this tutorial

Language	How to authenticate
Python or Go	Send sdp_suki_token and ambient_session_id as HTTP headers (sdp_provider_id when required)
TypeScript (browser)	Send Sec-WebSocket-Protocol: SukiAmbientAuth,<sdp_suki_token>,<ambient_session_id>

Important points to remember:

• If login fails because the provider is new, Register provider once, then log in again.
• Send audio at least every 25 seconds, or the server closes the stream. While paused, send KEEP_ALIVE every five seconds.
• Send raw LINEAR16 PCM. If you start from a WAV file, strip the 44-byte header first. Do not send binary WebSocket frames on this endpoint.
• Browser protocol order matches ambient: token, then session ID (SukiAmbientAuth,<sdp_suki_token>,<ambient_session_id>).
• Use the Form filling create response ambient_session_id as the Form filling session ID on the WebSocket.

Tutorial steps

Work through each step in order. Read the explanation, review the code example, then continue to the next step. The full code example is available in the accordion below.

1. Log In

   Call the Login API with your partner credentials to receive a suki_token. Send that token later as sdp_suki_token. If the provider is new, register once, then log in again.

   Check: Login returns HTTP 200 and a non-empty suki_token.

   Python

   def login(partner_id: str, partner_token: str, provider_id: Optional[str] = None) -> str:
   url = f"{BASE_URL}/api/v1/auth/login"
   payload = {"partner_id": partner_id, "partner_token": partner_token}
   if provider_id:
       payload["provider_id"] = provider_id
   r = requests.post(url, json=payload, timeout=60)
   expect_status(r, url, 200)
   return r.json()["suki_token"]
   
   token = login_with_register_fallback(
   PARTNER_ID, PARTNER_TOKEN, PROVIDER_ID, "Dr Example", "org-1"  # Example provider name / org only
   )
   
2. Create a Form Filling Session

   Create a session with the Form filling session API. The response field ambient_session_id is your Form filling session ID.

   Check: Create returns a non-empty ambient_session_id for this Form filling session.

   If it fails: Use the ambient_session_id from the Form filling create response as this session ID. Do not reuse an ambient clinical-notes session ID.

   Python

   def create_form_filling_session(suki_token: str, body: Optional[dict[str, str]] = None) -> str:
   url = f"{BASE_URL}/api/v1/form-filling/session/create"
   r = requests.post(url, headers=rest_headers(suki_token), json=body or {}, timeout=60)
   expect_status(r, url, 201)
   return r.json()["ambient_session_id"]
   
   ambient_session_id = create_form_filling_session(token)
   
3. Seed Form Filling Context

   Seed at least one form with the Form filling Context API before you end the session. This tutorial uses a Suki form_template_id. For partner-defined forms, send a schema instead. Refer to Dynamic Form filling.

   Check: At least one form is seeded (form_template_id or schema) before you end the session.

   If it fails: Seed at least one form with a form_template_id or schema before you call End.

   Python

   def seed_form_filling_context(suki_token: str, ambient_session_id: str, form_template_id: str) -> None:
   url = f"{BASE_URL}/api/v1/form-filling/session/{ambient_session_id}/context"
   payload = {
       "form_filling": {
           "values": [
               {"form_template_id": form_template_id},
           ]
       }
   }
   r = requests.post(url, headers=rest_headers(suki_token), json=payload, timeout=60)
   expect_status(r, url, 200)
   
   seed_form_filling_context(token, ambient_session_id, FORM_TEMPLATE_ID)
   
4. Open the WebSocket

   Open wss://sdp.suki.ai/ws/stream with the same ambient handshake. Authenticate with headers (Python/Go) or Sec-WebSocket-Protocol (browser).

   Check: The WebSocket opens on wss://sdp.suki.ai/ws/stream with the auth pattern for your language.

   Python

   ws_headers = [
   f"sdp_suki_token: {suki_token}",
   f"ambient_session_id: {ambient_session_id}",
   ]
   if SDP_PROVIDER_ID:
   ws_headers.append(f"sdp_provider_id: {SDP_PROVIDER_ID}")
   ws = websocket.create_connection(WS_URL, header=ws_headers, timeout=60)
   
5. Stream Audio

   Send START_TIME (Base64 of an RFC 3339 timestamp), then JSON AUDIO frames with Base64 LINEAR16 PCM (~100 ms chunks). End the segment with {"type":"AUDIO","data":"RU9G"}. Same wire format as ambient. Do not send binary WebSocket frames.

   Check: You sent START_TIME, then Base64 LINEAR16 PCM in JSON AUDIO frames, then RU9G. No binary WebSocket frames.

   Python

   rfc3339 = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
   send_json(ws, {"type": "START_TIME", "data": b64(rfc3339.encode("utf-8"))})
   
   # Send Base64 LINEAR16 PCM in JSON text frames. Do not send binary WebSocket frames.
   pcm = strip_wav_header(open(pcm_path, "rb").read())
   for i in range(0, len(pcm), CHUNK_BYTES):
   send_json(ws, {"type": "AUDIO", "data": b64(pcm[i : i + CHUNK_BYTES])})
   
   send_json(ws, {"type": "AUDIO", "data": "RU9G"})  # end marker (Base64 for EOF)
   
6. Optional Control Events

   While streaming, you can send PAUSE, RESUME, or KEEP_ALIVE (every five seconds while paused).

   Check: While paused, you send KEEP_ALIVE every five seconds.

   If it fails: Send KEEP_ALIVE every five seconds while you are paused, and send audio at least every 25 seconds while you are streaming.

   Python

   send_event(ws, "PAUSE")
   # While paused, send KEEP_ALIVE at least every five seconds.
   send_event(ws, "KEEP_ALIVE")
   send_event(ws, "RESUME")
   
7. End the Session and Retrieve Results

   Close the WebSocket, call End, poll status, then get structured data.

   Check: You called End, polled status, then retrieved structured data.

   Python

   end_form_filling_session(token, ambient_session_id)
   results = wait_for_results(token, ambient_session_id)
   print(results)
   
8. Handle API Errors

   Map REST failures using documented code, message, and details. See API error messages.

   Check: Your client maps REST failures using code, message, and details.

   If it fails: Call login again for a fresh suki_token, then see API errors.

   Python

   def hint_for_error(err: SukiApiError) -> str:
   if err.matches("invalid_sdp_token"):
       return "Missing, expired, or invalid sdp_suki_token. Call login again."
   if err.matches("provider_not_registered"):
       return "Provider is not registered. Call register once, then login again."
   return "See /form-filling-api-reference/error-messages-form-filling"
   

Full code example

Python

# Form filling streaming tutorial (Python)
# Flow: login → create Form filling session → seed template context → stream PCM over WebSocket → end → poll structured data
# Install: pip install requests websocket-client
#
# ambient_session_id from Form filling create is the Form filling session ID (same field name as Ambient; different ID).

import base64
import json
import time
from datetime import datetime, timezone
from typing import Any, Optional

import requests
import websocket

# ---------------------------------------------------------------------------
# Endpoints and streaming defaults
# ---------------------------------------------------------------------------
BASE_URL = "https://sdp.suki.ai"          # REST API host
WS_URL = "wss://sdp.suki.ai/ws/stream"    # Shared Ambient / Form filling WebSocket stream
CHUNK_BYTES = 3200                        # ~100 ms of 16 kHz mono 16-bit PCM
WAV_HEADER_BYTES = 44                     # Standard WAV header size to strip
KEEP_ALIVE_SEC = 5                        # Send KEEP_ALIVE this often while paused

# ---------------------------------------------------------------------------
# Replace these with your partner credentials
# ---------------------------------------------------------------------------
PARTNER_ID = "your-partner-id"  # Example only — replace with your partner ID
PARTNER_TOKEN = "eyJhbGciOiJSUzI1NiIsInR5cCI6IkpXVCJ9..."  # Example only — replace; keep on backend only
PROVIDER_ID = "provider-123"  # Example only — optional; required for Bearer / Single Auth Token partners
SDP_PROVIDER_ID = ""  # Optional. Leave empty unless your partnership requires sdp_provider_id

# This tutorial seeds a Suki template_id from GET /api/v1/info/suki-medical-form-templates.
# Replace this example UUID with a real template from your account.
FORM_TEMPLATE_ID = "YOUR_FORM_TEMPLATE_ID"  # Example placeholder — replace with template_id from GET /api/v1/info/suki-medical-form-templates

# ---------------------------------------------------------------------------
# Error helpers: map REST failures to the documented code / message / details shape
# ---------------------------------------------------------------------------
class SukiApiError(RuntimeError):
    """REST failure with the documented {code, message, details} body when present."""

    def __init__(
        self,
        http_status: int,
        url: str,
        *,
        code: Any = None,
        message: str = "",
        details: Any = None,
        raw_body: str = "",
    ) -> None:
        self.http_status = http_status
        self.url = url
        self.code = code
        self.message = message or ""
        self.details = details if details is not None else []
        self.raw_body = raw_body
        super().__init__(
            f"HTTP {http_status} {url}: code={code!r} message={self.message!r}"
        )

    def matches(self, error_id: str) -> bool:
        # Docs: match error id exactly or as a prefix (for example invalid_sdp_token: …).
        m = self.message.strip()
        return m == error_id or m.startswith(error_id + ":") or m.startswith(error_id + " ")

def parse_suki_error_body(raw: str) -> tuple[Any, str, Any]:
    try:
        data = json.loads(raw) if raw else {}
    except json.JSONDecodeError:
        return None, (raw or "")[:500], []
    if not isinstance(data, dict):
        return None, (raw or "")[:500], []
    message = data.get("message")
    return data.get("code"), str(message) if message is not None else "", data.get("details") or []

def raise_suki_api_error(response: requests.Response, url: str) -> None:
    raw = response.text or ""
    code, message, details = parse_suki_error_body(raw)
    raise SukiApiError(
        response.status_code,
        url,
        code=code,
        message=message,
        details=details,
        raw_body=raw,
    )

def expect_status(response: requests.Response, url: str, want: int) -> None:
    if response.status_code != want:
        raise_suki_api_error(response, url)

def hint_for_error(err: SukiApiError) -> str:
    if err.matches("invalid_sdp_token") or "invalid sdp token" in err.message.lower():
        return "Missing, expired, or invalid sdp_suki_token. Call login again."
    if err.matches("provider_not_registered"):
        return "Provider is not registered. Call register once, then login again."
    if err.http_status == 401:
        return "Unauthorized. Check partner credentials and sdp_suki_token."
    if err.http_status >= 500:
        return "Server error. Retry with backoff; see API error messages."
    return "See /form-filling-api-reference/error-messages-form-filling"

def b64(data: bytes) -> str:
    return base64.b64encode(data).decode("ascii")

def rest_headers(suki_token: str) -> dict[str, str]:
    headers = {
        "sdp_suki_token": suki_token,
        "Content-Type": "application/json",
    }
    # Include only when your partnership requires it (leave SDP_PROVIDER_ID empty otherwise).
    if SDP_PROVIDER_ID:
        headers["sdp_provider_id"] = SDP_PROVIDER_ID
    return headers

# ---------------------------------------------------------------------------
# Authentication (same as Ambient)
# ---------------------------------------------------------------------------
def login(partner_id: str, partner_token: str, provider_id: Optional[str] = None) -> str:
    """Exchange partner credentials for a Suki token (send later as sdp_suki_token)."""
    url = f"{BASE_URL}/api/v1/auth/login"
    payload: dict[str, Any] = {
        "partner_id": partner_id,
        "partner_token": partner_token,
    }
    if provider_id:
        payload["provider_id"] = provider_id
    r = requests.post(url, json=payload, timeout=60)
    expect_status(r, url, 200)
    token = r.json().get("suki_token")
    if not token:
        raise SukiApiError(200, url, message="login response missing suki_token")
    return token

def register_provider(
    partner_id: str,
    partner_token: str,
    provider_name: str,
    provider_org_id: str,
    provider_id: Optional[str] = None,
    provider_specialty: str = "CARDIOLOGY",
) -> None:
    """POST /api/v1/auth/register (one-time)."""
    url = f"{BASE_URL}/api/v1/auth/register"
    payload: dict[str, Any] = {
        "partner_id": partner_id,
        "partner_token": partner_token,
        "provider_name": provider_name,
        "provider_org_id": provider_org_id,
        "provider_specialty": provider_specialty,
    }
    if provider_id:
        payload["provider_id"] = provider_id
    r = requests.post(url, json=payload, timeout=60)
    if r.status_code not in (201, 409):
        raise_suki_api_error(r, url)

def login_with_register_fallback(
    partner_id: str,
    partner_token: str,
    provider_id: Optional[str],
    provider_name: str,
    provider_org_id: str,
) -> str:
    try:
        return login(partner_id, partner_token, provider_id)
    except SukiApiError as err:
        # Only register when the API indicates an unregistered provider.
        if err.matches("provider_not_registered"):
            print(hint_for_error(err))
            register_provider(
                partner_id,
                partner_token,
                provider_name=provider_name,
                provider_org_id=provider_org_id,
                provider_id=provider_id,
            )
            return login(partner_id, partner_token, provider_id)
        print(hint_for_error(err))
        raise

# ---------------------------------------------------------------------------
# Form filling session setup
# ---------------------------------------------------------------------------
def create_form_filling_session(suki_token: str, body: Optional[dict[str, str]] = None) -> str:
    """
    Create a Form filling session (POST /api/v1/form-filling/session/create, HTTP 201).
    Returns ambient_session_id: this is the Form filling session ID despite the field name.
    Never pass an Ambient clinical-note create ID here.
    """
    url = f"{BASE_URL}/api/v1/form-filling/session/create"
    r = requests.post(url, headers=rest_headers(suki_token), json=body or {}, timeout=60)
    expect_status(r, url, 201)
    sid = r.json().get("ambient_session_id")
    if not sid:
        raise SukiApiError(201, url, message="create response missing ambient_session_id")
    return sid

def seed_form_filling_context(suki_token: str, ambient_session_id: str, form_template_id: str) -> None:
    """
    Seed Form filling context before end.
    This tutorial uses a Suki template_id from GET /api/v1/info/suki-medical-form-templates.
    Replace the example UUID with a real template from your account.
    """
    url = f"{BASE_URL}/api/v1/form-filling/session/{ambient_session_id}/context"
    payload = {
        "form_filling": {
            "values": [
                {"form_template_id": form_template_id},
            ]
        }
    }
    r = requests.post(url, headers=rest_headers(suki_token), json=payload, timeout=60)
    expect_status(r, url, 200)

def strip_wav_header(raw: bytes) -> bytes:
    """If the file is WAV, drop the 44-byte header and keep only raw PCM bytes."""
    if len(raw) >= 12 and raw[:4] == b"RIFF" and raw[8:12] == b"WAVE":
        return raw[WAV_HEADER_BYTES:]
    return raw

def send_json(ws: websocket.WebSocket, obj: dict[str, Any]) -> None:
    ws.send(json.dumps(obj))

def send_event(ws: websocket.WebSocket, event: str) -> None:
    send_json(ws, {"type": "EVENT", "event": event})

# ---------------------------------------------------------------------------
# WebSocket streaming (same wire format as Ambient; auth headers, not Sec-WebSocket-Protocol)
# ---------------------------------------------------------------------------
def stream_pcm_file(
    suki_token: str,
    ambient_session_id: str,
    pcm_or_wav_path: str,
) -> None:
    """
    Open /ws/stream, send START_TIME, stream LINEAR16 PCM chunks, then end with RU9G.
    Auth uses HTTP headers: sdp_suki_token + ambient_session_id (+ sdp_provider_id when required).
    ambient_session_id must be the Form filling create ID.
    """
    try:
        ws = websocket.create_connection(
            WS_URL,
            header=(
                [
                    f"sdp_suki_token: {suki_token}",
                    f"ambient_session_id: {ambient_session_id}",
                ]
                + ([f"sdp_provider_id: {SDP_PROVIDER_ID}"] if SDP_PROVIDER_ID else [])
            ),
            timeout=60,
        )
    except websocket.WebSocketException as err:
        raise RuntimeError(f"WebSocket connect failed: {err}") from err

    try:
        # Mark the start of this audio segment (RFC3339 timestamp, base64-encoded).
        rfc3339 = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
        send_json(ws, {"type": "START_TIME", "data": b64(rfc3339.encode("utf-8"))})

        with open(pcm_or_wav_path, "rb") as f:
            pcm = strip_wav_header(f.read())

        # Send ~100 ms JSON AUDIO frames (base64 PCM). Do not send binary WebSocket frames.
        for i in range(0, len(pcm), CHUNK_BYTES):
            chunk = pcm[i : i + CHUNK_BYTES]
            send_json(ws, {"type": "AUDIO", "data": b64(chunk)})

        # Optional controls while streaming (same EVENT enum as Ambient):
        # send_event(ws, "PAUSE")
        # while paused: send_event(ws, "KEEP_ALIVE"); time.sleep(KEEP_ALIVE_SEC)
        # send_event(ws, "RESUME")
        # send_event(ws, "CANCEL")  # cancels the session; server closes the socket
        # ABORT is deprecated and is not handled by the stream handler

        # End marker for this stream segment (EOF as base64 text "RU9G").
        send_json(ws, {"type": "AUDIO", "data": "RU9G"})
    except (OSError, websocket.WebSocketException) as err:
        raise RuntimeError(f"WebSocket stream failed: {err}") from err
    finally:
        try:
            ws.close()
        except Exception:
            pass

# ---------------------------------------------------------------------------
# End session and fetch structured data over REST
# ---------------------------------------------------------------------------
def end_form_filling_session(suki_token: str, ambient_session_id: str) -> None:
    """End the Form filling session after the WebSocket stream is closed."""
    url = f"{BASE_URL}/api/v1/form-filling/session/{ambient_session_id}/end"
    r = requests.post(url, headers=rest_headers(suki_token), timeout=60)
    expect_status(r, url, 200)

def get_status(suki_token: str, ambient_session_id: str) -> str:
    url = f"{BASE_URL}/api/v1/form-filling/session/{ambient_session_id}/status"
    r = requests.get(url, headers=rest_headers(suki_token), timeout=60)
    expect_status(r, url, 200)
    return r.json()["status"]

def get_structured_data(suki_token: str, ambient_session_id: str) -> Any:
    """
    GET structured-data. OpenAPI/API samples wrap the payload under the structured_data key.
    """
    url = f"{BASE_URL}/api/v1/form-filling/session/{ambient_session_id}/structured-data"
    r = requests.get(url, headers=rest_headers(suki_token), timeout=60)
    expect_status(r, url, 200)
    data = r.json()
    if not isinstance(data, dict) or "structured_data" not in data:
        raise SukiApiError(200, url, message="response missing structured_data key")
    return data["structured_data"]

def wait_for_results(suki_token: str, ambient_session_id: str, poll_sec: float = 2.0) -> dict[str, Any]:
    """Poll session status until completed, failed, or aborted, then load structured data."""
    while True:
        status = get_status(suki_token, ambient_session_id)
        if status == "completed":
            return {
                "status": status,
                "structured_data": get_structured_data(suki_token, ambient_session_id),
            }
        if status in ("failed", "aborted"):
            return {"status": status, "structured_data": None}
        time.sleep(poll_sec)

# ---------------------------------------------------------------------------
# Run the full flow end to end
# ---------------------------------------------------------------------------
if __name__ == "__main__":
    try:
        # 1) Authenticate (register once if the provider is new).
        suki_token = login_with_register_fallback(
            PARTNER_ID,
            PARTNER_TOKEN,
            PROVIDER_ID,
            provider_name="Dr. John Smith",  # Example only — replace with the provider's name
            provider_org_id="org-123",  # Example only — replace with your org ID
        )
        # 2) Create the Form filling session and seed template context.
        #    ambient_session_id here is the Form filling session ID (not Ambient clinical create).
        ambient_session_id = create_form_filling_session(suki_token, {})
        seed_form_filling_context(suki_token, ambient_session_id, FORM_TEMPLATE_ID)
        # 3) Stream audio, then end the session and wait for structured data.
        stream_pcm_file(suki_token, ambient_session_id, "audio.wav")  # Example path — replace with your LINEAR16 PCM/WAV file
        end_form_filling_session(suki_token, ambient_session_id)
        print(wait_for_results(suki_token, ambient_session_id))
    except SukiApiError as err:
        print(hint_for_error(err))
        raise

Troubleshooting

Wrong Session ID on the WebSocket

Use the Form filling create response ambient_session_id as the Form filling session ID. Do not use an ambient clinical-notes session ID.

End Session Fails without Forms in Context

Seed at least one form in context before you call end. This tutorial uses a Suki form_template_id. You can also send a partner schema. Refer to Dynamic Form filling.

WebSocket Closes During a Pause

Send KEEP_ALIVE every five seconds while paused, and send audio at least every 25 seconds while streaming.

Auth or Stream Errors

Re-login for a fresh suki_token, and see API errors.

Summary

In this tutorial, you:

• Authenticated with the Suki API.
• Created a Form filling session and seeded form template context.
• Streamed LINEAR16 PCM audio over a WebSocket connection.
• Ended the session and retrieved structured form data.

For production, use the following endpoints:

• API: https://sdp.suki.ai.
• WebSocket: wss://sdp.suki.ai/ws/stream.

Other tutorials

Continue with another end-to-end tutorial.