feat(anthropic): add support for Genkit v2 streaming API

- Update streaming callback interface to use v2 API format with streamingRequested, sendChunk, and abortSignal
- Add apiVersion: 'v2' to model definition for compatibility
- Pass AbortSignal to Anthropic SDK client calls for proper request cancellation
- Update tests to reflect the new API changes
- Fix code indentation for better consistency

BREAKING CHANGE: This updates the plugin to work with Genkit v2 API which changes the streaming callback interface

Author: longfin

package-lock.json

@@ -26,7 +26,7 @@
   "author": "TheFireCo",
   "license": "Apache-2.0",
   "peerDependencies": {
-    "genkit": "^0.9.0 || ^1.0.0"
+    "genkit": "^1.15.0"
   },
   "dependencies": {
     "@anthropic-ai/sdk": "^0.39.0"

plugins/anthropic/package.json

@@ -915,11 +915,20 @@ describe('claudeRunner', () => {
       'claude-3-5-haiku',
       anthropicClient as unknown as Anthropic
     );
-    await runner({ messages: [] });
-    expect(anthropicClient.messages.create).toHaveBeenCalledWith({
-      model: 'claude-3-5-haiku-latest',
-      max_tokens: 4096,
-    });
+    const abortSignal = new AbortController().signal;
+    await runner(
+      { messages: [] },
+      { streamingRequested: false, sendChunk: () => {}, abortSignal }
+    );
+    expect(anthropicClient.messages.create).toHaveBeenCalledWith(
+      {
+        model: 'claude-3-5-haiku-latest',
+        max_tokens: 4096,
+      },
+      {
+        signal: abortSignal,
+      }
+    );
   });
 
   it('should correctly run streaming requests', async () => {
@@ -968,12 +977,21 @@ describe('claudeRunner', () => {
       'claude-3-5-haiku',
       anthropicClient as unknown as Anthropic
     );
-    await runner({ messages: [] }, streamingCallback);
-    expect(anthropicClient.messages.stream).toHaveBeenCalledWith({
-      model: 'claude-3-5-haiku-latest',
-      max_tokens: 4096,
-      stream: true,
-    });
+    const abortSignal = new AbortController().signal;
+    await runner(
+      { messages: [] },
+      { streamingRequested: true, sendChunk: streamingCallback, abortSignal }
+    );
+    expect(anthropicClient.messages.stream).toHaveBeenCalledWith(
+      {
+        model: 'claude-3-5-haiku-latest',
+        max_tokens: 4096,
+        stream: true,
+      },
+      {
+        signal: abortSignal,
+      }
+    );
   });
 });
 
@@ -995,6 +1013,7 @@ describe('claudeModel', () => {
     claudeModel(ai, 'claude-3-5-haiku', {} as Anthropic);
     expect(ai.defineModel).toHaveBeenCalledWith(
       {
+        apiVersion: 'v2',
         name: claude35Haiku.name,
         ...claude35Haiku.info,
         configSchema: claude35Haiku.configSchema,

plugins/anthropic/src/claude.test.ts

@@ -569,29 +569,40 @@ export function claudeRunner(
 ) {
   return async (
     request: GenerateRequest<typeof AnthropicConfigSchema>,
-    streamingCallback?: StreamingCallback<GenerateResponseChunkData>
+    {
+      streamingRequested,
+      sendChunk,
+      abortSignal,
+    }: {
+      streamingRequested: boolean;
+      sendChunk: StreamingCallback<GenerateResponseChunkData>;
+      abortSignal: AbortSignal;
+    }
   ): Promise<GenerateResponseData> => {
     let response: Message;
     const body = toAnthropicRequestBody(
       name,
       request,
-      !!streamingCallback,
+      streamingRequested,
       cacheSystemPrompt
     );
-    if (streamingCallback) {
-      const stream = client.messages.stream(body);
+
+    if (streamingRequested) {
+      const stream = client.messages.stream(body, { signal: abortSignal });
       for await (const chunk of stream) {
         const c = fromAnthropicContentBlockChunk(chunk);
         if (c) {
-          streamingCallback({
+          sendChunk({
             index: 0,
             content: [c],
           });
         }
       }
       response = await stream.finalMessage();
     } else {
-      response = (await client.messages.create(body)) as Message;
+      response = (await client.messages.create(body, {
+        signal: abortSignal,
+      })) as Message;
     }
     return fromAnthropicResponse(response);
   };
@@ -612,6 +623,7 @@ export function claudeModel(
 
   return ai.defineModel(
     {
+      apiVersion: 'v2',
       name: modelId,
       ...model.info,
       configSchema: model.configSchema,

plugins/anthropic/src/claude.ts

@@ -21,9 +21,15 @@ import type {
   MessageData,
   Part,
   Role,
+  StreamingCallback,
   ToolRequestPart,
 } from 'genkit';
-import { CandidateData, modelRef, ToolDefinition } from 'genkit/model';
+import {
+  CandidateData,
+  GenerateResponseChunkData,
+  modelRef,
+  ToolDefinition,
+} from 'genkit/model';
 import { AzureOpenAI } from 'openai';
 import {
   type ChatCompletion,
@@ -633,7 +639,10 @@ export function gptModel(ai: Genkit, name: string, client: AzureOpenAI) {
       ...model.info,
       configSchema: SUPPORTED_GPT_MODELS[name].configSchema,
     },
-    async (request, streamingCallback) => {
+    async (
+      request,
+      streamingCallback?: StreamingCallback<GenerateResponseChunkData>
+    ) => {
       let response: ChatCompletion;
       const body = toOpenAiRequestBody(name, request);
       if (streamingCallback) {

plugins/azure-openai/src/gpt.ts

@@ -20,9 +20,15 @@ import type {
   GenerateRequest,
   MessageData,
   Role,
+  StreamingCallback,
   ToolRequestPart,
 } from 'genkit';
-import type { ModelAction, ToolDefinition, CandidateData } from 'genkit/model';
+import type {
+  ModelAction,
+  ToolDefinition,
+  CandidateData,
+  GenerateResponseChunkData,
+} from 'genkit/model';
 import { modelRef } from 'genkit/model';
 import type { CohereClient } from 'cohere-ai';
 import { Cohere } from 'cohere-ai';
@@ -466,7 +472,7 @@ export function commandModel(
     },
     async (
       request,
-      streamingCallback
+      streamingCallback?: StreamingCallback<GenerateResponseChunkData>
     ): Promise<{
       candidates: CandidateData[];
       usage: {

plugins/cohere/src/command.ts

@@ -30,10 +30,16 @@ import {
   MessageData,
   Part,
   Role,
+  StreamingCallback,
   ToolRequestPart,
   z,
 } from 'genkit';
-import { CandidateData, modelRef, ToolDefinition } from 'genkit/model';
+import {
+  CandidateData,
+  GenerateResponseChunkData,
+  modelRef,
+  ToolDefinition,
+} from 'genkit/model';
 
 export const GroqConfigSchema = GenerationCommonConfigSchema.extend({
   stream: z.boolean().optional(),
@@ -583,7 +589,10 @@ export function groqModel(ai: Genkit, name: string, client: Groq) {
       ...model.info,
       configSchema: model.configSchema,
     },
-    async (request, streamingCallback) => {
+    async (
+      request,
+      streamingCallback?: StreamingCallback<GenerateResponseChunkData>
+    ) => {
       let response: ChatCompletion;
       const body = toGroqRequestBody(name, request);
       if (streamingCallback) {

plugins/groq/src/groq_models.ts

@@ -14,10 +14,11 @@
  * limitations under the License.
  */
 
-import { Message, Part } from 'genkit';
+import { Message, Part, StreamingCallback } from 'genkit';
 import {
   CandidateData,
   GenerateRequest,
+  GenerateResponseChunkData,
   MessageData,
   ModelAction,
   Role,
@@ -499,7 +500,10 @@ export function mistralModel(
       ...model.info,
       configSchema: SUPPORTED_MISTRAL_MODELS[name].configSchema,
     },
-    async (request, streamingCallback) => {
+    async (
+      request,
+      streamingCallback?: StreamingCallback<GenerateResponseChunkData>
+    ) => {
       let response: ChatCompletionResponse | CompletionChunk;
       const body = toMistralRequestBody(name, request);
       if (streamingCallback) {