core

Overview ¶

Package core implements Genkit actions and other essential machinery. This package is primarily intended for Genkit internals and for plugins. Genkit applications should use the genkit package.

Package core implements Genkit's foundational action system and runtime machinery.

This package is primarily intended for plugin developers and Genkit internals. Application developers should use the genkit package instead, which provides a higher-level, more convenient API.

Actions ¶

Actions are the fundamental building blocks of Genkit. Every operation - flows, model calls, tool invocations, retrieval - is implemented as an action. Actions provide:

• Type-safe input/output with JSON schema validation
• Automatic tracing and observability
• Consistent error handling
• Registration in the action registry

Define a non-streaming action:

action := core.NewActionOf(api.ActionTypeCustom, "myAction", nil,
	func(ctx context.Context, input string) (string, error) {
		return "processed: " + input, nil
	},
)
action.Register(registry)

result, err := action.Run(context.Background(), "hello")

Define a streaming action that sends chunks during execution:

streamingAction := core.NewStreamingActionOf(api.ActionTypeCustom, "countdown", nil,
	func(ctx context.Context, start int, cb core.StreamCallback[string]) (string, error) {
		for i := start; i > 0; i-- {
			if cb != nil {
				if err := cb(ctx, fmt.Sprintf("T-%d", i)); err != nil {
					return "", err
				}
			}
			time.Sleep(time.Second)
		}
		return "Liftoff!", nil
	},
)
streamingAction.Register(registry)

Flows ¶

Flows are user-defined actions that orchestrate AI operations. They are the primary way application developers define business logic in Genkit:

flow := core.NewFlow("myFlow",
	func(ctx context.Context, input string) (string, error) {
		// Use Run to create traced sub-steps
		result, err := core.Run(ctx, "step1", func() (string, error) {
			return process(input), nil
		})
		if err != nil {
			return "", err
		}
		return result, nil
	},
)
flow.Register(registry)

Streaming flows can send intermediate results to callers:

streamingFlow := core.NewStreamingFlow("generateReport",
	func(ctx context.Context, input Input, cb core.StreamCallback[Progress]) (Report, error) {
		for i := 0; i < 100; i += 10 {
			if cb != nil {
				cb(ctx, Progress{Percent: i})
			}
			// ... work ...
		}
		return Report{...}, nil
	},
)
streamingFlow.Register(registry)

Traced Steps with Run ¶

Use Run within flows to create traced sub-operations. Each Run call creates a span in the trace that's visible in the Genkit Developer UI:

result, err := core.Run(ctx, "fetchData", func() (Data, error) {
	return fetchFromAPI()
})

processed, err := core.Run(ctx, "processData", func() (Result, error) {
	return process(result)
})

Middleware ¶

Actions support middleware for cross-cutting concerns like logging, metrics, or authentication:

loggingMiddleware := func(next core.StreamingFunc[string, string, struct{}]) core.StreamingFunc[string, string, struct{}] {
	return func(ctx context.Context, input string, cb core.StreamCallback[struct{}]) (string, error) {
		log.Printf("Input: %s", input)
		output, err := next(ctx, input, cb)
		log.Printf("Output: %s, Error: %v", output, err)
		return output, err
	}
}

Chain multiple middleware together:

combined := core.ChainMiddleware(loggingMiddleware, metricsMiddleware)
wrappedFn := combined(originalFunc)

Schema Management ¶

Register JSON schemas for use in prompts and validation:

// Register a schema from a map
registry.RegisterSchema("Person", map[string]any{
	"type": "object",
	"properties": map[string]any{
		"name": map[string]any{"type": "string"},
		"age":  map[string]any{"type": "integer"},
	},
	"required": []any{"name"},
})

// Register a schema inferred from a Go type (recommended)
registry.RegisterSchema("Person", core.InferSchemaMap(Person{}))

Schemas can be referenced in .prompt files by name.

Plugin Development ¶

Plugins extend Genkit's functionality by providing models, tools, retrievers, and other capabilities. Implement the api.Plugin interface:

type MyPlugin struct {
	APIKey string
}

func (p *MyPlugin) Name() string {
	return "myplugin"
}

func (p *MyPlugin) Init(ctx context.Context) []api.Action {
	// Initialize the plugin and return actions to register
	model := ai.NewModelAction(...)
	tool := ai.NewTool(...)
	return []api.Action{model, tool}
}

For plugins that resolve actions dynamically (e.g., listing available models from an API), implement api.DynamicPlugin:

type DynamicModelPlugin struct{}

func (p *DynamicModelPlugin) ListActions(ctx context.Context) []api.ActionDesc {
	// Return descriptors of available actions
	return []api.ActionDesc{
		{Key: "/model/myplugin/model-a", Name: "model-a"},
		{Key: "/model/myplugin/model-b", Name: "model-b"},
	}
}

func (p *DynamicModelPlugin) ResolveAction(atype api.ActionType, id string) api.Action {
	// Create and return the action on demand
	return createModel(id)
}

Background Actions ¶

For long-running operations, use background actions that return immediately with an operation ID that can be polled for completion:

bgAction := core.NewBackgroundActionOf(api.ActionTypeCustom, "longTask",
	&core.BackgroundActionOptions[Input, Output]{
		// Check polls operation status; omitting Cancel means the
		// action does not support cancellation.
		Check: func(ctx context.Context, op *core.Operation[Output]) (*core.Operation[Output], error) {
			return checkOperationStatus(op)
		},
	},
	func(ctx context.Context, input Input) (*core.Operation[Output], error) {
		// Start the operation
		return startLongOperation(input)
	},
)
bgAction.Register(registry)

Error Handling ¶

Errors live in github.com/firebase/genkit/go/core/status. Classify a failure with a sentinel so callers can branch on it with errors.Is rather than by matching message text, and mark a message public only when it is safe to return to a client:

import "github.com/firebase/genkit/go/core/status"

Return user-facing errors with appropriate status codes:

if err := validate(input); err != nil {
	return nil, status.PublicErrorf(status.ErrInvalidArgument, "Invalid input").WithDetails(map[string]any{
		"field": "email",
		"error": err.Error(),
	})
}

For internal errors that should be logged but not exposed to users:

return nil, status.Errorf(status.ErrInternal, "database connection failed: %w", err)

Context ¶

Access action context for metadata and configuration:

ctx := core.FromContext(ctx)
if ctx != nil {
	// Access action-specific context values
}

Set action context for nested operations:

ctx = core.WithActionContext(ctx, core.ActionContext{
	"requestId": requestID,
})

For more information, see https://genkit.dev/docs/plugins

Package core provides base error types and utilities for Genkit.

The error surface in this file is deprecated in favour of github.com/firebase/genkit/go/core/status, which unifies the two error types below into one and adds sentinel classification so callers can branch with errors.Is instead of matching on message text. Everything here is an alias or a thin wrapper over that package: GenkitError and status.Error are the same type, so an errors.As for either finds errors raised by any part of Genkit, old or new.

package main

import (
	"fmt"

	"github.com/firebase/genkit/go/internal/registry"
)

func main() {
	r := registry.New()

	// Register a JSON schema defined as a map
	r.RegisterSchema("Address", map[string]any{
		"type": "object",
		"properties": map[string]any{
			"street": map[string]any{"type": "string"},
			"city":   map[string]any{"type": "string"},
			"zip":    map[string]any{"type": "string"},
		},
		"required": []any{"street", "city"},
	})

	fmt.Println("Schema registered: Address")
}

Output:
Schema registered: Address