streamcoreai

package module
v0.0.5 Latest Latest
Warning

This package is not in the latest version of its module.

Go to latest
Published: Aug 24, 2026 License: Apache-2.0 Imports: 19 Imported by: 0

README

github.com/streamcoreai/go-sdk

English | 简体中文

Go SDK for connecting to a StreamCoreAI server via WebRTC + WHIP.

When this module lives in a split repository, tag releases (for example v0.1.0) so downstream modules can require a version without a replace directive. See Repository layout.

Installation

go get github.com/streamcoreai/go-sdk

Quick Start

package main

import (
	"context"
	"fmt"
	"log"
	"os"
	"os/signal"

	streamcoreai "github.com/streamcoreai/go-sdk"
)

func main() {
	client := streamcoreai.NewClient(
		streamcoreai.Config{
			WHIPEndpoint: "http://localhost:8080/whip",
		},
		streamcoreai.EventHandler{
			OnStatusChange: func(status streamcoreai.ConnectionStatus) {
				fmt.Println("Status:", status)
			},
			OnTranscript: func(entry streamcoreai.TranscriptEntry, all []streamcoreai.TranscriptEntry) {
				fmt.Printf("[%s] %s\n", entry.Role, entry.Text)
			},
			OnError: func(err error) {
				log.Println("Error:", err)
			},
		},
	)

	ctx, cancel := signal.NotifyContext(context.Background(), os.Interrupt)
	defer cancel()

	if err := client.Connect(ctx); err != nil {
		log.Fatal(err)
	}
	defer client.Disconnect()

	// Send microphone audio to the agent.
	// (Replace with your actual audio capture — e.g. PortAudio)
	go func() {
		pcm := make([]int16, streamcoreai.FrameSize) // 20ms mono 48kHz
		for {
			// Fill pcm from your audio source...
			if err := client.SendPCM(pcm); err != nil {
				return
			}
		}
	}()

	// Receive agent audio and play it.
	go func() {
		pcm := make([]int16, streamcoreai.FrameSize)
		for {
			nSamples, err := client.RecvPCM(pcm)
			if err != nil {
				return
			}
			// Play pcm[:nSamples] through your audio output...
			_ = nSamples
		}
	}()

	<-ctx.Done()
}

API

streamcoreai.NewClient(cfg, events)

Creates a new client instance.

Config
Field Type Default Description
WHIPEndpoint string "http://localhost:8080/whip" WHIP signaling endpoint URL
Token string — JWT sent as Authorization: Bearer on the WHIP request
TokenURL string — Token endpoint; when set, a JWT is fetched before each connection (overrides Token)
APIKey string — Sent as Authorization: Bearer when fetching from TokenURL
ResourceID string — Who is on the call, forwarded to an external agent so it can scope memory to the person rather than the call. Sent in the token request body when TokenURL is set (the server signs it into the token), otherwise as an X-StreamCore-Resource-Id header
ICEServers []webrtc.ICEServer Google STUN server ICE server configuration
ReconnectAttempts int 3 ICE restarts while Disconnected; negative disables the phase
ReconnectDelay time.Duration 2s Wait before the first ICE restart, doubling each retry
ResumeAttempts int 2 Resume redials once the connection has Failed; negative disables the phase
ResumeDelay time.Duration 1s Wait before the first redial, doubling each retry
EventHandler

All callbacks are optional; leave unused fields nil.

Callback Signature Description
OnStatusChange func(status ConnectionStatus) Fired when connection status changes
OnTranscript func(entry TranscriptEntry, all []TranscriptEntry) Fired on new or updated transcript
OnAgentStateChange func(state AgentState) Fired when the agent starts listening, thinking, or speaking
OnTiming func(event TimingEvent) Fired with server-side pipeline timing info
OnError func(err error) Fired on connection or server errors
OnDataChannelMessage func(msg DataChannelMessage) Fired for every raw data channel message
OnData func(topic string, payload []byte) Fire-and-forget server data packet, payload already base64-decoded (movement.command carries locomotion commands)
Client Methods
Method Returns Description
Connect(ctx context.Context) error Establish WebRTC + WHIP session
Disconnect() — Tear down connection, free resources
Status() ConnectionStatus Current connection status
Transcript() []TranscriptEntry Full conversation history (copy)
SendPCM(pcm []int16) error Encode and send a 20ms PCM frame (960 samples, mono, 48kHz)
RecvPCM(pcm []int16) (int, error) Receive and decode agent audio into PCM (blocks until available)
Audio Constants
Constant Value Description
SampleRate 48000 Audio sample rate in Hz (Opus)
Channels 1 Number of audio channels (mono)
FrameSize 960 Samples per 20ms frame at 48kHz
Client Fields (advanced — for custom audio pipelines)
Field Type Description
LocalTrack *webrtc.TrackLocalStaticRTP Raw RTP track for sending audio to server
RemoteTrackCh chan *webrtc.TrackRemote Receives the agent's raw audio track
Types
type ConnectionStatus string // "idle", "connecting", "connected", "error", "disconnected"

type TranscriptEntry struct {
    Role    string // "user" or "assistant"
    Text    string
    Partial bool
}

type DataChannelMessage struct {
    Type    string // "transcript", "response", or "error"
    Text    string
    Final   bool
    Message string // for error type
}

Reconnection

A network change mid-call — a machine moving networks, a VPN toggle, a process suspended and resumed — kills the transport without ending the call. The client recovers it automatically and the conversation survives: the agent still knows who it is talking to and does not replay its greeting.

Recovery runs as a ladder of two phases:

Phase When Cost
ICE restart While the connection is Disconnected Invisible. Same PeerConnection, same DTLS, same tracks — just new candidates.
Resume redial Once the connection has Failed A full renegotiation and a moment of silence, but the server reattaches you to the same conversation.

ICE restart is tried first because it costs nothing. It stops being possible the moment the connection reaches Failed — the server has closed its peer by then — which is where a process that was paused, or offline for more than about 25 seconds, always lands. The resume phase recovers those.

Status goes StatusConnected → StatusReconnecting → StatusConnected:

client := streamcoreai.NewClient(streamcoreai.Config{
    WHIPEndpoint:      "http://localhost:8080/whip",
    ReconnectAttempts: 3,                // ICE restarts, 2s -> 4s -> 8s
    ReconnectDelay:    2 * time.Second,
    ResumeAttempts:    2,                // then redials,  1s -> 2s
    ResumeDelay:       time.Second,
}, streamcoreai.EventHandler{
    OnReconnect: func(e streamcoreai.ReconnectEvent) {
        log.Printf("%s %d/%d: %s", e.Phase, e.Attempt, e.MaxAttempts, e.Outcome)
        if e.Outcome == streamcoreai.ReconnectRecoveredWithoutHistory {
            log.Println("reconnected, but the agent has lost the conversation")
        }
    },
})

Handle ReconnectRecoveredWithoutHistory. It means the call is working but the server could not resume the session — usually because the client was away longer than session_grace_ms — so the agent has no memory of anything said before. Everything still functions, which is exactly why it goes unnoticed until the agent asks something it was already told.

Two details worth knowing:

  • The first ICE restart is deliberately delayed (ReconnectDelay, default 2s). Most drops are brief packet loss that ICE repairs unaided, and patching immediately would spend an attempt on a connection that was about to recover by itself.
  • Both phases share one deadline. Disconnected becomes Failed after roughly 25 seconds, and the server then holds the conversation for session_grace_ms (30s by default). Raising ReconnectAttempts spends budget the resume phase would otherwise have.

Set ReconnectAttempts or ResumeAttempts negative to disable either phase.

What this means for your audio loops. LocalTrack is deliberately not replaced across a redial — the same track is rebound to the new PeerConnection, so a goroutine writing RTP to it keeps working. Inbound audio is different: the server sends a new track, delivered on RemoteTrackCh. RecvPCM handles that for you (it re-acquires and keeps decoding, so a reconnect is a gap in the audio rather than an error). If you read RemoteTrackCh yourself, expect a second track after a reconnect and switch to it — the old one only returns read errors.

Audio I/O

Audio I/O

The SDK handles Opus encoding/decoding and RTP packetization internally. You only need to supply and consume raw PCM audio:

  • Sending audio: Capture PCM int16 samples (mono, 48kHz, 960 samples per frame) and call client.SendPCM(pcm)
  • Receiving audio: Call client.RecvPCM(pcm) to get decoded PCM samples from the agent

For PortAudio setup, use the exported constants: streamcoreai.SampleRate, streamcoreai.Channels, streamcoreai.FrameSize.

For advanced use cases that need direct access to the WebRTC tracks (custom codecs, raw RTP), the LocalTrack and RemoteTrackCh fields are still available.

For reference implementations, see the Go CLI example and Go TUI example.

License

MIT

Documentation

Index

Constants

View Source
const (
	// SampleRate is the audio sample rate in Hz required by Opus.
	SampleRate = 48000
	// Channels is the number of audio channels (mono).
	Channels = 1
	// FrameSize is the number of samples in a 20ms frame at 48kHz.
	FrameSize = 960
)
View Source
const ICEFragmentContentType = "application/trickle-ice-sdpfrag"

ICEFragmentContentType is the media type of an ICE restart body (RFC 8840, RFC 9725 §4.4).

View Source
const ResourceIDHeader = "X-StreamCore-Resource-Id"

ResourceIDHeader carries the caller identity when there is no token endpoint to sign one into a claim. For server-side clients, not browsers.

Variables

This section is empty.

Functions

This section is empty.

Types

type AgentState

type AgentState string

AgentState represents the server-reported state of the voice agent pipeline.

const (
	AgentListening AgentState = "listening"
	AgentThinking  AgentState = "thinking"
	AgentSpeaking  AgentState = "speaking"
)

type Client

type Client struct {

	// LocalTrack is the outbound audio track you write RTP packets to.
	// It is created during Connect() and available afterwards.
	LocalTrack *webrtc.TrackLocalStaticRTP

	// RemoteTrack receives inbound audio from the agent.
	// It is delivered via the RemoteTrackCh channel after the connection is established.
	RemoteTrackCh chan *webrtc.TrackRemote
	// contains filtered or unexported fields
}

Client manages a WebRTC connection to a Voice Agent server via WHIP signaling. It handles peer connection setup, data channel event handling, and provides access to the local/remote audio tracks for custom audio I/O.

func NewClient

func NewClient(cfg Config, events EventHandler) *Client

NewClient creates a new voice agent client with the given configuration and event handlers.

func (*Client) Connect

func (c *Client) Connect(ctx context.Context) error

Connect establishes a WebRTC connection to the voice agent server using WHIP. It creates a local audio track (Opus), performs WHIP signaling, and sets up the data channel for receiving transcript/response events.

After Connect returns, write audio to LocalTrack and read agent audio from RemoteTrackCh.

func (*Client) Disconnect

func (c *Client) Disconnect()

Disconnect tears down the WebRTC connection and frees resources.

func (*Client) RecvPCM

func (c *Client) RecvPCM(pcm []int16) (int, error)

RecvPCM blocks until a frame of audio is received from the agent, decodes the Opus payload, and writes PCM int16 samples into pcm. The pcm slice should have capacity for at least FrameSize (960) samples. Returns the number of decoded samples.

func (*Client) SendPCM

func (c *Client) SendPCM(pcm []int16) error

SendPCM encodes a 20ms frame of PCM int16 audio (mono, 48kHz, 960 samples) and sends it as an RTP/Opus packet to the voice agent server.

func (*Client) Status

func (c *Client) Status() ConnectionStatus

Status returns the current connection status.

func (*Client) Transcript

func (c *Client) Transcript() []TranscriptEntry

Transcript returns the current conversation transcript.

type Config

type Config struct {
	// WHIPEndpoint is the URL of the WHIP signaling endpoint.
	// Defaults to "http://localhost:8080/whip".
	WHIPEndpoint string

	// Token is an optional JWT token for authenticating with the WHIP endpoint.
	Token string

	// TokenURL is the URL of a token endpoint. If set, the client will POST
	// to this URL to fetch a JWT before each WHIP connection. Overrides Token.
	TokenURL string

	// APIKey is sent as a Bearer header when fetching a token from TokenURL.
	APIKey string

	// ResourceID is who is on the call: an app user ID, or the number a phone
	// call came from. The server passes it to an external agent, which can then
	// remember a caller across separate calls.
	//
	// With TokenURL set it goes in the token request body and the server signs
	// it into the token. Otherwise it goes as a header, which the server only
	// trusts when there is no signed claim.
	ResourceID string

	// ICEServers configures the ICE servers for the WebRTC connection.
	// Defaults to Google's public STUN server.
	ICEServers []webrtc.ICEServer

	// Metadata is an optional set of key-value pairs appended to the WHIP
	// endpoint as query parameters (e.g. direction=outbound).
	Metadata map[string]string

	// ReconnectAttempts is how many ICE restarts to attempt before giving up
	// on a dropped connection. Attempts are spaced by ReconnectDelay doubling
	// each time, and the whole sequence must finish inside the ~25 seconds it
	// takes the connection to go from disconnected to failed — past that the
	// server has closed the peer and only a fresh Connect recovers.
	// Defaults to 3. Negative disables automatic reconnection.
	ReconnectAttempts int

	// ReconnectDelay is the wait before the first ICE restart attempt,
	// doubling for each retry. The initial wait matters: most disconnected
	// transitions are brief packet loss that ICE repairs on its own, and
	// patching immediately would spend a restart on a connection that was
	// about to recover by itself. Defaults to 2s.
	ReconnectDelay time.Duration

	// ResumeAttempts is how many times to redial with the session's resume
	// token after ICE restart is no longer possible — which is the case as
	// soon as the connection reaches Failed, and the only case for a process
	// that was suspended or offline longer than ICE tolerates.
	//
	// A redial builds a new transport but reattaches to the same conversation,
	// so history and the agent's memory survive. The deadline is the server's
	// session_grace_ms (30s by default) measured from the drop, and the ICE
	// restart phase spends from the same budget. Defaults to 2. Negative
	// disables the resume phase.
	ResumeAttempts int

	// ResumeDelay is the wait before the first resume redial, doubling for
	// each retry. Shorter than ReconnectDelay because by this point the
	// connection is known dead — there is nothing left to wait out.
	// Defaults to 1s.
	ResumeDelay time.Duration
}

Config holds the configuration for a StreamCoreAIClient.

type ConnectionStatus

type ConnectionStatus string

ConnectionStatus represents the current state of the voice agent connection.

const (
	StatusIdle       ConnectionStatus = "idle"
	StatusConnecting ConnectionStatus = "connecting"
	StatusConnected  ConnectionStatus = "connected"
	// StatusReconnecting means the transport dropped and an ICE restart is in
	// flight. The session, and with it the conversation, is still alive on the
	// server — this is not a terminal state and usually resolves back to
	// StatusConnected.
	StatusReconnecting ConnectionStatus = "reconnecting"
	StatusError        ConnectionStatus = "error"
	StatusDisconnected ConnectionStatus = "disconnected"
)

type DataChannelMessage

type DataChannelMessage struct {
	Type    string `json:"type"` // "transcript", "response", "error", "timing", "state", or "data"
	Text    string `json:"text,omitempty"`
	Final   bool   `json:"final,omitempty"`
	Message string `json:"message,omitempty"` // for error type
	Stage   string `json:"stage,omitempty"`   // for timing type
	Ms      int    `json:"ms,omitempty"`      // for timing type
	State   string `json:"state,omitempty"`   // for state type
	Topic   string `json:"topic,omitempty"`   // for data type
	Payload string `json:"payload,omitempty"` // for data type; base64-encoded JSON
}

DataChannelMessage represents a message received on the data channel.

type EventHandler

type EventHandler struct {
	// OnStatusChange is called when the connection status changes.
	OnStatusChange func(status ConnectionStatus)

	// OnTranscript is called when a new or updated transcript entry is received.
	OnTranscript func(entry TranscriptEntry, all []TranscriptEntry)

	// OnError is called when an error occurs.
	OnError func(err error)

	// OnTiming is called when a timing/latency event is received from the server.
	OnTiming func(event TimingEvent)

	// OnAgentStateChange is called when the server reports an agent state transition.
	OnAgentStateChange func(state AgentState)

	// OnDataChannelMessage is called for every raw data channel message.
	// This is optional and useful for custom message handling.
	OnDataChannelMessage func(msg DataChannelMessage)

	// OnData is called for a topic-addressed packet from a device-side tool,
	// with the payload already base64-decoded. The server sends these
	// fire-and-forget — it has already told the model the action succeeded —
	// so there is nothing to reply to. Locomotion commands from the movement.*
	// tools arrive on the "movement.command" topic.
	OnData func(topic string, payload []byte)

	// OnReconnect is called for each recovery attempt and once when the
	// outcome is known, so a UI can distinguish a recoverable drop from a
	// lost call — and, via ReconnectRecoveredWithoutHistory, a reconnection
	// that kept the conversation from one that lost it.
	OnReconnect func(event ReconnectEvent)
}

EventHandler defines callbacks for voice agent events.

type ReconnectEvent

type ReconnectEvent struct {
	Attempt     int              // 1-based attempt number, counted within the phase
	MaxAttempts int              // attempts this phase makes before moving on
	Phase       ReconnectPhase   // which mechanism this attempt used
	Outcome     ReconnectOutcome // "attempting" while in flight, then the result
	Err         error            // why the attempt failed, when Outcome is ReconnectFailed
}

ReconnectEvent reports progress of the recovery sequence.

type ReconnectOutcome

type ReconnectOutcome string

ReconnectOutcome describes where a recovery attempt got to.

const (
	ReconnectAttempting ReconnectOutcome = "attempting"
	ReconnectRecovered  ReconnectOutcome = "recovered"
	// ReconnectRecoveredWithoutHistory means the call works but the server
	// could not resume the session, so the agent has forgotten the
	// conversation and will not know what was already said. Worth surfacing
	// to the user rather than only logging.
	ReconnectRecoveredWithoutHistory ReconnectOutcome = "recovered-without-history"
	ReconnectFailed                  ReconnectOutcome = "failed"
)

type ReconnectPhase

type ReconnectPhase string

ReconnectPhase identifies which recovery mechanism an attempt used.

const (
	// PhaseICERestart keeps the existing transport alive. Only possible while
	// the connection is Disconnected; nothing above the transport notices.
	PhaseICERestart ReconnectPhase = "ice-restart"
	// PhaseResume rebuilds the transport and reattaches it to the same
	// server-side conversation. The only option once the connection has Failed.
	PhaseResume ReconnectPhase = "resume"
)

type TimingEvent

type TimingEvent struct {
	Stage string `json:"stage"`
	Ms    int    `json:"ms"`
}

TimingEvent represents a single latency measurement from the server pipeline.

type TranscriptEntry

type TranscriptEntry struct {
	Role    string `json:"role"` // "user" or "assistant"
	Text    string `json:"text"`
	Partial bool   `json:"partial,omitempty"`
}

TranscriptEntry represents a single transcript message in the conversation.

type WHIPRestartError

type WHIPRestartError struct {
	Status int
	Body   string
	// CurrentETag is the tag the server reported as current, present on a 412.
	CurrentETag string
}

WHIPRestartError is returned when a PATCH is rejected. Status lets the caller tell a retryable failure from a terminal one.

func (*WHIPRestartError) Error

func (e *WHIPRestartError) Error() string

func (*WHIPRestartError) Retryable

func (e *WHIPRestartError) Retryable() bool

Retryable reports whether another attempt against the same session could still succeed. A 404 means the session was reaped, 409 that it has no peer to restart, and 405 that the server declines restarts entirely — only a redial recovers from those.

type WHIPRestartResult

type WHIPRestartResult struct {
	// Fragment is the server's sdpfrag, to fold into the stored remote
	// description.
	Fragment string
	// ETag is the rotated tag identifying the new ICE session.
	ETag string
}

WHIPRestartResult holds the server's reply to an ICE restart.

type WHIPResult

type WHIPResult struct {
	AnswerSDP  string
	SessionURL string
	// ETag identifies the ICE session (RFC 9725 §4.3.1) and is required to
	// PATCH it for an ICE restart.
	ETag string
	// ResumeToken is a single-use credential for reattaching a later redial to
	// this conversation. Empty when the server cannot resume — a realtime
	// (speech-to-speech) session, whose history lives inside the provider.
	ResumeToken string
	// ResumeStatus is "new", "resumed", or "expired". Anything but "resumed"
	// on a redial means the agent has no memory of the earlier conversation.
	ResumeStatus string
}

WHIPResult holds the response from a WHIP signaling exchange.

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL