feat(go): added experimental ways to define tools, interrupts, and streaming

[apascal07]

Adds experimental tool definition APIs (ai/exp) with simplified function signatures, typed interrupt/resume, and a runtime helper package for use inside tool functions.

These APIs are designed to replace the existing DefineTool and DefineMultipartTool in a future major release. The current implementation wraps the existing tool infrastructure for backwards compatibility, but the intent is to invert this -- the new APIs will become the primary implementation and the old ones will be removed.

Examples

Simple Tools

The new experimental tool constructor (genkitx.DefineTool, in the genkit/exp package) replaces both the stable DefineTool and DefineMultipartTool with a single API. The function receives a plain context.Context instead of ai.ToolContext, making tool functions easier to write and test. Multipart responses are handled via tool.AttachParts rather than requiring a separate function signature:

typeWeatherInputstruct {
Citystring`json:"city" jsonschema:"description=city name"`
}
weatherTool:=genkitx.DefineTool(g, "getWeather", "Fetches the weather for a given city",
func(ctx context.Context, inputWeatherInput) (string, error) {
ifinput.City=="Paris" {
return"Sunny, 25°C", nil
}
return"Cloudy, 18°C", nil
},
)
resp, _:=genkit.Generate(ctx, g,
ai.WithPrompt("What's the weather like in Paris?"),
ai.WithTools(weatherTool),
)
fmt.Println(resp.Text())

Use tool.AttachParts to return additional content (images, media) without changing the function signature:

genkitx.DefineTool(g, "screenshot", "Takes a screenshot",
func(ctx context.Context, inputScreenshotInput) (string, error) {
img:=takeScreenshot(input.URL)
tool.AttachParts(ctx, ai.NewMediaPart("image/png", img))
return"Screenshot captured", nil
},
)

---

Partial Tool Responses (Streaming Progress)

Tools can stream partial responses during execution for client-side display (e.g., progress indicators). Use tool.SendPartial inside any tool function -- it's best-effort and no-ops when streaming isn't available.

Define a status type shared between the tool and client:

typePipelineStatusstruct {
Stepstring`json:"step"`Progressint`json:"progress"`
}

Send partial responses from the tool:

processDataTool:=aix.NewTool("process_data", "Processes data from a source",
func(ctx context.Context, inputProcessInput) (string, error) {
tool.SendPartial(ctx, PipelineStatus{Step: "connecting", Progress: 0})
connect(input.DataSource)
tool.SendPartial(ctx, PipelineStatus{Step: "processing", Progress: 50})
results:=process(input.DataSource)
tool.SendPartial(ctx, PipelineStatus{Step: "done", Progress: 100})
returnfmt.Sprintf("Processed %d records", len(results)), nil
},
)

On the client side, partial tool responses arrive as ModelResponseChunks during streaming. Use Part.IsPartial() to distinguish progress updates from final tool results:

// GenerateStreamforchunk, _:=rangeai.GenerateStream(ctx, r, opts...) {
for_, p:=rangechunk.Chunk.ToolResponses() {
ifp.IsPartial() {
status:=p.ToolResponse.Output.(PipelineStatus)
fmt.Printf("[%s... %d%%]\n", status.Step, status.Progress)
}
}
fmt.Print(sv.Chunk.Text())
}
// Agent flowforchunk, _:=rangeconn.Receive() {
for_, p:=rangechunk.ModelChunk.ToolResponses() {
ifp.IsPartial() {
status:=p.ToolResponse.Output.(PipelineStatus)
fmt.Printf("[%s... %d%%]", status.Step, status.Progress)
}
}
fmt.Print(chunk.ModelChunk.Text())
}

---

Interruptible Tools

DefineInterruptibleTool adds typed interrupt/resume for human-in-the-loop workflows. The function receives a *Res parameter that is non-nil when the tool is being re-executed after an interrupt. Like DefineTool, multipart responses are supported via tool.AttachParts:

typeTransferInputstruct {
ToAccountstring`json:"toAccount"`Amountfloat64`json:"amount"`
}
typeTransferInterruptstruct {
Reasonstring`json:"reason"`Amountfloat64`json:"amount"`
}
typeConfirmationstruct {
Approvedbool`json:"approved"`
}
transferTool:=genkitx.DefineInterruptibleTool(g, "transfer",
"Transfers money to another account.",
func(ctx context.Context, inputTransferInput, confirm*Confirmation) (string, error) {
ifconfirm!=nil&&!confirm.Approved {
return"cancelled", nil
}
ifconfirm==nil&&input.Amount>100 {
return"", tool.Interrupt(TransferInterrupt{
Reason: "large_amount",
Amount: input.Amount,
})
}
return"completed", nil
},
)

Handle the interrupt on the caller side. Use transferTool.Resume for typed, compile-time-checked resume data:

resp, _:=genkit.Generate(ctx, g,
ai.WithPrompt("Transfer $200 to Alice"),
ai.WithTools(transferTool),
)
for_, interrupt:=rangeresp.Interrupts() {
meta, _:= tool.InterruptAs[TransferInterrupt](interrupt)
fmt.Printf("Transfer of $%.2f requires confirmation (reason: %s)\n", meta.Amount, meta.Reason)
// Resume: re-execute the tool with typed resume data.restart, _:=transferTool.Resume(interrupt, Confirmation{Approved: true})
resp, _=genkit.Generate(ctx, g,
ai.WithMessages(resp.History()...),
ai.WithTools(transferTool),
ai.WithToolRestarts(restart),
)
}

Or respond with a pre-computed result instead of re-executing the tool:

// Respond: provide the tool's output directly, skipping re-execution.response, _:=transferTool.Respond(interrupt, "cancelled by user")
resp, _=genkit.Generate(ctx, g,
ai.WithMessages(resp.History()...),
ai.WithTools(transferTool),
ai.WithToolResponses(response),
)

When the caller does not have access to the tool definition (e.g., in a generic interrupt handler), use the definition-free helpers tool.Resume and tool.Respond:

restart, _:=tool.Resume(interrupt, Confirmation{Approved: true})
// orresponse, _:=tool.Respond(interrupt, "cancelled by user")

---

Interrupts in an Agent Conversation

Interruptible tools work the same way inside an agent (genkit/exp): the agent streams the model's reply and pauses when a tool interrupts, and the client resumes over the same connection. Back a prompt agent with the transfer tool from above, drive a turn, and resume until it settles (a resumed turn can interrupt again):

agent:=genkitx.DefinePromptAgent[any](g, "banker",
aix.WithSessionStore(localstore.NewInMemorySessionStore[any]()),
)
conn, _:=agent.Connect(ctx)
deferconn.Close()
conn.SendText("Transfer $200 to Alice")
for {
// Stream one turn: print the reply and collect any tool interrupts.varinterrupts []*ai.Partvarend*aix.TurnEndforchunk, err:=rangeconn.Receive() {
iferr!=nil {
log.Fatal(err)
}
ifchunk.ModelChunk!=nil {
fmt.Print(chunk.ModelChunk.Text())
interrupts=append(interrupts, chunk.ModelChunk.Interrupts()...)
}
ifchunk.TurnEnd!=nil { // FinishReason is "interrupted" when pausedend=chunk.TurnEndbreak
}
}
ifend==nil||len(interrupts) ==0 {
break// the turn settled on its own (or the stream closed)
}
// Resolve each interrupt and resume. tool.Resume re-executes the tool with// typed data; tool.Respond would instead supply a result without re-running.resume:=&aix.ToolResume{}
for_, interrupt:=rangeinterrupts {
meta, _:= tool.InterruptAs[TransferInterrupt](interrupt)
fmt.Printf("\n[%s] $%.2f -- approve?\n", meta.Reason, meta.Amount)
restart, _:=transferTool.Resume(interrupt, Confirmation{Approved: true})
resume.Restart=append(resume.Restart, restart)
}
conn.SendResume(resume) // the resumed turn streams on the next Receive
}

---

API Reference

Runtime Helpers (ai/exp/tool -- experimental)

The tool package provides functions for use inside tool functions:

// Interrupt pauses tool execution and sends data to the caller.funcInterrupt(dataany) error// InterruptAs extracts typed interrupt data from an interrupted tool request part.funcInterruptAs[Tany](p*ai.Part) (T, bool)
// Resume creates a restart part for resuming an interrupted tool call.// Does not require access to the tool definition.funcResume[Resany](interruptedPart*ai.Part, dataRes) (*ai.Part, error)
// Respond creates a tool response part for an interrupted tool request.// Provides a pre-computed result instead of re-executing the tool.// Does not require access to the tool definition.funcRespond(interruptedPart*ai.Part, outputany) (*ai.Part, error)
// AttachParts attaches additional content parts (e.g., media) to the tool's// response without changing the function signature.funcAttachParts(ctx context.Context, parts...*ai.Part)
// OriginalInput returns the original input if the caller replaced it during restart.funcOriginalInput[Inany](ctx context.Context) (In, bool)
// SendPartial streams a partial tool response during execution (e.g., progress).// Best-effort: no-ops when streaming is not available.funcSendPartial(ctx context.Context, outputany)
// SendChunk streams a raw ModelResponseChunk during tool execution.// Unlike SendPartial (which wraps data as a partial tool response),// this gives the tool full control over the chunk contents.// Best-effort: no-ops when streaming is not available.funcSendChunk(ctx context.Context, chunk*ai.ModelResponseChunk)

Partial Tool Response Helpers (ai package)

// Part.IsPartial reports whether a Part is a partial (streaming) tool response.func (p*Part) IsPartial() bool// NewPartialToolResponsePart creates a Part marked as a partial tool response.funcNewPartialToolResponsePart(r*ToolResponse) *Part// ModelResponseChunk.ToolResponses returns tool response parts from a chunk.// Use Part.IsPartial() to distinguish progress updates from final results.func (c*ModelResponseChunk) ToolResponses() []*Part

Tool Definitions (ai/exp -- experimental)

Define & Register

// DefineTool creates a tool with a plain context.Context signature and registers it.funcDefineTool[In, Outany](
rRegistry, name, descriptionstring,
fnToolFunc[In, Out],
opts...ai.ToolOption,
) *Tool[In, Out]
// DefineInterruptibleTool creates a tool with typed interrupt/resume and registers it.funcDefineInterruptibleTool[In, Out, Resany](
rRegistry, name, descriptionstring,
fnInterruptibleToolFunc[In, Out, Res],
opts...ai.ToolOption,
) *InterruptibleTool[In, Out, Res]

Create Without Registering

// NewTool creates an unregistered tool.funcNewTool[In, Outany](name, descriptionstring, fnToolFunc[In, Out], opts...ai.ToolOption) *Tool[In, Out]
// NewInterruptibleTool creates an unregistered interruptible tool.funcNewInterruptibleTool[In, Out, Resumeany](name, descriptionstring, fnInterruptibleToolFunc[In, Out, Resume], opts...ai.ToolOption) *InterruptibleTool[In, Out, Resume]

Genkit Convenience Functions (genkit/exp -- experimental)

The genkit/exp package (aliased genkitx) wraps the registry-level constructors above to register through a *genkit.Genkit instead of a raw Registry. The registry stays internal; these reach it through an internal bridge, so no public registry accessor is added. The same package also hosts the experimental agent constructors (DefineAgent, DefinePromptAgent, DefineCustomAgent).

// DefineTool registers a simple tool via the Genkit instance.funcDefineTool[In, Outany](g*genkit.Genkit, name, descriptionstring, fn aix.ToolFunc[In, Out], opts...ai.ToolOption) *aix.Tool[In, Out]
// DefineInterruptibleTool registers an interruptible tool via the Genkit instance.funcDefineInterruptibleTool[In, Out, Resumeany](g*genkit.Genkit, name, descriptionstring, fn aix.InterruptibleToolFunc[In, Out, Resume], opts...ai.ToolOption) *aix.InterruptibleTool[In, Out, Resume]

Function Signatures

// ToolFunc -- simple tools.typeToolFunc[In, Outany] =func(ctx context.Context, inputIn) (Out, error)
// InterruptibleToolFunc -- tools with typed resume.// The resumed parameter is non-nil when the tool is re-executed after an interrupt.typeInterruptibleToolFunc[In, Out, Resumeany] =func(ctx context.Context, inputIn, res*Resume) (Out, error)

InterruptibleTool[In, Out, Res]

// Resume creates a restart part for this tool with typed data.// Validates that the interrupted part belongs to this tool.func (*InterruptibleTool) Resume(interruptedPart*ai.Part, dataRes) (*ai.Part, error)
// Respond creates a tool response part for this tool with typed output.// Provides a pre-computed result instead of re-executing the tool.// Validates that the interrupted part belongs to this tool.func (*InterruptibleTool) Respond(interruptedPart*ai.Part, outputOut) (*ai.Part, error)

Tool[In, Out]

Implements ai.Tool -- usable anywhere a tool is accepted (ai.WithTools, etc.).

func (*Tool) Name() stringfunc (*Tool) Definition() *ai.ToolDefinitionfunc (*Tool) RunRaw(ctx context.Context, inputany) (any, error)
func (*Tool) RunRawMultipart(ctx context.Context, inputany) (*ai.MultipartToolResponse, error)
func (*Tool) Respond(toolReq*ai.Part, outputDataany, opts*ai.RespondOptions) *ai.Partfunc (*Tool) Register(r api.Registry)