genkit 0.11.0 | Dart package

AI SDK for Dart • LLM Framework • AI Agent Toolkit

Build production-ready AI-powered applications in Dart with a unified interface for text generation, structured output, tool calling, and agentic workflows.

Documentation • API Reference • Discord

Installation #

dart pub add genkit
# For usage with Google AI (Gemini):
dart pub add genkit_google_genai

Quick Start (Framework) #

Get up and running in under a minute:

import 'package:genkit/genkit.dart';
import 'package:genkit_google_genai/genkit_google_genai.dart';

void main() async {
  final ai = Genkit(plugins: [googleAI()]);

  final response = await ai.generate(
    model: googleAI.gemini('gemini-2.5-flash'),
    prompt: 'Why is Dart a great language for AI applications?',
  );

  print(response.text);
}

export GEMINI_API_KEY="your-api-key"
dart run main.dart

Features #

Genkit Dart gives you everything you need to build AI applications with confidence.

Generate Text #

Call any model with a simple, unified API:

final response = await ai.generate(
  model: googleAI.gemini('gemini-2.5-flash'),
  prompt: 'Explain quantum computing in simple terms.',
);
print(response.text);

Stream Responses #

Stream text as it's generated for responsive user experiences:

final stream = ai.generateStream(
  model: googleAI.gemini('gemini-2.5-flash'),
  prompt: 'Write a short story about a robot learning to paint.',
);

await for (final chunk in stream) {
  print(chunk.text);
}

Embed Text #

Turn text into vector embeddings for search and retrieval tasks:

final embeddings = await ai.embedMany(
  documents: [
    DocumentData(content: [TextPart(text: 'Hello world')]),
  ],
  embedder: googleAI.textEmbedding('text-embedding-004'),
);

print(embeddings.first.embedding);

Define Tools #

Give models the ability to take actions and access external data:

// Define schemas for tool input
@Schema()
abstract class $WeatherInput {
  String get location;
}

// ... run build_runner to generate WeatherInputType ...

final weatherTool = ai.defineTool(
  name: 'getWeather',
  description: 'Gets the current weather for a location',
  inputSchema: WeatherInput.$schema,
  fn: (input, _) async {
    // Call your weather API here
    return 'Weather in ${input.location}: 72°F and sunny';
  },
);

final response = await ai.generate(
  model: googleAI.gemini('gemini-2.5-flash'),
  prompt: 'What\'s the weather like in San Francisco?',
  toolNames: ['getWeather'],
);
print(response.text);

Define Flows #

Wrap your AI logic in flows for better observability, testing, and deployment:

final jokeFlow = ai.defineFlow(
  name: 'tellJoke',
  inputSchema: .string(),
  outputSchema: .string(),
  fn: (topic, _) async {
    final response = await ai.generate(
      model: googleAI.gemini('gemini-2.5-flash'),
      prompt: 'Tell me a joke about $topic',
    );
    return response.text;
  },
);

final joke = await jokeFlow('programming');
print(joke);

Streaming Flows #

Stream data from your flows using context.sendChunk:

final streamStory = ai.defineFlow(
  name: 'streamStory',
  inputSchema: .string(),
  outputSchema: .string(),
  streamSchema: .string(),
  fn: (topic, context) async {
    final stream = ai.generateStream(
      model: googleAI.gemini('gemini-2.5-flash'),
      prompt: 'Write a story about $topic',
    );

    await for (final chunk in stream) {
      context.sendChunk(chunk.text);
    }
    return 'Story complete';
  },
);

Interrupts #

Interrupts allow a flow or model to pause execution and wait for external input (human-in-the-loop). A tool can trigger an interrupt, which pauses the ai.generate loop and returns control to the caller with the interrupt metadata.

Triggering an Interrupt

To trigger an interrupt, call context.interrupt() within a tool definition:

@Schema()
abstract class $AskUserInput {
  String get question;
}

// ... run build_runner ...

final askUser = ai.defineTool(
  name: 'askUser',
  description: 'Ask the user a clarifying question.',
  inputSchema: AskUserInput.$schema,
  fn: (input, context) async {
    context.interrupt(input.question);
  },
);

Resuming from an Interrupt

When ai.generate is interrupted, it returns a response with finishReason == FinishReason.interrupted. You can then resume the generation by providing either a direct response to the tool or by requesting to restart the tool execution:

Provide a response (`interruptRespond`)

Use interruptRespond when you have the data the tool was waiting for (e.g., a user's confirmation or additional input):

var response = await ai.generate(
  prompt: 'Research Genkit support for Dart',
  toolNames: ['askUser', 'webSearch'],
);

while (response.finishReason == FinishReason.interrupted) {
  for (final interrupt in response.interrupts) {
    if (interrupt.toolRequest.name == 'askUser') {
      final question = interrupt.metadata?['interrupt'] as String;
      
      // Prompt user for answer...
      final answer = 'Dart support is coming soon!';

      response = await ai.generate(
        messages: response.messages,
        toolNames: ['askUser', 'webSearch'],
        interruptRespond: [
          InterruptResponse(interrupt.toolRequestPart!, answer),
        ],
      );
    }
  }
}

Restart the tool (`interruptRestart`)

Use interruptRestart when the environment has changed or the user has taken an action that allows the tool to proceed on its own. You can pass additional metadata to the tool (e.g., an approval token), which the tool can then access via context.metadata:

final confirmAction = ai.defineTool(
  name: 'confirmAction',
  description: 'A tool that requires approval metadata',
  fn: (input, context) async {
    // Access the metadata passed during the initial call or restart
    if (context.metadata['approved'] != true) {
      context.interrupt('Approval required');
    }
    return 'Action confirmed';
  },
);

// ...

var response = await ai.generate(
  prompt: 'Delete the database',
  toolNames: ['confirmAction'],
);

if (response.finishReason == FinishReason.interrupted) {
  // Use UI to get user confirmation...
  final confirmed = true; 

  if (confirmed) {
    // Resume generation and use `withMetadata` to inject the approval
    response = await ai.generate(
      messages: response.messages,
      toolNames: ['confirmAction'],
      interruptRestart: [
        response.interrupts.first.toolRequestPart!.withMetadata({'approved': true}),
      ],
    );
  }
}

Middleware #

Intercept and modify requests and responses with middleware. Genkit provides built-in middleware like retry for robust error handling. Check out the genkit_middleware package for more specialized middleware like tool approval, filesystem access, and skills.

Retry Middleware

Automatically retry failed requests with exponential backoff and jitter:

final ai = Genkit(
  plugins: [
    googleAI(),
    RetryPlugin(), // Required for retry middleware
  ],
);

final response = await ai.generate(
  model: googleAI.gemini('gemini-2.5-flash'),
  prompt: 'Reliable request',
  use: [
    retry(
      maxRetries: 3,
      retryModel: true, // Retry model validation errors (default: true)
      retryTools: false, // Retry tool execution errors (default: false)
      statuses: [StatusCodes.UNAVAILABLE], // Retry only on specific errors
    ),
  ],
);

Development Tools #

Genkit CLI #

Use the Genkit CLI to run your app with tracing and a local development UI:

curl -sL cli.genkit.dev | bash
genkit start -- dart run main.dart

Developer UI #

The local developer UI lets you:

Test flows with different inputs interactively
Inspect traces to debug complex multi-step operations
View traces of your generative AI applications

Client SDK #

This package also provides a Dart client for interacting with remote Genkit flows, enabling type-safe communication for both unary and streaming operations.

Getting Started #

Import the client library and define your remote actions.

import 'package:genkit/client.dart';

### Defining Remote Actions

Remote actions represent a remote Genkit action (like flows, models and prompts) that can be invoked or streamed. You use generated `*Type` classes (from `@Schema`) to ensure type safety.

#### Creating a remote action
final stringAction = defineRemoteAction(
  url: 'http://localhost:3400/my-flow',
  inputSchema: .string(),
  outputSchema: .string(),
);

// Create a remote action for custom objects
final customAction = defineRemoteAction(
  url: 'http://localhost:3400/custom-flow',
  inputSchema: MyInput.$schema,
  outputSchema: MyOutput.$schema,
);

The code assumes that you have my-flow and custom-flow deployed at those URLs. See https://genkit.dev/docs/deploy-node/ or https://genkit.dev/go/docs/deploy/ for details.

Calling Actions #

Example: String to String

final action = defineRemoteAction(
  url: 'http://localhost:3400/echo-string',
  inputSchema: .string(),
  outputSchema: .string(),
);

try {
  final response = await action(input: 'Hello from Dart!');
  print('Flow Response: $response');
} catch (e) {
  print('Error calling flow: $e');
}

Example: Custom Object Input and Output

First, define your schemas and run build_runner to generate the types.

@Schema()
abstract class $MyInput {
  String get message;
  int get count;
}

@Schema()
abstract class $MyOutput {
  String get reply;
  int get newCount;
}

// ... run build_runner ...

Then define and call the action:

final action = defineRemoteAction(
  url: 'http://localhost:3400/process-object',
  inputSchema: MyInput.$schema,
  outputSchema: MyOutput.$schema,
);

final input = MyInput(message: 'Process this data', count: 10);

try {
  final output = await action(input: input);
  print('Flow Response: ${output.reply}, ${output.newCount}');
} catch (e) {
  print('Error calling flow: $e');
}

Calling Streaming Flows #

Use the stream method for flows that stream multiple chunks of data and then return a final response. Specify streamSchema to handle typed chunks.

Example 1: Using `onResult` (Recommended)

final streamAction = defineRemoteAction(
  url: 'http://localhost:3400/stream-story',
  inputSchema: .string(),
  outputSchema: .string(),
  streamSchema: .string(),
);

try {
  final stream = streamAction.stream(
    input: 'Tell me a short story about a Dart developer.',
  );

  print('Streaming chunks:');
  await for (final chunk in stream) {
    print('Chunk: $chunk'); // chunk is String
  }

  // Use onResult to asynchronously get the final response.
  final finalResult = await stream.onResult;
  print('\nFinal Response: $finalResult');
} catch (e) {
  print('Error calling streamFlow: $e');
}

Example: Custom Object Streaming

@Schema()
abstract class $StreamChunk {
  String get content;
}

// ... generated StreamChunkType ...

final streamAction = defineRemoteAction(
  url: 'http://localhost:3400/stream-process',
  inputSchema: MyInput.$schema,
  outputSchema: MyOutput.$schema,
  streamSchema: StreamChunk.$schema,
);

final input = MyInput(message: 'Stream this data', count: 5);

try {
  final stream = streamAction.stream(input: input);

  print('Streaming chunks:');
  await for (final chunk in stream) {
    print('Chunk: ${chunk.content}');
  }

  final finalResult = await stream.onResult;
  print('\nFinal Response: ${finalResult.reply}');
} catch (e) {
  print('Error calling streaming flow: $e');
}

Custom Headers #

You can provide custom headers for individual requests:

final response = await action(
  input: 'test input',
  headers: {'Authorization': 'Bearer your-token'},
);

// For streaming
final streamResponse = action.stream(
  input: 'test input',
  headers: {'Authorization': 'Bearer your-token'},
);

You can also set default headers when creating the remote action:

final action = defineRemoteAction(
  url: 'http://localhost:3400/my-flow',
  inputSchema: .string(),
  outputSchema: .string(),
  defaultHeaders: {'Authorization': 'Bearer your-token'},
);

Error Handling #

The library throws GenkitException for various error conditions:

try {
  final result = await action(input: 'test');
} on GenkitException catch (e) {
  print('Genkit error: ${e.message}');
  print('Status code: ${e.statusCode}');
  print('Details: ${e.details}');
} catch (e) {
  print('Other error: $e');
}

Advanced: Manual Data Conversion #

For advanced use cases where generated types are not available, you can still use fromResponse and fromStreamChunk for manual conversion:

final action = defineRemoteAction(
  url: 'http://localhost:3400/my-flow',
  fromResponse: (data) => data as String,
);

Working with Genkit Data Objects #

When interacting with Genkit models, you'll often work with a set of standardized data classes that represent the inputs and outputs of generative models.

Example: Streaming with Genkit Data Objects

import 'package:genkit/client.dart';

final generateFlow = defineRemoteAction(
  url: 'http://localhost:3400/generate',
  inputSchema: ModelRequest.$schema,
  outputSchema: ModelResponse.$schema,
  streamSchema: ModelResponseChunk.$schema,
);

final stream = generateFlow.stream(
  input: ModelRequest(
    messages: [Message(role: Role.user, content: [TextPart(text: "hello")])],
  ),
);

print('Streaming chunks:');
await for (final chunk in stream) {
  print('Chunk: ${chunk.text}');
}

final finalResult = await stream.onResult;
print('Final Response: ${finalResult.text}');

Remote Models #

You can also define and use remotely deployed models as if they were local models using defineRemoteModel. This is particularly useful when you have models hosted via genkit_shelf or other compatible Genkit servers.

final remoteModel = ai.defineRemoteModel(
  name: 'my-remote-model',
  url: 'http://localhost:3400/my-model',
  // Optional: Provide custom headers dynamically based on context
  headers: (context) {
    return {'Authorization': 'Bearer ${context['token']}'};
  },
);

final response = await ai.generate(
  model: remoteModel,
  prompt: 'say hello',
  context: {'token': 'my-secret-token'},
);

print(response.text);

Check out the genkit_shelf package for details on how to host remote models and flows.

Genkit Lite API #

For lightweight applications or scripts where you only need basic model orchestration without the full Genkit framework (no registries, flows, or Dev UI), you can use the Lite API.

import 'package:genkit/lite.dart' as lite;
// Need to import genkit for standard components like RetryMiddleware
import 'package:genkit/genkit.dart';
import 'package:genkit_google_genai/genkit_google_genai.dart';

void main() async {
  // 1. Initialize the plugin directly
  final gemini = googleAI();

  // 2. Direct generation call without a Genkit instance
  final response = await lite.generate(
    model: gemini.model('gemini-2.5-flash'),
    prompt: 'Hello from Lite API!',
    // Middleware objects are used directly in the Lite API
    use: [
      RetryMiddleware(maxRetries: 2),
    ],
  );

  print(response.text);
}

Built by Google with contributions from the Open Source Community

genkit 0.11.0 genkit: ^0.11.0 copied to clipboard

Metadata