realtime

Documentation

Index

• func HandleSessions(server *RealtimeServer) http.HandlerFunc
• type Adapter
• type AudioDeltaData
• type ErrorData
• type EventType
• type GeminiRealtimeAdapter
  • func NewGeminiRealtimeAdapter(session *Session, messageSaver threads.MessageSaver, eventBus *events.EventBus) (*GeminiRealtimeAdapter, error)
  • func (a *GeminiRealtimeAdapter) Close() error
  • func (a *GeminiRealtimeAdapter) GetResumptionToken() string
  • func (a *GeminiRealtimeAdapter) HandleControl(action string, data map[string]interface{}) error
  • func (a *GeminiRealtimeAdapter) ReceiveEvents() <-chan UnifiedEvent
  • func (a *GeminiRealtimeAdapter) SendAudio(base64Audio string) error
  • func (a *GeminiRealtimeAdapter) SendText(text string) error
• type Message
• type OpenAIRealtimeAdapter
  • func NewOpenAIRealtimeAdapter(session *Session, messageSaver threads.MessageSaver, eventBus *events.EventBus) (*OpenAIRealtimeAdapter, error)
  • func (a *OpenAIRealtimeAdapter) Close() error
  • func (a *OpenAIRealtimeAdapter) HandleControl(action string, data map[string]interface{}) error
  • func (a *OpenAIRealtimeAdapter) ReceiveEvents() <-chan UnifiedEvent
  • func (a *OpenAIRealtimeAdapter) SendAudio(base64Audio string) error
  • func (a *OpenAIRealtimeAdapter) SendText(text string) error
• type RealtimeServer
  • func NewServer(db *sql.DB, messageSaver threads.MessageSaver, eventBus *events.EventBus) *RealtimeServer
  • func (s *RealtimeServer) GetSessions() []*Session
  • func (s *RealtimeServer) HandleWebSocket(w http.ResponseWriter, r *http.Request)
• type Session
  • func (s *Session) AddPendingTool(execution *ToolExecution)
  • func (s *Session) Close()
  • func (s *Session) GetPendingTool(callID string) (*ToolExecution, bool)
  • func (s *Session) SetTurnState(state string)
  • func (s *Session) UpdateLastActivity()
  • func (s *Session) UpdateToolStatus(callID, status string)
• type SessionState
• type ToolCallData
• type ToolExecution
• type ToolResultData
• type TranscriptData
• type UnifiedEvent

Functions

func HandleSessions

func HandleSessions(server *RealtimeServer) http.HandlerFunc

HandleSessions handles HTTP requests for session management

Types

type Adapter

type Adapter interface {
	SendAudio(base64Audio string) error
	SendText(text string) error
	HandleControl(action string, data map[string]interface{}) error
	ReceiveEvents() <-chan UnifiedEvent
	Close() error
}

Adapter interface for different realtime providers

type AudioDeltaData

type AudioDeltaData struct {
	Format     string `json:"format"`      // "pcm16"
	SampleRate int    `json:"sample_rate"` // 24000
	Chunk      string `json:"chunk"`       // Base64-encoded audio
}

AudioDeltaData contains audio chunk data

type ErrorData

type ErrorData struct {
	Code    string `json:"code"`
	Message string `json:"message"`
}

ErrorData contains error information

type EventType

type EventType string

EventType represents the type of unified event

const (
	EventTypeAudioDelta     EventType = "audio_delta"
	EventTypeAudioComplete  EventType = "audio_complete"
	EventTypeTranscript     EventType = "transcript"
	EventTypeToolCall       EventType = "tool_call"
	EventTypeToolResult     EventType = "tool_result"
	EventTypeTurnStart      EventType = "turn_start"
	EventTypeTurnEnd        EventType = "turn_end"
	EventTypeError          EventType = "error"
	EventTypeSessionCreated EventType = "session_created"
)

type GeminiRealtimeAdapter

type GeminiRealtimeAdapter struct {
	// contains filtered or unexported fields
}

GeminiRealtimeAdapter implements the Adapter interface for Gemini Live API

func NewGeminiRealtimeAdapter

func NewGeminiRealtimeAdapter(session *Session, messageSaver threads.MessageSaver,
	eventBus *events.EventBus) (*GeminiRealtimeAdapter, error)

NewGeminiRealtimeAdapter creates a new Gemini Live adapter

func (*GeminiRealtimeAdapter) Close

func (a *GeminiRealtimeAdapter) Close() error

Close closes the adapter

func (*GeminiRealtimeAdapter) GetResumptionToken

func (a *GeminiRealtimeAdapter) GetResumptionToken() string

GetResumptionToken returns the session resumption token (Gemini-specific)

func (*GeminiRealtimeAdapter) HandleControl

func (a *GeminiRealtimeAdapter) HandleControl(action string, data map[string]interface{}) error

HandleControl handles control messages

func (*GeminiRealtimeAdapter) ReceiveEvents

func (a *GeminiRealtimeAdapter) ReceiveEvents() <-chan UnifiedEvent

ReceiveEvents returns channel for receiving events

func (*GeminiRealtimeAdapter) SendAudio

func (a *GeminiRealtimeAdapter) SendAudio(base64Audio string) error

func (*GeminiRealtimeAdapter) SendText

func (a *GeminiRealtimeAdapter) SendText(text string) error

SendText sends text message to Gemini

type Message

type Message struct {
	Type      string      `json:"type"`       // "audio", "text", "control", event types
	SessionID string      `json:"session_id"` // Session identifier
	Data      interface{} `json:"data"`       // Payload (varies by type)
	Timestamp int64       `json:"timestamp"`  // Unix milliseconds
}

Message represents a unified message format for client-server communication

type OpenAIRealtimeAdapter

type OpenAIRealtimeAdapter struct {
	// contains filtered or unexported fields
}

OpenAIRealtimeAdapter implements the Adapter interface for OpenAI Realtime API

func NewOpenAIRealtimeAdapter

func NewOpenAIRealtimeAdapter(session *Session, messageSaver threads.MessageSaver,
	eventBus *events.EventBus) (*OpenAIRealtimeAdapter, error)

NewOpenAIRealtimeAdapter creates a new OpenAI Realtime adapter

func (*OpenAIRealtimeAdapter) Close

func (a *OpenAIRealtimeAdapter) Close() error

Close closes the adapter

func (*OpenAIRealtimeAdapter) HandleControl

func (a *OpenAIRealtimeAdapter) HandleControl(action string, data map[string]interface{}) error

HandleControl handles control messages

func (*OpenAIRealtimeAdapter) ReceiveEvents

func (a *OpenAIRealtimeAdapter) ReceiveEvents() <-chan UnifiedEvent

ReceiveEvents returns channel for receiving events

func (*OpenAIRealtimeAdapter) SendAudio

func (a *OpenAIRealtimeAdapter) SendAudio(base64Audio string) error

SendAudio sends audio chunk to OpenAI

func (*OpenAIRealtimeAdapter) SendText

func (a *OpenAIRealtimeAdapter) SendText(text string) error

SendText sends text message to OpenAI

type RealtimeServer

type RealtimeServer struct {
	// contains filtered or unexported fields
}

RealtimeServer manages WebSocket connections for real-time voice communication

func NewServer

func NewServer(db *sql.DB, messageSaver threads.MessageSaver, eventBus *events.EventBus) *RealtimeServer

NewServer creates a new realtime server instance

func (*RealtimeServer) GetSessions

func (s *RealtimeServer) GetSessions() []*Session

GetSessions returns all active sessions

func (*RealtimeServer) HandleWebSocket

func (s *RealtimeServer) HandleWebSocket(w http.ResponseWriter, r *http.Request)

HandleWebSocket handles WebSocket connections for real-time communication Auto-detects format: JSON (browser/app) or Binary (telephony like Vonage/Twilio)

type Session

type Session struct {
	ID       string
	ThreadID string
	Provider string // "openai-realtime"

	// Tool execution state
	PendingTools map[string]*ToolExecution

	State        SessionState
	CreatedAt    time.Time
	LastActivity time.Time
	// contains filtered or unexported fields
}

Session represents an active realtime voice session

func (*Session) AddPendingTool

func (s *Session) AddPendingTool(execution *ToolExecution)

AddPendingTool adds a tool execution to pending state

func (*Session) Close

func (s *Session) Close()

Close closes the session connections

func (*Session) GetPendingTool

func (s *Session) GetPendingTool(callID string) (*ToolExecution, bool)

GetPendingTool retrieves a pending tool execution

func (*Session) SetTurnState

func (s *Session) SetTurnState(state string)

SetTurnState sets the current turn state

func (*Session) UpdateLastActivity

func (s *Session) UpdateLastActivity()

UpdateLastActivity updates the last activity timestamp

func (*Session) UpdateToolStatus

func (s *Session) UpdateToolStatus(callID, status string)

UpdateToolStatus updates the status of a tool execution

type SessionState

type SessionState struct {
	IsActive          bool
	TurnState         string // "idle", "user_speaking", "assistant_speaking", "tool_executing"
	ConversationTurns int
	CurrentResponseID string
}

SessionState tracks the current state of a session

type ToolCallData

type ToolCallData struct {
	ID    string                 `json:"id"`
	Name  string                 `json:"name"`
	Input map[string]interface{} `json:"input"`
}

ToolCallData contains tool invocation information

type ToolExecution

type ToolExecution struct {
	CallID    string
	Name      string
	Arguments map[string]interface{}
	Status    string // "pending", "executing", "completed", "failed"
	Result    interface{}
	Error     error
	StartTime time.Time
	EndTime   time.Time
}

ToolExecution represents a tool call in progress

type ToolResultData

type ToolResultData struct {
	CallID string      `json:"call_id"`
	Name   string      `json:"name"`
	Result interface{} `json:"result"`
	Error  error       `json:"error,omitempty"`
}

ToolResultData contains tool execution result

type TranscriptData

type TranscriptData struct {
	Role    string `json:"role"`    // "user" or "assistant"
	Content string `json:"content"` // Transcribed text
}

TranscriptData contains transcription information

type UnifiedEvent

type UnifiedEvent struct {
	Type      EventType   `json:"type"`
	SessionID string      `json:"session_id,omitempty"`
	Timestamp time.Time   `json:"timestamp"`
	Data      interface{} `json:"data"`
}

UnifiedEvent represents a provider-agnostic event