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

# Amazon Nova Sonic (Bedrock)

> Bidirectional speech-to-speech with Amazon Nova Sonic through TrueFoundry via HTTP/2 AWS EventStream (not the WebSocket /live path).

Amazon Nova Sonic is Bedrock's speech-to-speech realtime model. Through TrueFoundry it uses a **dedicated HTTP/2 bidirectional EventStream** endpoint — not the WebSocket [Live / Realtime API](/docs/ai-gateway/live-api.md) (`/live/...`) used by Gemini, OpenAI, and Azure.

| Approach | Path |
| - | - |
| Nova Sonic (this page) | `POST /bedrock-stream/{providerAccount}/model/{modelId}/invoke-with-bidirectional-stream` |
| Other realtime providers | `wss://{GATEWAY_HOST}/live/{providerAccountName}` |

## Prerequisites

1. An [AWS Bedrock](/docs/ai-gateway/aws-bedrock.md) model account on TrueFoundry.
2. A model with type **Realtime** and Bedrock model id such as `amazon.nova-2-sonic-v1:0` (or `amazon.nova-2-5-sonic`).
3. IAM permission for bidirectional invoke on that model (see [IAM](#iam) below).
4. Your TrueFoundry API key and the gateway **streaming** base URL.

> [!NOTE]
>
> **Display name vs model id:** TrueFoundry sanitizes `:` in Bedrock ids for the UI display name (for example `amazon.nova-2-sonic-v1-0`). The streaming URL and Bedrock SDK still use the real model id with the colon (`amazon.nova-2-sonic-v1:0`).

> [!NOTE]
>
> **Streaming host:** Self-hosted gateways often expose Nova Sonic on a dedicated TLS port (default **8443**). Set `GATEWAY_BASE` to that HTTPS origin (for example `https://gateway.example.com:8443`). On managed TrueFoundry, use the Bedrock streaming base URL for your tenant.

## Endpoint and auth

```text theme={"dark"}
POST {GATEWAY_BASE}/bedrock-stream/{providerAccount}/model/{modelId}/invoke-with-bidirectional-stream
```

| Header | Value |
| - | - |
| `Authorization` | `Bearer <TFY_API_KEY>` (or `x-tfy-api-key: <TFY_API_KEY>`) |
| `Content-Type` | `application/vnd.amazon.eventstream` |
| `Accept` | `application/vnd.amazon.eventstream` |

The gateway authenticates you with the TrueFoundry API key, then signs upstream Bedrock traffic with the provider account's AWS credentials (SigV4 or Bedrock API key).

Request/response bodies are **AWS EventStream** frames. Chunk payloads are JSON with a base64 `bytes` field; each frame's `:content-type` must be `application/json` (not `application/octet-stream`), or Bedrock returns `modelStreamErrorException`.

## Add the model

1. Open your Bedrock model account → add a model.
2. Set **Model ID** to the Bedrock id (for example `amazon.nova-2-sonic-v1:0`).
3. Set **Model type** to **Realtime**.
4. Save. Use the account name as `{providerAccount}` and the Bedrock model id as `{modelId}` in the path (URL-encode `:` as `%3A` if you build the URL by hand).

## Code example (Python)

Requires Python 3.12+. Uses the experimental Bedrock Runtime SDK against the gateway streaming base. Authenticate with your TrueFoundry API key via `AWS_BEARER_TOKEN_BEDROCK` (same header shape as Bedrock API keys).

```python Python lines theme={"dark"}
"""
Amazon Nova Sonic via TrueFoundry Gateway (HTTP/2 bidirectional EventStream)

pip install aws_sdk_bedrock_runtime smithy-aws-core pyaudio
"""
import asyncio
import base64
import json
import os
import uuid

import pyaudio
from aws_sdk_bedrock_runtime.client import (
    BedrockRuntimeClient,
    InvokeModelWithBidirectionalStreamOperationInput,
)
from aws_sdk_bedrock_runtime.config import Config
from aws_sdk_bedrock_runtime.models import (
    BidirectionalInputPayloadPart,
    InvokeModelWithBidirectionalStreamInputChunk,
)

API_KEY = "your-truefoundry-api-key"
GATEWAY_BASE = "https://{GATEWAY_STREAMING_HOST}"  # e.g. https://gateway.example.com:8443
PROVIDER_ACCOUNT = "tfy-ai-bedrock"  # Bedrock account name on TrueFoundry
MODEL_ID = "amazon.nova-2-sonic-v1:0"
REGION = "us-east-1"

os.environ["AWS_BEARER_TOKEN_BEDROCK"] = API_KEY

INPUT_SAMPLE_RATE = 16000
OUTPUT_SAMPLE_RATE = 24000
CHANNELS = 1
FORMAT = pyaudio.paInt16
CHUNK_SIZE = 1024


class GatewayNovaSonic:
    def __init__(self):
        self.client = None
        self.stream = None
        self.response_task = None
        self.is_active = False
        self.prompt_name = str(uuid.uuid4())
        self.content_name = str(uuid.uuid4())
        self.audio_content_name = str(uuid.uuid4())
        self.audio_queue: asyncio.Queue[bytes] = asyncio.Queue()
        self.role = None
        self.display_assistant_text = False

    def _initialize_client(self):
        # SDK appends /model/{model_id}/invoke-with-bidirectional-stream
        config = Config(
            endpoint_uri=f"{GATEWAY_BASE.rstrip('/')}/bedrock-stream/{PROVIDER_ACCOUNT}",
            region=REGION,
        )
        self.client = BedrockRuntimeClient(config=config)

    async def send_event(self, event_json: str):
        event = InvokeModelWithBidirectionalStreamInputChunk(
            value=BidirectionalInputPayloadPart(bytes_=event_json.encode("utf-8"))
        )
        await self.stream.input_stream.send(event)

    async def start_session(self):
        if not self.client:
            self._initialize_client()

        self.stream = await self.client.invoke_model_with_bidirectional_stream(
            InvokeModelWithBidirectionalStreamOperationInput(model_id=MODEL_ID)
        )
        self.is_active = True

        await self.send_event(
            json.dumps(
                {
                    "event": {
                        "sessionStart": {
                            "inferenceConfiguration": {
                                "maxTokens": 1024,
                                "topP": 0.9,
                                "temperature": 0.7,
                            }
                        }
                    }
                }
            )
        )
        await self.send_event(
            json.dumps(
                {
                    "event": {
                        "promptStart": {
                            "promptName": self.prompt_name,
                            "textOutputConfiguration": {"mediaType": "text/plain"},
                            "audioOutputConfiguration": {
                                "mediaType": "audio/lpcm",
                                "sampleRateHertz": OUTPUT_SAMPLE_RATE,
                                "sampleSizeBits": 16,
                                "channelCount": 1,
                                "voiceId": "matthew",
                                "encoding": "base64",
                                "audioType": "SPEECH",
                            },
                        }
                    }
                }
            )
        )
        await self.send_event(
            json.dumps(
                {
                    "event": {
                        "contentStart": {
                            "promptName": self.prompt_name,
                            "contentName": self.content_name,
                            "type": "TEXT",
                            "interactive": False,
                            "role": "SYSTEM",
                            "textInputConfiguration": {"mediaType": "text/plain"},
                        }
                    }
                }
            )
        )
        await self.send_event(
            json.dumps(
                {
                    "event": {
                        "textInput": {
                            "promptName": self.prompt_name,
                            "contentName": self.content_name,
                            "content": "You are a friendly assistant. Keep spoken replies short.",
                        }
                    }
                }
            )
        )
        await self.send_event(
            json.dumps(
                {
                    "event": {
                        "contentEnd": {
                            "promptName": self.prompt_name,
                            "contentName": self.content_name,
                        }
                    }
                }
            )
        )
        self.response_task = asyncio.create_task(self._process_responses())
        print("Connected to TrueFoundry Nova Sonic streaming endpoint")

    async def start_audio_input(self):
        await self.send_event(
            json.dumps(
                {
                    "event": {
                        "contentStart": {
                            "promptName": self.prompt_name,
                            "contentName": self.audio_content_name,
                            "type": "AUDIO",
                            "interactive": True,
                            "role": "USER",
                            "audioInputConfiguration": {
                                "mediaType": "audio/lpcm",
                                "sampleRateHertz": INPUT_SAMPLE_RATE,
                                "sampleSizeBits": 16,
                                "channelCount": 1,
                                "audioType": "SPEECH",
                                "encoding": "base64",
                            },
                        }
                    }
                }
            )
        )

    async def send_audio_chunk(self, audio_bytes: bytes):
        if not self.is_active:
            return
        await self.send_event(
            json.dumps(
                {
                    "event": {
                        "audioInput": {
                            "promptName": self.prompt_name,
                            "contentName": self.audio_content_name,
                            "content": base64.b64encode(audio_bytes).decode("utf-8"),
                        }
                    }
                }
            )
        )

    async def end_audio_input(self):
        await self.send_event(
            json.dumps(
                {
                    "event": {
                        "contentEnd": {
                            "promptName": self.prompt_name,
                            "contentName": self.audio_content_name,
                        }
                    }
                }
            )
        )

    async def end_session(self):
        if not self.is_active:
            return
        await self.send_event(json.dumps({"event": {"promptEnd": {"promptName": self.prompt_name}}}))
        await self.send_event(json.dumps({"event": {"sessionEnd": {}}}))
        await self.stream.input_stream.close()

    async def _process_responses(self):
        try:
            while self.is_active:
                output = await self.stream.await_output()
                result = await output[1].receive()
                if not (result.value and result.value.bytes_):
                    continue
                data = json.loads(result.value.bytes_.decode("utf-8"))
                event = data.get("event") or {}
                if "contentStart" in event:
                    self.role = event["contentStart"].get("role")
                    fields = event["contentStart"].get("additionalModelFields")
                    if fields:
                        stage = json.loads(fields).get("generationStage")
                        self.display_assistant_text = stage == "SPECULATIVE"
                    else:
                        self.display_assistant_text = False
                elif "textOutput" in event:
                    text = event["textOutput"]["content"]
                    if self.role == "ASSISTANT" and self.display_assistant_text:
                        print(f"Assistant: {text}")
                    elif self.role == "USER":
                        print(f"User: {text}")
                elif "audioOutput" in event:
                    await self.audio_queue.put(base64.b64decode(event["audioOutput"]["content"]))
        except Exception as exc:
            print(f"Error processing responses: {exc}")

    async def play_audio(self):
        p = pyaudio.PyAudio()
        stream = p.open(format=FORMAT, channels=CHANNELS, rate=OUTPUT_SAMPLE_RATE, output=True)
        try:
            while self.is_active:
                stream.write(await self.audio_queue.get())
        finally:
            stream.stop_stream()
            stream.close()
            p.terminate()

    async def capture_audio(self):
        p = pyaudio.PyAudio()
        stream = p.open(
            format=FORMAT,
            channels=CHANNELS,
            rate=INPUT_SAMPLE_RATE,
            input=True,
            frames_per_buffer=CHUNK_SIZE,
        )
        print("Speak into your microphone. Press Enter to stop...")
        await self.start_audio_input()
        try:
            while self.is_active:
                await self.send_audio_chunk(stream.read(CHUNK_SIZE, exception_on_overflow=False))
                await asyncio.sleep(0.01)
        finally:
            stream.stop_stream()
            stream.close()
            p.terminate()
            await self.end_audio_input()


async def main():
    client = GatewayNovaSonic()
    await client.start_session()
    playback = asyncio.create_task(client.play_audio())
    capture = asyncio.create_task(client.capture_audio())
    await asyncio.get_event_loop().run_in_executor(None, input)

    for task in (playback, capture):
        if not task.done():
            task.cancel()
    await asyncio.gather(playback, capture, return_exceptions=True)

    await client.end_session()
    client.is_active = False
    if client.response_task and not client.response_task.done():
        client.response_task.cancel()
    print("Session ended")


if __name__ == "__main__":
    asyncio.run(main())
```

## Session shape (high level)

Nova Sonic expects an ordered EventStream of JSON events, including:

1. `sessionStart` → `promptStart`
2. Optional system `TEXT` (`contentStart` / `textInput` / `contentEnd`)
3. User `AUDIO` (`contentStart` / `audioInput` chunks / `contentEnd`) — at least one audio content is required
4. After the turn: `promptEnd` → `sessionEnd`

See [Amazon Nova Sonic](https://docs.aws.amazon.com/nova/latest/userguide/speech.html) and [InvokeModelWithBidirectionalStream](https://docs.aws.amazon.com/bedrock/latest/APIReference/API_runtime_InvokeModelWithBidirectionalStream.html) for the full event schema.

## IAM

Grant the Bedrock identity used by your provider account permission to invoke bidirectional streams, for example:

```json JSON lines theme={"dark"}
{
  "Effect": "Allow",
  "Action": [
    "bedrock:InvokeModel",
    "bedrock:InvokeModelWithResponseStream",
    "bedrock:InvokeModelWithBidirectionalStream"
  ],
  "Resource": [
    "arn:aws:bedrock:*::foundation-model/amazon.nova-2-sonic-v1:0",
    "arn:aws:bedrock:*::foundation-model/amazon.nova-2-5-sonic",
    "arn:aws:bedrock:*:<aws-account-id>:inference-profile/*"
  ]
}
```

Adjust ARNs for the model ids and regions you use. See [AWS Bedrock authentication](/docs/ai-gateway/aws-bedrock-authentication) for how TrueFoundry attaches credentials.

## Limitations

* Not available on the WebSocket `/live` path.
* Clients must speak AWS EventStream (HTTP/2). Ordinary REST/SSE clients and most Postman setups cannot drive a full duplex session; use the Bedrock Runtime SDK or an HTTP/2 EventStream client.
* Tool calling for Nova Sonic follows Bedrock's native schema, not the OpenAI Realtime tool-calling examples on the Live API pages.

## References

* [AWS Bedrock on TrueFoundry](/docs/ai-gateway/aws-bedrock.md)
* [Live / Realtime API (WebSocket)](/docs/ai-gateway/live-api.md)
* [Amazon Nova Sonic user guide](https://docs.aws.amazon.com/nova/latest/userguide/speech.html)
* [InvokeModelWithBidirectionalStream API](https://docs.aws.amazon.com/bedrock/latest/APIReference/API_runtime_InvokeModelWithBidirectionalStream.html)


This documentation is built and hosted on [Mintlify](https://mintlify.com), a developer documentation platform.