feat(api): add messages endpoint with streaming helpers (#235)

More information here: https://docs.anthropic.com/claude/reference/messages_post

.stats.yml

@@ -1 +1 @@
-configured_endpoints: 1
+configured_endpoints: 2

api.md

@@ -9,3 +9,27 @@ Types:
 Methods:
 
 - <code title="post /v1/complete">client.completions.<a href="./src/resources/completions.ts">create</a>({ ...params }) -> Completion</code>
+
+# Beta
+
+## Messages
+
+Types:
+
+- <code><a href="./src/resources/beta/messages.ts">ContentBlock</a></code>
+- <code><a href="./src/resources/beta/messages.ts">ContentBlockDeltaEvent</a></code>
+- <code><a href="./src/resources/beta/messages.ts">ContentBlockStartEvent</a></code>
+- <code><a href="./src/resources/beta/messages.ts">ContentBlockStopEvent</a></code>
+- <code><a href="./src/resources/beta/messages.ts">Message</a></code>
+- <code><a href="./src/resources/beta/messages.ts">MessageDeltaEvent</a></code>
+- <code><a href="./src/resources/beta/messages.ts">MessageParam</a></code>
+- <code><a href="./src/resources/beta/messages.ts">MessageStartEvent</a></code>
+- <code><a href="./src/resources/beta/messages.ts">MessageStopEvent</a></code>
+- <code><a href="./src/resources/beta/messages.ts">MessageStreamEvent</a></code>
+- <code><a href="./src/resources/beta/messages.ts">TextBlock</a></code>
+- <code><a href="./src/resources/beta/messages.ts">TextDelta</a></code>
+
+Methods:
+
+- <code title="post /v1/messages">client.beta.messages.<a href="./src/resources/beta/messages.ts">create</a>({ ...params }) -> Message</code>
+- <code>client.beta.messages.<a href="./src/resources/beta/messages.ts">stream</a>(body, options?) -> MessageStream</code>

examples/raw-streaming.ts

@@ -0,0 +1,25 @@
+#!/usr/bin/env -S npm run tsn -T
+
+import Anthropic from '@anthropic-ai/sdk';
+
+const client = new Anthropic(); // gets API Key from environment variable ANTHROPIC_API_KEY
+
+async function main() {
+  const question = 'Hey Claude! How can I recursively list all files in a directory in Rust?';
+
+  const stream = await client.completions.create({
+    prompt: `${Anthropic.HUMAN_PROMPT}${question}${Anthropic.AI_PROMPT}:`,
+    model: 'claude-2.1',
+    stream: true,
+    max_tokens_to_sample: 500,
+  });
+
+  for await (const completion of stream) {
+    process.stdout.write(completion.completion);
+  }
+}
+
+main().catch((err) => {
+  console.error(err);
+  process.exit(1);
+});

examples/streaming.ts

@@ -5,18 +5,28 @@ import Anthropic from '@anthropic-ai/sdk';
 const client = new Anthropic(); // gets API Key from environment variable ANTHROPIC_API_KEY
 
 async function main() {
-  const question = 'Hey Claude! How can I recursively list all files in a directory in Rust?';
+  const stream = client.beta.messages
+    .stream({
+      messages: [
+        {
+          role: 'user',
+          content: `Hey Claude! How can I recursively list all files in a directory in Rust?`,
+        },
+      ],
+      model: 'claude-2.1',
+      max_tokens: 1024,
+    })
+    // Once a content block is fully streamed, this event will fire
+    .on('contentBlock', (content) => console.log('contentBlock', content))
+    // Once a message is fully streamed, this event will fire
+    .on('message', (message) => console.log('message', message));
 
-  const stream = await client.completions.create({
-    prompt: `${Anthropic.HUMAN_PROMPT}${question}${Anthropic.AI_PROMPT}:`,
-    model: 'claude-2.1',
-    stream: true,
-    max_tokens_to_sample: 500,
-  });
-
-  for await (const completion of stream) {
-    process.stdout.write(completion.completion);
+  for await (const event of stream) {
+    console.log('event', event);
   }
+
+  const message = await stream.finalMessage();
+  console.log('finalMessage', message);
 }
 
 main().catch((err) => {

helpers.md

@@ -0,0 +1,96 @@
+# Chat Completion Helpers
+
+## Streaming Responses
+
+```ts
+anthropic.beta.messages.stream({ … }, options?): MessageStream
+```
+
+`anthropic.beta.messages.stream()` returns a `MessageStream`, which emits events, has an async
+iterator, and exposes helper methods to accumulate stream events into a convenient shape and make it easy to reason
+about the conversation.
+
+Alternatively, you can use `anthropic.beta.messages.create({ stream: true, … })` which returns an async
+iterable of the chunks in the stream and uses less memory (most notably, it does not accumulate a message
+object for you).
+
+If you need to cancel a stream, you can `break` from a `for await` loop or call `stream.abort()`.
+
+See an example of streaming helpers in action in [`examples/streaming.ts`](examples/streaming.ts).
+
+## MessageStream API
+
+### Events
+
+#### `.on('connect', () => …)`
+
+The first event that is fired when the connection with the Anthropic API is established.
+
+#### `.on('streamEvent', (event: MessageStreamEvent, snapshot: Message) => …)`
+
+The event fired when a stream event is received from the API. Not fired when it is not streaming. The snapshot
+returns an accumulated `Message` which is progressively built-up over events.
+
+#### `.on('text', (textDelta: string, textSnapshot: string) => …)`
+
+The event fired when a text delta is sent by the API. The second parameter returns a `textSnapshot`.
+
+#### `.on('message', (message: Message) => …)`
+
+The event fired when a message is done being streamed by the API. Corresponds to the `message_stop` SSE event.
+
+#### `.on('contentBlock', (content: ContentBlock) => …)`
+
+The event fired when a content block is done being streamed by the API. Corresponds to the
+`content_block_stop` SSE event.
+
+#### `.on('finalMessage', (message: Message) => …)`
+
+The event fired for the final message. Currently this is equivalent to the `message` event, but is fired after
+it.
+
+#### `.on('error', (error: AnthropicError) => …)`
+
+The event fired when an error is encountered while streaming.
+
+#### `.on('abort', (error: APIUserAbortError) => …)`
+
+The event fired when the stream receives a signal to abort.
+
+#### `.on('end', () => …)`
+
+The last event fired in the stream.
+
+### Methods
+
+#### `.abort()`
+
+Aborts the runner and the streaming request, equivalent to `.controller.abort()`. Calling `.abort()` on a
+`MessageStream` will also abort any in-flight network requests.
+
+#### `await .done()`
+
+An empty promise which resolves when the stream is done.
+
+#### `.currentMessage`
+
+Returns the current state of the message that is being accumulated, or `undefined` if there is no such
+message.
+
+#### `await .finalMessage()`
+
+A promise which resolves with the last message received from the API. Throws if no such message exists.
+
+#### `await .finalText()`
+
+A promise which resolves with the text of the last message received from the API.
+
+### Fields
+
+#### `.messages`
+
+A mutable array of all messages in the conversation.
+
+#### `.controller`
+
+The underlying `AbortController` for the runner.

src/index.ts

@@ -121,6 +121,7 @@ export class Anthropic extends Core.APIClient {
   }
 
   completions: API.Completions = new API.Completions(this);
+  beta: API.Beta = new API.Beta(this);
 
   protected override defaultQuery(): Core.DefaultQuery | undefined {
     return this._options.defaultQuery;
@@ -234,6 +235,8 @@ export namespace Anthropic {
   export import CompletionCreateParams = API.CompletionCreateParams;
   export import CompletionCreateParamsNonStreaming = API.CompletionCreateParamsNonStreaming;
   export import CompletionCreateParamsStreaming = API.CompletionCreateParamsStreaming;
+
+  export import Beta = API.Beta;
 }
 
 export default Anthropic;