Messaging SDK

The conversation stream

Open the bidirectional ProcessConversation stream and handle inbound events
View as Markdown

The primary RPC is ProcessConversation, a bidirectional stream. The sidecar pushes incoming user messages, feedback, and audio. Your agent pushes back status updates, content chunks, errors, and other responses on the same stream.

import { MessagingClient, AgentResponse } from '@astropods/messaging';
const client = new MessagingClient('localhost:9090');
await client.connect();
const conversation = client.createConversationStream();
conversation.on('response', (resp: AgentResponse) => {
  if (resp.incomingMessage) {
    const m = resp.incomingMessage;
    conversation.sendContentChunk(m.conversationId, {
      type: 'END',
      content: `you said: ${m.content}`,
    });
  }
});
conversation.on('error', (err) => console.error('stream error', err));
conversation.on('reconnecting', (info) => console.warn('reconnecting', info));

ConversationStream events

EventPayloadNotes
responseAgentResponseInbound event from the sidecar.
audioConfigAudioStreamConfigConvenience: emitted in addition to response when audio config arrives.
audioChunkAudioChunkConvenience: emitted in addition to response when an audio chunk arrives.
reconnecting{ attempt, reason, delayMs }Before each retry delay.
reconnected{ attempt }After a successful stream recreation.
errorErrorNon-retryable error OR max retries exceeded.
end—Only on intentional close(), not on unexpected drop.

ConversationStream send methods

MethodWhat it sends
sendMessage(message)A Message.
sendFeedback(feedback)A PlatformFeedback.
sendAgentConfig(config)An AgentConfig.
sendAgentResponse(response)An AgentResponse (typed, any variant).
sendContentChunk(conversationId, chunk)Convenience: wraps ContentChunk in AgentResponse.
sendStatusUpdate(conversationId, status)Convenience: wraps StatusUpdate.
sendTranscript(conversationId, text, msgId?, lang?)Convenience: wraps Transcript.
sendAudioConfig(config)AudioStreamConfig upstream.
sendAudioChunk(chunk)AudioChunk upstream.
endAudio()Sends { done: true } to mark segment end.
end()Closes the stream intentionally.

Next steps