Compare commits

..

1 Commits

Author SHA1 Message Date
Codex e4c8e8cf5e Correlate agent trace spans
Harness (E2E) / Harnesses (mock LLM) (push) Waiting to run
Harness (E2E) / Provider harnesses (live LLM conformance) (push) Waiting to run
Lint / golangci-lint (push) Waiting to run
Run Tests / Unit Tests (push) Waiting to run
Run Tests / Etcd Integration Tests (push) Waiting to run
2026-06-25 22:09:21 +00:00
67 changed files with 268 additions and 1710 deletions
+2 -19
View File
@@ -27,13 +27,11 @@ jobs:
cache: true
- name: Build
run: go build ./...
- name: 0→1 scaffold contract
run: go test ./cmd/micro/cli/new -run TestZeroToOneContract -count=1
- name: Universe end-to-end (asserts; exits non-zero on failure)
run: go run ./internal/harness/universe
- name: Agent-flow harness
run: go run ./internal/harness/agent-flow
- name: 0→hero plan-delegate workflow harness
- name: Plan-delegate harness
run: go run ./internal/harness/plan-delegate
harness-live:
@@ -59,19 +57,4 @@ jobs:
MISTRAL_API_KEY: ${{ secrets.MISTRAL_API_KEY }}
TOGETHER_API_KEY: ${{ secrets.TOGETHER_API_KEY }}
ATLASCLOUD_API_KEY: ${{ secrets.ATLASCLOUD_API_KEY }}
run: |
go run ./internal/harness/provider-conformance \
-require-configured \
-summary-json provider-conformance-summary.json \
-summary-markdown provider-conformance-summary.md \
-capabilities-markdown provider-capabilities.md
- name: Upload provider conformance summary
if: always()
uses: actions/upload-artifact@v4
with:
name: provider-conformance
path: |
provider-conformance-summary.json
provider-conformance-summary.md
provider-capabilities.md
if-no-files-found: ignore
run: go run ./internal/harness/provider-conformance -require-configured
-1
View File
@@ -7,7 +7,6 @@ on:
types:
- opened
- reopened
- synchronize
branches:
- "**"
jobs:
+1 -1
View File
@@ -47,7 +47,7 @@ test-coverage:
harness:
go run ./internal/harness/universe
go run ./internal/harness/agent-flow
go run ./internal/harness/plan-delegate # 0→hero: services + agents + flow + plan/delegate
go run ./internal/harness/plan-delegate
# Run the same harnesses against every configured live provider. Providers
# without API keys are skipped; configured providers must pass.
+3 -5
View File
@@ -2,9 +2,7 @@
Go Micro is an **agent harness** and service framework for Go.
A harness is the runtime around an agent: the tools it can call, the memory it keeps, the guardrails that bound it, the workflows that trigger it, the services it depends on, and the protocols other agents use to reach it.
Go Micro gives you the harness as Go code. Build an agent and it gets a model, memory, tools, planning, delegation, guardrails, and service discovery; it is reachable over [MCP](https://modelcontextprotocol.io/) and [A2A](https://a2a-protocol.org). Write services and every endpoint becomes an AI-callable tool. Orchestrate the deterministic parts with durable flows. Agents, services, and flows share one runtime because an agent is a distributed system, and building one is building a service.
A harness is the runtime around an agent: the tools it can call, the memory it keeps, the guardrails that bound it, the workflows that trigger it, the services it depends on, and the protocols other agents use to reach it. Go Micro gives you that harness as Go code. Build an agent and it gets a model, memory, tools, planning, delegation, guardrails, and service discovery; it is reachable over [MCP](https://modelcontextprotocol.io/) and [A2A](https://a2a-protocol.org). Write services and every endpoint becomes an AI-callable tool. Orchestrate the deterministic parts with durable flows. Agents, services, and flows share one runtime because an agent is a distributed system, and building one is building a service.
## Sponsors
@@ -255,9 +253,9 @@ Every endpoint is an AI-callable tool — and it can be a *paid* tool. Go Micro
```bash
# Charge for tool calls at the MCP gateway (off unless you set a pay-to address)
micro mcp serve --x402_pay_to 0xYourAddress --x402_network solana --x402_amount 10000
micro mcp serve --x402-pay-to 0xYourAddress --x402-network solana --x402-amount 10000
# Per-tool amounts via a config file
micro mcp serve --x402_config x402.json
micro mcp serve --x402-config x402.json
```
See the [Payments (x402) guide](internal/website/docs/guides/x402-payments.md).
+4 -10
View File
@@ -136,7 +136,7 @@ func (a *agentImpl) setup() {
a.tools = ai.NewTools(a.opts.Registry, ai.ToolClient(a.opts.Client))
modelOpts = append(modelOpts, ai.WithToolHandler(a.toolHandler()))
a.model = ai.New(a.opts.Provider, modelOpts...)
if a.model != nil {
if a.opts.TraceProvider != nil && a.model != nil {
a.model = a.tracedModel(a.model)
}
@@ -168,10 +168,6 @@ func (a *agentImpl) stateStore() store.Store {
// Ask sends a message and returns the agent's response.
// This is the programmatic API for direct use.
func (a *agentImpl) Ask(ctx context.Context, message string) (*Response, error) {
return a.ask(ctx, message, a.parentRunID)
}
func (a *agentImpl) ask(ctx context.Context, message, parentRunID string) (*Response, error) {
a.mu.Lock()
defer a.mu.Unlock()
@@ -192,7 +188,7 @@ func (a *agentImpl) ask(ctx context.Context, message, parentRunID string) (*Resp
a.runID = uuid.New().String()
ctx = ai.WithRunInfo(ctx, ai.RunInfo{
RunID: a.runID,
ParentID: parentRunID,
ParentID: a.parentRunID,
Agent: a.opts.Name,
})
ctx, endRun := a.startRun(ctx, message)
@@ -232,21 +228,19 @@ func (a *agentImpl) ask(ctx context.Context, message, parentRunID string) (*Resp
ToolCalls: resp.ToolCalls,
Agent: a.opts.Name,
RunID: a.runID,
ParentID: parentRunID,
ParentID: a.parentRunID,
}, nil
}
// Chat implements the proto AgentHandler interface for RPC.
// @example {"message": "What tasks are overdue?"}
func (a *agentImpl) Chat(ctx context.Context, req *pb.ChatRequest, rsp *pb.ChatResponse) error {
resp, err := a.ask(ctx, req.Message, req.ParentId)
resp, err := a.Ask(ctx, req.Message)
if err != nil {
return err
}
rsp.Reply = resp.Reply
rsp.Agent = resp.Agent
rsp.RunId = resp.RunID
rsp.ParentId = resp.ParentID
for _, tc := range resp.ToolCalls {
input, _ := json.Marshal(tc.Input)
rsp.ToolCalls = append(rsp.ToolCalls, &pb.ToolCall{
-49
View File
@@ -1,11 +1,7 @@
package agent
import (
"context"
"testing"
pb "go-micro.dev/v6/agent/proto"
"go-micro.dev/v6/ai"
)
func TestNew(t *testing.T) {
@@ -38,51 +34,6 @@ func TestNew(t *testing.T) {
}
}
func TestChatResponseIncludesRunIDs(t *testing.T) {
fakeGen = func(ctx context.Context, opts ai.Options, req *ai.Request) (*ai.Response, error) {
return &ai.Response{Reply: "ok"}, nil
}
defer func() { fakeGen = nil }()
a := newTestAgent(Name("chat-run"))
var rsp pb.ChatResponse
if err := a.Chat(context.Background(), &pb.ChatRequest{Message: "hello"}, &rsp); err != nil {
t.Fatalf("Chat: %v", err)
}
if rsp.RunId == "" {
t.Fatal("Chat response RunId is empty")
}
if rsp.Agent != "chat-run" {
t.Errorf("Agent = %q, want chat-run", rsp.Agent)
}
if rsp.ParentId != "" {
t.Errorf("ParentId = %q, want empty", rsp.ParentId)
}
}
func TestChatRequestParentIDPropagatesToResponse(t *testing.T) {
fakeGen = func(ctx context.Context, opts ai.Options, req *ai.Request) (*ai.Response, error) {
info, ok := ai.RunInfoFrom(ctx)
if !ok {
t.Fatal("RunInfo missing from model context")
}
if info.ParentID != "flow-run-123" {
t.Fatalf("RunInfo.ParentID = %q, want flow-run-123", info.ParentID)
}
return &ai.Response{Reply: "ok"}, nil
}
defer func() { fakeGen = nil }()
a := newTestAgent(Name("chat-child"))
var rsp pb.ChatResponse
if err := a.Chat(context.Background(), &pb.ChatRequest{Message: "hello", ParentId: "flow-run-123"}, &rsp); err != nil {
t.Fatalf("Chat: %v", err)
}
if rsp.ParentId != "flow-run-123" {
t.Errorf("ParentId = %q, want flow-run-123", rsp.ParentId)
}
}
func TestBuildPrompt(t *testing.T) {
// Custom prompt
a := New(Name("test"), Prompt("custom prompt")).(*agentImpl)
-16
View File
@@ -108,7 +108,6 @@ func (a *agentImpl) toolHandler() ai.ToolHandler {
h = a.loopWrap(h)
h = a.stepWrap(h)
h = a.planWrap(h)
h = contextWrap(h)
h = a.traceTool(h)
for i := len(a.opts.wrappers) - 1; i >= 0; i-- {
h = a.opts.wrappers[i](h)
@@ -116,21 +115,6 @@ func (a *agentImpl) toolHandler() ai.ToolHandler {
return h
}
// contextWrap stops tool execution promptly when the Ask context has
// already been canceled or its deadline has expired. This keeps guardrail
// bookkeeping and side-effecting tools from running after the caller has
// abandoned the agent run.
func contextWrap(next ai.ToolHandler) ai.ToolHandler {
return func(ctx context.Context, call ai.ToolCall) ai.ToolResult {
select {
case <-ctx.Done():
return errResult(call.ID, ctx.Err().Error())
default:
}
return next(ctx, call)
}
}
// baseHandler executes a tool call: a developer custom tool, the built-in
// delegate, or an RPC to the service. It is the innermost handler.
func (a *agentImpl) baseHandler() ai.ToolHandler {
+2 -3
View File
@@ -250,9 +250,8 @@ func WithTool(name, description string, properties map[string]any, handler ToolF
}
}
// TraceProvider enables OpenTelemetry tracing for agent runs. The persisted
// run timeline is recorded even when TraceProvider is nil; trace/span IDs are
// added only when a provider is configured.
// TraceProvider enables OpenTelemetry tracing for agent runs. When nil,
// agent tracing and run timeline recording are disabled.
func TraceProvider(tp trace.TracerProvider) Option {
return func(o *Options) { o.TraceProvider = tp }
}
+12 -47
View File
@@ -62,10 +62,6 @@ type RunListOptions struct {
// Status, when set, keeps only runs with the matching status
// (for example "running", "done", "error", or "refused").
Status string
// TraceID, when set, keeps only runs correlated with this trace id.
// A prefix is accepted so operators can paste the shortened trace id
// printed by `micro runs`.
TraceID string
// Limit, when positive, returns the most recently updated runs up to
// the limit. Limited results are ordered newest first.
Limit int
@@ -92,23 +88,13 @@ func (a *agentImpl) tracer() trace.Tracer {
}
func (a *agentImpl) startRun(ctx context.Context, message string) (context.Context, func(error)) {
info, _ := ai.RunInfoFrom(ctx)
start := time.Now()
if a.opts.TraceProvider == nil {
a.recordRunEvent(RunEvent{Time: start, RunID: info.RunID, ParentID: info.ParentID, Agent: info.Agent, Kind: "run", Name: message})
return ctx, func(err error) {
latency := time.Since(start).Milliseconds()
if err != nil {
a.recordRunEvent(RunEvent{Time: time.Now(), RunID: info.RunID, ParentID: info.ParentID, Agent: info.Agent, Kind: "error", LatencyMS: latency, Error: err.Error()})
return
}
a.recordRunEvent(RunEvent{Time: time.Now(), RunID: info.RunID, ParentID: info.ParentID, Agent: info.Agent, Kind: "done", LatencyMS: latency})
}
return ctx, func(error) {}
}
info, _ := ai.RunInfoFrom(ctx)
ctx, span := a.tracer().Start(ctx, spanNameRun, trace.WithSpanKind(trace.SpanKindInternal), trace.WithAttributes(
attribute.String(AttrRunID, info.RunID), attribute.String(AttrParentRunID, info.ParentID), attribute.String(AttrAgentName, info.Agent)))
start := time.Now()
a.recordSpanEvent(span, RunEvent{Time: start, RunID: info.RunID, ParentID: info.ParentID, Agent: info.Agent, Kind: "run", Name: message})
return ctx, func(err error) {
latency := time.Since(start).Milliseconds()
@@ -132,26 +118,12 @@ type tracedModel struct {
func (a *agentImpl) tracedModel(m ai.Model) ai.Model { return &tracedModel{Model: m, a: a} }
func (m *tracedModel) Generate(ctx context.Context, req *ai.Request, opts ...ai.GenerateOption) (*ai.Response, error) {
if m.a.opts.TraceProvider == nil {
return m.Model.Generate(ctx, req, opts...)
}
info, _ := ai.RunInfoFrom(ctx)
provider := m.String()
model := m.Options().Model
start := time.Now()
if m.a.opts.TraceProvider == nil {
resp, err := m.Model.Generate(ctx, req, opts...)
dur := time.Since(start).Milliseconds()
usage := ai.Usage{}
if resp != nil {
usage = resp.Usage
}
e := RunEvent{Time: time.Now(), RunID: info.RunID, ParentID: info.ParentID, Agent: info.Agent, Kind: "model", Provider: provider, Model: model, LatencyMS: dur, Tokens: usage}
if err != nil {
e.Error = err.Error()
}
m.a.recordRunEvent(e)
return resp, err
}
ctx, span := m.a.tracer().Start(ctx, spanNameModelCall, trace.WithAttributes(
attribute.String(AttrRunID, info.RunID),
attribute.String(AttrParentRunID, info.ParentID),
@@ -159,6 +131,7 @@ func (m *tracedModel) Generate(ctx context.Context, req *ai.Request, opts ...ai.
attribute.String(AttrProvider, provider),
attribute.String(AttrModel, model),
))
start := time.Now()
resp, err := m.Model.Generate(ctx, req, opts...)
dur := time.Since(start).Milliseconds()
attrs := []attribute.KeyValue{attribute.Int64(AttrLatencyMS, dur)}
@@ -197,17 +170,11 @@ func appendUsage(attrs []attribute.KeyValue, u ai.Usage) []attribute.KeyValue {
}
func (a *agentImpl) traceTool(next ai.ToolHandler) ai.ToolHandler {
if a.opts.TraceProvider == nil {
return next
}
return func(ctx context.Context, call ai.ToolCall) ai.ToolResult {
info, _ := ai.RunInfoFrom(ctx)
start := time.Now()
if a.opts.TraceProvider == nil {
res := next(ctx, call)
dur := time.Since(start).Milliseconds()
a.recordRunEvent(RunEvent{Time: time.Now(), RunID: info.RunID, ParentID: info.ParentID, Agent: info.Agent, Kind: "tool", Name: call.Name, LatencyMS: dur, Refused: res.Refused, Error: resultError(res)})
return res
}
ctx, span := a.tracer().Start(ctx, spanNameToolCall, trace.WithAttributes(
attribute.String(AttrRunID, info.RunID),
attribute.String(AttrParentRunID, info.ParentID),
@@ -215,6 +182,7 @@ func (a *agentImpl) traceTool(next ai.ToolHandler) ai.ToolHandler {
attribute.String(AttrToolName, call.Name),
attribute.Bool(AttrDelegate, call.Name == toolDelegate),
))
start := time.Now()
res := next(ctx, call)
dur := time.Since(start).Milliseconds()
attrs := []attribute.KeyValue{attribute.Int64(AttrLatencyMS, dur)}
@@ -257,7 +225,7 @@ func (a *agentImpl) recordSpanEvent(span trace.Span, e RunEvent) {
}
func (a *agentImpl) recordRunEvent(e RunEvent) {
if e.RunID == "" {
if a.opts.TraceProvider == nil || e.RunID == "" {
return
}
b, _ := json.Marshal(e)
@@ -336,9 +304,6 @@ func ListRunSummariesWithOptions(s store.Store, agentName string, opts RunListOp
if opts.Status != "" && summary.Status != opts.Status {
continue
}
if opts.TraceID != "" && !strings.HasPrefix(summary.TraceID, opts.TraceID) {
continue
}
summaries = append(summaries, summary)
}
if opts.Limit > 0 {
+7 -33
View File
@@ -117,7 +117,7 @@ func spanAttributes(attrs []attribute.KeyValue) map[string]string {
return out
}
func TestAgentRunTimelineRecordsModelAndToolWithoutTraceProvider(t *testing.T) {
func TestAgentOpenTelemetryNoopWhenUnconfigured(t *testing.T) {
st := store.NewMemoryStore()
a := New(Name("runner-noop"), Provider("oteltest"), WithStore(st), WithTool("probe", "probe", nil, func(context.Context, map[string]any) (string, error) { return "ok", nil }))
if _, err := a.Ask(context.Background(), "hello"); err != nil {
@@ -127,37 +127,11 @@ func TestAgentRunTimelineRecordsModelAndToolWithoutTraceProvider(t *testing.T) {
if err != nil {
t.Fatal(err)
}
if len(keys) == 0 {
t.Fatal("expected run timeline without TraceProvider")
if len(keys) != 0 {
t.Fatalf("expected no run timeline without TraceProvider, got %v", keys)
}
summaries, err := ListRunSummaries(st, "runner-noop")
if err != nil {
t.Fatal(err)
}
if len(summaries) != 1 {
t.Fatalf("got %d summaries, want 1", len(summaries))
}
if summaries[0].Status != "done" || summaries[0].LastKind != "done" {
t.Fatalf("unexpected summary without TraceProvider: %#v", summaries[0])
}
if summaries[0].TraceID != "" || summaries[0].SpanID != "" {
t.Fatalf("unexpected trace correlation without TraceProvider: %#v", summaries[0])
}
events, err := LoadRunEvents(st, "runner-noop", summaries[0].RunID)
if err != nil {
t.Fatal(err)
}
seen := map[string]bool{"run": false, "model": false, "tool": false, "done": false}
for _, e := range events {
seen[e.Kind] = true
if e.TraceID != "" || e.SpanID != "" {
t.Fatalf("event has trace correlation without TraceProvider: %#v", e)
}
}
for kind, ok := range seen {
if !ok {
t.Fatalf("missing %s event in timeline: %#v", kind, events)
}
if _, ok := a.(*agentImpl).model.(*tracedModel); ok {
t.Fatal("model should not be wrapped when TraceProvider is nil")
}
}
@@ -236,7 +210,7 @@ func TestListRunSummariesWithOptionsFiltersAndLimits(t *testing.T) {
events := []RunEvent{
{Time: time.Unix(0, 1), RunID: "run-old", Agent: "runner", Kind: "run"},
{Time: time.Unix(0, 2), RunID: "run-old", Agent: "runner", Kind: "done"},
{Time: time.Unix(0, 3), RunID: "run-new", Agent: "runner", TraceID: "abcdef1234567890", Kind: "run"},
{Time: time.Unix(0, 3), RunID: "run-new", Agent: "runner", Kind: "run"},
{Time: time.Unix(0, 4), RunID: "run-new", Agent: "runner", Kind: "error", Error: "boom"},
}
for _, e := range events {
@@ -249,7 +223,7 @@ func TestListRunSummariesWithOptionsFiltersAndLimits(t *testing.T) {
}
}
got, err := ListRunSummariesWithOptions(st, "runner", RunListOptions{Status: "error", TraceID: "abcdef", Limit: 1})
got, err := ListRunSummariesWithOptions(st, "runner", RunListOptions{Status: "error", Limit: 1})
if err != nil {
t.Fatal(err)
}
+40 -70
View File
@@ -2,7 +2,7 @@
// versions:
// protoc-gen-go v1.36.11
// protoc v3.21.12
// source: agent/proto/agent.proto
// source: proto/agent.proto
package agent
@@ -22,17 +22,15 @@ const (
)
type ChatRequest struct {
state protoimpl.MessageState `protogen:"open.v1"`
Message string `protobuf:"bytes,1,opt,name=message,proto3" json:"message,omitempty"`
// parent_id correlates this chat with the workflow or agent run that dispatched it.
ParentId string `protobuf:"bytes,2,opt,name=parent_id,json=parentId,proto3" json:"parent_id,omitempty"`
state protoimpl.MessageState `protogen:"open.v1"`
Message string `protobuf:"bytes,1,opt,name=message,proto3" json:"message,omitempty"`
unknownFields protoimpl.UnknownFields
sizeCache protoimpl.SizeCache
}
func (x *ChatRequest) Reset() {
*x = ChatRequest{}
mi := &file_agent_proto_agent_proto_msgTypes[0]
mi := &file_proto_agent_proto_msgTypes[0]
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
ms.StoreMessageInfo(mi)
}
@@ -44,7 +42,7 @@ func (x *ChatRequest) String() string {
func (*ChatRequest) ProtoMessage() {}
func (x *ChatRequest) ProtoReflect() protoreflect.Message {
mi := &file_agent_proto_agent_proto_msgTypes[0]
mi := &file_proto_agent_proto_msgTypes[0]
if x != nil {
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
if ms.LoadMessageInfo() == nil {
@@ -57,7 +55,7 @@ func (x *ChatRequest) ProtoReflect() protoreflect.Message {
// Deprecated: Use ChatRequest.ProtoReflect.Descriptor instead.
func (*ChatRequest) Descriptor() ([]byte, []int) {
return file_agent_proto_agent_proto_rawDescGZIP(), []int{0}
return file_proto_agent_proto_rawDescGZIP(), []int{0}
}
func (x *ChatRequest) GetMessage() string {
@@ -67,29 +65,18 @@ func (x *ChatRequest) GetMessage() string {
return ""
}
func (x *ChatRequest) GetParentId() string {
if x != nil {
return x.ParentId
}
return ""
}
type ChatResponse struct {
state protoimpl.MessageState `protogen:"open.v1"`
Reply string `protobuf:"bytes,1,opt,name=reply,proto3" json:"reply,omitempty"`
Agent string `protobuf:"bytes,2,opt,name=agent,proto3" json:"agent,omitempty"`
ToolCalls []*ToolCall `protobuf:"bytes,3,rep,name=tool_calls,json=toolCalls,proto3" json:"tool_calls,omitempty"`
// run_id correlates this chat response with tool calls, traces, and run history.
RunId string `protobuf:"bytes,4,opt,name=run_id,json=runId,proto3" json:"run_id,omitempty"`
// parent_id is set when this response belongs to a delegated sub-agent run.
ParentId string `protobuf:"bytes,5,opt,name=parent_id,json=parentId,proto3" json:"parent_id,omitempty"`
state protoimpl.MessageState `protogen:"open.v1"`
Reply string `protobuf:"bytes,1,opt,name=reply,proto3" json:"reply,omitempty"`
Agent string `protobuf:"bytes,2,opt,name=agent,proto3" json:"agent,omitempty"`
ToolCalls []*ToolCall `protobuf:"bytes,3,rep,name=tool_calls,json=toolCalls,proto3" json:"tool_calls,omitempty"`
unknownFields protoimpl.UnknownFields
sizeCache protoimpl.SizeCache
}
func (x *ChatResponse) Reset() {
*x = ChatResponse{}
mi := &file_agent_proto_agent_proto_msgTypes[1]
mi := &file_proto_agent_proto_msgTypes[1]
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
ms.StoreMessageInfo(mi)
}
@@ -101,7 +88,7 @@ func (x *ChatResponse) String() string {
func (*ChatResponse) ProtoMessage() {}
func (x *ChatResponse) ProtoReflect() protoreflect.Message {
mi := &file_agent_proto_agent_proto_msgTypes[1]
mi := &file_proto_agent_proto_msgTypes[1]
if x != nil {
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
if ms.LoadMessageInfo() == nil {
@@ -114,7 +101,7 @@ func (x *ChatResponse) ProtoReflect() protoreflect.Message {
// Deprecated: Use ChatResponse.ProtoReflect.Descriptor instead.
func (*ChatResponse) Descriptor() ([]byte, []int) {
return file_agent_proto_agent_proto_rawDescGZIP(), []int{1}
return file_proto_agent_proto_rawDescGZIP(), []int{1}
}
func (x *ChatResponse) GetReply() string {
@@ -138,20 +125,6 @@ func (x *ChatResponse) GetToolCalls() []*ToolCall {
return nil
}
func (x *ChatResponse) GetRunId() string {
if x != nil {
return x.RunId
}
return ""
}
func (x *ChatResponse) GetParentId() string {
if x != nil {
return x.ParentId
}
return ""
}
type ToolCall struct {
state protoimpl.MessageState `protogen:"open.v1"`
Id string `protobuf:"bytes,1,opt,name=id,proto3" json:"id,omitempty"`
@@ -164,7 +137,7 @@ type ToolCall struct {
func (x *ToolCall) Reset() {
*x = ToolCall{}
mi := &file_agent_proto_agent_proto_msgTypes[2]
mi := &file_proto_agent_proto_msgTypes[2]
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
ms.StoreMessageInfo(mi)
}
@@ -176,7 +149,7 @@ func (x *ToolCall) String() string {
func (*ToolCall) ProtoMessage() {}
func (x *ToolCall) ProtoReflect() protoreflect.Message {
mi := &file_agent_proto_agent_proto_msgTypes[2]
mi := &file_proto_agent_proto_msgTypes[2]
if x != nil {
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
if ms.LoadMessageInfo() == nil {
@@ -189,7 +162,7 @@ func (x *ToolCall) ProtoReflect() protoreflect.Message {
// Deprecated: Use ToolCall.ProtoReflect.Descriptor instead.
func (*ToolCall) Descriptor() ([]byte, []int) {
return file_agent_proto_agent_proto_rawDescGZIP(), []int{2}
return file_proto_agent_proto_rawDescGZIP(), []int{2}
}
func (x *ToolCall) GetId() string {
@@ -220,21 +193,18 @@ func (x *ToolCall) GetResult() string {
return ""
}
var File_agent_proto_agent_proto protoreflect.FileDescriptor
var File_proto_agent_proto protoreflect.FileDescriptor
const file_agent_proto_agent_proto_rawDesc = "" +
const file_proto_agent_proto_rawDesc = "" +
"\n" +
"\x17agent/proto/agent.proto\x12\x05agent\"D\n" +
"\x11proto/agent.proto\x12\x05agent\"'\n" +
"\vChatRequest\x12\x18\n" +
"\amessage\x18\x01 \x01(\tR\amessage\x12\x1b\n" +
"\tparent_id\x18\x02 \x01(\tR\bparentId\"\x9e\x01\n" +
"\amessage\x18\x01 \x01(\tR\amessage\"j\n" +
"\fChatResponse\x12\x14\n" +
"\x05reply\x18\x01 \x01(\tR\x05reply\x12\x14\n" +
"\x05agent\x18\x02 \x01(\tR\x05agent\x12.\n" +
"\n" +
"tool_calls\x18\x03 \x03(\v2\x0f.agent.ToolCallR\ttoolCalls\x12\x15\n" +
"\x06run_id\x18\x04 \x01(\tR\x05runId\x12\x1b\n" +
"\tparent_id\x18\x05 \x01(\tR\bparentId\"\\\n" +
"tool_calls\x18\x03 \x03(\v2\x0f.agent.ToolCallR\ttoolCalls\"\\\n" +
"\bToolCall\x12\x0e\n" +
"\x02id\x18\x01 \x01(\tR\x02id\x12\x12\n" +
"\x04name\x18\x02 \x01(\tR\x04name\x12\x14\n" +
@@ -244,24 +214,24 @@ const file_agent_proto_agent_proto_rawDesc = "" +
"\x04Chat\x12\x12.agent.ChatRequest\x1a\x13.agent.ChatResponse\"\x00B\x0fZ\r./proto;agentb\x06proto3"
var (
file_agent_proto_agent_proto_rawDescOnce sync.Once
file_agent_proto_agent_proto_rawDescData []byte
file_proto_agent_proto_rawDescOnce sync.Once
file_proto_agent_proto_rawDescData []byte
)
func file_agent_proto_agent_proto_rawDescGZIP() []byte {
file_agent_proto_agent_proto_rawDescOnce.Do(func() {
file_agent_proto_agent_proto_rawDescData = protoimpl.X.CompressGZIP(unsafe.Slice(unsafe.StringData(file_agent_proto_agent_proto_rawDesc), len(file_agent_proto_agent_proto_rawDesc)))
func file_proto_agent_proto_rawDescGZIP() []byte {
file_proto_agent_proto_rawDescOnce.Do(func() {
file_proto_agent_proto_rawDescData = protoimpl.X.CompressGZIP(unsafe.Slice(unsafe.StringData(file_proto_agent_proto_rawDesc), len(file_proto_agent_proto_rawDesc)))
})
return file_agent_proto_agent_proto_rawDescData
return file_proto_agent_proto_rawDescData
}
var file_agent_proto_agent_proto_msgTypes = make([]protoimpl.MessageInfo, 3)
var file_agent_proto_agent_proto_goTypes = []any{
var file_proto_agent_proto_msgTypes = make([]protoimpl.MessageInfo, 3)
var file_proto_agent_proto_goTypes = []any{
(*ChatRequest)(nil), // 0: agent.ChatRequest
(*ChatResponse)(nil), // 1: agent.ChatResponse
(*ToolCall)(nil), // 2: agent.ToolCall
}
var file_agent_proto_agent_proto_depIdxs = []int32{
var file_proto_agent_proto_depIdxs = []int32{
2, // 0: agent.ChatResponse.tool_calls:type_name -> agent.ToolCall
0, // 1: agent.Agent.Chat:input_type -> agent.ChatRequest
1, // 2: agent.Agent.Chat:output_type -> agent.ChatResponse
@@ -272,26 +242,26 @@ var file_agent_proto_agent_proto_depIdxs = []int32{
0, // [0:1] is the sub-list for field type_name
}
func init() { file_agent_proto_agent_proto_init() }
func file_agent_proto_agent_proto_init() {
if File_agent_proto_agent_proto != nil {
func init() { file_proto_agent_proto_init() }
func file_proto_agent_proto_init() {
if File_proto_agent_proto != nil {
return
}
type x struct{}
out := protoimpl.TypeBuilder{
File: protoimpl.DescBuilder{
GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
RawDescriptor: unsafe.Slice(unsafe.StringData(file_agent_proto_agent_proto_rawDesc), len(file_agent_proto_agent_proto_rawDesc)),
RawDescriptor: unsafe.Slice(unsafe.StringData(file_proto_agent_proto_rawDesc), len(file_proto_agent_proto_rawDesc)),
NumEnums: 0,
NumMessages: 3,
NumExtensions: 0,
NumServices: 1,
},
GoTypes: file_agent_proto_agent_proto_goTypes,
DependencyIndexes: file_agent_proto_agent_proto_depIdxs,
MessageInfos: file_agent_proto_agent_proto_msgTypes,
GoTypes: file_proto_agent_proto_goTypes,
DependencyIndexes: file_proto_agent_proto_depIdxs,
MessageInfos: file_proto_agent_proto_msgTypes,
}.Build()
File_agent_proto_agent_proto = out.File
file_agent_proto_agent_proto_goTypes = nil
file_agent_proto_agent_proto_depIdxs = nil
File_proto_agent_proto = out.File
file_proto_agent_proto_goTypes = nil
file_proto_agent_proto_depIdxs = nil
}
+1 -1
View File
@@ -1,5 +1,5 @@
// Code generated by protoc-gen-micro. DO NOT EDIT.
// source: agent/proto/agent.proto
// source: proto/agent.proto
package agent
-8
View File
@@ -11,20 +11,12 @@ service Agent {
message ChatRequest {
string message = 1;
// parent_id correlates this chat with the workflow or agent run that dispatched it.
string parent_id = 2;
}
message ChatResponse {
string reply = 1;
string agent = 2;
repeated ToolCall tool_calls = 3;
// run_id correlates this chat response with tool calls, traces, and run history.
string run_id = 4;
// parent_id is set when this response belongs to a delegated sub-agent run.
string parent_id = 5;
}
message ToolCall {
-27
View File
@@ -3,7 +3,6 @@ package agent
import (
"context"
"errors"
"strings"
"testing"
"time"
@@ -76,29 +75,3 @@ func TestAskRetriesTransientErrorsThenSurfacesStructuredError(t *testing.T) {
t.Fatalf("model attempts = %d, want 2", attempts)
}
}
func TestCanceledAskContextSkipsToolExecution(t *testing.T) {
fakeGen = func(ctx context.Context, opts ai.Options, req *ai.Request) (*ai.Response, error) {
if opts.ToolHandler == nil {
t.Fatal("missing tool handler")
}
canceled, cancel := context.WithCancel(ctx)
cancel()
res := opts.ToolHandler(canceled, ai.ToolCall{ID: "call-1", Name: toolPlan, Input: map[string]any{
"steps": []any{map[string]any{"task": "should not persist", "status": "pending"}},
}})
if !strings.Contains(res.Content, context.Canceled.Error()) {
t.Fatalf("tool result = %q, want cancellation error", res.Content)
}
return &ai.Response{Reply: "ok"}, nil
}
defer func() { fakeGen = nil }()
a := newTestAgent(Name("cancel-tools"))
if _, err := a.Ask(context.Background(), "try a canceled tool"); err != nil {
t.Fatalf("Ask: %v", err)
}
if plan := a.loadPlan(); plan != "" {
t.Fatalf("plan persisted after canceled tool context: %q", plan)
}
}
-16
View File
@@ -188,22 +188,6 @@ type Response struct {
- `ToolCalls`: List of tools the model requested (if any)
- `Answer`: The final answer after tools are executed (only set if ToolHandler is provided)
## Provider capability matrix
The CLI can print the provider capabilities registered in the current build:
```bash
micro ai providers
```
For automation and docs generation, emit the same matrix as stable JSON:
```bash
micro ai providers --json
```
It reports support from Go Micro's provider registry, so the matrix reflects the model, image, and video interfaces available to this binary rather than external provider marketing claims.
## Supported Providers
### Anthropic Claude
+8 -32
View File
@@ -5,7 +5,7 @@ import "sort"
// CapabilityRow is one deterministic row in a provider capability matrix.
type CapabilityRow struct {
// Provider is the registered provider name.
Provider string `json:"provider"`
Provider string
Capabilities
}
@@ -14,15 +14,11 @@ type CapabilityRow struct {
// provider marketing claims, so it reflects what this build can actually use.
type Capabilities struct {
// Model reports whether ai.New can construct a chat/text model provider.
Model bool `json:"model"`
Model bool
// Image reports whether ai.NewImage can construct an image model provider.
Image bool `json:"image"`
Image bool
// Video reports whether ai.NewVideo can construct a video model provider.
Video bool `json:"video"`
// Stream reports whether the provider has registered end-to-end token streaming.
// Providers that only satisfy the Model interface with ErrStreamingUnsupported
// leave this false until their Stream implementation is usable.
Stream bool `json:"stream"`
Video bool
}
// ProviderCapabilities reports the capabilities registered for provider.
@@ -30,13 +26,11 @@ func ProviderCapabilities(provider string) Capabilities {
_, hasModel := providers[provider]
_, hasImage := imageProviders[provider]
_, hasVideo := videoProviders[provider]
_, hasStream := streamProviders[provider]
return Capabilities{
Model: hasModel,
Image: hasImage,
Video: hasVideo,
Stream: hasStream,
Model: hasModel,
Image: hasImage,
Video: hasVideo,
}
}
@@ -55,9 +49,6 @@ func CapabilityMatrix() map[string]Capabilities {
for name := range videoProviders {
names[name] = struct{}{}
}
for name := range streamProviders {
names[name] = struct{}{}
}
matrix := make(map[string]Capabilities, len(names))
for name := range names {
@@ -81,17 +72,8 @@ func CapabilityRows() []CapabilityRow {
return rows
}
// RegisterStream records that provider has a usable Stream implementation.
// Providers should call this from init alongside Register once Stream returns
// chunks instead of ErrStreamingUnsupported.
func RegisterStream(provider string) {
streamProviders[provider] = struct{}{}
}
var streamProviders = make(map[string]struct{})
// RegisteredProviders returns the registered provider names in sorted order.
// kind may be "model", "image", "video", "stream", or empty for the union of all
// kind may be "model", "image", "video", or empty for the union of all
// provider registries.
func RegisteredProviders(kind string) []string {
names := map[string]struct{}{}
@@ -109,18 +91,12 @@ func RegisteredProviders(kind string) []string {
for name := range r {
names[name] = struct{}{}
}
case map[string]struct{}:
for name := range r {
names[name] = struct{}{}
}
}
}
switch kind {
case "model":
add(providers)
case "stream":
add(streamProviders)
case "image":
add(imageProviders)
case "video":
-20
View File
@@ -32,12 +32,6 @@ func TestRegisteredProviders(t *testing.T) {
if !reflect.DeepEqual(got, want) {
t.Fatalf("RegisteredProviders(video) = %#v, want %#v", got, want)
}
got = ai.RegisteredProviders("stream")
want = []string{}
if !reflect.DeepEqual(got, want) {
t.Fatalf("RegisteredProviders(stream) = %#v, want %#v", got, want)
}
}
func TestCapabilityRows(t *testing.T) {
@@ -79,17 +73,3 @@ func TestCapabilityMatrix(t *testing.T) {
t.Fatalf("ProviderCapabilities(missing) = %#v", caps)
}
}
func TestRegisterStream(t *testing.T) {
ai.RegisterStream("test-stream")
if caps := ai.ProviderCapabilities("test-stream"); caps != (ai.Capabilities{Stream: true}) {
t.Fatalf("ProviderCapabilities(test-stream) = %#v", caps)
}
got := ai.RegisteredProviders("stream")
want := []string{"test-stream"}
if !reflect.DeepEqual(got, want) {
t.Fatalf("RegisteredProviders(stream) = %#v, want %#v", got, want)
}
}
-96
View File
@@ -1,96 +0,0 @@
package ai
import (
"context"
"errors"
"testing"
"time"
)
type retryModel struct {
generate func(context.Context, *Request, ...GenerateOption) (*Response, error)
}
func (m retryModel) Init(...Option) error { return nil }
func (m retryModel) Options() Options { return Options{} }
func (m retryModel) Generate(ctx context.Context, req *Request, opts ...GenerateOption) (*Response, error) {
return m.generate(ctx, req, opts...)
}
func (m retryModel) Stream(context.Context, *Request, ...GenerateOption) (Stream, error) {
return nil, ErrStreamingUnsupported
}
func (m retryModel) String() string { return "retry-test" }
func TestGenerateWithRetryRetriesTransientErrors(t *testing.T) {
attempts := 0
model := retryModel{generate: func(context.Context, *Request, ...GenerateOption) (*Response, error) {
attempts++
if attempts == 1 {
return nil, errors.New("temporary provider outage")
}
return &Response{Reply: "ok"}, nil
}}
resp, err := GenerateWithRetry(context.Background(), model, &Request{Prompt: "hi"}, GeneratePolicy{
MaxAttempts: 2,
Backoff: time.Millisecond,
})
if err != nil {
t.Fatalf("GenerateWithRetry returned error: %v", err)
}
if resp.Reply != "ok" {
t.Fatalf("response reply = %q, want ok", resp.Reply)
}
if attempts != 2 {
t.Fatalf("attempts = %d, want 2", attempts)
}
}
func TestGenerateWithRetryDoesNotRetryCallerCancellation(t *testing.T) {
ctx, cancel := context.WithCancel(context.Background())
attempts := 0
model := retryModel{generate: func(context.Context, *Request, ...GenerateOption) (*Response, error) {
attempts++
cancel()
return nil, errors.New("temporary provider outage")
}}
_, err := GenerateWithRetry(ctx, model, &Request{Prompt: "hi"}, GeneratePolicy{
MaxAttempts: 3,
Backoff: time.Millisecond,
})
if !errors.Is(err, context.Canceled) {
t.Fatalf("error = %v, want context.Canceled", err)
}
if attempts != 1 {
t.Fatalf("attempts = %d, want 1", attempts)
}
}
func TestGenerateWithRetryHonorsPerAttemptTimeout(t *testing.T) {
attempts := 0
model := retryModel{generate: func(ctx context.Context, _ *Request, _ ...GenerateOption) (*Response, error) {
attempts++
<-ctx.Done()
return nil, ctx.Err()
}}
_, err := GenerateWithRetry(context.Background(), model, &Request{Prompt: "hi"}, GeneratePolicy{
Timeout: time.Millisecond,
MaxAttempts: 2,
Backoff: time.Millisecond,
})
var retryErr *RetryError
if !errors.As(err, &retryErr) {
t.Fatalf("error = %T %[1]v, want RetryError", err)
}
if retryErr.Attempts != 2 {
t.Fatalf("retry attempts = %d, want 2", retryErr.Attempts)
}
if !errors.Is(err, context.DeadlineExceeded) {
t.Fatalf("error = %v, want context.DeadlineExceeded", err)
}
if attempts != 2 {
t.Fatalf("attempts = %d, want 2", attempts)
}
}
-72
View File
@@ -1,72 +0,0 @@
package ai
import (
"encoding/json"
"fmt"
"io"
"github.com/urfave/cli/v2"
goai "go-micro.dev/v6/ai"
_ "go-micro.dev/v6/ai/anthropic"
_ "go-micro.dev/v6/ai/atlascloud"
_ "go-micro.dev/v6/ai/gemini"
_ "go-micro.dev/v6/ai/groq"
_ "go-micro.dev/v6/ai/mistral"
_ "go-micro.dev/v6/ai/openai"
_ "go-micro.dev/v6/ai/together"
"go-micro.dev/v6/cmd"
)
func init() {
cmd.Register(&cli.Command{
Name: "ai",
Usage: "Inspect AI provider support",
Subcommands: []*cli.Command{{
Name: "providers",
Usage: "Print the registered AI provider capability matrix",
Flags: []cli.Flag{
&cli.BoolFlag{
Name: "json",
Usage: "Print the capability matrix as JSON",
},
},
Action: providersAction,
}},
})
}
func providersAction(c *cli.Context) error {
rows := goai.CapabilityRows()
if c.Bool("json") {
return writeProviderJSON(c.App.Writer, rows)
}
writeProviderMatrix(c.App.Writer, rows)
return nil
}
func writeProviderJSON(w io.Writer, rows []goai.CapabilityRow) error {
enc := json.NewEncoder(w)
enc.SetIndent("", " ")
return enc.Encode(rows)
}
func writeProviderMatrix(w io.Writer, rows []goai.CapabilityRow) {
const check = "✓"
fmt.Fprintln(w, "Provider Model Image Video")
fmt.Fprintln(w, "-------- ----- ----- -----")
for _, row := range rows {
fmt.Fprintf(w, "%-11s %-6s %-6s %-6s\n",
row.Provider,
mark(row.Model, check),
mark(row.Image, check),
mark(row.Video, check),
)
}
}
func mark(ok bool, value string) string {
if ok {
return value
}
return "-"
}
-53
View File
@@ -1,53 +0,0 @@
package ai
import (
"bytes"
"encoding/json"
"strings"
"testing"
goai "go-micro.dev/v6/ai"
)
func TestWriteProviderMatrix(t *testing.T) {
rows := []goai.CapabilityRow{
{Provider: "atlascloud", Capabilities: goai.Capabilities{Model: true, Image: true, Video: true}},
{Provider: "openai", Capabilities: goai.Capabilities{Model: true, Image: true}},
}
var out bytes.Buffer
writeProviderMatrix(&out, rows)
got := out.String()
for _, want := range []string{
"Provider Model Image Video",
"atlascloud ✓ ✓ ✓",
"openai ✓ ✓ -",
} {
if !strings.Contains(got, want) {
t.Fatalf("matrix output missing %q:\n%s", want, got)
}
}
}
func TestWriteProviderJSON(t *testing.T) {
rows := []goai.CapabilityRow{
{Provider: "openai", Capabilities: goai.Capabilities{Model: true, Image: true}},
}
var out bytes.Buffer
if err := writeProviderJSON(&out, rows); err != nil {
t.Fatalf("writeProviderJSON returned error: %v", err)
}
var got []goai.CapabilityRow
if err := json.Unmarshal(out.Bytes(), &got); err != nil {
t.Fatalf("JSON output did not decode: %v\n%s", err, out.String())
}
if len(got) != 1 || got[0].Provider != "openai" || !got[0].Model || !got[0].Image || got[0].Video {
t.Fatalf("decoded JSON = %#v, want openai model+image", got)
}
if !strings.HasSuffix(out.String(), "\n") {
t.Fatalf("JSON output should end with newline: %q", out.String())
}
}
+1 -2
View File
@@ -135,13 +135,12 @@ func runFlags() []cli.Flag {
return []cli.Flag{
&cli.BoolFlag{Name: "json", Usage: "Print run data as JSON for automation"},
&cli.StringFlag{Name: "status", Usage: "Only show runs with this status (running, done, error, refused)"},
&cli.StringFlag{Name: "trace", Usage: "Only show runs whose trace id matches this full id or prefix"},
&cli.IntFlag{Name: "limit", Usage: "Show the most recently updated N runs"},
}
}
func runOptions(c *cli.Context) goagent.RunListOptions {
return goagent.RunListOptions{Status: c.String("status"), TraceID: c.String("trace"), Limit: c.Int("limit")}
return goagent.RunListOptions{Status: c.String("status"), Limit: c.Int("limit")}
}
func printRunIndex(name string, opts goagent.RunListOptions, asJSON bool) error {
-76
View File
@@ -9,7 +9,6 @@ import (
"io"
"os"
"os/signal"
"sort"
"syscall"
"github.com/urfave/cli/v2"
@@ -76,10 +75,6 @@ Examples:
ArgsUsage: "[name]",
Flags: []cli.Flag{
&cli.BoolFlag{Name: "json", Usage: "Print durable run history as JSON for automation"},
&cli.BoolFlag{Name: "pending", Usage: "Only show runs that have not completed"},
&cli.StringFlag{Name: "status", Usage: "Only show runs with this status (running, done, failed)"},
&cli.IntFlag{Name: "limit", Usage: "Show the most recently updated N runs"},
&cli.StringFlag{Name: "stage", Usage: "Only show runs currently checkpointed at this stage"},
},
Action: flowRuns,
},
@@ -134,83 +129,19 @@ func flowRuns(c *cli.Context) error {
if err != nil {
return err
}
opts, err := validateFlowRunOptions(flowRunOptions{Pending: c.Bool("pending"), Status: c.String("status"), Stage: c.String("stage"), Limit: c.Int("limit")})
if err != nil {
return err
}
runs = filterFlowRuns(runs, opts)
if len(runs) == 0 {
if c.Bool("pending") {
fmt.Printf(" No pending runs recorded for flow %q.\n", name)
return nil
}
fmt.Printf(" No runs recorded for flow %q.\n", name)
return nil
}
return writeFlowRuns(os.Stdout, runs, c.Bool("json"))
}
type flowRunOptions struct {
Pending bool
Status string
Stage string
Limit int
}
func validateFlowRunOptions(opts flowRunOptions) (flowRunOptions, error) {
switch opts.Status {
case "", "running", "done", "failed":
default:
return opts, fmt.Errorf("invalid run status %q: expected running, done, or failed", opts.Status)
}
if opts.Limit < 0 {
return opts, fmt.Errorf("invalid limit %d: expected a non-negative value", opts.Limit)
}
return opts, nil
}
func filterFlowRuns(runs []aiflow.Run, opts flowRunOptions) []aiflow.Run {
if len(runs) == 0 {
return nil
}
filtered := make([]aiflow.Run, 0, len(runs))
for _, run := range runs {
if opts.Pending && run.Status == "done" {
continue
}
if opts.Status != "" && run.Status != opts.Status {
continue
}
if opts.Stage != "" && run.State.Stage != opts.Stage {
continue
}
filtered = append(filtered, run)
}
if opts.Limit > 0 && len(filtered) > opts.Limit {
sort.SliceStable(filtered, func(i, j int) bool {
if filtered[i].Updated.Equal(filtered[j].Updated) {
return filtered[i].Started.Before(filtered[j].Started)
}
return filtered[i].Updated.Before(filtered[j].Updated)
})
start := len(filtered) - opts.Limit
limited := append([]aiflow.Run(nil), filtered[start:]...)
return limited
}
return filtered
}
func pendingFlowRuns(runs []aiflow.Run) []aiflow.Run {
return filterFlowRuns(runs, flowRunOptions{Pending: true})
}
func writeFlowRuns(w io.Writer, runs []aiflow.Run, asJSON bool) error {
if asJSON {
enc := json.NewEncoder(w)
enc.SetIndent("", " ")
return enc.Encode(runs)
}
fmt.Fprintf(w, " %d run%s\n", len(runs), plural(len(runs)))
for _, r := range runs {
id := r.ID
if len(id) > 8 {
@@ -233,13 +164,6 @@ func writeFlowRuns(w io.Writer, runs []aiflow.Run, asJSON bool) error {
return nil
}
func plural(n int) string {
if n == 1 {
return ""
}
return "s"
}
func flowFlags() []cli.Flag {
return []cli.Flag{
&cli.StringFlag{Name: "trigger", Usage: "Broker topic to subscribe to", EnvVars: []string{"MICRO_FLOW_TRIGGER"}},
-99
View File
@@ -29,7 +29,6 @@ func TestWriteFlowRunsIncludesStepDetails(t *testing.T) {
}
got := out.String()
for _, want := range []string{
"1 run",
"12345678 failed stage=charge",
"updated=2026-06-24T12:30:00Z",
"- reserve done attempts=1",
@@ -41,20 +40,6 @@ func TestWriteFlowRunsIncludesStepDetails(t *testing.T) {
}
}
func TestValidateFlowRunOptionsRejectsInvalidStatus(t *testing.T) {
_, err := validateFlowRunOptions(flowRunOptions{Status: "stuck"})
if err == nil || !strings.Contains(err.Error(), "invalid run status") {
t.Fatalf("expected invalid status error, got %v", err)
}
}
func TestValidateFlowRunOptionsRejectsNegativeLimit(t *testing.T) {
_, err := validateFlowRunOptions(flowRunOptions{Limit: -1})
if err == nil || !strings.Contains(err.Error(), "invalid limit") {
t.Fatalf("expected invalid limit error, got %v", err)
}
}
func TestWriteFlowRunsJSON(t *testing.T) {
runs := []aiflow.Run{{ID: "run-1", Flow: "checkout", Status: "done"}}
@@ -70,87 +55,3 @@ func TestWriteFlowRunsJSON(t *testing.T) {
t.Fatalf("decoded runs = %+v", got)
}
}
func TestPendingFlowRunsFiltersCompletedRuns(t *testing.T) {
runs := []aiflow.Run{
{ID: "run-1", Status: "done"},
{ID: "run-2", Status: "failed"},
{ID: "run-3", Status: "running"},
}
got := pendingFlowRuns(runs)
if len(got) != 2 {
t.Fatalf("pendingFlowRuns returned %d runs, want 2: %+v", len(got), got)
}
if got[0].ID != "run-2" || got[1].ID != "run-3" {
t.Fatalf("pending runs = %+v", got)
}
}
func TestFilterFlowRunsStatus(t *testing.T) {
runs := []aiflow.Run{
{ID: "run-1", Status: "done"},
{ID: "run-2", Status: "failed"},
{ID: "run-3", Status: "running"},
{ID: "run-4", Status: "failed"},
}
got := filterFlowRuns(runs, flowRunOptions{Status: "failed"})
if len(got) != 2 {
t.Fatalf("filterFlowRuns returned %d runs, want 2: %+v", len(got), got)
}
if got[0].ID != "run-2" || got[1].ID != "run-4" {
t.Fatalf("failed runs = %+v", got)
}
}
func TestFilterFlowRunsStage(t *testing.T) {
runs := []aiflow.Run{
{ID: "run-1", Status: "failed", State: aiflow.State{Stage: "reserve"}},
{ID: "run-2", Status: "failed", State: aiflow.State{Stage: "charge"}},
{ID: "run-3", Status: "running", State: aiflow.State{Stage: "charge"}},
{ID: "run-4", Status: "done", State: aiflow.State{}},
}
got := filterFlowRuns(runs, flowRunOptions{Stage: "charge"})
if len(got) != 2 {
t.Fatalf("filterFlowRuns returned %d runs, want 2: %+v", len(got), got)
}
if got[0].ID != "run-2" || got[1].ID != "run-3" {
t.Fatalf("charge-stage runs = %+v", got)
}
}
func TestFilterFlowRunsLimitKeepsNewestRuns(t *testing.T) {
runs := []aiflow.Run{
{ID: "run-1", Status: "done"},
{ID: "run-2", Status: "failed"},
{ID: "run-3", Status: "running"},
}
got := filterFlowRuns(runs, flowRunOptions{Limit: 2})
if len(got) != 2 {
t.Fatalf("filterFlowRuns returned %d runs, want 2: %+v", len(got), got)
}
if got[0].ID != "run-2" || got[1].ID != "run-3" {
t.Fatalf("limited runs = %+v", got)
}
}
func TestFilterFlowRunsCombinesPendingStatusAndLimit(t *testing.T) {
runs := []aiflow.Run{
{ID: "run-1", Status: "failed"},
{ID: "run-2", Status: "done"},
{ID: "run-3", Status: "failed"},
{ID: "run-4", Status: "running"},
{ID: "run-5", Status: "failed"},
}
got := filterFlowRuns(runs, flowRunOptions{Pending: true, Status: "failed", Limit: 2})
if len(got) != 2 {
t.Fatalf("filterFlowRuns returned %d runs, want 2: %+v", len(got), got)
}
if got[0].ID != "run-3" || got[1].ID != "run-5" {
t.Fatalf("filtered runs = %+v", got)
}
}
-1
View File
@@ -5,7 +5,6 @@ import (
"go-micro.dev/v6/cmd"
_ "go-micro.dev/v6/cmd/micro/a2a"
_ "go-micro.dev/v6/cmd/micro/ai"
_ "go-micro.dev/v6/cmd/micro/api"
_ "go-micro.dev/v6/cmd/micro/chat"
_ "go-micro.dev/v6/cmd/micro/cli"
+1 -11
View File
@@ -2,7 +2,6 @@ package flow
import (
"context"
"encoding/json"
"testing"
"go-micro.dev/v6/client"
@@ -26,17 +25,11 @@ func (c *fakeClient) Call(ctx context.Context, req client.Request, rsp interface
func TestExecuteDispatchesToAgent(t *testing.T) {
f := New("welcome", Agent("comms"), Prompt("welcome {{.Data}}"))
var svc, ep, parentID string
var svc, ep string
f.client = &fakeClient{
Client: client.DefaultClient,
callFn: func(req client.Request, rsp interface{}) error {
svc, ep = req.Service(), req.Endpoint()
reqFrame := req.Body().(*codecbytes.Frame)
var body map[string]string
if err := json.Unmarshal(reqFrame.Data, &body); err != nil {
t.Fatalf("request body: %v", err)
}
parentID = body["parent_id"]
frame := rsp.(*codecbytes.Frame)
frame.Data = []byte(`{"reply":"welcomed bob","agent":"comms"}`)
return nil
@@ -50,9 +43,6 @@ func TestExecuteDispatchesToAgent(t *testing.T) {
if svc != "comms" || ep != "Agent.Chat" {
t.Errorf("dispatched to %s.%s, want comms.Agent.Chat", svc, ep)
}
if parentID == "" {
t.Fatal("dispatch request parent_id is empty")
}
results := f.Results()
if len(results) != 1 {
+1 -5
View File
@@ -205,9 +205,6 @@ func (f *Flow) Execute(ctx context.Context, data string) error {
return err
}
runID := uuid.New().String()
ctx = ai.WithRunInfo(ctx, ai.RunInfo{RunID: runID, Agent: f.name})
start := time.Now()
prompt := data
@@ -280,8 +277,7 @@ func (f *Flow) Execute(ctx context.Context, data string) error {
// callAgent hands the rendered prompt to a registered agent's Agent.Chat
// endpoint over RPC and returns its reply.
func (f *Flow) callAgent(ctx context.Context, name, message string) (string, error) {
info, _ := ai.RunInfoFrom(ctx)
body, _ := json.Marshal(map[string]string{"message": message, "parent_id": info.RunID})
body, _ := json.Marshal(map[string]string{"message": message})
req := f.client.NewRequest(name, "Agent.Chat", &codecbytes.Frame{Data: body})
var rsp codecbytes.Frame
if err := f.client.Call(ctx, req, &rsp); err != nil {
+1 -12
View File
@@ -1,10 +1,6 @@
package flow
import (
"time"
"go.opentelemetry.io/otel/trace"
)
import "time"
// Options configures a Flow.
type Options struct {
@@ -44,8 +40,6 @@ type Options struct {
// Checkpoint is the durability backend for stepped runs. Nil with
// steps present means a store-backed default; set it to swap backends.
Checkpoint Checkpoint
// TraceProvider emits OpenTelemetry spans for stepped flow runs.
TraceProvider trace.TracerProvider
// DeleteOnSuccess removes a run's checkpoint when it completes
// successfully. Failed runs are always retained. Default: retain all.
DeleteOnSuccess bool
@@ -138,8 +132,3 @@ func WithCheckpoint(c Checkpoint) Option {
func DeleteOnSuccess() Option {
return func(o *Options) { o.DeleteOnSuccess = true }
}
// TraceProvider enables OpenTelemetry spans for stepped flow runs and steps.
func TraceProvider(tp trace.TracerProvider) Option {
return func(o *Options) { o.TraceProvider = tp }
}
-80
View File
@@ -1,80 +0,0 @@
package flow
import (
"context"
"time"
"go-micro.dev/v6/ai"
"go.opentelemetry.io/otel/attribute"
"go.opentelemetry.io/otel/codes"
"go.opentelemetry.io/otel/trace"
)
const flowInstrumentationName = "go-micro.dev/v6/flow"
const (
spanNameFlowRun = "flow.run"
spanNameFlowStep = "flow.step"
AttrFlowRunID = "flow.run.id"
AttrFlowName = "flow.name"
AttrFlowStepName = "flow.step.name"
AttrFlowStatus = "flow.status"
AttrFlowAttempts = "flow.step.attempts"
AttrFlowLatencyMS = "flow.latency_ms"
)
func (f *Flow) tracer() trace.Tracer {
return f.opts.TraceProvider.Tracer(flowInstrumentationName)
}
func (f *Flow) startRunSpan(ctx context.Context, run Run) (context.Context, func(Run, error)) {
if f.opts.TraceProvider == nil {
return ctx, func(Run, error) {}
}
ctx, span := f.tracer().Start(ctx, spanNameFlowRun, trace.WithSpanKind(trace.SpanKindInternal), trace.WithAttributes(
attribute.String(AttrFlowRunID, run.ID),
attribute.String(AttrFlowName, f.name),
attribute.String(AttrFlowStatus, run.Status),
))
start := time.Now()
return ctx, func(done Run, err error) {
span.SetAttributes(
attribute.String(AttrFlowStatus, done.Status),
attribute.Int64(AttrFlowLatencyMS, time.Since(start).Milliseconds()),
)
if err != nil {
span.RecordError(err)
span.SetStatus(codes.Error, err.Error())
} else {
span.SetStatus(codes.Ok, "")
}
span.End()
}
}
func (f *Flow) runStepSpan(ctx context.Context, step Step, in State) (State, int, error) {
if f.opts.TraceProvider == nil {
return f.runStep(ctx, step, in)
}
info, _ := ai.RunInfoFrom(ctx)
ctx, span := f.tracer().Start(ctx, spanNameFlowStep, trace.WithAttributes(
attribute.String(AttrFlowRunID, info.RunID),
attribute.String(AttrFlowName, f.name),
attribute.String(AttrFlowStepName, step.Name),
))
start := time.Now()
out, attempts, err := f.runStep(ctx, step, in)
span.SetAttributes(
attribute.Int(AttrFlowAttempts, attempts),
attribute.Int64(AttrFlowLatencyMS, time.Since(start).Milliseconds()),
)
if err != nil {
span.RecordError(err)
span.SetStatus(codes.Error, err.Error())
} else {
span.SetStatus(codes.Ok, "")
}
span.End()
return out, attempts, err
}
-70
View File
@@ -1,70 +0,0 @@
package flow
import (
"context"
"testing"
"go-micro.dev/v6/store"
"go.opentelemetry.io/otel/attribute"
"go.opentelemetry.io/otel/sdk/trace"
"go.opentelemetry.io/otel/sdk/trace/tracetest"
)
func TestFlowOpenTelemetrySpans(t *testing.T) {
exp := tracetest.NewInMemoryExporter()
tp := trace.NewTracerProvider(trace.WithSyncer(exp))
step := Step{Name: "inspect", Run: func(ctx context.Context, in State) (State, error) {
in.Data = []byte("done")
return in, nil
}}
f := New("observed", WithCheckpoint(StoreCheckpoint(store.NewMemoryStore(), "observed")), TraceProvider(tp), Steps(step))
if err := f.Execute(context.Background(), "start"); err != nil {
t.Fatal(err)
}
spans := exp.GetSpans().Snapshots()
seen := map[string]bool{spanNameFlowRun: false, spanNameFlowStep: false}
var runID string
for _, span := range spans {
attrs := flowSpanAttributes(span.Attributes())
switch span.Name() {
case spanNameFlowRun:
seen[spanNameFlowRun] = true
runID = attrs[AttrFlowRunID]
if attrs[AttrFlowName] != "observed" || attrs[AttrFlowStatus] != "done" {
t.Fatalf("run span attributes = %#v", attrs)
}
case spanNameFlowStep:
seen[spanNameFlowStep] = true
if attrs[AttrFlowName] != "observed" || attrs[AttrFlowStepName] != "inspect" {
t.Fatalf("step span attributes = %#v", attrs)
}
}
}
for name, ok := range seen {
if !ok {
t.Fatalf("span %s not emitted; got %d spans", name, len(spans))
}
}
if runID == "" {
t.Fatal("run span missing run id")
}
for _, span := range spans {
if span.Name() != spanNameFlowStep {
continue
}
attrs := flowSpanAttributes(span.Attributes())
if attrs[AttrFlowRunID] != runID {
t.Fatalf("step span run id = %q, want %q", attrs[AttrFlowRunID], runID)
}
}
}
func flowSpanAttributes(attrs []attribute.KeyValue) map[string]string {
out := make(map[string]string, len(attrs))
for _, attr := range attrs {
out[string(attr.Key)] = attr.Value.AsString()
}
return out
}
+9 -51
View File
@@ -240,8 +240,7 @@ func Dispatch(name string) StepFunc {
if d := depsFrom(ctx); d != nil && d.client != nil {
cl = d.client
}
info, _ := ai.RunInfoFrom(ctx)
body, _ := json.Marshal(map[string]string{"message": in.String(), "parent_id": info.RunID})
body, _ := json.Marshal(map[string]string{"message": in.String()})
req := cl.NewRequest(name, "Agent.Chat", &codecbytes.Frame{Data: body})
var rsp codecbytes.Frame
if err := cl.Call(ctx, req, &rsp); err != nil {
@@ -306,9 +305,6 @@ func LLM(prompt string) StepFunc {
// startRun begins a fresh run of the flow's steps with the given input.
func (f *Flow) startRun(ctx context.Context, data string) (Run, error) {
if err := validateSteps(f.opts.Steps); err != nil {
return Run{}, err
}
run := Run{
ID: uuid.New().String(),
Flow: f.name,
@@ -325,9 +321,6 @@ func (f *Flow) startRun(ctx context.Context, data string) (Run, error) {
// Resume continues a persisted run by id, picking up at the step it
// stopped on. Completed runs are a no-op.
func (f *Flow) Resume(ctx context.Context, runID string) error {
if err := validateSteps(f.opts.Steps); err != nil {
return err
}
if f.checkpoint == nil {
return fmt.Errorf("flow %s has no checkpoint configured", f.name)
}
@@ -391,9 +384,6 @@ func (f *Flow) runFrom(ctx context.Context, run Run) (Run, error) {
steps := f.opts.Steps
ctx = withDeps(ctx, &runDeps{client: f.client, model: f.model, tools: f.toolSet})
ctx = ai.WithRunInfo(ctx, ai.RunInfo{RunID: run.ID, Agent: f.name})
ctx, finishSpan := f.startRunSpan(ctx, run)
var spanErr error
defer func() { finishSpan(run, spanErr) }()
start := stepIndex(steps, run.State.Stage)
if start < 0 {
@@ -408,22 +398,15 @@ func (f *Flow) runFrom(ctx context.Context, run Run) (Run, error) {
step := steps[i]
run.State.Stage = step.Name
run.Steps[i].Status = "in_progress"
if err := f.save(ctx, run); err != nil {
spanErr = err
return run, err
}
f.save(ctx, run)
out, attempts, err := f.runStepSpan(ctx, step, run.State)
out, attempts, err := f.runStep(ctx, step, run.State)
run.Steps[i].Attempts = attempts
if err != nil {
spanErr = err
run.Steps[i].Status = "failed"
run.Steps[i].Error = err.Error()
run.Status = "failed"
if saveErr := f.save(ctx, run); saveErr != nil {
spanErr = saveErr
return run, fmt.Errorf("%w; additionally failed to checkpoint failed run: %v", err, saveErr)
}
f.save(ctx, run)
f.record(resultFromRun(f.opts.TriggerTopic, run))
f.log.Logf(logger.ErrorLevel, "Flow %s run %s failed at step %q: %v", f.name, run.ID, step.Name, err)
return run, err
@@ -437,22 +420,13 @@ func (f *Flow) runFrom(ctx context.Context, run Run) (Run, error) {
} else {
run.State.Stage = ""
}
if err := f.save(ctx, run); err != nil {
spanErr = err
return run, err
}
f.save(ctx, run)
}
run.Status = "done"
if err := f.save(ctx, run); err != nil {
spanErr = err
return run, err
}
f.save(ctx, run)
if f.opts.DeleteOnSuccess && f.checkpoint != nil {
if err := f.checkpoint.Delete(ctx, run.ID); err != nil {
spanErr = err
return run, err
}
_ = f.checkpoint.Delete(ctx, run.ID)
}
f.record(resultFromRun(f.opts.TriggerTopic, run))
f.log.Logf(logger.InfoLevel, "Flow %s run %s completed (%d steps)", f.name, run.ID, len(steps))
@@ -494,29 +468,13 @@ func (f *Flow) runStep(ctx context.Context, step Step, in State) (State, int, er
return in, retries + 1, lastErr
}
func (f *Flow) save(ctx context.Context, run Run) error {
func (f *Flow) save(ctx context.Context, run Run) {
if f.checkpoint == nil {
return nil
return
}
if err := f.checkpoint.Save(ctx, run); err != nil {
f.log.Logf(logger.ErrorLevel, "Flow %s checkpoint save: %v", f.name, err)
return fmt.Errorf("flow %s checkpoint save: %w", f.name, err)
}
return nil
}
func validateSteps(steps []Step) error {
seen := make(map[string]struct{}, len(steps))
for i, step := range steps {
if step.Name == "" {
return fmt.Errorf("flow: step %d has an empty name", i)
}
if _, ok := seen[step.Name]; ok {
return fmt.Errorf("flow: duplicate step name %q", step.Name)
}
seen[step.Name] = struct{}{}
}
return nil
}
func stepIndex(steps []Step, name string) int {
-88
View File
@@ -318,41 +318,6 @@ func TestFlowStepRetryStopsOnCancel(t *testing.T) {
}
}
func TestFlowStepNamesMustBeUnique(t *testing.T) {
step := Step{Name: "work", Run: func(_ context.Context, in State) (State, error) {
return in, nil
}}
f := New("duplicate-steps",
WithCheckpoint(StoreCheckpoint(store.NewMemoryStore(), "duplicate-steps")),
Steps(step, step),
)
err := f.Execute(context.Background(), "")
if err == nil {
t.Fatal("expected duplicate step names to fail validation")
}
if got, want := err.Error(), `flow: duplicate step name "work"`; got != want {
t.Fatalf("error = %q, want %q", got, want)
}
}
func TestFlowStepNamesMustBeNonEmpty(t *testing.T) {
f := New("empty-step-name",
WithCheckpoint(StoreCheckpoint(store.NewMemoryStore(), "empty-step-name")),
Steps(Step{Name: "", Run: func(_ context.Context, in State) (State, error) {
return in, nil
}}),
)
err := f.Execute(context.Background(), "")
if err == nil {
t.Fatal("expected an empty step name to fail validation")
}
if got, want := err.Error(), "flow: step 0 has an empty name"; got != want {
t.Fatalf("error = %q, want %q", got, want)
}
}
// A step with no Run function is reported as a configuration error rather
// than panicking the run.
func TestFlowStepNilRun(t *testing.T) {
@@ -417,59 +382,6 @@ func TestStoreCheckpointHonorsCanceledContext(t *testing.T) {
}
}
type failingCheckpoint struct {
err error
}
func (c failingCheckpoint) Save(context.Context, Run) error { return c.err }
func (c failingCheckpoint) Load(context.Context, string) (Run, bool, error) {
return Run{}, false, c.err
}
func (c failingCheckpoint) Delete(context.Context, string) error { return c.err }
func (c failingCheckpoint) List(context.Context) ([]Run, error) { return nil, c.err }
func TestFlowCheckpointSaveFailureStopsRun(t *testing.T) {
checkpointErr := errors.New("checkpoint unavailable")
var ran bool
f := New("checkpoint-fails",
WithCheckpoint(failingCheckpoint{err: checkpointErr}),
Steps(Step{Name: "work", Run: func(_ context.Context, in State) (State, error) {
ran = true
return in, nil
}}),
)
err := f.Execute(context.Background(), "start")
if !errors.Is(err, checkpointErr) {
t.Fatalf("Execute error = %v, want checkpoint error", err)
}
if ran {
t.Fatal("step ran even though the in-progress checkpoint failed")
}
}
func TestFlowDeleteOnSuccessFailureIsReturned(t *testing.T) {
checkpointErr := errors.New("delete unavailable")
cp := &deleteFailCheckpoint{Checkpoint: StoreCheckpoint(store.NewMemoryStore(), "delete-fails"), err: checkpointErr}
f := New("delete-fails",
WithCheckpoint(cp),
DeleteOnSuccess(),
Steps(appendStep("work")),
)
err := f.Execute(context.Background(), "")
if !errors.Is(err, checkpointErr) {
t.Fatalf("Execute error = %v, want delete error", err)
}
}
type deleteFailCheckpoint struct {
Checkpoint
err error
}
func (c *deleteFailCheckpoint) Delete(context.Context, string) error { return c.err }
func TestStateSetScan(t *testing.T) {
var s State
type payload struct {
+13 -57
View File
@@ -18,11 +18,10 @@
// BaseURL: "https://agents.example.com",
// })
//
// Scope of this version: the JSON-RPC binding — `message/send`
// (returns a completed Task), `message/stream` (SSE with the completed
// Task event), `tasks/get`, and Agent Card discovery. Multi-turn
// `input-required`, `tasks/resubscribe`, and push notifications are
// advertised as unsupported and are follow-ups.
// Scope of this version: the synchronous JSON-RPC binding — `message/send`
// (returns a completed Task), `tasks/get`, and Agent Card discovery.
// Streaming (`message/stream`), multi-turn `input-required`, and push
// notifications are advertised as unsupported and are follow-ups.
package a2a
import (
@@ -315,7 +314,7 @@ func Card(name, url, description string, services []string) AgentCard {
URL: url,
Version: "1.0.0",
ProtocolVersion: protocolVersion,
Capabilities: Capabilities{Streaming: true, PushNotifications: false},
Capabilities: Capabilities{Streaming: false, PushNotifications: false},
// The agent converses over a single Chat endpoint; advertise that
// as one skill, tagged with the services it manages.
DefaultInputModes: []string{"text/plain"},
@@ -417,15 +416,13 @@ func (d *dispatcher) serve(w http.ResponseWriter, r *http.Request, invoke Invoke
switch req.Method {
case "message/send":
d.send(requestContext(r.Context()), w, req, invoke)
case "message/stream":
d.stream(requestContext(r.Context()), w, req, invoke)
case "tasks/get":
d.get(w, req)
case "tasks/cancel":
// v1 tasks complete synchronously, so they're already terminal.
writeRPC(w, req.ID, nil, &rpcError{Code: errNotCancelable, Message: "task is not cancelable"})
case "tasks/resubscribe":
writeRPC(w, req.ID, nil, &rpcError{Code: errMethodNotFound, Message: "resubscribe is not supported"})
case "message/stream", "tasks/resubscribe":
writeRPC(w, req.ID, nil, &rpcError{Code: errMethodNotFound, Message: "streaming is not supported"})
default:
writeRPC(w, req.ID, nil, &rpcError{Code: errMethodNotFound, Message: "method not found: " + req.Method})
}
@@ -436,38 +433,15 @@ type sendParams struct {
}
func (d *dispatcher) send(ctx context.Context, w http.ResponseWriter, req rpcRequest, invoke Invoke) {
task, e := d.run(ctx, req.Params, invoke)
if e != nil {
writeRPC(w, req.ID, nil, e)
return
}
writeRPC(w, req.ID, task, nil)
}
func (d *dispatcher) stream(ctx context.Context, w http.ResponseWriter, req rpcRequest, invoke Invoke) {
task, e := d.run(ctx, req.Params, invoke)
if e != nil {
writeRPC(w, req.ID, nil, e)
return
}
w.Header().Set("Content-Type", "text/event-stream")
w.Header().Set("Cache-Control", "no-cache")
w.Header().Set("Connection", "keep-alive")
w.WriteHeader(http.StatusOK)
_ = json.NewEncoder(sseWriter{w: w}).Encode(rpcResponse{JSONRPC: "2.0", ID: req.ID, Result: task})
if f, ok := w.(http.Flusher); ok {
f.Flush()
}
}
func (d *dispatcher) run(ctx context.Context, params json.RawMessage, invoke Invoke) (*Task, *rpcError) {
var p sendParams
if err := json.Unmarshal(params, &p); err != nil {
return nil, &rpcError{Code: errInvalidParams, Message: "invalid params"}
if err := json.Unmarshal(req.Params, &p); err != nil {
writeRPC(w, req.ID, nil, &rpcError{Code: errInvalidParams, Message: "invalid params"})
return
}
text := textOf(p.Message.Parts)
if text == "" {
return nil, &rpcError{Code: errInvalidParams, Message: "message has no text part"}
writeRPC(w, req.ID, nil, &rpcError{Code: errInvalidParams, Message: "message has no text part"})
return
}
reply, err := invoke(ctx, text)
@@ -490,7 +464,7 @@ func (d *dispatcher) run(ctx context.Context, params json.RawMessage, invoke Inv
task.Artifacts = []Artifact{textArtifact(reply)}
}
d.store(task)
return task, nil
writeRPC(w, req.ID, task, nil)
}
type getParams struct {
@@ -591,24 +565,6 @@ func requestContext(parent context.Context) context.Context {
return ctx
}
type sseWriter struct {
w http.ResponseWriter
}
func (s sseWriter) Write(p []byte) (int, error) {
if _, err := s.w.Write([]byte("data: ")); err != nil {
return 0, err
}
n, err := s.w.Write(p)
if err != nil {
return n, err
}
if _, err := s.w.Write([]byte("\n")); err != nil {
return n, err
}
return n, nil
}
func writeJSON(w http.ResponseWriter, status int, v any) {
w.Header().Set("Content-Type", "application/json")
w.WriteHeader(status)
+2 -41
View File
@@ -4,10 +4,8 @@ import (
"bytes"
"context"
"encoding/json"
"fmt"
"net/http"
"net/http/httptest"
"strings"
"testing"
"time"
@@ -46,7 +44,6 @@ func newGatewayWithAgent(t *testing.T) (*httptest.Server, func()) {
srv := server.NewServer(
server.Name("echo"),
server.Address("127.0.0.1:0"),
server.Registry(reg),
server.Metadata(map[string]string{"type": "agent", "services": ""}),
)
@@ -148,42 +145,6 @@ func TestMessageSendUsesRequestContext(t *testing.T) {
}
}
func TestMessageStream(t *testing.T) {
ts, cleanup := newGatewayWithAgent(t)
defer cleanup()
body := `{"jsonrpc":"2.0","id":1,"method":"message/stream","params":{"message":{"role":"user","parts":[{"kind":"text","text":"ping"}],"kind":"message"}}}`
resp, err := http.Post(ts.URL+"/agents/echo", "application/json", bytes.NewBufferString(body))
if err != nil {
t.Fatalf("post: %v", err)
}
defer resp.Body.Close()
if ct := resp.Header.Get("Content-Type"); !strings.HasPrefix(ct, "text/event-stream") {
t.Fatalf("content-type = %q, want text/event-stream", ct)
}
var line string
if _, err := fmt.Fscan(resp.Body, &line); err != nil {
t.Fatalf("read event prefix: %v", err)
}
if line != "data:" {
t.Fatalf("event prefix = %q, want data:", line)
}
var out struct {
Result Task `json:"result"`
Error *rpcError `json:"error"`
}
if err := json.NewDecoder(resp.Body).Decode(&out); err != nil {
t.Fatalf("decode event: %v", err)
}
if out.Error != nil {
t.Fatalf("rpc error: %+v", out.Error)
}
if out.Result.Status.State != stateCompleted || len(out.Result.Artifacts) != 1 || textOf(out.Result.Artifacts[0].Parts) != "pong" {
t.Fatalf("streamed task = %+v", out.Result)
}
}
func TestUnknownMethod(t *testing.T) {
ts, cleanup := newGatewayWithAgent(t)
defer cleanup()
@@ -191,9 +152,9 @@ func TestUnknownMethod(t *testing.T) {
var resp struct {
Error *rpcError `json:"error"`
}
rpc(t, ts.URL+"/agents/echo", `{"jsonrpc":"2.0","id":1,"method":"tasks/resubscribe","params":{}}`, &resp)
rpc(t, ts.URL+"/agents/echo", `{"jsonrpc":"2.0","id":1,"method":"message/stream","params":{}}`, &resp)
if resp.Error == nil || resp.Error.Code != errMethodNotFound {
t.Errorf("expected method-not-found for resubscribe, got %+v", resp.Error)
t.Errorf("expected method-not-found for streaming, got %+v", resp.Error)
}
}
+25 -75
View File
@@ -23,15 +23,8 @@ import (
"sync"
"time"
"go-micro.dev/v6/agent"
"go-micro.dev/v6"
"go-micro.dev/v6/ai"
"go-micro.dev/v6/broker"
"go-micro.dev/v6/client"
"go-micro.dev/v6/flow"
"go-micro.dev/v6/registry"
"go-micro.dev/v6/selector"
"go-micro.dev/v6/service"
"go-micro.dev/v6/store"
)
// ---------------------------------------------------------------------------
@@ -243,95 +236,52 @@ func main() {
fmt.Printf("\n\033[1mPlan & Delegate — live integration harness (provider: %s)\033[0m\n", *provider)
fmt.Print("Real services, registry, RPC, agent loop, store, delegation.\n\n")
reg := registry.NewMemoryRegistry()
cl := client.NewClient(client.Registry(reg), client.Selector(selector.NewSelector(selector.Registry(reg))))
mem := store.NewMemoryStore()
// Real services.
taskSvc := new(TaskService)
task := service.New(service.Name("task"), service.Address("127.0.0.1:0"), service.Registry(reg), service.Client(cl))
if err := task.Handle(taskSvc); err != nil {
fmt.Println("task handle:", err)
os.Exit(1)
}
task := micro.NewService("task")
task.Handle(new(TaskService))
go task.Run()
notifySvc := new(NotifyService)
notify := service.New(service.Name("notify"), service.Address("127.0.0.1:0"), service.Registry(reg), service.Client(cl))
if err := notify.Handle(notifySvc); err != nil {
fmt.Println("notify handle:", err)
os.Exit(1)
}
notify := micro.NewService("notify")
notify.Handle(new(NotifyService))
go notify.Run()
// Real comms agent (owns notify), registered so delegate reaches it over RPC.
comms := agent.New(
agent.Name("comms"),
agent.Address("127.0.0.1:0"),
agent.Services("notify"),
agent.Prompt("You handle outbound notifications. Use the notify service."),
agent.Provider(*provider), agent.APIKey(apiKey),
agent.WithRegistry(reg), agent.WithClient(cl), agent.WithStore(mem),
comms := micro.NewAgent("comms",
micro.AgentServices("notify"),
micro.AgentPrompt("You handle outbound notifications. Use the notify service."),
micro.AgentProvider(*provider),
micro.AgentAPIKey(apiKey),
)
go comms.Run()
defer comms.Stop()
// Real conductor agent (owns task), registered so the flow can reach it over RPC.
conductor := agent.New(
agent.Name("conductor"),
agent.Address("127.0.0.1:0"),
agent.Services("task"),
agent.Prompt("You coordinate launch work. Plan first, create tasks, and delegate notifications to the \"comms\" agent."),
agent.Provider(*provider), agent.APIKey(apiKey),
agent.WithRegistry(reg), agent.WithClient(cl), agent.WithStore(mem),
// Real conductor agent (owns task).
conductor := micro.NewAgent("conductor",
micro.AgentServices("task"),
micro.AgentPrompt("You coordinate launch work. Plan first, create tasks, and delegate notifications to the \"comms\" agent."),
micro.AgentProvider(*provider),
micro.AgentAPIKey(apiKey),
)
go conductor.Run()
defer conductor.Stop()
fmt.Println("waiting for services + agents to register...")
waitForService := func(name string) {
deadline := time.Now().Add(5 * time.Second)
for time.Now().Before(deadline) {
if svcs, err := reg.GetService(name); err == nil && len(svcs) > 0 && len(svcs[0].Nodes) > 0 {
return
}
time.Sleep(20 * time.Millisecond)
}
}
for _, name := range []string{"task", "notify", "comms", "conductor"} {
waitForService(name)
}
fmt.Println("waiting for services + comms agent to register...")
time.Sleep(3 * time.Second)
f := flow.New("zero-to-hero",
flow.Agent("conductor"),
flow.Prompt("Create three launch tasks (Design, Build, Ship), then make sure owner@acme.com is notified: {{.Data}}"),
)
if err := f.Register(reg, broker.DefaultBroker, cl); err != nil {
fmt.Println("flow register:", err)
os.Exit(1)
}
fmt.Print("\n\033[1m> prompt:\033[0m Create three launch tasks (Design, Build, Ship), then make sure owner@acme.com is notified.\n\n")
fmt.Print("\n\033[1m> flow:\033[0m services + agents + workflow + plan/delegate, no API key.\n\n")
if err := f.Execute(context.Background(), "launch readiness"); err != nil {
resp, err := conductor.Ask(context.Background(),
"Create three launch tasks: Design, Build, and Ship. Then make sure owner@acme.com is notified that the launch plan is ready.")
if err != nil {
fmt.Println("\033[31merror:\033[0m", err)
os.Exit(1)
}
if rs := f.Results(); len(rs) > 0 {
fmt.Println("\n\033[1m< conductor reply:\033[0m", rs[len(rs)-1].Reply)
}
fmt.Println("\n\033[1m< conductor reply:\033[0m", resp.Reply)
// Prove plan was persisted to the real store.
if recs, _ := store.Scope(mem, "agent", "conductor").Read("plan"); len(recs) > 0 {
if recs, _ := conductor.Options().Store.Read("agent/conductor/plan"); len(recs) > 0 {
fmt.Printf("\n\033[1mstored plan (agent/conductor/plan):\033[0m %s\n", string(recs[0].Value))
} else {
fmt.Println("\n\033[31m! plan was not persisted\033[0m")
os.Exit(1)
}
if taskSvc.count() != 3 || notifySvc.count() != 1 {
fmt.Printf("\n\033[31m! unexpected side effects: tasks=%d notify=%d\033[0m\n", taskSvc.count(), notifySvc.count())
os.Exit(1)
}
fmt.Println("\n\033[32m✓ 0→hero flow complete (services → agents → workflow)\033[0m")
fmt.Println("\n\033[32m✓ end-to-end flow complete\033[0m")
}
+2 -119
View File
@@ -14,7 +14,6 @@ package main
import (
"context"
"encoding/json"
"flag"
"fmt"
"os"
@@ -50,9 +49,6 @@ func main() {
timeoutFlag := flag.Duration("timeout", 10*time.Minute, "timeout per provider/harness run")
requireConfiguredFlag := flag.Bool("require-configured", false, "fail when a selected live provider is missing an API key")
capabilitiesFlag := flag.Bool("capabilities", true, "print the registered provider capability matrix before running conformance")
summaryJSONFlag := flag.String("summary-json", "", "write a machine-readable conformance summary to this path")
summaryMarkdownFlag := flag.String("summary-markdown", "", "write a human-readable conformance summary to this path")
capabilityMarkdownFlag := flag.String("capabilities-markdown", "", "write the registered provider capability matrix as a Markdown table")
flag.Parse()
providers := splitCSV(*providersFlag)
@@ -65,26 +61,17 @@ func main() {
if *capabilitiesFlag {
printCapabilityMatrix()
}
if *capabilityMarkdownFlag != "" {
if err := writeCapabilityMarkdown(*capabilityMarkdownFlag, ai.CapabilityRows()); err != nil {
fmt.Fprintf(os.Stderr, "write capabilities markdown: %v\n", err)
os.Exit(1)
}
}
var ran, skipped, failed int
var results []conformanceResult
for _, provider := range providers {
if provider != "mock" && providerKey(provider) == "" {
msg := fmt.Sprintf("set MICRO_AI_API_KEY or %s", providerEnv[provider])
if *requireConfiguredFlag {
fmt.Printf("FAIL %s: missing API key (%s)\n", provider, msg)
failed++
results = append(results, conformanceResult{Provider: provider, Status: statusFailed, Error: "missing API key: " + msg})
} else {
fmt.Printf("- %s: skipped (%s)\n", provider, msg)
skipped++
results = append(results, conformanceResult{Provider: provider, Status: statusSkipped, Error: msg})
}
continue
}
@@ -94,127 +81,23 @@ func main() {
if err := runHarness(provider, harness, *timeoutFlag); err != nil {
fmt.Printf("FAIL %s / %s: %v\n", provider, harness, err)
failed++
results = append(results, conformanceResult{Provider: provider, Harness: harness, Status: statusFailed, Error: err.Error()})
continue
}
ran++
results = append(results, conformanceResult{Provider: provider, Harness: harness, Status: statusPassed})
}
}
fmt.Printf("\nprovider conformance: %d passed, %d skipped providers, %d failed\n", ran, skipped, failed)
summary := conformanceSummary{
Providers: providers,
Harnesses: harnesses,
Capabilities: ai.CapabilityRows(),
Results: results,
Passed: ran,
Skipped: skipped,
Failed: failed,
}
if *summaryJSONFlag != "" {
if err := writeSummaryJSON(*summaryJSONFlag, summary); err != nil {
fmt.Fprintf(os.Stderr, "write summary: %v\n", err)
os.Exit(1)
}
}
if *summaryMarkdownFlag != "" {
if err := writeSummaryMarkdown(*summaryMarkdownFlag, summary); err != nil {
fmt.Fprintf(os.Stderr, "write summary markdown: %v\n", err)
os.Exit(1)
}
}
if failed > 0 {
os.Exit(1)
}
}
const (
statusPassed = "passed"
statusSkipped = "skipped"
statusFailed = "failed"
)
type conformanceResult struct {
Provider string `json:"provider"`
Harness string `json:"harness,omitempty"`
Status string `json:"status"`
Error string `json:"error,omitempty"`
}
type conformanceSummary struct {
Providers []string `json:"providers"`
Harnesses []string `json:"harnesses"`
Capabilities []ai.CapabilityRow `json:"capabilities"`
Results []conformanceResult `json:"results"`
Passed int `json:"passed"`
Skipped int `json:"skipped"`
Failed int `json:"failed"`
}
func writeSummaryJSON(path string, summary conformanceSummary) error {
b, err := json.MarshalIndent(summary, "", " ")
if err != nil {
return err
}
b = append(b, '\n')
return os.WriteFile(path, b, 0o644)
}
func writeCapabilityMarkdown(path string, rows []ai.CapabilityRow) error {
return os.WriteFile(path, []byte(capabilityMarkdown(rows)), 0o644)
}
func writeSummaryMarkdown(path string, summary conformanceSummary) error {
var b strings.Builder
b.WriteString("# Provider conformance summary\n\n")
fmt.Fprintf(&b, "Passed: %d. Skipped providers: %d. Failed: %d.\n\n", summary.Passed, summary.Skipped, summary.Failed)
b.WriteString("## Capability matrix\n\n")
b.WriteString(capabilityMarkdown(summary.Capabilities))
b.WriteString("\n## Harness results\n\n")
b.WriteString("| Provider | Harness | Status | Detail |\n")
b.WriteString("| --- | --- | --- | --- |\n")
for _, result := range summary.Results {
harness := result.Harness
if harness == "" {
harness = "—"
}
fmt.Fprintf(&b, "| %s | %s | %s | %s |\n", result.Provider, harness, result.Status, markdownCell(result.Error))
}
return os.WriteFile(path, []byte(b.String()), 0o644)
}
func capabilityMarkdown(rows []ai.CapabilityRow) string {
var b strings.Builder
b.WriteString("| Provider | Model | Image | Video | Streaming |\n")
b.WriteString("| --- | --- | --- | --- | --- |\n")
for _, row := range rows {
fmt.Fprintf(&b, "| %s | %s | %s | %s | %s |\n", row.Provider, mark(row.Model), mark(row.Image), mark(row.Video), mark(row.Stream))
}
return b.String()
}
func markdownCell(s string) string {
if s == "" {
return "—"
}
s = strings.ReplaceAll(s, "|", "\\|")
s = strings.ReplaceAll(s, "\n", "<br>")
return s
}
func mark(ok bool) string {
if ok {
return "✅"
}
return "—"
}
func printCapabilityMatrix() {
fmt.Println("Provider capability matrix:")
fmt.Println("provider model image video stream")
fmt.Println("provider model image video")
for _, row := range ai.CapabilityRows() {
fmt.Printf("%-12s %-5s %-5s %-5s %-6s\n", row.Provider, yesNo(row.Model), yesNo(row.Image), yesNo(row.Video), yesNo(row.Stream))
fmt.Printf("%-12s %-5s %-5s %-5s\n", row.Provider, yesNo(row.Model), yesNo(row.Image), yesNo(row.Video))
}
fmt.Println()
}
@@ -1,9 +1,6 @@
package main
import (
"encoding/json"
"os"
"path/filepath"
"strings"
"testing"
@@ -55,95 +52,3 @@ func TestCapabilityMatrixHasRegisteredProviders(t *testing.T) {
t.Fatalf("CapabilityRows = %#v, want openai row", rows)
}
}
func TestWriteCapabilityMarkdown(t *testing.T) {
path := filepath.Join(t.TempDir(), "capabilities.md")
rows := []ai.CapabilityRow{
{Provider: "mock", Capabilities: ai.Capabilities{Model: true}},
{Provider: "vision", Capabilities: ai.Capabilities{Image: true, Video: true}},
}
if err := writeCapabilityMarkdown(path, rows); err != nil {
t.Fatalf("writeCapabilityMarkdown returned error: %v", err)
}
b, err := os.ReadFile(path)
if err != nil {
t.Fatalf("read capabilities markdown: %v", err)
}
got := string(b)
for _, want := range []string{
"| Provider | Model | Image | Video | Streaming |",
"| mock | ✅ | — | — | — |",
"| vision | — | ✅ | ✅ | — |",
} {
if !strings.Contains(got, want) {
t.Fatalf("capabilities markdown = %q, want row %q", got, want)
}
}
}
func TestWriteSummaryMarkdown(t *testing.T) {
path := filepath.Join(t.TempDir(), "summary.md")
summary := conformanceSummary{
Capabilities: []ai.CapabilityRow{{Provider: "mock", Capabilities: ai.Capabilities{Model: true}}},
Results: []conformanceResult{
{Provider: "mock", Harness: "agent-flow", Status: statusPassed},
{Provider: "live", Status: statusSkipped, Error: "missing | key"},
},
Passed: 1,
Skipped: 1,
}
if err := writeSummaryMarkdown(path, summary); err != nil {
t.Fatalf("writeSummaryMarkdown returned error: %v", err)
}
b, err := os.ReadFile(path)
if err != nil {
t.Fatalf("read summary markdown: %v", err)
}
got := string(b)
for _, want := range []string{
"# Provider conformance summary",
"Passed: 1. Skipped providers: 1. Failed: 0.",
"| mock | ✅ | — | — | — |",
"| mock | agent-flow | passed | — |",
"| live | — | skipped | missing \\| key |",
} {
if !strings.Contains(got, want) {
t.Fatalf("summary markdown = %q, want %q", got, want)
}
}
}
func TestWriteSummaryJSON(t *testing.T) {
path := filepath.Join(t.TempDir(), "summary.json")
summary := conformanceSummary{
Providers: []string{"mock"},
Harnesses: []string{"provider-conformance"},
Results: []conformanceResult{{
Provider: "mock",
Harness: "provider-conformance",
Status: statusPassed,
}},
Passed: 1,
}
if err := writeSummaryJSON(path, summary); err != nil {
t.Fatalf("writeSummaryJSON returned error: %v", err)
}
b, err := os.ReadFile(path)
if err != nil {
t.Fatalf("read summary: %v", err)
}
if !strings.HasSuffix(string(b), "\n") {
t.Fatalf("summary JSON should end with newline: %q", b)
}
var got conformanceSummary
if err := json.Unmarshal(b, &got); err != nil {
t.Fatalf("summary JSON did not decode: %v", err)
}
if got.Passed != 1 || len(got.Results) != 1 || got.Results[0].Status != statusPassed {
t.Fatalf("summary JSON decoded as %#v, want one passed result", got)
}
}
-3
View File
@@ -50,8 +50,6 @@ guides:
url: /docs/guides/atlascloud-integration.html
- title: AI Provider Guide
url: /docs/guides/ai-provider-guide.html
- title: Provider Conformance
url: /docs/guides/provider-conformance.html
- title: Comparison
url: /docs/guides/comparison.html
- title: Migration Guides
@@ -82,7 +80,6 @@ search_order:
- /docs/plugins.html
- /docs/examples/
- /docs/examples/realworld/
- /docs/guides/provider-conformance.html
- /docs/guides/comparison.html
- /docs/guides/migration/
- /docs/architecture/
+2 -1
View File
@@ -66,7 +66,8 @@ import (
)
func main() {
service := micro.NewService("example",
service := micro.NewService(
micro.Name("example"),
)
service.Init()
if err := service.Run(); err != nil {
@@ -31,11 +31,12 @@ Use **mDNS as the default registry** for service discovery.
```go
// Default - uses mDNS automatically
svc := micro.NewService("myservice")
svc := micro.NewService(micro.Name("myservice"))
// Production - swap to Consul
reg := consul.NewConsulRegistry()
svc := micro.NewService("myservice",
svc := micro.NewService(
micro.Name("myservice"),
micro.Registry(reg),
)
```
@@ -29,7 +29,7 @@ Implement **progressive configuration** where:
### Level 1: Zero Config (Development)
```go
svc := micro.NewService("hello")
svc := micro.NewService(micro.Name("hello"))
svc.Run()
```
@@ -62,7 +62,8 @@ b := nats.NewNatsBroker(
nats.DrainConnection(),
)
svc := micro.NewService("myservice",
svc := micro.NewService(
micro.Name("myservice"),
micro.Version("1.2.3"),
micro.Registry(reg),
micro.Broker(b),
+3 -3
View File
@@ -38,7 +38,7 @@ import (
)
func main() {
service := micro.NewService("publisher")
service := micro.NewService()
service.Init()
// Publish a message
@@ -73,7 +73,7 @@ import (
func main() {
b := bnats.NewNatsBroker()
svc := micro.NewService("publisher", micro.Broker(b))
svc := micro.NewService(micro.Broker(b))
svc.Init()
svc.Run()
}
@@ -88,7 +88,7 @@ import (
func main() {
b := rabbitmq.NewBroker()
svc := micro.NewService("publisher", micro.Broker(b))
svc := micro.NewService(micro.Broker(b))
svc.Init()
svc.Run()
}
+2 -1
View File
@@ -35,7 +35,8 @@ func (g *Greeter) Hello(ctx context.Context, req *struct{}, rsp *struct{Msg stri
}
func main() {
service := micro.NewService("greeter",
service := micro.NewService(
micro.Name("greeter"),
)
service.Init()
micro.RegisterHandler(service.Server(), new(Greeter))
+2 -1
View File
@@ -53,7 +53,8 @@ No code changes required. The framework internally wires the selected implementa
## Equivalent Code Configuration
```go
service := micro.NewService("helloworld",
service := micro.NewService(
micro.Name("helloworld"),
micro.Broker(nats.NewBroker()),
micro.Transport(natstransport.NewTransport()),
micro.Registry(consul.NewRegistry(registry.Addrs("127.0.0.1:8500"))),
@@ -54,7 +54,8 @@ curl -XPOST \
Set a fixed address:
```go
svc := micro.NewService("helloworld",
svc := micro.NewService(
micro.Name("helloworld"),
micro.Address(":8080"),
)
```
@@ -20,7 +20,7 @@ import (
func main() {
b := bnats.NewNatsBroker()
svc := micro.NewService("nats-pubsub", micro.Broker(b))
svc := micro.NewService(micro.Broker(b))
svc.Init()
// subscribe
@@ -79,7 +79,8 @@ func main() {
}
defer db.Close()
svc := micro.NewService("users",
svc := micro.NewService(
micro.Name("users"),
micro.Version("1.0.0"),
)
@@ -171,7 +172,8 @@ func main() {
}
defer db.Close()
svc := micro.NewService("orders",
svc := micro.NewService(
micro.Name("orders"),
micro.Version("1.0.0"),
)
@@ -252,7 +254,8 @@ func (g *Gateway) CreateOrder(w http.ResponseWriter, r *http.Request) {
}
func main() {
svc := micro.NewService("api.gateway",
svc := micro.NewService(
micro.Name("api.gateway"),
)
svc.Init()
@@ -34,7 +34,8 @@ import (
)
func main() {
svc := micro.NewService("myservice",
svc := micro.NewService(
micro.Name("myservice"),
micro.BeforeStop(func() error {
logger.Info("Service stopping, running cleanup...")
return cleanup()
@@ -232,7 +233,8 @@ func main() {
app.AddWorker(&Worker{name: "cleanup"})
app.AddWorker(&Worker{name: "metrics"})
svc := micro.NewService("myservice",
svc := micro.NewService(
micro.Name("myservice"),
micro.BeforeStop(func() error {
ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
defer cancel()
@@ -18,7 +18,7 @@ import (
func main() {
reg := consul.NewConsulRegistry()
svc := micro.NewService("consul-registry", micro.Registry(reg))
svc := micro.NewService(micro.Registry(reg))
svc.Init()
svc.Run()
}
@@ -20,7 +20,7 @@ import (
func main() {
st := postgres.NewStore()
svc := micro.NewService("postgres-store", micro.Store(st))
svc := micro.NewService(micro.Store(st))
svc.Init()
_ = store.Write(&store.Record{Key: "foo", Value: []byte("bar")})
@@ -18,7 +18,7 @@ import (
func main() {
t := tnats.NewTransport()
svc := micro.NewService("nats-transport", micro.Transport(t))
svc := micro.NewService(micro.Transport(t))
svc.Init()
svc.Run()
}
+3 -4
View File
@@ -132,10 +132,9 @@ yet terminal, it polls `tasks/get` until it completes.
## Scope
This is the JSON-RPC binding for completed-task execution:
This is the synchronous JSON-RPC binding:
- **`message/send`** runs the agent and returns a completed `Task`.
- **`message/stream`** streams the completed `Task` as an SSE `data:` event, giving A2A clients a streaming-compatible path while the underlying agent call remains synchronous.
- **`tasks/get`** returns a recent task by id.
- **Agent Card** discovery, generated from the registry.
@@ -143,11 +142,11 @@ Both directions work: the gateway exposes your agents, and `a2a.Client` (via `fl
Not yet supported (advertised as such on the card, so clients negotiate correctly):
- **`tasks/resubscribe`** for reconnecting to a live stream.
- **`message/stream`** (SSE streaming) and `tasks/resubscribe`.
- Multi-turn `input-required` tasks.
- Push notifications.
These are the natural follow-ups; the completed-task binding is what makes a Go Micro agent both reachable from, and able to reach, the A2A ecosystem today.
These are the natural follow-ups; the synchronous binding is what makes a Go Micro agent both reachable from, and able to reach, the A2A ecosystem today.
## See also
@@ -71,5 +71,4 @@ harness. When that loop has to touch production, you do.
- [Agent Loops](agent-loops.html) — run-until-done, with a ceiling
- [Plan & Delegate](plan-delegate.html)
- [Agent Guardrails](agent-guardrails.html)
- [Provider Conformance](provider-conformance.html) — verified provider behavior
- [Roadmap](/docs/roadmap.html)
@@ -30,21 +30,21 @@ imports are linked in:
```go
for _, row := range ai.CapabilityRows() {
fmt.Printf("%s: chat=%t image=%t video=%t stream=%t\n", row.Provider, row.Model, row.Image, row.Video, row.Stream)
fmt.Printf("%s: chat=%t image=%t video=%t\n", row.Provider, row.Model, row.Image, row.Video)
}
```
The built-in providers currently register these capability interfaces:
| Provider | Chat/text (`ai.Model`) | Image (`ai.ImageModel`) | Video (`ai.VideoModel`) | Streaming (`ai.Stream`) |
| --- | --- | --- | --- | --- |
| `anthropic` | Yes | No | No | No |
| `atlascloud` | Yes | Yes | Yes | No |
| `gemini` | Yes | No | No | No |
| `groq` | Yes | No | No | No |
| `mistral` | Yes | No | No | No |
| `openai` | Yes | Yes | No | No |
| `together` | Yes | No | No | No |
| Provider | Chat/text (`ai.Model`) | Image (`ai.ImageModel`) | Video (`ai.VideoModel`) |
| --- | --- | --- | --- |
| `anthropic` | Yes | No | No |
| `atlascloud` | Yes | Yes | Yes |
| `gemini` | Yes | No | No |
| `groq` | Yes | No | No |
| `mistral` | Yes | No | No |
| `openai` | Yes | Yes | No |
| `together` | Yes | No | No |
## Step 1: Implement the `ai.Model` Interface
+4 -4
View File
@@ -97,7 +97,7 @@ func (s *MyService) DoThing(ctx context.Context, req *Request, rsp *Response) er
}
func main() {
svc := micro.NewService("myservice")
svc := micro.NewService(micro.Name("myservice"))
svc.Init()
svc.Handle(new(MyService))
svc.Run()
@@ -140,7 +140,7 @@ import (
grpcClient "go-micro.dev/v6/client/grpc"
)
svc := micro.NewService("myservice",
svc := micro.NewService(
micro.Server(grpcServer.NewServer()),
micro.Client(grpcClient.NewClient()),
)
@@ -234,11 +234,11 @@ mesh / runtime and let ADK (or any A2A agent) plug into it.
**Go Micro**: Built-in with plugins
```go
// Zero-config for dev
svc := micro.NewService("myservice")
svc := micro.NewService(micro.Name("myservice"))
// Consul for production
reg := consul.NewRegistry()
svc := micro.NewService("myservice", micro.Registry(reg))
svc := micro.NewService(micro.Registry(reg))
```
**go-kit**: Bring your own
@@ -19,7 +19,8 @@ The gRPC **transport** uses the gRPC protocol as a communication layer, similar
import "go-micro.dev/v6/transport/grpc"
t := grpc.NewTransport()
service := micro.NewService("helloworld",
service := micro.NewService(
micro.Name("helloworld"),
micro.Transport(t),
)
```
@@ -41,12 +42,15 @@ import (
grpcClient "go-micro.dev/v6/client/grpc"
)
service := micro.NewService("helloworld",
micro.Server(grpcServer.NewServer()),
service := micro.NewService(
micro.Server(grpcServer.NewServer()), // Server must come before Name
micro.Client(grpcClient.NewClient()),
micro.Name("helloworld"),
)
```
> **Important**: The `micro.Server()` option must be specified **before** `micro.Name()`. This is because `micro.Name()` sets the name on the current server, and if `micro.Server()` comes after, it replaces the server with a new one that has no name set.
## When to Use Which
| Use Case | Solution |
@@ -119,8 +123,9 @@ func (s *Say) Hello(ctx context.Context, req *pb.Request, rsp *pb.Response) erro
func main() {
// Create service with gRPC server for native gRPC compatibility
// Note: Server must be set before Name to ensure the name is applied to the gRPC server
service := micro.NewService("helloworld",
service := micro.NewService(
micro.Server(grpcServer.NewServer()),
micro.Name("helloworld"),
micro.Address(":8080"),
)
@@ -153,8 +158,9 @@ import (
func main() {
// Create service with gRPC client
service := micro.NewService("helloworld.client",
service := micro.NewService(
micro.Client(grpcClient.NewClient()),
micro.Name("helloworld.client"),
)
service.Init()
@@ -201,9 +207,10 @@ import (
)
func main() {
service := micro.NewService("helloworld",
micro.Server(grpcServer.NewServer()),
service := micro.NewService(
micro.Server(grpcServer.NewServer()), // Server first
micro.Client(grpcClient.NewClient()),
micro.Name("helloworld"), // Name after Server
micro.Address(":8080"),
)
@@ -232,7 +239,7 @@ ERROR:
```go
// Wrong - uses transport
t := grpc.NewTransport()
service := micro.NewService("helloworld",
service := micro.NewService(
micro.Transport(t),
)
```
@@ -243,7 +250,7 @@ To:
// Correct - uses server
import grpcServer "go-micro.dev/v6/server/grpc"
service := micro.NewService("helloworld",
service := micro.NewService(
micro.Server(grpcServer.NewServer()),
)
```
@@ -263,12 +270,36 @@ import "go-micro.dev/v6/server/grpc"
import "go-micro.dev/v6/client/grpc"
```
### Service Name vs Package Name
### Option Ordering Issue
When creating a client to call another service, use the **service name** passed to `micro.NewService`, not the proto package name:
If the gRPC server is working but your service has no name or is not being found in the registry:
**Cause**: The `micro.Server()` option is specified **after** `micro.Name()`.
When options are processed, `micro.Name()` sets the name on the current server. If `micro.Server()` comes later, it replaces the server with a new one that doesn't have the name set.
**Solution**: Always specify `micro.Server()` **before** `micro.Name()`:
```go
// If the server was started with micro.NewService("helloworld", ...)
// Wrong - server replaces the one with the name set
service := micro.NewService(
micro.Name("helloworld"), // Sets name on default server
micro.Server(grpcServer.NewServer()), // Replaces server, name is lost!
)
// Correct - name is set on the gRPC server
service := micro.NewService(
micro.Server(grpcServer.NewServer()), // Set server first
micro.Name("helloworld"), // Name is now applied to gRPC server
)
```
### Service Name vs Package Name
When creating a client to call another service, use the **service name** (set via `micro.Name()`), not the proto package name:
```go
// If the server was started with micro.Name("helloworld")
sayService := pb.NewSayService("helloworld", service.Client()) // Use service name
// NOT the package name from the proto file
+2 -1
View File
@@ -319,7 +319,8 @@ import (
)
func main() {
service := micro.NewService("tasks",
service := micro.NewService(
micro.Name("tasks"),
micro.Address(":8081"),
)
service.Init()
@@ -128,7 +128,8 @@ func (s *Greeter) SayHello(ctx context.Context, req *pb.HelloRequest, rsp *pb.He
}
func main() {
svc := micro.NewService("greeter",
svc := micro.NewService(
micro.Name("greeter"),
)
svc.Init()
@@ -158,7 +159,7 @@ rsp, err := client.SayHello(context.Background(), &pb.HelloRequest{Name: "John"}
**Go Micro client:**
```go
svc := micro.NewService("client")
svc := micro.NewService(micro.Name("client"))
svc.Init()
client := pb.NewGreeterService("greeter", svc.Client())
@@ -184,7 +185,8 @@ import (
grpcserver "go-micro.dev/v6/server/grpc"
)
svc := micro.NewService("greeter",
svc := micro.NewService(
micro.Name("greeter"),
micro.Client(grpcclient.NewClient()),
micro.Server(grpcserver.NewServer()),
)
@@ -266,7 +268,8 @@ defer client.Agent().ServiceDeregister("greeter-1")
import "go-micro.dev/v6/registry/consul"
reg := consul.NewConsulRegistry()
svc := micro.NewService("greeter",
svc := micro.NewService(
micro.Name("greeter"),
micro.Registry(reg),
)
@@ -290,7 +293,7 @@ svc.Run()
import "go-micro.dev/v6/selector"
// Client-side load balancing built-in
svc := micro.NewService("greeter",
svc := micro.NewService(
micro.Selector(selector.NewSelector(
selector.SetStrategy(selector.RoundRobin),
)),
@@ -359,10 +362,11 @@ lis, _ := net.Listen("tcp", ":50051")
**Go Micro**: Automatic or explicit
```go
// Let Go Micro choose
svc := micro.NewService("greeter")
svc := micro.NewService(micro.Name("greeter"))
// Or specify
svc := micro.NewService("greeter",
svc := micro.NewService(
micro.Name("greeter"),
micro.Address(":50051"),
)
```
@@ -386,7 +390,7 @@ Ensure both use protobuf:
```go
import "go-micro.dev/v6/codec/proto"
svc := micro.NewService("greeter",
svc := micro.NewService(
micro.Codec("application/protobuf", proto.Marshaler{}),
)
```
@@ -1,109 +0,0 @@
---
layout: default
---
# Provider Conformance Matrix
Go Micro treats model providers as interchangeable pieces of the same agent
harness: services expose tools, agents reason over them, and workflows stitch the
work together. The conformance harness keeps that promise honest by running the
same deterministic services → agents → workflows scenarios against every
configured provider.
The live harness is in `internal/harness/provider-conformance`. It skips
providers without API keys by default, so it is safe to run locally, and it fails
when any configured provider breaks the shared contract.
```sh
go run ./internal/harness/provider-conformance
```
For a no-key smoke test of the same harness wiring, run the mock provider:
```sh
go run ./internal/harness/provider-conformance -providers mock
```
## Status legend
| Status | Meaning |
| --- | --- |
| ✅ Verified | Covered by the provider-conformance harness for configured live providers. |
| ⚠️ Unverified | Implemented in the public API, but not yet exercised by provider conformance. |
| — Unsupported | Not exposed by that provider integration today. |
## Harness coverage by capability
These rows describe what the conformance harness verifies today. A provider is
considered conformant when the configured-key run passes all selected harnesses.
| Capability | Harness coverage | Notes |
| --- | --- | --- |
| Simple generation | ✅ Verified | Each harness asks the provider to produce an agent response through `ai.Model`. |
| Service tool calls | ✅ Verified | Harness services are discovered and invoked as model-selected tools. |
| Multi-step tool use | ✅ Verified | The `universe` and `plan-delegate` harnesses require more than one service/tool action. |
| `plan` | ✅ Verified | `plan-delegate` verifies that the conductor agent stores a plan in scoped state. |
| `delegate` | ✅ Verified | `plan-delegate` verifies agent-to-agent delegation over real RPC. |
| Guardrail/stop behavior | ✅ Verified | `universe` runs with guardrails enabled and asserts the guarded path completes. |
| Streaming | ⚠️ Unverified | `ai.Model.Stream` exists on the interface, but end-to-end streaming conformance is a roadmap item. |
| Structured errors | ⚠️ Unverified | Error handling is covered by normal test suites, but provider conformance does not yet compare structured provider errors. |
## Provider capability matrix
This matrix combines the registered provider interfaces with the conformance
coverage above. The chat/text column is the harness path: when the provider has a
configured key, the conformance command exercises the verified rows in the
previous section.
| Provider | Chat/text agent harness | Image | Video | Streaming | Structured errors |
| --- | --- | --- | --- | --- | --- |
| `anthropic` | ✅ Verified when configured | — Unsupported | — Unsupported | ⚠️ Unverified | ⚠️ Unverified |
| `openai` | ✅ Verified when configured | ✅ Registered | — Unsupported | ⚠️ Unverified | ⚠️ Unverified |
| `gemini` | ✅ Verified when configured | — Unsupported | — Unsupported | ⚠️ Unverified | ⚠️ Unverified |
| `groq` | ✅ Verified when configured | — Unsupported | — Unsupported | ⚠️ Unverified | ⚠️ Unverified |
| `mistral` | ✅ Verified when configured | — Unsupported | — Unsupported | ⚠️ Unverified | ⚠️ Unverified |
| `together` | ✅ Verified when configured | — Unsupported | — Unsupported | ⚠️ Unverified | ⚠️ Unverified |
| `atlascloud` | ✅ Verified when configured | ✅ Registered | ✅ Registered | ⚠️ Unverified | ⚠️ Unverified |
## Running a focused check
Use `-providers` to select a provider and `-harnesses` to narrow the scenario:
```sh
go run ./internal/harness/provider-conformance \
-providers openai,anthropic \
-harnesses agent-flow,plan-delegate
```
By default missing live-provider keys are reported as skips. Add
`-require-configured` in CI when a selected provider must be present:
```sh
go run ./internal/harness/provider-conformance \
-providers openai \
-require-configured
```
The command also prints the registered model, image, and video provider
capabilities before running conformance. Disable that with `-capabilities=false`
when you only want pass/fail output.
For automation, add `-summary-json` to capture the selected providers,
harnesses, registered capability rows, and pass/skip/fail results in a stable
machine-readable file. Add `-capabilities-markdown` when you also want a
ready-to-publish Markdown support table for release notes, docs, or issue
updates:
```sh
go run ./internal/harness/provider-conformance \
-providers mock \
-summary-json provider-conformance-summary.json \
-capabilities-markdown provider-capabilities.md
```
## Related docs
- [The Agent Harness](agent-harness.html)
- [Agents and Workflows](agents-and-workflows.html)
- [AI Provider Guide](ai-provider-guide.html)
- [Roadmap](/docs/roadmap.html)
@@ -1,47 +0,0 @@
package guides_test
import (
"fmt"
"os"
"path/filepath"
"runtime"
"strings"
"testing"
"go-micro.dev/v6/ai"
_ "go-micro.dev/v6/ai/anthropic"
_ "go-micro.dev/v6/ai/atlascloud"
_ "go-micro.dev/v6/ai/gemini"
_ "go-micro.dev/v6/ai/groq"
_ "go-micro.dev/v6/ai/mistral"
_ "go-micro.dev/v6/ai/openai"
_ "go-micro.dev/v6/ai/together"
)
func TestAIProviderGuideCapabilityMatrixMatchesRegistry(t *testing.T) {
_, filename, _, ok := runtime.Caller(0)
if !ok {
t.Fatal("runtime.Caller failed")
}
guidePath := filepath.Join(filepath.Dir(filename), "ai-provider-guide.md")
b, err := os.ReadFile(guidePath)
if err != nil {
t.Fatalf("read AI provider guide: %v", err)
}
guide := string(b)
for _, row := range ai.CapabilityRows() {
want := fmt.Sprintf("| `%s` | %s | %s | %s | %s |", row.Provider, yesNo(row.Model), yesNo(row.Image), yesNo(row.Video), yesNo(row.Stream))
if !strings.Contains(guide, want) {
t.Fatalf("AI provider guide capability matrix is stale; missing row %q", want)
}
}
}
func yesNo(ok bool) string {
if ok {
return "Yes"
}
return "No"
}
@@ -42,10 +42,10 @@ Because every endpoint is already an MCP tool, the gateway is where you charge.
```bash
micro mcp serve --address :3000 \
--x402_pay_to 0xYourAddress \
--x402_network solana \
--x402_amount 10000 \
--x402_facilitator https://facilitator.example
--x402-pay-to 0xYourAddress \
--x402-network solana \
--x402-amount 10000 \
--x402-facilitator https://facilitator.example
```
## A shoppable catalog
@@ -69,7 +69,7 @@ Free tools carry no `payment` block. This is the foundation for a tool marketpla
Different tools can cost different amounts. Pricing is an **operator** concern — the payTo address is the operator's, and amounts change without redeploying anyone's service — so it's configured at the gateway with a file, the same way per-tool scopes and rate limits are. Point the gateway at an x402 config:
```bash
micro mcp serve --address :3000 --x402_config x402.json
micro mcp serve --address :3000 --x402-config x402.json
```
```json
@@ -85,7 +85,7 @@ micro mcp serve --address :3000 --x402_config x402.json
}
```
`amount` is the default (here `"0"` — free unless priced), and `amounts` sets per-tool overrides keyed by tool name. There is no "pricing" abstraction; it's the x402 `amount`, resolved per tool, in the protocol's own vocabulary. `micro mcp serve` accepts the file via `--x402_config`; the standalone gateway accepts the same file via `--x402-config` or the `X402_CONFIG` environment variable.
`amount` is the default (here `"0"` — free unless priced), and `amounts` sets per-tool overrides keyed by tool name. There is no "pricing" abstraction; it's the x402 `amount`, resolved per tool, in the protocol's own vocabulary. The standalone gateway accepts the same file via `--x402-config` or the `X402_CONFIG` environment variable.
## Paying for tools (the consumer side)
+4 -4
View File
@@ -34,7 +34,7 @@ about the framework.
- [Transport](transport.html)
- [Store](store.html)
- [Plugins](plugins.html)
- [Examples](examples/)
- [Examples](examples/index.md)
## Development & Deployment
@@ -52,9 +52,9 @@ about the framework.
## Advanced
- [Framework Comparison](guides/comparison.html)
- [Architecture Decisions](architecture/)
- [Real-World Examples](examples/realworld/)
- [Migration Guides](guides/migration/)
- [Architecture Decisions](architecture/index.md)
- [Real-World Examples](examples/realworld/index.md)
- [Migration Guides](guides/migration/index.md)
- [Observability](observability.html)
- [Contributing](contributing.html)
- [Roadmap](roadmap.html)
+1 -1
View File
@@ -98,7 +98,7 @@ Choose based on your deployment:
import "go-micro.dev/v6/server/grpc"
// Use gRPC for better performance
service := micro.NewService("performance-example",
service := micro.NewService(
micro.Server(grpc.NewServer()),
)
```
+8 -8
View File
@@ -26,7 +26,7 @@ import (
func main() {
reg := consul.NewConsulRegistry()
svc := micro.NewService("plugin-example",
svc := micro.NewService(
micro.Registry(reg),
)
svc.Init()
@@ -43,7 +43,7 @@ import (
func main() {
reg := etcd.NewRegistry()
svc := micro.NewService("plugin-example", micro.Registry(reg))
svc := micro.NewService(micro.Registry(reg))
svc.Init()
svc.Run()
}
@@ -60,7 +60,7 @@ import (
func main() {
b := bnats.NewNatsBroker()
svc := micro.NewService("plugin-example", micro.Broker(b))
svc := micro.NewService(micro.Broker(b))
svc.Init()
svc.Run()
}
@@ -75,7 +75,7 @@ import (
func main() {
b := rabbitmq.NewBroker()
svc := micro.NewService("plugin-example", micro.Broker(b))
svc := micro.NewService(micro.Broker(b))
svc.Init()
svc.Run()
}
@@ -90,7 +90,7 @@ import (
func main() {
t := tnats.NewTransport()
svc := micro.NewService("plugin-example", micro.Transport(t))
svc := micro.NewService(micro.Transport(t))
svc.Init()
svc.Run()
}
@@ -108,7 +108,7 @@ import (
)
func main() {
svc := micro.NewService("plugin-example",
svc := micro.NewService(
micro.Server(grpcServer.NewServer()),
micro.Client(grpcClient.NewClient()),
)
@@ -130,7 +130,7 @@ import (
func main() {
st := postgres.NewStore()
svc := micro.NewService("plugin-example", micro.Store(st))
svc := micro.NewService(micro.Store(st))
svc.Init()
svc.Run()
}
@@ -145,7 +145,7 @@ import (
func main() {
st := natsjskv.NewStore()
svc := micro.NewService("plugin-example", micro.Store(st))
svc := micro.NewService(micro.Store(st))
svc.Init()
svc.Run()
}
+1 -1
View File
@@ -53,7 +53,7 @@ import (
func main() {
reg := consul.NewRegistry()
service := micro.NewService("registry-example",
service := micro.NewService(
micro.Registry(reg),
)
service.Init()
+3 -3
View File
@@ -36,7 +36,7 @@ import (
)
func main() {
service := micro.NewService("store-example")
service := micro.NewService()
service.Init()
// Write a record
@@ -64,7 +64,7 @@ import (
func main() {
st := postgres.NewStore()
svc := micro.NewService("store-example", micro.Store(st))
svc := micro.NewService(micro.Store(st))
svc.Init()
svc.Run()
}
@@ -79,7 +79,7 @@ import (
func main() {
st := natsjskv.NewStore()
svc := micro.NewService("store-example", micro.Store(st))
svc := micro.NewService(micro.Store(st))
svc.Init()
svc.Run()
}
+5 -3
View File
@@ -31,9 +31,11 @@ import (
grpcClient "go-micro.dev/v6/client/grpc"
)
service := micro.NewService("myservice",
// Important: Server must be specified before Name
service := micro.NewService(
micro.Server(grpcServer.NewServer()),
micro.Client(grpcClient.NewClient()),
micro.Name("myservice"),
)
```
@@ -57,7 +59,7 @@ import (
func main() {
t := grpc.NewTransport()
service := micro.NewService("transport-example",
service := micro.NewService(
micro.Transport(t),
)
service.Init()
@@ -74,7 +76,7 @@ import (
func main() {
t := tnats.NewTransport()
service := micro.NewService("transport-example", micro.Transport(t))
service := micro.NewService(micro.Transport(t))
service.Init()
service.Run()
}
-4
View File
@@ -12,7 +12,6 @@ import (
"go-micro.dev/v6/server"
"go-micro.dev/v6/service"
"go-micro.dev/v6/store"
"go.opentelemetry.io/otel/trace"
)
type serviceKey struct{}
@@ -202,9 +201,6 @@ func FlowWithCheckpoint(c Checkpoint) FlowOption { return flow.WithCheckpoint(c)
// are always kept). Default: retain all.
func FlowDeleteOnSuccess() FlowOption { return flow.DeleteOnSuccess() }
// FlowTraceProvider enables OpenTelemetry spans for stepped flow runs and steps.
func FlowTraceProvider(tp trace.TracerProvider) FlowOption { return flow.TraceProvider(tp) }
// FlowCall is a step action: an RPC to a service endpoint, sending the
// state data as the request and storing the response.
func FlowCall(service, endpoint string) FlowStepFunc { return flow.Call(service, endpoint) }