> For clean Markdown of any page, append .md to the page URL.
> For a complete documentation index, see https://docs.astropods.com/llms.txt.
> For AI client integration (Claude Code, Cursor, etc.), connect to the MCP server at https://docs.astropods.com/_mcp/server.

# The conversation stream

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.

#### Node

```typescript
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**

| Event          | Payload                        | Notes                                                                       |
| -------------- | ------------------------------ | --------------------------------------------------------------------------- |
| `response`     | `AgentResponse`                | Inbound event from the sidecar.                                             |
| `audioConfig`  | `AudioStreamConfig`            | Convenience: emitted in addition to `response` when audio config arrives.   |
| `audioChunk`   | `AudioChunk`                   | Convenience: 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.                                       |
| `error`        | `Error`                        | Non-retryable error OR max retries exceeded.                                |
| `end`          | —                              | Only on intentional `close()`, not on unexpected drop.                      |

**`ConversationStream` send methods**

| Method                                                | What 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.                      |

#### Python

Bidi streaming in `grpcio` is symmetric: pass a generator of `ConversationRequest`s and iterate the returned generator of `AgentResponse`s. A common pattern uses a `queue.Queue` to push outbound messages from anywhere in the program.

```python
import queue, grpc
from astropods_messaging import (
    AgentMessagingStub, ConversationRequest, AgentResponse, ContentChunk,
)

channel = grpc.insecure_channel("localhost:9090")
stub = AgentMessagingStub(channel)

outbound: queue.Queue[ConversationRequest] = queue.Queue()

def requests():
    while True:
        yield outbound.get()

for resp in stub.ProcessConversation(requests()):
    if resp.HasField("incoming_message"):
        m = resp.incoming_message
        reply = AgentResponse(
            conversation_id=m.conversation_id,
            content=ContentChunk(
                type=ContentChunk.END,
                content=f"you said: {m.content}",
            ),
        )
        outbound.put(ConversationRequest(agent_response=reply))
```

For inbound payloads, switch on `resp.WhichOneof("payload")` and read the matching field. Use `resp.HasField("incoming_message")` for individual checks.

**`AgentResponse` payload variants** (server → agent)

| `WhichOneof("payload")` | Type                | Notes                               |
| ----------------------- | ------------------- | ----------------------------------- |
| `incoming_message`      | `Message`           | The inbound message itself.         |
| `feedback`              | `PlatformFeedback`  | User feedback event.                |
| `audio_config`          | `AudioStreamConfig` | Audio session start.                |
| `audio_chunk`           | `AudioChunk`        | Raw audio bytes.                    |
| `status`                | `StatusUpdate`      | Relayed echoes (rare).              |
| `content`               | `ContentChunk`      | Relayed echoes (rare).              |
| ...                     | ...                 | Any of the outbound variants below. |

**Outbound (`ConversationRequest`) variants** (agent → sidecar)

| Field            | Type                | Notes                                |
| ---------------- | ------------------- | ------------------------------------ |
| `message`        | `Message`           | Forward an agent-originated message. |
| `feedback`       | `PlatformFeedback`  | Relay or fabricate feedback.         |
| `agent_config`   | `AgentConfig`       | Announce capabilities on startup.    |
| `agent_response` | `AgentResponse`     | **Main path.** Any agent response.   |
| `audio_config`   | `AudioStreamConfig` | Upstream audio session start.        |
| `audio`          | `AudioChunk`        | Upstream audio bytes.                |

## Next steps

* [Inbound message anatomy](/messaging-sdk/messages): the shape of what arrives
* [Sending a response](/messaging-sdk/responses): every response variant
* [Reconnection](/messaging-sdk/reconnection): surviving a dropped stream