feat(bedrock): stream Converse tool arguments

API.md

@@ -332,6 +332,7 @@ composable integration value:
 | `streamfold/vercel-ai` | `vercelAI` |
 | `streamfold/openai` | `openAI` |
 | `streamfold/anthropic` | `anthropic` |
+| `streamfold/bedrock` | `bedrock` |
 | `streamfold/gemini` | `gemini` |
 | `streamfold/langchain` | `langchain` |
 | `streamfold/ag-ui` | `agUI` |
@@ -379,6 +380,14 @@ boundaries, `tool-input-end` continues to complete a call immediately.
 `tool-call` and UI `tool-input-available` events also work without preceding
 deltas. Repeated completion events do not emit duplicate results.
 
+For Amazon Bedrock, pass the `stream` returned by `ConverseStreamCommand` to
+`readStructured` with `integration: bedrock`. The adapter correlates tool-use
+blocks by `contentBlockIndex` and exposes `toolUseId` as the call ID. Text,
+reasoning, and metadata events are ignored. Calls complete at their block stop
+or EOF; malformed or truncated JSON fails and releases active parsers. Service
+error events also fail the integration. AWS SDK types are accepted directly,
+but the AWS SDK is not a runtime dependency of Streamfold.
+
 The Gemini adapter accepts the Interactions API's `arguments_delta` events
 and the older `arguments` / `partial_arguments` shape. The empty object at
 `step.start` is a placeholder while deltas arrive. A call with no argument

CHANGELOG.md

@@ -2,6 +2,7 @@
 
 ## Unreleased
 
+- Add a Bedrock ConverseStream adapter for streaming tool-use arguments.
 - Accept Anthropic stable and beta SDK message event interfaces directly in
   the Anthropic adapter and managed reader.
 - Accept OpenAI SDK response stream event types directly in the OpenAI adapter

README.md

@@ -79,6 +79,7 @@ Streamfold includes isolated adapters for decoded SDK events:
 | `streamfold/vercel-ai` | Vercel AI SDK |
 | `streamfold/openai` | OpenAI Responses |
 | `streamfold/anthropic` | Anthropic Messages |
+| `streamfold/bedrock` | Amazon Bedrock ConverseStream |
 | `streamfold/gemini` | Gemini Interactions |
 | `streamfold/langchain` | LangChain message chunks |
 | `streamfold/ag-ui` | AG-UI events |

examples/tsconfig.json

@@ -5,5 +5,5 @@
     "checkJs": true,
     "skipLibCheck": true
   },
-  "files": ["assistant-ui.mjs", "vercel-ai.mjs", "custom.mjs", "../tests/langchain-sdk.ts", "../tests/gemini-sdk.ts", "../tests/openai-sdk.ts", "../tests/anthropic-sdk.ts"]
+  "files": ["assistant-ui.mjs", "vercel-ai.mjs", "custom.mjs", "../tests/langchain-sdk.ts", "../tests/gemini-sdk.ts", "../tests/openai-sdk.ts", "../tests/anthropic-sdk.ts", "../tests/bedrock-sdk.ts"]
 }

package.json

@@ -32,6 +32,7 @@
   },
   "devDependencies": {
     "@anthropic-ai/sdk": "1.3.0",
+    "@aws-sdk/client-bedrock-runtime": "3.1159.0",
     "@google/genai": "2.31.0",
     "@langchain/core": "1.3.0",
     "ai": "7.0.152",

packages/core/package.json

@@ -28,6 +28,10 @@
       "types": "./src/anthropic.d.ts",
       "default": "./src/anthropic.js"
     },
+    "./bedrock": {
+      "types": "./src/bedrock.d.ts",
+      "default": "./src/bedrock.js"
+    },
     "./assistant-ui": {
       "types": "./src/assistant-ui.d.ts",
       "default": "./src/assistant-ui.js"

packages/core/src/bedrock.d.ts

@@ -0,0 +1,31 @@
+import type {
+  BatchEventStructuredStream,
+  BatchStructuredStreamIntegration,
+  StructuredStreamIntegrationOptions,
+  StructuredStreamPool,
+} from "./index.js";
+
+export type BedrockToolCallEvent =
+  | {
+      readonly contentBlockStart?: {
+        readonly contentBlockIndex?: number | undefined;
+        readonly start?: {
+          readonly toolUse?: { readonly toolUseId?: string | undefined } | undefined;
+        } | undefined;
+      } | undefined;
+      readonly contentBlockDelta?: {
+        readonly contentBlockIndex?: number | undefined;
+        readonly delta?: {
+          readonly toolUse?: { readonly input?: string | undefined } | undefined;
+        } | undefined;
+      } | undefined;
+      readonly contentBlockStop?: { readonly contentBlockIndex?: number | undefined } | undefined;
+    }
+  | { readonly [key: string]: unknown };
+
+export const bedrock: BatchStructuredStreamIntegration<BedrockToolCallEvent, string>;
+
+export function createStructuredStream(
+  pool?: StructuredStreamPool<string>,
+  options?: StructuredStreamIntegrationOptions,
+): BatchEventStructuredStream<BedrockToolCallEvent, string>;

packages/core/src/bedrock.js

@@ -0,0 +1,51 @@
+import { createStructuredStreamPool } from "./index.js";
+import { createIntegration } from "./internal/integration.js";
+
+export const bedrock = (pool = createStructuredStreamPool(), options) => {
+  const idsByIndex = new Map();
+  return createIntegration(
+    pool,
+    (event, calls) => {
+      const failure = event.internalServerException ?? event.modelStreamErrorException ??
+        event.validationException ?? event.throttlingException ?? event.serviceUnavailableException;
+      if (failure) throw failure instanceof Error ? failure : new Error(failure.message ?? "Bedrock stream failed");
+
+      const start = event.contentBlockStart;
+      if (start?.start?.toolUse) {
+        if (start.contentBlockIndex === undefined) {
+          calls.unmatched("Tool block start has no content block index");
+          return undefined;
+        }
+        const id = start.start.toolUse.toolUseId;
+        if (id !== undefined) idsByIndex.set(start.contentBlockIndex, id);
+        return calls.start(id);
+      }
+      const delta = event.contentBlockDelta;
+      if (delta?.delta?.toolUse) {
+        const id = idsByIndex.get(delta.contentBlockIndex);
+        if (id === undefined) {
+          calls.unmatched("Arguments delta has no matching tool block index");
+          return undefined;
+        }
+        return calls.push(id, delta.delta.toolUse.input ?? "");
+      }
+      const stop = event.contentBlockStop;
+      if (stop) {
+        const id = idsByIndex.get(stop.contentBlockIndex);
+        if (id !== undefined) {
+          idsByIndex.delete(stop.contentBlockIndex);
+          return calls.complete(id);
+        }
+      }
+      return undefined;
+    },
+    (complete) => {
+      for (const id of idsByIndex.values()) complete(id);
+      idsByIndex.clear();
+    },
+    "bedrock",
+    options,
+  );
+};
+
+export { bedrock as createStructuredStream };

packages/core/src/bedrock.test.js

@@ -0,0 +1,60 @@
+import assert from "node:assert/strict";
+import test from "node:test";
+import { createStructuredStreamPool, readStructured } from "./index.js";
+import { bedrock } from "./bedrock.js";
+
+const start = (index, id) => ({ contentBlockStart: { contentBlockIndex: index, start: { toolUse: { toolUseId: id, name: "weather" } } } });
+const delta = (index, input) => ({ contentBlockDelta: { contentBlockIndex: index, delta: { toolUse: { input } } } });
+const stop = (index) => ({ contentBlockStop: { contentBlockIndex: index } });
+
+test("Bedrock correlates interleaved tool blocks and ignores other content", async () => {
+  const events = [
+    { messageStart: { role: "assistant" } },
+    { contentBlockDelta: { contentBlockIndex: 0, delta: { text: "Checking weather" } } },
+    stop(0), start(1, "paris"), start(2, "tokyo"),
+    delta(1, '{"city":"Pa'), delta(2, '{"city":"Tokyo"}'), delta(1, 'ris"}'),
+    stop(2), stop(1), stop(1), { messageStop: { stopReason: "tool_use" } },
+    { metadata: { usage: { inputTokens: 10, outputTokens: 20 } } },
+  ];
+  const updates = [];
+  for await (const update of readStructured(events, { integration: bedrock, limits: { snapshots: "immutable" } })) updates.push(update);
+  assert.deepEqual(updates.map(({ type, id }) => [type, id]), [
+    ["start", "paris"], ["start", "tokyo"], ["update", "paris"], ["update", "tokyo"],
+    ["update", "paris"], ["complete", "tokyo"], ["complete", "paris"],
+  ]);
+  assert.deepEqual(updates[2].partialValue, { city: "Pa" });
+  assert.deepEqual(updates.slice(-2).map(({ value }) => value), [{ city: "Tokyo" }, { city: "Paris" }]);
+});
+
+test("Bedrock finalizes complete pending inputs at EOF and preserves manual history", () => {
+  const stream = bedrock();
+  stream.push(start(0, "empty"));
+  stream.push(delta(0, "{}"));
+  assert.deepEqual(stream.finish().map(({ id, value }) => ({ id, value })), [{ id: "empty", value: {} }]);
+  assert.equal(stream.finish().length, 1);
+});
+
+test("Bedrock rejects truncation, parser limits, and service errors without retaining active calls", () => {
+  for (const [options, input, ending, message] of [
+    [{}, '{"city":', stop(0), /complete|incomplete/i],
+    [{ maxBytes: 4 }, '{"city":"Paris"}', stop(0), /maxBytes/],
+    [{}, '{"city":"Paris"}', { modelStreamErrorException: { message: "model failed" } }, /model failed/],
+  ]) {
+    const pool = createStructuredStreamPool(options);
+    const stream = bedrock(pool);
+    stream.push(start(0, "first"));
+    stream.push(start(1, "second"));
+    assert.throws(() => { stream.push(delta(0, input)); stream.push(ending); }, message);
+    assert.equal(pool.size, 0);
+    assert.throws(() => stream.finish(), message);
+  }
+});
+
+test("Bedrock diagnoses uncorrelated deltas and missing block indexes", () => {
+  const diagnostics = [];
+  const stream = bedrock(undefined, { onDiagnostic: (diagnostic) => diagnostics.push(diagnostic) });
+  stream.push({ contentBlockStart: { start: { toolUse: { toolUseId: "missing-index" } } } });
+  stream.push(delta(8, "{}"));
+  assert.deepEqual(stream.finish(), []);
+  assert.equal(diagnostics.filter(({ code }) => code === "UNMATCHED_TOOL_EVENT").length, 2);
+});