Struct MessageStream

pub struct MessageStream { /* private fields */ }

A streaming response from the Anthropic API.

MessageStream provides an event-driven interface for processing streaming responses from Claude. It accumulates message content from incremental updates and provides both callback-based and async iteration APIs.

Examples

Callback-based processing:

let client = Anthropic::new("your-api-key")?;
let stream = client.messages().create_stream(
    MessageCreateBuilder::new("claude-3-5-sonnet-latest", 1024)
        .user("Write a story about AI")
        .stream(true)
        .build()
).await?;

let final_message = stream
    .on_text(|delta, _snapshot| {
        print!("{}", delta);
    })
    .on_error(|error| {
        eprintln!("Stream error: {}", error);
    })
    .final_message().await?;

Async iteration:

let client = Anthropic::new("your-api-key")?;
let mut stream = client.messages().create_stream(
    MessageCreateBuilder::new("claude-3-5-sonnet-latest", 1024)
        .user("Tell me a joke")
        .stream(true)
        .build()
).await?;

while let Some(event) = stream.next().await {
    match event? {
        MessageStreamEvent::ContentBlockDelta { delta, .. } => {
            // Process incremental content
        }
        MessageStreamEvent::MessageStop => break,
        _ => {}
    }
}

Implementations

impl MessageStream

pub fn new(response: Response, request_id: Option<String>) -> Self

Create a new MessageStream from an HTTP response.

This is typically called internally by the SDK when creating streaming requests.

pub fn from_http_stream(http_stream: HttpStreamClient) -> Result<Self>

Create a new MessageStream from an HttpStreamClient.

This connects a real HTTP stream to the MessageStream, providing proper streaming functionality for real-time response processing.

pub fn on_text<F>(self, callback: F) -> Self

where F: Fn(&str, &str) + Send + Sync + 'static,

Register a callback for text delta events.

The callback receives two parameters:

• delta: The new text being appended
• snapshot: The current accumulated text

Examples

stream.on_text(|delta, snapshot| {
    print!("{}", delta);
    println!("Total so far: {}", snapshot);
});

pub fn on_stream_event<F>(self, callback: F) -> Self

where F: Fn(&MessageStreamEvent, &Message) + Send + Sync + 'static,

Register a callback for stream events.

This provides access to all raw stream events and the current message snapshot.

Examples

stream.on_stream_event(|event, snapshot| {
    match event {
        MessageStreamEvent::ContentBlockStart { .. } => {
            println!("New content block started");
        }
        _ => {}
    }
});

pub fn on_message<F>(self, callback: F) -> Self

where F: Fn(&Message) + Send + Sync + 'static,

Register a callback for when a complete message is received.

Examples

stream.on_message(|message| {
    println!("Received message: {:?}", message);
});

pub fn on_final_message<F>(self, callback: F) -> Self

where F: Fn(&Message) + Send + Sync + 'static,

Register a callback for when the final message is complete.

Examples

stream.on_final_message(|message| {
    println!("Final message: {:?}", message);
});

pub fn on_error<F>(self, callback: F) -> Self

where F: Fn(&AnthropicError) + Send + Sync + 'static,

Register a callback for errors.

Examples

stream.on_error(|error| {
    eprintln!("Stream error: {}", error);
});

pub fn on_end<F>(self, callback: F) -> Self

where F: Fn() + Send + Sync + 'static,

Register a callback for when the stream ends.

Examples

stream.on_end(|| {
    println!("Stream ended");
});

pub async fn final_message(self) -> Result<Message>

Wait for the stream to complete and return the final message.

This method will block until the stream ends and return the accumulated message.

Examples

let final_message = stream.final_message().await?;
println!("Claude said: {:?}", final_message.content);

pub async fn done(self) -> Result<()>

Wait for the stream to complete without returning the message.

This is useful when you’re processing events with callbacks and just need to wait for completion.

Examples

stream.on_text(|delta, _| print!("{}", delta))
    .done().await?;
println!("\nStream completed!");

pub fn current_message(&self) -> Option<Message>

Get the current accumulated message snapshot.

Returns None if the stream hasn’t started or no message has been received yet.

pub fn ended(&self) -> bool

Check if the stream has ended.

pub fn errored(&self) -> bool

Check if an error occurred.

pub fn aborted(&self) -> bool

Check if the stream was aborted.

pub fn response(&self) -> Option<&Response>

Get the response metadata.

pub fn request_id(&self) -> Option<&str>

Get the request ID.

pub fn abort(&self)

Abort the stream.

This will cancel the underlying HTTP request and mark the stream as aborted.

Trait Implementations

impl Stream for MessageStream

type Item = Result<MessageStreamEvent, AnthropicError>

Values yielded by the stream.

fn poll_next( self: Pin<&mut Self>, cx: &mut Context<'_>, ) -> Poll<Option<Self::Item>>

Attempt to pull out the next value of this stream, registering the current task for wakeup if the value is not yet available, and returning None if the stream is exhausted.

fn size_hint(&self) -> (usize, Option<usize>)

Returns the bounds on the remaining length of the stream.

impl<'pin> Unpin for MessageStream

where PinnedFieldsOf<__MessageStream<'pin>>: Unpin,