A coworking framework for Firebase Genkit (Go) that gives AI agents pluggable capabilities for autonomous and interactive work.
go get github.com/TheSlowpes/genkit-cowork
package main
import (
"context"
"github.com/TheSlowpes/genkit-cowork/genkit-cowork/flows"
"github.com/TheSlowpes/genkit-cowork/genkit-cowork/memory"
"github.com/firebase/genkit/go/ai"
"github.com/firebase/genkit/go/genkit"
)
func main() {
ctx := context.Background()
g := genkit.Init(ctx, genkit.WithDefaultModel("googleai/gemini-2.0-flash"))
store := memory.NewSession()
// ...
}Each tool is a standalone ai.Tool that works with any Genkit instance:
import "github.com/TheSlowpes/genkit-cowork/genkit-cowork/tools"
bashTool := tools.NewBashTool(g, "/working/dir",
tools.WithCommandPrefix("#!/bin/bash\nset -e\n"),
)
readTool := tools.NewReadTool(g, "/working/dir")
editTool := tools.NewEditTool(g, "/working/dir")
writeTool := tools.NewWriteTool(g, "/working/dir")HandleMessageFlow is a session-backed chat flow. It loads or creates session state, runs the Genkit-backed agent loop, and persists every new model/tool message produced by the run:
messageFlow := flows.NewHandleMessageFlow(g, store,
flows.WithCustomAgentConfig(flows.AgentLoopConfig{
Model: "googleai/gemini-2.0-flash",
Tools: []string{"bash", "read", "edit", "write"},
MaxTurns: 25,
}),
)
output, err := messageFlow.Run(ctx, &flows.HandleMessageInput{
SessionID: "session-1",
TenantID: "tenant-1",
Origin: memory.UIMessage,
Content: ai.Message{Role: ai.RoleUser, Content: []*ai.Part{ai.NewTextPart("Hello")}},
})The agent loop delegates tool execution, interrupt annotation, resume handling,
and max-turn enforcement to genkit.Generate. MaxTurns is passed through to
Genkit; when it is zero, cowork applies a generous bounded default.
When a tool interrupts, HandleMessageOutput.FinishReason is
ai.FinishReasonInterrupted and Interrupts contains the interrupted tool
request parts. The interrupted model message is persisted in session history, so
a later request can resume without resending user content:
first, err := messageFlow.Run(ctx, &flows.HandleMessageInput{
SessionID: "session-1",
TenantID: "tenant-1",
Origin: memory.UIMessage,
Content: *ai.NewUserTextMessage("Deploy the app"),
})
if err != nil {
return err
}
if first.FinishReason == ai.FinishReasonInterrupted {
responses := make([]*ai.Part, 0, len(first.Interrupts))
for _, interrupt := range first.Interrupts {
responses = append(responses, ai.NewToolResponsePart(&ai.ToolResponse{
Name: interrupt.ToolRequest.Name,
Ref: interrupt.ToolRequest.Ref,
Output: map[string]any{"approved": true},
}))
}
resumed, err := messageFlow.Run(ctx, &flows.HandleMessageInput{
SessionID: "session-1",
TenantID: "tenant-1",
ToolResponses: responses,
})
_ = resumed
return err
}Use ToolRestarts instead of ToolResponses when the original tool should be
re-executed with resume metadata or replacement input.
Heartbeat runs the same Genkit-backed agent loop on a schedule against existing session state, classifying results as ack, alert, skipped, or error:
heartbeat := flows.NewHeartbeat(g, store, flows.HeartbeatConfig{
Interval: 5 * time.Minute,
SessionID: "heartbeat-session",
TenantID: "tenant-1",
ActiveHours: &flows.ActiveHours{
Start: "09:00",
End: "17:00",
Timezone: "America/New_York",
},
Delivery: flows.DefaultHeartbeatDelivery(),
Target: flows.HeartbeatTargetLast,
To: memory.WhatsAppMessage,
AgentConfig: &flows.AgentLoopConfig{
Model: "googleai/gemini-2.0-flash",
},
}, flows.WithHeartbeatOnResult(func(output *flows.HeartbeatOutput) {
if output.ShouldDeliver {
// Forward to sendReply flow
}
}))
heartbeat.Start(ctx)
defer heartbeat.Stop()SendReplyFlow routes agent output to external channels via the ChannelHandler interface:
// Implement ChannelHandler for your channels
type WhatsAppHandler struct { /* ... */ }
func (h *WhatsAppHandler) Setup(ctx context.Context, tenantID string) error { /* ... */ }
func (h *WhatsAppHandler) SendReply(ctx context.Context, input *flows.SendReplyInput) error { /* ... */ }
func (h *WhatsAppHandler) Acknowledge(ctx context.Context, input *flows.AcknowledgeInput) error { /* ... */ }
// Register handlers by channel
senders := map[memory.MessageOrigin]flows.ChannelHandler{
memory.WhatsAppMessage: &WhatsAppHandler{},
memory.ZoomMessage: &ZoomHandler{},
}
// Run per-tenant channel setup (webhooks, token verification, etc.)
if err := flows.SetupSenders(ctx, "tenant-1", senders); err != nil {
log.Fatal(err)
}
// Create the flow
replyFlow := flows.NewSendReplyFlow(g, senders)heartbeat := flows.NewHeartbeat(g, store, cfg,
flows.WithHeartbeatOnResult(func(output *flows.HeartbeatOutput) {
if !output.ShouldDeliver {
return
}
replyFlow.Run(ctx, &flows.SendReplyInput{
SessionID: output.SessionID,
Sender: flows.Sender{TenantID: cfg.TenantID, DisplayName: "Agent"},
Content: output.Response,
Channel: cfg.To,
Target: cfg.Target,
Destination: flows.Destination{
ChatID: "resolved-chat-id",
},
})
}),
)The EventBus provides typed lifecycle events for agent, turn, message, and tool execution stages:
bus := flows.NewEventBus()
flows.Subscribe[flows.ToolExecutionContext](bus, flows.ToolExecutionEnd,
func(ctx context.Context, event *flows.Event[flows.ToolExecutionContext]) error {
log.Printf("tool %s completed in %s", event.Data.ToolName, event.Data.Duration)
return nil
},
)
// Pass the bus to flows
messageFlow := flows.NewHandleMessageFlow(g, store,
flows.WithHandleMessageEventBus(bus),
)
heartbeat := flows.NewHeartbeat(g, store, cfg,
flows.WithHeartbeatEventBus(bus),
)The framework is built around four pillars:
┌─────────────────────────────────────────────────┐
│ genkit-cowork │
│ │
│ ┌───────────┐ ┌───────────┐ │
│ │ Flows │ │ Tools │ │
│ │ │ │ │ │
│ │ agent │ │ bash │ │
│ │ loop │ │ read │ │
│ │ message │ │ edit │ │
│ │ heartbeat│ │ write │ │
│ │ reply │ │ │ │
│ └───────────┘ └───────────┘ │
│ │
│ ┌───────────┐ ┌───────────┐ │
│ │ Memory │ │ Skills │ │
│ │ │ │ │ │
│ │ sessions │ │ discover │ │
│ │ file │ │ list │ │
│ │ vector │ │ resolve │ │
│ └───────────┘ └───────────┘ │
│ │
│ ┌─────────────────┐ │
│ │ Genkit Runtime │ │
│ └─────────────────┘ │
└─────────────────────────────────────────────────┘
Each pillar can be adopted independently. Use the full framework or pick individual pieces:
- Tools only — register
NewBashTool,NewReadTool,NewEditTool,NewWriteToolwith any Genkit instance. - Flows only — use the agent loop, message handling, heartbeat, or reply flows.
- Memory only — use
NewSessionwith in-memory, file-backed, or vector-augmented operators. - Skills only — register the
Skillsplugin to discover and serve domain knowledge. - Mix and match — combine pillars based on your use case.
| Flow | Registration | Purpose |
|---|---|---|
agentLoop |
genkit.NewFlow (internal) |
Core Genkit-backed model/tool loop |
handleMessage |
genkit.DefineFlow |
Session-backed chat |
heartbeat |
genkit.DefineFlow |
Scheduled background monitoring |
sendReply |
genkit.DefineFlow |
Channel-routed reply delivery |
| Tool | Constructor | Description |
|---|---|---|
| Bash | NewBashTool(g, cwd, ...opts) |
Shell command execution with timeout, env, spawn hooks |
| Read | NewReadTool(g, cwd, ...opts) |
File/image reading with pagination, truncation, auto-resize |
| Edit | NewEditTool(g, cwd, ...opts) |
Find-and-replace with exact/fuzzy matching, unified diff |
| Write | NewWriteTool(g, cwd, ...opts) |
File creation with auto-mkdir, operator interface |
Memory is implemented through a Session store plus pluggable SessionOperator backends.
| Type | Constructor / API | Description |
|---|---|---|
| Session store | NewSession(...opts) |
Implements session.Store[SessionState] for Genkit flows |
| Persistence mode | WithPersistenceMode(mode, n) |
Load behavior: All, SlidingWindow, TailEndsPruning, TokenBudget |
| Media asset store | WithMediaAssetStore(store) |
Normalizes media data URI parts into persisted files and tracks SessionAsset metadata |
| Tenant binding | WithTenantID(id) / ForTenant(id) |
Scopes Get/Save operations to a tenant in a session.Store-compatible API |
| Ledger validation | ValidateSessionLedger(state) |
Validates append-only sequencing and immutable-prefix constraints for a session ledger |
| Replay window | MessagesForTurn(state, turn) |
Reconstructs the exact message slice represented by a persisted turn sequence range |
| In-memory backend | default (defaultSessionOperator) |
Process-local map-based state storage |
| File backend | NewFileSessionOperator(rootDir, tenantID) |
Durable JSON state at rootDir/{tenantID}/{sessionID}/state.json |
| Vector wrapper | NewVectorOperator(base, backend, rootDir) |
Wraps a base operator and indexes new messages for semantic retrieval |
| Local vector backend | NewLocalVecBackend(g, name, cfg) |
localvec-based implementation of VectorBackend |
| File media asset backend | NewFileMediaAssetStore(rootDir) |
Filesystem-backed asset persistence at rootDir/{tenantID}/{sessionID}/assets with document loading and chunk index files |
NewFileSessionOperator provides durable session state with per-session locks, atomic writes, append-only validation, and tenant consistency checks.
fileOp := memory.NewFileSessionOperator("./data/sessions", "tenant-1")VectorOperator composes on top of a base SessionOperator. It indexes new messages by messageID and supports semantic lookup with Search.
fileOp := memory.NewFileSessionOperator("./data/sessions", "tenant-1")
vecBackend, _ := memory.NewLocalVecBackend(g, "session-memory", memory.LocalVecConfig{
Embedder: embedder, // any ai.Embedder
IndexDir: "./data/localvec", // localvec persistence directory
})
vecOp := memory.NewVectorOperator(fileOp, vecBackend, "./data/sessions")
store := memory.NewSession(
memory.WithCustomSessionOperator(vecOp),
memory.WithTenantID("tenant-1"),
)
results, err := vecOp.Search(ctx, "tenant-1", "session-1", "customer asked about invoice", 5)
_ = results
_ = errUse WithMediaAssetStore to persist media/data URI parts to disk and track
SessionAsset metadata in session state.
assetStore := memory.NewFileMediaAssetStore("./data/assets")
store := memory.NewSession(
memory.WithTenantID("tenant-1"),
memory.WithMediaAssetStore(assetStore),
)
// During Session.Save, data URI media parts are normalized to absolute files
// under ./data/assets/{tenantID}/{sessionID}/assets.Each SessionMessage stores MessageID, Origin, Kind, Content, and Timestamp.
Kindis auto-derived when missing: tool-role messages becomeinstrumental, others default toepisodic.- Additional kinds (
semantic,procedural) are available for higher-level memory workflows. Session.Saveauto-fills missingMessageIDandTimestamp.
examples/pgvector/main.goshows pgvector wiring for session memory only by wrapping the Genkit PostgreSQL plugin as amemory.VectorBackendand plugging it intomemory.NewVectorOperator.
Skills are domain-specific knowledge modules discovered from a directory of SKILL.md files. The skills system is implemented as a Genkit plugin.
import "github.com/TheSlowpes/genkit-cowork/genkit-cowork/plugins/skills"
g := genkit.Init(ctx,
genkit.WithDefaultModel("googleai/gemini-2.0-flash"),
genkit.WithPlugins(&skills.Skills{
SkillsDir: "./skills", // optional; falls back to default search paths
AllowedSkills: []string{"my-skill"}, // optional whitelist; all skills exposed when empty
}),
)When SkillsDir is not set, the plugin searches for the first existing directory among: ./skills, ./SKILLS, ./.agent/skills, ./agent/skills, ./docs/skills. If none are found, the plugin starts with an empty skill set and does not panic.
After Init, register the tool with a Genkit instance:
s := &skills.Skills{SkillsDir: "./skills"}
// after genkit.Init with the plugin...
skillTool := s.SkillTool(g)Each skill lives in a subdirectory and must contain a SKILL.md file with YAML frontmatter:
---
name: my-skill
description: What the skill does
license: MIT
metadata:
key: value
---
Skill content in Markdown...The plugin provides a single tool for agent use:
| Tool | Name | Description |
|---|---|---|
SkillTool(g) |
resolve-skill |
Lists all available skills in the tool description; accepts a skill name and returns the full SKILL.md body and metadata |
On Init, the plugin resolves the skills directory (checking defaultSkillsDirs when SkillsDir is unset), then scans top-level subdirectories for SKILL.md files, parses frontmatter, validates required fields (name, description), and catalogs all files in the skill directory (including one level of subdirectories). Invalid skills are silently skipped. Skill body content is loaded lazily only when resolve-skill is called.
When AllowedSkills is non-empty, only skills whose names appear in that list are exposed by SkillTool and ListSkills.
- Functional options — all constructors accept variadic option functions for clean, extensible configuration.
- Operator interfaces —
BashOperator,ReadOperator,EditOperator,WriteOperator,SessionOperator,AgentLoopOperatorabstract I/O behind interfaces for testability and sandboxing. - Hook system — lifecycle hooks (spawn hooks, event bus) allow interception and mutation before operations run.
genkit-cowork/
├── flows/ # Flow definitions
│ ├── agent_loop.go # Core Genkit-backed model/tool loop
│ ├── agent_loop_recorder.go # Event and turn metadata recorder middleware
│ ├── message.go # Session-backed message handling
│ ├── heartbeat.go # Scheduled heartbeat runner
│ ├── heartbeat_config.go # Heartbeat configuration types
│ ├── heartbeat_result.go # Result parsing and classification
│ ├── reply.go # Channel-routed reply delivery
│ ├── event.go # EventBus and typed lifecycle events
│ └── event_context.go # Event payload types
├── tools/ # Tool definitions
│ ├── bash.go # Shell command execution
│ ├── read.go # File/image reading
│ ├── edit.go # Find-and-replace editing
│ ├── write.go # File creation with auto-mkdir
│ ├── edit_diff.go # Text normalization, fuzzy matching, diffs
│ ├── diff.go # LCS-based line diff algorithm
│ ├── truncate.go # Output truncation utilities
│ ├── path.go # Path resolution utilities
│ ├── constants.go # Output limits
│ └── memory_retrieval.go # Tenant/session memory retrieval tools
├── plugins/ # Genkit plugins
│ └── skills/ # Skill discovery and serving
│ ├── skills.go # Plugin struct, Init, tool registration
│ ├── skill_parser.go # SKILL.md frontmatter parsing
│ └── skill_scanner.go # Directory scanning, skill discovery
├── media/ # MIME detection and processing
│ ├── mime.go # Image MIME detection, image resizing
│ ├── registry.go # MIME to DocumentProcessor registry
│ ├── documents.go # DocumentProcessor interface
│ ├── shared.go # Shared chunking and processor helpers
│ └── *_processor.go # One file per MIME-specific DocumentProcessor
├── memory/ # Session persistence and retrieval
│ ├── sessions.go # Session store, message models, persistence modes
│ ├── turns.go # Turn ledger records and ledger validation
│ ├── snapshots.go # State snapshot records and checksum support
│ ├── assets.go # Session asset model and media asset store interface
│ ├── file_assets.go # Filesystem media asset store implementation
│ ├── file_sessions.go # File-backed SessionOperator (JSON + atomic write)
│ ├── vector_sessions.go # VectorOperator wrapper + semantic search
│ └── vector_backend.go # Vector backend interface + localvec backend
└── utils/ # Shared utilities
├── shell.go # Shell environment management
└── path_segment.go # Path segment validation for filesystem safety