feat(ai): add bedrock messages transport

packages/ai/src/cache-policy.ts

@@ -40,6 +40,7 @@ const RESPECTS_INLINE_HINTS = new Set([
   "anthropic-messages",
   "google-vertex-messages",
   "bedrock-converse",
+  "bedrock-messages",
   "openrouter",
 ])
 

packages/ai/src/protocols/bedrock-messages.ts

@@ -0,0 +1,91 @@
+import { Effect, Encoding, Schema, Struct } from "effect"
+import { Headers } from "effect/unstable/http"
+import { AIError } from "../schema/index.js"
+import { Route } from "../route/client.js"
+import { Endpoint } from "../route/endpoint.js"
+import { Protocol } from "../route/protocol.js"
+import { classifyProviderFailure } from "../provider-error.js"
+import { AnthropicMessages } from "./anthropic-messages.js"
+import { BedrockEventStream } from "./bedrock-event-stream.js"
+import { BedrockAuth } from "./utils/bedrock-auth.js"
+import { JsonObject, ProviderShared } from "./shared.js"
+
+const ID = "bedrock-messages"
+const VERSION = "bedrock-2023-05-31"
+const Body = Schema.Struct({
+  ...Struct.omit(AnthropicMessages.AnthropicMessagesBody.fields, ["model", "stream"]),
+  anthropic_version: Schema.Literal(VERSION),
+  anthropic_beta: Schema.optional(Schema.Array(Schema.String)),
+})
+const Event = Schema.Struct({
+  chunk: Schema.optional(Schema.Struct({ bytes: Schema.String })),
+  exception: Schema.optional(
+    Schema.Struct({
+      type: Schema.String,
+      details: Schema.StructWithRest(
+        Schema.Struct({ message: Schema.optional(Schema.String), originalMessage: Schema.optional(Schema.String) }),
+        [JsonObject],
+      ),
+    }),
+  ),
+})
+
+export const protocol = Protocol.make({
+  id: ID,
+  body: {
+    schema: Body,
+    from: Effect.fn("BedrockMessages.fromRequest")(function* (request) {
+      const body = yield* AnthropicMessages.protocol.body.from(request)
+      const headers = Headers.fromInput(request.http?.headers)
+      const betas = new Set(
+        (headers["anthropic-beta"] ?? "")
+          .split(",")
+          .map((value) => value.trim())
+          .filter(Boolean),
+      )
+      return {
+        ...Struct.omit(body, ["model", "stream"]),
+        anthropic_version: VERSION,
+        anthropic_beta: betas.size ? [...betas] : undefined,
+      } satisfies typeof Body.Type
+    }),
+  },
+  stream: {
+    event: Event,
+    initial: AnthropicMessages.protocol.stream.initial,
+    step: Effect.fn("BedrockMessages.step")(function* (state, event) {
+      if (event.exception)
+        return yield* new AIError({
+          reason: classifyProviderFailure({
+            message: event.exception.details.message ?? event.exception.details.originalMessage ?? event.exception.type,
+            rawBody: ProviderShared.encodeJson(event),
+          }),
+        })
+      if (!event.chunk) return yield* ProviderShared.eventError(ID, "Bedrock Messages event is missing its chunk")
+      const text = yield* Effect.fromResult(Encoding.decodeBase64String(event.chunk.bytes)).pipe(
+        Effect.mapError((cause) =>
+          ProviderShared.eventError(ID, "Invalid Bedrock Messages chunk encoding", undefined, cause),
+        ),
+      )
+      const decoded = yield* Schema.decodeUnknownEffect(AnthropicMessages.protocol.stream.event)(text).pipe(
+        Effect.mapError((cause) => ProviderShared.eventError(ID, "Invalid Bedrock Messages event", undefined, cause)),
+      )
+      return yield* AnthropicMessages.protocol.stream.step(state, decoded)
+    }),
+  },
+})
+
+export const route = Route.make({
+  id: ID,
+  provider: "amazon-bedrock",
+  providerMetadataKey: "anthropic",
+  protocol,
+  endpoint: Endpoint.path(
+    ({ request }) => `/model/${encodeURIComponent(request.model.id)}/invoke-with-response-stream`,
+    { baseURL: "https://bedrock-runtime.us-east-1.amazonaws.com" },
+  ),
+  auth: BedrockAuth.auth,
+  framing: BedrockEventStream.framing(ID),
+})
+
+export * as BedrockMessages from "./bedrock-messages.js"

packages/ai/src/protocols/index.ts

@@ -1,5 +1,6 @@
 export * as AnthropicMessages from "./anthropic-messages.js"
 export * as BedrockConverse from "./bedrock-converse.js"
+export { BedrockMessages } from "./bedrock-messages.js"
 export * as Gemini from "./gemini.js"
 export * as MistralChat from "./mistral-chat.js"
 export * as OpenAIChat from "./openai-chat.js"

packages/ai/src/providers/amazon-bedrock.ts

@@ -1,9 +1,11 @@
-import type { RouteDefaultsInput } from "../route/client.js"
+import type { Route, RouteDefaultsInput } from "../route/client.js"
 import { Auth } from "../route/auth.js"
 import type { ProviderPackage } from "../provider-package.js"
 import { ProviderID, type ModelID } from "../schema/index.js"
 import * as BedrockConverse from "../protocols/bedrock-converse.js"
 import type { BedrockCredentials } from "../protocols/bedrock-converse.js"
+import { BedrockMessages } from "../protocols/bedrock-messages.js"
+import type { AnthropicMessages } from "../protocols/anthropic-messages.js"
 
 export const id = ProviderID.make("amazon-bedrock")
 
@@ -25,38 +27,40 @@ export interface Settings extends ProviderPackage.Settings {
   readonly region?: string
   readonly topP?: number
 }
-export const routes = [BedrockConverse.route]
+export const routes = [BedrockConverse.route, BedrockMessages.route]
 
 const bedrockBaseURL = (region: string) => `https://bedrock-runtime.${region}.amazonaws.com`
 
-const configuredRoute = (input: Config) => {
+const configuredRoute = <Body, Prepared>(route: Route<Body, Prepared>, input: Config) => {
   const { apiKey, credentials, region, baseURL, ...rest } = input
   const resolvedRegion = region ?? credentials?.region ?? "us-east-1"
-  return BedrockConverse.route.with({
+  return route.with({
     ...rest,
     provider: id,
-    providerMetadataKey: "bedrock",
+    providerMetadataKey: route.providerMetadataKey,
     endpoint: { baseURL: baseURL ?? bedrockBaseURL(resolvedRegion) },
     auth: apiKey === undefined ? BedrockConverse.sigV4Auth(credentials) : Auth.bearer(apiKey),
   })
 }
 
 export const configure = (input: Config = {}) => {
-  const route = configuredRoute(input)
+  const route = configuredRoute(BedrockConverse.route, input)
+  const messages = configuredRoute(BedrockMessages.route, input)
   return {
     id,
     model: (modelID: string | ModelID) => route.model({ id: modelID }),
+    messages: (modelID: string | ModelID) => messages.model<AnthropicMessages.ProviderOptionsInput>({ id: modelID }),
     configure,
   }
 }
 
 export const provider = configure()
-export const model: ProviderPackage.Definition<Settings>["model"] = (modelID, settings) => {
+const config = (settings: Settings): Config => {
   if (settings.auth === "bearer" && settings.apiKey === undefined)
     throw new Error("Amazon Bedrock bearer auth requires apiKey")
   if (settings.auth === "sigv4" && settings.apiKey !== undefined)
     throw new Error("Amazon Bedrock SigV4 auth does not accept apiKey")
-  return configure({
+  return {
     apiKey: settings.auth === "sigv4" ? undefined : settings.apiKey,
     baseURL: settings.baseURL,
     credentials: settings.credentials,
@@ -64,5 +68,13 @@ export const model: ProviderPackage.Definition<Settings>["model"] = (modelID, se
     headers: settings.headers === undefined ? undefined : { ...settings.headers },
     http: settings.body === undefined ? undefined : { body: { ...settings.body } },
     region: settings.region,
-  }).model(modelID)
+  }
 }
+
+export const model: ProviderPackage.Definition<Settings>["model"] = (modelID, settings) =>
+  configure(config(settings)).model(modelID)
+export const messagesModel: ProviderPackage.Definition<
+  Settings & { readonly providerOptions?: AnthropicMessages.ProviderOptionsInput },
+  AnthropicMessages.ProviderOptionsInput
+>["model"] = (modelID, settings) =>
+  configure({ ...config(settings), providerOptions: settings.providerOptions }).messages(modelID)

packages/ai/src/providers/amazon-bedrock/messages.ts

@@ -0,0 +1 @@
+export { messagesModel as model } from "../amazon-bedrock.js"

packages/ai/test/cache-policy.test.ts

@@ -126,6 +126,38 @@ describe("applyCachePolicy", () => {
     }),
   )
 
+  for (const fixture of [
+    { name: "default", cache: undefined, control: { type: "ephemeral" } },
+    { name: "auto", cache: "auto", control: { type: "ephemeral" } },
+    {
+      name: "explicit one-hour",
+      cache: { tools: true, system: true, messages: { tail: 1 }, ttlSeconds: 3600 },
+      control: { type: "ephemeral", ttl: "1h" },
+    },
+    { name: "disabled", cache: "none", control: undefined },
+  ] as const) {
+    it.effect(`Bedrock Messages respects ${fixture.name} caching`, () =>
+      Effect.gen(function* () {
+        const prepared = yield* compileRequest(
+          LLM.request({
+            model: AmazonBedrock.configure({ apiKey: "test" }).messages("anthropic.claude-opus-4-6-v1"),
+            system: "Stable instructions",
+            tools: [
+              { name: "lookup", description: "Look up a value", inputSchema: { type: "object", properties: {} } },
+            ],
+            prompt: "hello",
+            cache: fixture.cache,
+          }),
+        )
+        expect(prepared.body).toMatchObject({
+          tools: [{ name: "lookup", cache_control: fixture.control }],
+          system: [{ type: "text", text: "Stable instructions", cache_control: fixture.control }],
+          messages: [{ role: "user", content: [{ type: "text", text: "hello", cache_control: fixture.control }] }],
+        })
+      }),
+    )
+  }
+
   it.effect("'auto' is a no-op on Gemini (out-of-band caching protocol)", () =>
     Effect.gen(function* () {
       const prepared = yield* compileRequest(

packages/ai/test/provider/bedrock-messages-basic.test.ts

@@ -0,0 +1,93 @@
+import { EventStreamCodec } from "@smithy/eventstream-codec"
+import { fromUtf8, toUtf8 } from "@smithy/util-utf8"
+import { expect } from "bun:test"
+import { Effect } from "effect"
+import { LLM, LLMRequest, Message } from "../../src/index.js"
+import { LLMClient } from "../../src/route/client.js"
+import { AmazonBedrock } from "../../src/providers/index.js"
+import { testEffect } from "../lib/effect.js"
+import { dynamicResponse, fixedResponse } from "../lib/http.js"
+
+const codec = new EventStreamCodec(toUtf8, fromUtf8)
+const response = Buffer.concat(
+  [
+    { type: "message_start", message: { usage: { input_tokens: 10 } } },
+    { type: "content_block_start", index: 0, content_block: { type: "text", text: "" } },
+    { type: "content_block_delta", index: 0, delta: { type: "text_delta", text: "Hello" } },
+    { type: "content_block_stop", index: 0 },
+    { type: "message_delta", delta: { stop_reason: "end_turn" }, usage: { output_tokens: 2 } },
+    { type: "message_stop" },
+  ].map((event) =>
+    codec.encode({
+      headers: {
+        ":message-type": { type: "string", value: "event" },
+        ":event-type": { type: "string", value: "chunk" },
+      },
+      body: new TextEncoder().encode(JSON.stringify({ bytes: Buffer.from(JSON.stringify(event)).toString("base64") })),
+    }),
+  ),
+)
+
+for (const auth of [
+  { apiKey: "test" },
+  { credentials: { accessKeyId: "test", secretAccessKey: "test", region: "us-west-2" } },
+]) {
+  testEffect(
+    dynamicResponse(({ request, text, respond }) =>
+      Effect.sync(() => {
+        expect(request.url).toBe(
+          "https://bedrock-runtime.us-west-2.amazonaws.com/model/anthropic.claude-opus-4-6-v1%3A0/invoke-with-response-stream",
+        )
+        expect(request.headers.authorization).toStartWith(auth.apiKey ? "Bearer test" : "AWS4-HMAC-SHA256")
+        const body = JSON.parse(text)
+        expect(body.model).toBeUndefined()
+        expect(body.stream).toBeUndefined()
+        expect(body.anthropic_version).toBe("bedrock-2023-05-31")
+        expect(body.anthropic_beta).toEqual(["existing-beta"])
+        if (body.messages.length > 1) expect(body.messages[1].content).toEqual([{ type: "text", text: "Hello" }])
+        return respond(response, { headers: { "content-type": "application/vnd.amazon.eventstream" } })
+      }),
+    ),
+  ).effect(`Bedrock Messages text round trip with ${auth.apiKey ? "bearer" : "SigV4"} authentication`, () =>
+    Effect.gen(function* () {
+      const provider = AmazonBedrock.configure({ ...auth, region: "us-west-2" })
+      expect(provider.model("fixture").route.id).toBe("bedrock-converse")
+      const request = LLM.request({
+        model: provider.messages("anthropic.claude-opus-4-6-v1:0"),
+        prompt: "hello",
+        http: { headers: { "anthropic-beta": "existing-beta, existing-beta" } },
+      })
+      const first = yield* LLMClient.generate(request)
+      expect(first.text).toBe("Hello")
+      expect(first.usage?.totalTokens).toBe(12)
+      yield* LLMClient.generate(
+        LLMRequest.update(request, {
+          messages: [...request.messages, first.message, Message.user("continue")],
+        }),
+      )
+    }),
+  )
+}
+
+testEffect(
+  fixedResponse(
+    codec.encode({
+      headers: {
+        ":message-type": { type: "string", value: "exception" },
+        ":exception-type": { type: "string", value: "throttlingException" },
+      },
+      body: new TextEncoder().encode(JSON.stringify({ message: "Too many requests", trace: "keep-original" })),
+    }),
+  ),
+).effect("Bedrock Messages retains the original exception frame", () =>
+  Effect.gen(function* () {
+    const error = yield* LLMClient.generate(
+      LLM.request({
+        model: AmazonBedrock.configure({ apiKey: "test" }).messages("claude"),
+        prompt: "hello",
+      }),
+    ).pipe(Effect.flip)
+    expect(error.reason.body).toContain("keep-original")
+    expect(error.reason.http?.status).toBe(200)
+  }),
+)