fix(streaming): correct error message serialisation (#524)

fixes #346
anthropics · Sep 9, 2024 · 45b0018 · 45b0018
1 parent 02f0fc5
commit 45b0018
Show file tree

Hide file tree

Showing 3 changed files with 33 additions and 9 deletions.
diff --git a/src/error.ts b/src/error.ts
@@ -52,7 +52,7 @@ export class APIError extends AnthropicError {
     headers: Headers | undefined,
   ) {
     if (!status) {
-      return new APIConnectionError({ cause: castToError(errorResponse) });
+      return new APIConnectionError({ message, cause: castToError(errorResponse) });
     }
 
     const error = errorResponse as Record<string, any>;
@@ -104,7 +104,7 @@ export class APIUserAbortError extends APIError {
 export class APIConnectionError extends APIError {
   override readonly status: undefined = undefined;
 
-  constructor({ message, cause }: { message?: string; cause?: Error | undefined }) {
+  constructor({ message, cause }: { message?: string | undefined; cause?: Error | undefined }) {
     super(undefined, undefined, message || 'Connection error.', undefined);
     // in some environments the 'cause' property is already declared
     // @ts-ignore

diff --git a/src/streaming.ts b/src/streaming.ts
@@ -1,7 +1,7 @@
 import { ReadableStream, type Response } from './_shims/index';
 import { AnthropicError } from './error';
 
-import { safeJSON, createResponseHeaders } from '@anthropic-ai/sdk/core';
+import { createResponseHeaders } from '@anthropic-ai/sdk/core';
 import { APIError } from '@anthropic-ai/sdk/error';
 
 type Bytes = string | ArrayBuffer | Uint8Array | Buffer | null | undefined;
@@ -65,11 +65,12 @@ export class Stream<Item> implements AsyncIterable<Item> {
           }
 
           if (sse.event === 'error') {
-            const errText = sse.data;
-            const errJSON = safeJSON(errText);
-            const errMessage = errJSON ? undefined : errText;
-
-            throw APIError.generate(undefined, errJSON, errMessage, createResponseHeaders(response.headers));
+            throw APIError.generate(
+              undefined,
+              `SSE Error: ${sse.data}`,
+              sse.data,
+              createResponseHeaders(response.headers),
+            );
           }
         }
         done = true;

diff --git a/tests/streaming.test.ts b/tests/streaming.test.ts
@@ -1,7 +1,8 @@
 import { Response } from 'node-fetch';
 import { PassThrough } from 'stream';
 import assert from 'assert';
-import { _iterSSEMessages, _decodeChunks as decodeChunks } from '@anthropic-ai/sdk/streaming';
+import { Stream, _iterSSEMessages, _decodeChunks as decodeChunks } from '@anthropic-ai/sdk/streaming';
+import { APIConnectionError } from '@anthropic-ai/sdk/error';
 
 describe('line decoder', () => {
   test('basic', () => {
@@ -247,6 +248,28 @@ describe('streaming decoding', () => {
   });
 });
 
+test('error handling', async () => {
+  async function* body(): AsyncGenerator<Buffer> {
+    yield Buffer.from('event: error\n');
+    yield Buffer.from('data: {"type":"error","error":{"type":"overloaded_error","message":"Overloaded"}}');
+    yield Buffer.from('\n\n');
+  }
+
+  const stream = Stream.fromSSEResponse(new Response(await iteratorToStream(body())), new AbortController());
+
+  const err = expect(
+    (async () => {
+      for await (const _event of stream) {
+      }
+    })(),
+  ).rejects;
+
+  await err.toMatchInlineSnapshot(
+    `[Error: {"type":"error","error":{"type":"overloaded_error","message":"Overloaded"}}]`,
+  );
+  await err.toBeInstanceOf(APIConnectionError);
+});
+
 async function iteratorToStream(iterator: AsyncGenerator<any>): Promise<PassThrough> {
   const parts: unknown[] = [];