gateway/mcp: configurable initialize (serverInfo, protocol version) + named not-found (#3335)

* gateway/mcp: configurable initialize + named not-found

NewHandler takes HandlerOptions: WithServerInfo(name, version) and WithProtocolVersion set the initialize response so a product can brand the endpoint instead of advertising go-micro-mcp/2024-11-05. ManualResolver's not-found error now includes the tool name.

* gateway/mcp: configurable initialize + named not-found

NewHandler takes HandlerOptions: WithServerInfo(name, version) and WithProtocolVersion set the initialize response so a product can brand the endpoint instead of advertising go-micro-mcp/2024-11-05. ManualResolver's not-found error now includes the tool name.

agent/otel.go

@@ -31,6 +31,8 @@ const (
 	AttrInputTokens    = "agent.tokens.input"
 	AttrOutputTokens   = "agent.tokens.output"
 	AttrTotalTokens    = "agent.tokens.total"
+	AttrAttempt        = "agent.model.attempt"
+	AttrMaxAttempts    = "agent.model.max_attempts"
 	AttrToolName       = "agent.tool.name"
 	AttrDelegate       = "agent.delegate"
 	AttrGuardrailBlock = "agent.guardrail.block"
@@ -38,20 +40,22 @@ const (
 )
 
 type RunEvent struct {
-	Time      time.Time `json:"time"`
-	RunID     string    `json:"run_id"`
-	ParentID  string    `json:"parent_id,omitempty"`
-	TraceID   string    `json:"trace_id,omitempty"`
-	SpanID    string    `json:"span_id,omitempty"`
-	Agent     string    `json:"agent"`
-	Kind      string    `json:"kind"`
-	Name      string    `json:"name,omitempty"`
-	Provider  string    `json:"provider,omitempty"`
-	Model     string    `json:"model,omitempty"`
-	LatencyMS int64     `json:"latency_ms,omitempty"`
-	Tokens    Usage     `json:"tokens,omitempty"`
-	Refused   string    `json:"refused,omitempty"`
-	Error     string    `json:"error,omitempty"`
+	Time        time.Time `json:"time"`
+	RunID       string    `json:"run_id"`
+	ParentID    string    `json:"parent_id,omitempty"`
+	TraceID     string    `json:"trace_id,omitempty"`
+	SpanID      string    `json:"span_id,omitempty"`
+	Agent       string    `json:"agent"`
+	Kind        string    `json:"kind"`
+	Name        string    `json:"name,omitempty"`
+	Provider    string    `json:"provider,omitempty"`
+	Model       string    `json:"model,omitempty"`
+	Attempt     int       `json:"attempt,omitempty"`
+	MaxAttempts int       `json:"max_attempts,omitempty"`
+	LatencyMS   int64     `json:"latency_ms,omitempty"`
+	Tokens      Usage     `json:"tokens,omitempty"`
+	Refused     string    `json:"refused,omitempty"`
+	Error       string    `json:"error,omitempty"`
 }
 
 type Usage = ai.Usage
@@ -144,7 +148,7 @@ func (m *tracedModel) Generate(ctx context.Context, req *ai.Request, opts ...ai.
 		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}
+		e := RunEvent{Time: time.Now(), RunID: info.RunID, ParentID: info.ParentID, Agent: info.Agent, Kind: "model", Provider: provider, Model: model, Attempt: info.Attempt, MaxAttempts: info.MaxAttempts, LatencyMS: dur, Tokens: usage}
 		if err != nil {
 			e.Error = err.Error()
 		}
@@ -162,6 +166,12 @@ func (m *tracedModel) Generate(ctx context.Context, req *ai.Request, opts ...ai.
 	resp, err := m.Model.Generate(ctx, req, opts...)
 	dur := time.Since(start).Milliseconds()
 	attrs := []attribute.KeyValue{attribute.Int64(AttrLatencyMS, dur)}
+	if info.Attempt > 0 {
+		attrs = append(attrs, attribute.Int(AttrAttempt, info.Attempt))
+	}
+	if info.MaxAttempts > 0 {
+		attrs = append(attrs, attribute.Int(AttrMaxAttempts, info.MaxAttempts))
+	}
 	usage := ai.Usage{}
 	if resp != nil {
 		usage = resp.Usage
@@ -175,7 +185,7 @@ func (m *tracedModel) Generate(ctx context.Context, req *ai.Request, opts ...ai.
 		span.SetStatus(codes.Ok, "")
 	}
 	span.End()
-	e := RunEvent{Time: time.Now(), RunID: info.RunID, ParentID: info.ParentID, Agent: info.Agent, Kind: "model", Provider: provider, Model: model, LatencyMS: dur, Tokens: usage}
+	e := RunEvent{Time: time.Now(), RunID: info.RunID, ParentID: info.ParentID, Agent: info.Agent, Kind: "model", Provider: provider, Model: model, Attempt: info.Attempt, MaxAttempts: info.MaxAttempts, LatencyMS: dur, Tokens: usage}
 	if err != nil {
 		e.Error = err.Error()
 	}
@@ -274,6 +284,12 @@ func runEventAttributes(e RunEvent) []attribute.KeyValue {
 	if e.Model != "" {
 		attrs = append(attrs, attribute.String(AttrModel, e.Model))
 	}
+	if e.Attempt > 0 {
+		attrs = append(attrs, attribute.Int(AttrAttempt, e.Attempt))
+	}
+	if e.MaxAttempts > 0 {
+		attrs = append(attrs, attribute.Int(AttrMaxAttempts, e.MaxAttempts))
+	}
 	if e.LatencyMS > 0 {
 		attrs = append(attrs, attribute.Int64(AttrLatencyMS, e.LatencyMS))
 	}

agent/otel_test.go

@@ -3,6 +3,7 @@ package agent
 import (
 	"context"
 	"encoding/json"
+	"errors"
 	"fmt"
 	"strings"
 	"testing"
@@ -11,10 +12,13 @@ import (
 	"go-micro.dev/v6/ai"
 	"go-micro.dev/v6/store"
 	"go.opentelemetry.io/otel/attribute"
+	"go.opentelemetry.io/otel/codes"
 	"go.opentelemetry.io/otel/sdk/trace"
 	"go.opentelemetry.io/otel/sdk/trace/tracetest"
 )
 
+const codesError = codes.Error
+
 type otelTestModel struct{ opts ai.Options }
 
 func (m *otelTestModel) Init(opts ...ai.Option) error {
@@ -89,6 +93,9 @@ func TestAgentOpenTelemetrySpans(t *testing.T) {
 		if attrs[AttrRunID] != runID || attrs[AttrAgentName] != "runner" {
 			t.Fatalf("%s missing run correlation attributes: %#v", s.Name(), attrs)
 		}
+		if s.Name() == spanNameModelCall && (attrs[AttrAttempt] != "1" || attrs[AttrMaxAttempts] != "1") {
+			t.Fatalf("model span missing attempt attributes: %#v", attrs)
+		}
 	}
 	keys, err := store.Scope(st, "agent", "runner").List(store.ListPrefix("runs/"))
 	if err != nil {
@@ -125,6 +132,77 @@ func TestAgentOpenTelemetrySpans(t *testing.T) {
 	}
 }
 
+type failingOtelModel struct{ opts ai.Options }
+
+func (m *failingOtelModel) Init(opts ...ai.Option) error {
+	for _, o := range opts {
+		o(&m.opts)
+	}
+	return nil
+}
+func (m *failingOtelModel) Options() ai.Options { return m.opts }
+func (m *failingOtelModel) String() string      { return "otelfail" }
+func (m *failingOtelModel) Stream(context.Context, *ai.Request, ...ai.GenerateOption) (ai.Stream, error) {
+	return nil, nil
+}
+func (m *failingOtelModel) Generate(context.Context, *ai.Request, ...ai.GenerateOption) (*ai.Response, error) {
+	return nil, errors.New("provider exploded")
+}
+
+func init() {
+	ai.Register("otelfail", func(opts ...ai.Option) ai.Model { return &failingOtelModel{opts: ai.NewOptions(opts...)} })
+}
+
+func TestAgentOpenTelemetrySpansModelFailure(t *testing.T) {
+	exp := tracetest.NewInMemoryExporter()
+	tp := trace.NewTracerProvider(trace.WithSyncer(exp))
+	st := store.NewMemoryStore()
+	a := New(Name("failing-runner"), Provider("otelfail"), WithStore(st), TraceProvider(tp))
+	if _, err := a.Ask(context.Background(), "hello"); err == nil {
+		t.Fatal("Ask succeeded, want provider error")
+	}
+
+	spans := exp.GetSpans().Snapshots()
+	var sawRunError, sawModelError bool
+	for _, s := range spans {
+		attrs := spanAttributes(s.Attributes())
+		switch s.Name() {
+		case spanNameRun:
+			if attrs[AttrAgentName] == "failing-runner" && s.Status().Code == codesError {
+				sawRunError = true
+			}
+		case spanNameModelCall:
+			if attrs[AttrAgentName] == "failing-runner" && attrs[AttrAttempt] == "1" && s.Status().Code == codesError {
+				sawModelError = true
+			}
+		}
+	}
+	if !sawRunError || !sawModelError {
+		t.Fatalf("missing error spans: run=%v model=%v spans=%d", sawRunError, sawModelError, len(spans))
+	}
+
+	summaries, err := ListRunSummaries(st, "failing-runner")
+	if err != nil {
+		t.Fatal(err)
+	}
+	if len(summaries) != 1 || summaries[0].Status != "error" || summaries[0].LastError == "" {
+		t.Fatalf("unexpected failure summary: %#v", summaries)
+	}
+	events, err := LoadRunEvents(st, "failing-runner", summaries[0].RunID)
+	if err != nil {
+		t.Fatal(err)
+	}
+	var sawModelEvent bool
+	for _, event := range events {
+		if event.Kind == "model" && event.Attempt == 1 && event.MaxAttempts == 1 && event.Error != "" {
+			sawModelEvent = true
+		}
+	}
+	if !sawModelEvent {
+		t.Fatalf("missing failed model event with attempt metadata: %#v", events)
+	}
+}
+
 func spanEventHasRunInfo(events []trace.Event, name, runID, agentName string) bool {
 	for _, event := range events {
 		if event.Name != name {

ai/atlascloud/atlascloud.go

@@ -43,6 +43,7 @@ func init() {
 	ai.RegisterVideo("atlascloud", func(opts ...ai.Option) ai.VideoModel {
 		return NewProvider(opts...)
 	})
+	ai.RegisterStream("atlascloud")
 }
 
 // Provider implements the ai.Model interface for Atlas Cloud.

ai/atlascloud/atlascloud_test.go

@@ -2,7 +2,11 @@ package atlascloud
 
 import (
 	"context"
+	"encoding/json"
 	"errors"
+	"io"
+	"net/http"
+	"net/http/httptest"
 	"testing"
 
 	"go-micro.dev/v6/ai"
@@ -81,16 +85,58 @@ func TestProvider_Generate_NoAPIKey(t *testing.T) {
 	}
 }
 
-func TestProvider_Stream_NotImplemented(t *testing.T) {
-	p := NewProvider()
+func TestProvider_Stream(t *testing.T) {
+	var sawStream, sawIncludeUsage bool
+	ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		if r.URL.Path != "/v1/chat/completions" {
+			t.Errorf("path = %s, want /v1/chat/completions", r.URL.Path)
+		}
+		var body map[string]any
+		if err := json.NewDecoder(r.Body).Decode(&body); err != nil {
+			t.Fatalf("decode request: %v", err)
+		}
+		sawStream, _ = body["stream"].(bool)
+		if so, ok := body["stream_options"].(map[string]any); ok {
+			sawIncludeUsage, _ = so["include_usage"].(bool)
+		}
+		w.Header().Set("Content-Type", "text/event-stream")
+		_, _ = w.Write([]byte("data: {\"choices\":[{\"delta\":{\"content\":\"hel\"}}]}\n\n"))
+		_, _ = w.Write([]byte("data: {\"choices\":[{\"delta\":{\"content\":\"lo\"}}]}\n\n"))
+		_, _ = w.Write([]byte("data: {\"choices\":[],\"usage\":{\"prompt_tokens\":7,\"completion_tokens\":2,\"total_tokens\":9}}\n\n"))
+		_, _ = w.Write([]byte("data: [DONE]\n\n"))
+	}))
+	defer ts.Close()
 
-	req := &ai.Request{
-		Prompt: "Hello",
+	p := NewProvider(ai.WithAPIKey("test-key"), ai.WithBaseURL(ts.URL))
+	stream, err := p.Stream(context.Background(), &ai.Request{Prompt: "Hello"})
+	if err != nil {
+		t.Fatalf("Stream returned error: %v", err)
+	}
+	defer stream.Close()
+	if !sawStream {
+		t.Fatal("stream request did not set stream=true")
+	}
+	if !sawIncludeUsage {
+		t.Fatal("stream request did not set stream_options.include_usage=true")
 	}
 
-	_, err := p.Stream(context.Background(), req)
-	if !errors.Is(err, ai.ErrStreamingUnsupported) {
-		t.Fatalf("Stream error = %v, want ErrStreamingUnsupported", err)
+	first, err := stream.Recv()
+	if err != nil || first.Reply != "hel" {
+		t.Fatalf("first chunk = %#v, %v; want hel", first, err)
+	}
+	second, err := stream.Recv()
+	if err != nil || second.Reply != "lo" {
+		t.Fatalf("second chunk = %#v, %v; want lo", second, err)
+	}
+	usage, err := stream.Recv()
+	if err != nil {
+		t.Fatalf("usage chunk error: %v", err)
+	}
+	if usage.Usage.TotalTokens != 9 || usage.Usage.InputTokens != 7 || usage.Usage.OutputTokens != 2 {
+		t.Fatalf("usage = %#v; want input=7 output=2 total=9", usage.Usage)
+	}
+	if _, err := stream.Recv(); !errors.Is(err, io.EOF) {
+		t.Fatalf("final error = %v, want EOF", err)
 	}
 }
 

ai/capabilities_test.go

@@ -34,7 +34,7 @@ func TestRegisteredProviders(t *testing.T) {
 	}
 
 	got = ai.RegisteredProviders("stream")
-	want = []string{"openai"}
+	want = []string{"atlascloud", "groq", "mistral", "openai", "together"}
 	if !reflect.DeepEqual(got, want) {
 		t.Fatalf("RegisteredProviders(stream) = %#v, want %#v", got, want)
 	}
@@ -44,12 +44,12 @@ func TestCapabilityRows(t *testing.T) {
 	got := ai.CapabilityRows()
 	want := []ai.CapabilityRow{
 		{Provider: "anthropic", Capabilities: ai.Capabilities{Model: true}},
-		{Provider: "atlascloud", Capabilities: ai.Capabilities{Model: true, Image: true, Video: true}},
+		{Provider: "atlascloud", Capabilities: ai.Capabilities{Model: true, Image: true, Video: true, Stream: true}},
 		{Provider: "gemini", Capabilities: ai.Capabilities{Model: true}},
-		{Provider: "groq", Capabilities: ai.Capabilities{Model: true}},
-		{Provider: "mistral", Capabilities: ai.Capabilities{Model: true}},
+		{Provider: "groq", Capabilities: ai.Capabilities{Model: true, Stream: true}},
+		{Provider: "mistral", Capabilities: ai.Capabilities{Model: true, Stream: true}},
 		{Provider: "openai", Capabilities: ai.Capabilities{Model: true, Image: true, Stream: true}},
-		{Provider: "together", Capabilities: ai.Capabilities{Model: true}},
+		{Provider: "together", Capabilities: ai.Capabilities{Model: true, Stream: true}},
 	}
 	if !reflect.DeepEqual(got, want) {
 		t.Fatalf("CapabilityRows() = %#v, want %#v", got, want)
@@ -72,7 +72,7 @@ func TestCapabilityMatrix(t *testing.T) {
 	if caps := ai.ProviderCapabilities("openai"); caps != (ai.Capabilities{Model: true, Image: true, Stream: true}) {
 		t.Fatalf("ProviderCapabilities(openai) = %#v", caps)
 	}
-	if caps := ai.ProviderCapabilities("atlascloud"); caps != (ai.Capabilities{Model: true, Image: true, Video: true}) {
+	if caps := ai.ProviderCapabilities("atlascloud"); caps != (ai.Capabilities{Model: true, Image: true, Video: true, Stream: true}) {
 		t.Fatalf("ProviderCapabilities(atlascloud) = %#v", caps)
 	}
 	if caps := ai.ProviderCapabilities("missing"); caps != (ai.Capabilities{}) {
@@ -88,7 +88,7 @@ func TestRegisterStream(t *testing.T) {
 	}
 
 	got := ai.RegisteredProviders("stream")
-	want := []string{"openai", "test-stream"}
+	want := []string{"atlascloud", "groq", "mistral", "openai", "test-stream", "together"}
 	if !reflect.DeepEqual(got, want) {
 		t.Fatalf("RegisteredProviders(stream) = %#v, want %#v", got, want)
 	}

ai/groq/groq.go

@@ -22,12 +22,14 @@ import (
 	"strings"
 
 	"go-micro.dev/v6/ai"
+	"go-micro.dev/v6/ai/internal/openaiapi"
 )
 
 func init() {
 	ai.Register("groq", func(opts ...ai.Option) ai.Model {
 		return NewProvider(opts...)
 	})
+	ai.RegisterStream("groq")
 }
 
 type Provider struct {
@@ -119,7 +121,7 @@ func (p *Provider) Generate(ctx context.Context, req *ai.Request, opts ...ai.Gen
 }
 
 func (p *Provider) Stream(ctx context.Context, req *ai.Request, opts ...ai.GenerateOption) (ai.Stream, error) {
-	return nil, fmt.Errorf("%w: groq provider", ai.ErrStreamingUnsupported)
+	return openaiapi.Stream(ctx, p.opts, req, "/v1/chat/completions")
 }
 
 func (p *Provider) callAPI(ctx context.Context, req map[string]any) (*ai.Response, map[string]any, error) {

ai/groq/groq_test.go

@@ -2,7 +2,11 @@ package groq
 
 import (
 	"context"
+	"encoding/json"
 	"errors"
+	"io"
+	"net/http"
+	"net/http/httptest"
 	"testing"
 
 	"go-micro.dev/v6/ai"
@@ -40,9 +44,44 @@ func TestProvider_Generate_NoAPIKey(t *testing.T) {
 	}
 }
 
-func TestProvider_Stream_NotImplemented(t *testing.T) {
-	if _, err := NewProvider().Stream(context.Background(), &ai.Request{Prompt: "hi"}); !errors.Is(err, ai.ErrStreamingUnsupported) {
-		t.Fatalf("Stream error = %v, want ErrStreamingUnsupported", err)
+func TestProvider_Stream(t *testing.T) {
+	var sawStream bool
+	ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		if r.URL.Path != "/v1/chat/completions" {
+			t.Fatalf("path = %s, want /v1/chat/completions", r.URL.Path)
+		}
+		var body map[string]any
+		if err := json.NewDecoder(r.Body).Decode(&body); err != nil {
+			t.Fatalf("decode request: %v", err)
+		}
+		sawStream, _ = body["stream"].(bool)
+		w.Header().Set("Content-Type", "text/event-stream")
+		_, _ = w.Write([]byte("data: {\"choices\":[{\"delta\":{\"content\":\"hel\"}}]}\n\n"))
+		_, _ = w.Write([]byte("data: {\"choices\":[{\"delta\":{\"content\":\"lo\"}}]}\n\n"))
+		_, _ = w.Write([]byte("data: [DONE]\n\n"))
+	}))
+	defer ts.Close()
+
+	p := NewProvider(ai.WithAPIKey("test-key"), ai.WithBaseURL(ts.URL))
+	stream, err := p.Stream(context.Background(), &ai.Request{Prompt: "Hello"})
+	if err != nil {
+		t.Fatalf("Stream returned error: %v", err)
+	}
+	defer stream.Close()
+	if !sawStream {
+		t.Fatal("stream request did not set stream=true")
+	}
+
+	first, err := stream.Recv()
+	if err != nil || first.Reply != "hel" {
+		t.Fatalf("first chunk = %#v, %v; want hel", first, err)
+	}
+	second, err := stream.Recv()
+	if err != nil || second.Reply != "lo" {
+		t.Fatalf("second chunk = %#v, %v; want lo", second, err)
+	}
+	if _, err := stream.Recv(); !errors.Is(err, io.EOF) {
+		t.Fatalf("final error = %v, want EOF", err)
 	}
 }
 

ai/internal/openaiapi/stream.go

@@ -0,0 +1,117 @@
+package openaiapi
+
+import (
+	"bufio"
+	"bytes"
+	"context"
+	"encoding/json"
+	"fmt"
+	"io"
+	"net/http"
+	"strings"
+
+	"go-micro.dev/v6/ai"
+)
+
+// Stream opens an OpenAI-compatible chat completions SSE stream.
+func Stream(ctx context.Context, opts ai.Options, req *ai.Request, basePath string) (ai.Stream, error) {
+	messages := []map[string]any{{"role": "system", "content": req.SystemPrompt}}
+	for _, m := range req.Messages {
+		messages = append(messages, map[string]any{"role": m.Role, "content": m.Content})
+	}
+	if req.Prompt != "" {
+		messages = append(messages, map[string]any{"role": "user", "content": req.Prompt})
+	}
+	apiReq := map[string]any{
+		"model":          opts.Model,
+		"messages":       messages,
+		"stream":         true,
+		"stream_options": map[string]any{"include_usage": true},
+	}
+	if opts.MaxTokens > 0 {
+		apiReq["max_tokens"] = opts.MaxTokens
+	}
+	reqBody, err := json.Marshal(apiReq)
+	if err != nil {
+		return nil, fmt.Errorf("failed to marshal stream request: %w", err)
+	}
+	apiURL := strings.TrimRight(opts.BaseURL, "/") + basePath
+	httpReq, err := http.NewRequestWithContext(ctx, http.MethodPost, apiURL, bytes.NewReader(reqBody))
+	if err != nil {
+		return nil, fmt.Errorf("failed to create stream request: %w", err)
+	}
+	httpReq.Header.Set("Content-Type", "application/json")
+	httpReq.Header.Set("Accept", "text/event-stream")
+	httpReq.Header.Set("Authorization", "Bearer "+opts.APIKey)
+
+	httpResp, err := http.DefaultClient.Do(httpReq)
+	if err != nil {
+		return nil, fmt.Errorf("stream API request failed: %w", err)
+	}
+	if httpResp.StatusCode != http.StatusOK {
+		defer httpResp.Body.Close()
+		respBody, _ := io.ReadAll(httpResp.Body)
+		return nil, fmt.Errorf("stream API error (%s): %s", httpResp.Status, string(respBody))
+	}
+	return &StreamReader{body: httpResp.Body, scanner: bufio.NewScanner(httpResp.Body)}, nil
+}
+
+// StreamReader reads OpenAI-compatible server-sent event chunks.
+type StreamReader struct {
+	body    io.ReadCloser
+	scanner *bufio.Scanner
+	closed  bool
+}
+
+func (s *StreamReader) Recv() (*ai.Response, error) {
+	for s.scanner.Scan() {
+		line := strings.TrimSpace(s.scanner.Text())
+		if line == "" || strings.HasPrefix(line, ":") {
+			continue
+		}
+		if !strings.HasPrefix(line, "data:") {
+			continue
+		}
+		data := strings.TrimSpace(strings.TrimPrefix(line, "data:"))
+		if data == "[DONE]" {
+			return nil, io.EOF
+		}
+		var chunk struct {
+			Choices []struct {
+				Delta struct {
+					Content string `json:"content"`
+				} `json:"delta"`
+			} `json:"choices"`
+			Usage *struct {
+				PromptTokens     int `json:"prompt_tokens"`
+				CompletionTokens int `json:"completion_tokens"`
+				TotalTokens      int `json:"total_tokens"`
+			} `json:"usage"`
+		}
+		if err := json.Unmarshal([]byte(data), &chunk); err != nil {
+			return nil, fmt.Errorf("failed to parse stream chunk: %w", err)
+		}
+		if len(chunk.Choices) > 0 && chunk.Choices[0].Delta.Content != "" {
+			return &ai.Response{Reply: chunk.Choices[0].Delta.Content}, nil
+		}
+		if chunk.Usage != nil {
+			return &ai.Response{Usage: ai.Usage{
+				InputTokens:  chunk.Usage.PromptTokens,
+				OutputTokens: chunk.Usage.CompletionTokens,
+				TotalTokens:  chunk.Usage.TotalTokens,
+			}}, nil
+		}
+	}
+	if err := s.scanner.Err(); err != nil {
+		return nil, err
+	}
+	return nil, io.EOF
+}
+
+func (s *StreamReader) Close() error {
+	if s.closed {
+		return nil
+	}
+	s.closed = true
+	return s.body.Close()
+}

ai/mistral/mistral.go

@@ -22,12 +22,14 @@ import (
 	"strings"
 
 	"go-micro.dev/v6/ai"
+	"go-micro.dev/v6/ai/internal/openaiapi"
 )
 
 func init() {
 	ai.Register("mistral", func(opts ...ai.Option) ai.Model {
 		return NewProvider(opts...)
 	})
+	ai.RegisterStream("mistral")
 }
 
 type Provider struct {
@@ -119,7 +121,7 @@ func (p *Provider) Generate(ctx context.Context, req *ai.Request, opts ...ai.Gen
 }
 
 func (p *Provider) Stream(ctx context.Context, req *ai.Request, opts ...ai.GenerateOption) (ai.Stream, error) {
-	return nil, fmt.Errorf("%w: mistral provider", ai.ErrStreamingUnsupported)
+	return openaiapi.Stream(ctx, p.opts, req, "/v1/chat/completions")
 }
 
 func (p *Provider) callAPI(ctx context.Context, req map[string]any) (*ai.Response, map[string]any, error) {

ai/mistral/mistral_test.go

@@ -2,7 +2,11 @@ package mistral
 
 import (
 	"context"
+	"encoding/json"
 	"errors"
+	"io"
+	"net/http"
+	"net/http/httptest"
 	"testing"
 
 	"go-micro.dev/v6/ai"
@@ -40,9 +44,44 @@ func TestProvider_Generate_NoAPIKey(t *testing.T) {
 	}
 }
 
-func TestProvider_Stream_NotImplemented(t *testing.T) {
-	if _, err := NewProvider().Stream(context.Background(), &ai.Request{Prompt: "hi"}); !errors.Is(err, ai.ErrStreamingUnsupported) {
-		t.Fatalf("Stream error = %v, want ErrStreamingUnsupported", err)
+func TestProvider_Stream(t *testing.T) {
+	var sawStream bool
+	ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		if r.URL.Path != "/v1/chat/completions" {
+			t.Fatalf("path = %s, want /v1/chat/completions", r.URL.Path)
+		}
+		var body map[string]any
+		if err := json.NewDecoder(r.Body).Decode(&body); err != nil {
+			t.Fatalf("decode request: %v", err)
+		}
+		sawStream, _ = body["stream"].(bool)
+		w.Header().Set("Content-Type", "text/event-stream")
+		_, _ = w.Write([]byte("data: {\"choices\":[{\"delta\":{\"content\":\"hel\"}}]}\n\n"))
+		_, _ = w.Write([]byte("data: {\"choices\":[{\"delta\":{\"content\":\"lo\"}}]}\n\n"))
+		_, _ = w.Write([]byte("data: [DONE]\n\n"))
+	}))
+	defer ts.Close()
+
+	p := NewProvider(ai.WithAPIKey("test-key"), ai.WithBaseURL(ts.URL))
+	stream, err := p.Stream(context.Background(), &ai.Request{Prompt: "Hello"})
+	if err != nil {
+		t.Fatalf("Stream returned error: %v", err)
+	}
+	defer stream.Close()
+	if !sawStream {
+		t.Fatal("stream request did not set stream=true")
+	}
+
+	first, err := stream.Recv()
+	if err != nil || first.Reply != "hel" {
+		t.Fatalf("first chunk = %#v, %v; want hel", first, err)
+	}
+	second, err := stream.Recv()
+	if err != nil || second.Reply != "lo" {
+		t.Fatalf("second chunk = %#v, %v; want lo", second, err)
+	}
+	if _, err := stream.Recv(); !errors.Is(err, io.EOF) {
+		t.Fatalf("final error = %v, want EOF", err)
 	}
 }
 

ai/together/together.go

@@ -22,12 +22,14 @@ import (
 	"strings"
 
 	"go-micro.dev/v6/ai"
+	"go-micro.dev/v6/ai/internal/openaiapi"
 )
 
 func init() {
 	ai.Register("together", func(opts ...ai.Option) ai.Model {
 		return NewProvider(opts...)
 	})
+	ai.RegisterStream("together")
 }
 
 type Provider struct {
@@ -119,7 +121,7 @@ func (p *Provider) Generate(ctx context.Context, req *ai.Request, opts ...ai.Gen
 }
 
 func (p *Provider) Stream(ctx context.Context, req *ai.Request, opts ...ai.GenerateOption) (ai.Stream, error) {
-	return nil, fmt.Errorf("%w: together provider", ai.ErrStreamingUnsupported)
+	return openaiapi.Stream(ctx, p.opts, req, "/v1/chat/completions")
 }
 
 func (p *Provider) callAPI(ctx context.Context, req map[string]any) (*ai.Response, map[string]any, error) {

ai/together/together_test.go

@@ -2,7 +2,11 @@ package together
 
 import (
 	"context"
+	"encoding/json"
 	"errors"
+	"io"
+	"net/http"
+	"net/http/httptest"
 	"testing"
 
 	"go-micro.dev/v6/ai"
@@ -40,9 +44,44 @@ func TestProvider_Generate_NoAPIKey(t *testing.T) {
 	}
 }
 
-func TestProvider_Stream_NotImplemented(t *testing.T) {
-	if _, err := NewProvider().Stream(context.Background(), &ai.Request{Prompt: "hi"}); !errors.Is(err, ai.ErrStreamingUnsupported) {
-		t.Fatalf("Stream error = %v, want ErrStreamingUnsupported", err)
+func TestProvider_Stream(t *testing.T) {
+	var sawStream bool
+	ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		if r.URL.Path != "/v1/chat/completions" {
+			t.Fatalf("path = %s, want /v1/chat/completions", r.URL.Path)
+		}
+		var body map[string]any
+		if err := json.NewDecoder(r.Body).Decode(&body); err != nil {
+			t.Fatalf("decode request: %v", err)
+		}
+		sawStream, _ = body["stream"].(bool)
+		w.Header().Set("Content-Type", "text/event-stream")
+		_, _ = w.Write([]byte("data: {\"choices\":[{\"delta\":{\"content\":\"hel\"}}]}\n\n"))
+		_, _ = w.Write([]byte("data: {\"choices\":[{\"delta\":{\"content\":\"lo\"}}]}\n\n"))
+		_, _ = w.Write([]byte("data: [DONE]\n\n"))
+	}))
+	defer ts.Close()
+
+	p := NewProvider(ai.WithAPIKey("test-key"), ai.WithBaseURL(ts.URL))
+	stream, err := p.Stream(context.Background(), &ai.Request{Prompt: "Hello"})
+	if err != nil {
+		t.Fatalf("Stream returned error: %v", err)
+	}
+	defer stream.Close()
+	if !sawStream {
+		t.Fatal("stream request did not set stream=true")
+	}
+
+	first, err := stream.Recv()
+	if err != nil || first.Reply != "hel" {
+		t.Fatalf("first chunk = %#v, %v; want hel", first, err)
+	}
+	second, err := stream.Recv()
+	if err != nil || second.Reply != "lo" {
+		t.Fatalf("second chunk = %#v, %v; want lo", second, err)
+	}
+	if _, err := stream.Recv(); !errors.Is(err, io.EOF) {
+		t.Fatalf("final error = %v, want EOF", err)
 	}
 }
 

cmd/micro/inspect/inspect.go

@@ -0,0 +1,169 @@
+// Package inspect registers the 'micro inspect' CLI command.
+package inspect
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"io"
+	"os"
+
+	"github.com/urfave/cli/v2"
+	goagent "go-micro.dev/v6/agent"
+	"go-micro.dev/v6/cmd"
+	aiflow "go-micro.dev/v6/flow"
+	"go-micro.dev/v6/store"
+)
+
+func init() {
+	cmd.Register(&cli.Command{
+		Name:  "inspect",
+		Usage: "Inspect recent agent and workflow activity",
+		Description: `Inspect is the CLI checkpoint in the local scaffold → run → chat → inspect loop.
+It reads durable local run history, so it works after the agent or flow has stopped.`,
+		Subcommands: []*cli.Command{
+			{
+				Name:      "agent",
+				Usage:     "Show recent recorded runs for an agent",
+				ArgsUsage: "[agent]",
+				Flags:     inspectAgentFlags(),
+				Action:    inspectAgent,
+			},
+			{
+				Name:      "flow",
+				Usage:     "Show durable run history for a flow",
+				ArgsUsage: "[flow]",
+				Flags:     inspectFlowFlags(),
+				Action:    inspectFlow,
+			},
+		},
+	})
+}
+
+func inspectAgentFlags() []cli.Flag {
+	return []cli.Flag{
+		&cli.BoolFlag{Name: "json", Usage: "Print run summaries 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 inspectFlowFlags() []cli.Flag {
+	return []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"},
+	}
+}
+
+func inspectAgent(c *cli.Context) error {
+	name := c.Args().First()
+	if name == "" {
+		return fmt.Errorf("agent name required: micro inspect agent <name>")
+	}
+	opts := goagent.RunListOptions{Status: c.String("status"), TraceID: c.String("trace"), Limit: c.Int("limit")}
+	runs, err := goagent.ListRunSummariesWithOptions(store.DefaultStore, name, opts)
+	if err != nil {
+		return err
+	}
+	return writeAgentInspection(os.Stdout, name, runs, c.Bool("json"))
+}
+
+func writeAgentInspection(w io.Writer, name string, runs []goagent.RunSummary, asJSON bool) error {
+	if asJSON {
+		enc := json.NewEncoder(w)
+		enc.SetIndent("", "  ")
+		return enc.Encode(runs)
+	}
+	if len(runs) == 0 {
+		fmt.Fprintf(w, "  No agent runs recorded for %q. After chatting, try: micro inspect agent %s\n", name, name)
+		return nil
+	}
+	fmt.Fprintf(w, "  Agent %q runs\n", name)
+	for _, run := range runs {
+		fmt.Fprintf(w, "  %s  status=%s  events=%d  last=%s", run.RunID, run.Status, run.Events, run.LastKind)
+		if run.LastError != "" {
+			fmt.Fprintf(w, "  error=%q", run.LastError)
+		}
+		if run.TraceID != "" {
+			fmt.Fprintf(w, "  trace=%s", shortID(run.TraceID))
+		}
+		fmt.Fprintln(w)
+	}
+	return nil
+}
+
+func inspectFlow(c *cli.Context) error {
+	name := c.Args().First()
+	if name == "" {
+		return fmt.Errorf("flow name required: micro inspect flow <name>")
+	}
+	runs, err := aiflow.StoreCheckpoint(nil, name).List(context.Background())
+	if err != nil {
+		return err
+	}
+	runs = filterFlowInspection(runs, c.Bool("pending"), c.String("status"), c.String("stage"), c.Int("limit"))
+	return writeFlowInspection(os.Stdout, name, runs, c.Bool("json"), c.Bool("pending"))
+}
+
+func filterFlowInspection(runs []aiflow.Run, pending bool, status, stage string, limit int) []aiflow.Run {
+	filtered := make([]aiflow.Run, 0, len(runs))
+	for _, run := range runs {
+		if pending && run.Status == "done" {
+			continue
+		}
+		if status != "" && run.Status != status {
+			continue
+		}
+		if stage != "" && run.State.Stage != stage {
+			continue
+		}
+		filtered = append(filtered, run)
+	}
+	if limit > 0 && len(filtered) > limit {
+		return filtered[len(filtered)-limit:]
+	}
+	return filtered
+}
+
+func writeFlowInspection(w io.Writer, name string, runs []aiflow.Run, asJSON, pending bool) error {
+	if asJSON {
+		enc := json.NewEncoder(w)
+		enc.SetIndent("", "  ")
+		return enc.Encode(runs)
+	}
+	if len(runs) == 0 {
+		if pending {
+			fmt.Fprintf(w, "  No pending flow runs recorded for %q.\n", name)
+			return nil
+		}
+		fmt.Fprintf(w, "  No flow runs recorded for %q. After executing a durable flow, try: micro inspect flow %s\n", name, name)
+		return nil
+	}
+	fmt.Fprintf(w, "  Flow %q runs\n", name)
+	for _, run := range runs {
+		stage := run.State.Stage
+		if stage == "" {
+			stage = "-"
+		}
+		fmt.Fprintf(w, "  %s  status=%s  stage=%s  steps=%d", shortID(run.ID), run.Status, stage, len(run.Steps))
+		for _, step := range run.Steps {
+			if step.Error != "" {
+				fmt.Fprintf(w, "  error=%q", step.Error)
+				break
+			}
+		}
+		fmt.Fprintln(w)
+	}
+	return nil
+}
+
+func shortID(id string) string {
+	if len(id) <= 12 {
+		return id
+	}
+	return id[:12]
+}

cmd/micro/inspect/inspect_test.go

@@ -0,0 +1,64 @@
+package inspect
+
+import (
+	"bytes"
+	"encoding/json"
+	"strings"
+	"testing"
+
+	goagent "go-micro.dev/v6/agent"
+	aiflow "go-micro.dev/v6/flow"
+)
+
+func TestWriteAgentInspectionIncludesActionableBreadcrumbs(t *testing.T) {
+	runs := []goagent.RunSummary{{RunID: "run-1", Status: "error", Events: 4, LastKind: "tool", LastError: "boom", TraceID: "1234567890abcdef"}}
+	var out bytes.Buffer
+	if err := writeAgentInspection(&out, "support", runs, false); err != nil {
+		t.Fatal(err)
+	}
+	got := out.String()
+	for _, want := range []string{"Agent \"support\" runs", "run-1", "status=error", "events=4", "last=tool", `error="boom"`, "trace=1234567890ab"} {
+		if !strings.Contains(got, want) {
+			t.Fatalf("output missing %q:\n%s", want, got)
+		}
+	}
+}
+
+func TestWriteAgentInspectionEmptyStateNamesInspectCommand(t *testing.T) {
+	var out bytes.Buffer
+	if err := writeAgentInspection(&out, "support", nil, false); err != nil {
+		t.Fatal(err)
+	}
+	if got := out.String(); !strings.Contains(got, "micro inspect agent support") {
+		t.Fatalf("empty state missing next step: %q", got)
+	}
+}
+
+func TestWriteFlowInspectionIncludesFailedStepBreadcrumb(t *testing.T) {
+	runs := []aiflow.Run{{ID: "1234567890abcdef", Status: "failed", State: aiflow.State{Stage: "charge"}, Steps: []aiflow.StepRecord{{Name: "charge", Status: "failed", Error: "card declined"}}}}
+	var out bytes.Buffer
+	if err := writeFlowInspection(&out, "checkout", runs, false, false); err != nil {
+		t.Fatal(err)
+	}
+	got := out.String()
+	for _, want := range []string{"Flow \"checkout\" runs", "1234567890ab", "status=failed", "stage=charge", "steps=1", `error="card declined"`} {
+		if !strings.Contains(got, want) {
+			t.Fatalf("output missing %q:\n%s", want, got)
+		}
+	}
+}
+
+func TestWriteFlowInspectionJSON(t *testing.T) {
+	runs := []aiflow.Run{{ID: "run-1", Flow: "checkout", Status: "done"}}
+	var out bytes.Buffer
+	if err := writeFlowInspection(&out, "checkout", runs, true, false); err != nil {
+		t.Fatal(err)
+	}
+	var got []aiflow.Run
+	if err := json.Unmarshal(out.Bytes(), &got); err != nil {
+		t.Fatalf("invalid JSON: %v\n%s", err, out.String())
+	}
+	if len(got) != 1 || got[0].ID != "run-1" || got[0].Status != "done" {
+		t.Fatalf("decoded runs = %+v", got)
+	}
+}

cmd/micro/main.go

@@ -12,6 +12,7 @@ import (
 	_ "go-micro.dev/v6/cmd/micro/cli/build"
 	_ "go-micro.dev/v6/cmd/micro/cli/deploy"
 	_ "go-micro.dev/v6/cmd/micro/flow"
+	_ "go-micro.dev/v6/cmd/micro/inspect"
 	_ "go-micro.dev/v6/cmd/micro/mcp"
 	_ "go-micro.dev/v6/cmd/micro/resource"
 	_ "go-micro.dev/v6/cmd/micro/run"

cmd/micro/zero_to_hero_cli_test.go

@@ -19,7 +19,7 @@ func TestZeroToHeroCLIBoundaries(t *testing.T) {
 		}
 	}
 
-	for _, want := range []string{"run", "chat", "flow"} {
+	for _, want := range []string{"run", "chat", "flow", "inspect"} {
 		if !commands[want] {
 			t.Fatalf("missing %q command", want)
 		}
@@ -27,4 +27,7 @@ func TestZeroToHeroCLIBoundaries(t *testing.T) {
 	if !subcommands["flow"]["runs"] {
 		t.Fatal("missing inspect boundary: flow runs")
 	}
+	if !subcommands["inspect"]["agent"] || !subcommands["inspect"]["flow"] {
+		t.Fatal("missing inspect boundary: inspect agent/flow")
+	}
 }

gateway/mcp/httpjsonrpc.go

@@ -0,0 +1,129 @@
+package mcp
+
+import (
+	"encoding/json"
+	"net/http"
+	"strings"
+)
+
+// HandlerOption configures NewHandler.
+type HandlerOption func(*handlerOptions)
+
+type handlerOptions struct {
+	serverName, serverVersion, protocolVersion string
+}
+
+// WithServerInfo sets the name/version advertised in the initialize response.
+func WithServerInfo(name, version string) HandlerOption {
+	return func(o *handlerOptions) { o.serverName, o.serverVersion = name, version }
+}
+
+// WithProtocolVersion sets the MCP protocol version advertised in initialize.
+func WithProtocolVersion(v string) HandlerOption {
+	return func(o *handlerOptions) { o.protocolVersion = v }
+}
+
+// NewHandler returns an http.Handler serving the MCP protocol over HTTP as
+// JSON-RPC 2.0 (initialize, ping, notifications/*, tools/list, tools/call),
+// backed by the resolver. Mount it on your own server (e.g. POST /mcp): the
+// gateway provides the protocol; you keep your routes, middleware and any
+// human-facing docs page.
+func NewHandler(r Resolver, opts ...HandlerOption) http.Handler {
+	o := handlerOptions{serverName: "go-micro-mcp", serverVersion: "1.0.0", protocolVersion: "2024-11-05"}
+	for _, fn := range opts {
+		fn(&o)
+	}
+	return http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) {
+		if req.Method != http.MethodPost {
+			http.Error(w, "method not allowed", http.StatusMethodNotAllowed)
+			return
+		}
+		var rpc struct {
+			JSONRPC string          `json:"jsonrpc"`
+			ID      json.RawMessage `json:"id"`
+			Method  string          `json:"method"`
+			Params  json.RawMessage `json:"params"`
+		}
+		if err := json.NewDecoder(req.Body).Decode(&rpc); err != nil {
+			writeRPCError(w, nil, ParseError, "Parse error", err.Error())
+			return
+		}
+
+		// Notifications (and any id-less request) expect no response body.
+		if strings.HasPrefix(rpc.Method, "notifications/") || len(rpc.ID) == 0 {
+			w.WriteHeader(http.StatusNoContent)
+			return
+		}
+
+		ctx := req.Context()
+		switch rpc.Method {
+		case "initialize":
+			writeRPCResult(w, rpc.ID, map[string]interface{}{
+				"protocolVersion": o.protocolVersion,
+				"capabilities":    map[string]interface{}{"tools": map[string]interface{}{}},
+				"serverInfo":      map[string]interface{}{"name": o.serverName, "version": o.serverVersion},
+			})
+		case "ping":
+			writeRPCResult(w, rpc.ID, map[string]interface{}{})
+		case "tools/list":
+			tools, err := r.List(ctx)
+			if err != nil {
+				writeRPCError(w, rpc.ID, InternalError, "Failed to list tools", err.Error())
+				return
+			}
+			list := make([]map[string]interface{}, 0, len(tools))
+			for _, t := range tools {
+				list = append(list, map[string]interface{}{
+					"name": t.Name, "description": t.Description, "inputSchema": t.InputSchema,
+				})
+			}
+			writeRPCResult(w, rpc.ID, map[string]interface{}{"tools": list})
+		case "tools/call":
+			var p struct {
+				Name      string                 `json:"name"`
+				Arguments map[string]interface{} `json:"arguments"`
+			}
+			if err := json.Unmarshal(rpc.Params, &p); err != nil {
+				writeRPCError(w, rpc.ID, InvalidParams, "Invalid params", err.Error())
+				return
+			}
+			res, err := r.Call(ctx, p.Name, p.Arguments)
+			if err != nil {
+				// Protocol/pre-check failure -> JSON-RPC error. An *RPCError
+				// carries a specific code; anything else is InternalError.
+				if rpcErr, ok := err.(*RPCError); ok {
+					writeRPCError(w, rpc.ID, rpcErr.Code, rpcErr.Message, rpcErr.Data)
+				} else {
+					writeRPCError(w, rpc.ID, InternalError, "Tool call failed", err.Error())
+				}
+				return
+			}
+			result := map[string]interface{}{
+				"content": []map[string]interface{}{{"type": "text", "text": res.Text}},
+			}
+			if res.IsError {
+				result["isError"] = true
+			}
+			writeRPCResult(w, rpc.ID, result)
+		default:
+			writeRPCError(w, rpc.ID, MethodNotFound, "Method not found", rpc.Method)
+		}
+	})
+}
+
+func writeRPCResult(w http.ResponseWriter, id json.RawMessage, result interface{}) {
+	w.Header().Set("Content-Type", "application/json")
+	_ = json.NewEncoder(w).Encode(map[string]interface{}{"jsonrpc": "2.0", "id": rawOrNull(id), "result": result})
+}
+
+func writeRPCError(w http.ResponseWriter, id json.RawMessage, code int, msg string, data interface{}) {
+	w.Header().Set("Content-Type", "application/json")
+	_ = json.NewEncoder(w).Encode(map[string]interface{}{"jsonrpc": "2.0", "id": rawOrNull(id), "error": map[string]interface{}{"code": code, "message": msg, "data": data}})
+}
+
+func rawOrNull(id json.RawMessage) interface{} {
+	if len(id) == 0 {
+		return nil
+	}
+	return id
+}

gateway/mcp/resolver.go

@@ -0,0 +1,131 @@
+package mcp
+
+import (
+	"context"
+	"sync"
+
+	"go-micro.dev/v6/ai"
+	"go-micro.dev/v6/client"
+	"go-micro.dev/v6/registry"
+)
+
+// CallResult is the outcome of a successful tool dispatch. A tool that ran but
+// produced an error sets IsError — per the MCP spec this is returned as a
+// tools/call result with isError:true, not a JSON-RPC protocol error.
+type CallResult struct {
+	Text    string
+	IsError bool
+}
+
+// Error lets the package's RPCError (see stdio.go) be returned by a resolver
+// to signal a protocol/pre-check failure with a specific JSON-RPC code; the
+// handler maps it straight to the JSON-RPC error.
+func (e *RPCError) Error() string { return e.Message }
+
+// ToolFunc executes a manually-registered tool. Return a *CallResult for tool
+// outcomes (set IsError for tool-level failures); return a non-nil error — an
+// *RPCError for a specific code — for protocol/pre-check failures.
+type ToolFunc func(ctx context.Context, args map[string]any) (*CallResult, error)
+
+// Resolver supplies the gateway's tools and executes calls. Swapping the
+// resolver changes where tools come from without touching the MCP protocol or
+// transport:
+//
+//   - NewManualResolver:   tools you register explicitly (full product control,
+//     including tools that are not go-micro services, executed via your own
+//     logic — auth, metering, …).
+//   - NewRegistryResolver: tools auto-discovered from registered services.
+//
+// The built-in store/broker tools are intentionally NOT exposed by any
+// resolver — they remain a development convenience on the legacy Serve() path.
+type Resolver interface {
+	// List returns the current tool catalog.
+	List(ctx context.Context) ([]Tool, error)
+	// Call executes a tool by name with JSON arguments.
+	Call(ctx context.Context, name string, args map[string]any) (*CallResult, error)
+}
+
+// ManualResolver exposes an explicitly-registered set of tools.
+type ManualResolver struct {
+	mu    sync.RWMutex
+	order []Tool
+	funcs map[string]ToolFunc
+}
+
+// NewManualResolver returns an empty manual resolver.
+func NewManualResolver() *ManualResolver {
+	return &ManualResolver{funcs: map[string]ToolFunc{}}
+}
+
+// Add registers (or replaces) a tool and its handler. Returns the resolver for
+// chaining.
+func (m *ManualResolver) Add(t Tool, fn ToolFunc) *ManualResolver {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	if _, ok := m.funcs[t.Name]; ok {
+		for i := range m.order {
+			if m.order[i].Name == t.Name {
+				m.order[i] = t
+			}
+		}
+	} else {
+		m.order = append(m.order, t)
+	}
+	m.funcs[t.Name] = fn
+	return m
+}
+
+// List returns the registered tools.
+func (m *ManualResolver) List(_ context.Context) ([]Tool, error) {
+	m.mu.RLock()
+	defer m.mu.RUnlock()
+	out := make([]Tool, len(m.order))
+	copy(out, m.order)
+	return out, nil
+}
+
+// Call runs the handler registered for name.
+func (m *ManualResolver) Call(ctx context.Context, name string, args map[string]any) (*CallResult, error) {
+	m.mu.RLock()
+	fn, ok := m.funcs[name]
+	m.mu.RUnlock()
+	if !ok {
+		return nil, &RPCError{Code: InvalidParams, Message: "Tool not found: " + name, Data: name}
+	}
+	return fn(ctx, args)
+}
+
+// RegistryResolver auto-discovers tools from registered go-micro services and
+// executes them over RPC. It exposes only services — never the internal
+// store/broker tools.
+type RegistryResolver struct {
+	tools *ai.Tools
+}
+
+// NewRegistryResolver discovers services from reg and calls them with cl.
+func NewRegistryResolver(reg registry.Registry, cl client.Client) *RegistryResolver {
+	return &RegistryResolver{tools: ai.NewTools(reg, ai.ToolClient(cl))}
+}
+
+// List discovers the current service tools.
+func (r *RegistryResolver) List(_ context.Context) ([]Tool, error) {
+	discovered, err := r.tools.Discover()
+	if err != nil {
+		return nil, err
+	}
+	out := make([]Tool, 0, len(discovered))
+	for _, t := range discovered {
+		out = append(out, Tool{
+			Name:        t.Name,
+			Description: t.Description,
+			InputSchema: map[string]interface{}{"type": "object", "properties": t.Properties},
+		})
+	}
+	return out, nil
+}
+
+// Call executes a discovered service tool.
+func (r *RegistryResolver) Call(ctx context.Context, name string, args map[string]any) (*CallResult, error) {
+	res := r.tools.Handler()(ctx, ai.ToolCall{ID: "1", Name: name, Input: args})
+	return &CallResult{Text: res.Content}, nil
+}

gateway/mcp/resolver_test.go

@@ -0,0 +1,61 @@
+package mcp
+
+import (
+	"context"
+	"encoding/json"
+	"net/http"
+	"net/http/httptest"
+	"strings"
+	"testing"
+)
+
+func TestManualResolverHandler(t *testing.T) {
+	res := NewManualResolver().
+		Add(Tool{Name: "echo", Description: "echoes text"},
+			func(_ context.Context, args map[string]interface{}) (*CallResult, error) {
+				s, _ := args["text"].(string)
+				return &CallResult{Text: "you said: " + s}, nil
+			}).
+		Add(Tool{Name: "boom", Description: "errors"},
+			func(_ context.Context, _ map[string]interface{}) (*CallResult, error) {
+				return &CallResult{Text: "kaboom", IsError: true}, nil
+			}).
+		Add(Tool{Name: "blocked", Description: "coded error"},
+			func(_ context.Context, _ map[string]interface{}) (*CallResult, error) {
+				return nil, &RPCError{Code: -32000, Message: "insufficient credits"}
+			})
+
+	ts := httptest.NewServer(NewHandler(res))
+	defer ts.Close()
+	rpc := func(body string) (int, map[string]interface{}) {
+		resp, _ := http.Post(ts.URL, "application/json", strings.NewReader(body))
+		defer resp.Body.Close()
+		var out map[string]interface{}
+		json.NewDecoder(resp.Body).Decode(&out)
+		return resp.StatusCode, out
+	}
+
+	if _, out := rpc(`{"jsonrpc":"2.0","id":1,"method":"tools/list"}`); len(out["result"].(map[string]interface{})["tools"].([]interface{})) != 3 {
+		t.Fatalf("tools/list: %v", out)
+	}
+	// tool result
+	_, out := rpc(`{"jsonrpc":"2.0","id":2,"method":"tools/call","params":{"name":"echo","arguments":{"text":"hi"}}}`)
+	if out["result"].(map[string]interface{})["content"].([]interface{})[0].(map[string]interface{})["text"] != "you said: hi" {
+		t.Fatalf("echo: %v", out)
+	}
+	// tool-level error -> isError result, NOT protocol error
+	_, out = rpc(`{"jsonrpc":"2.0","id":3,"method":"tools/call","params":{"name":"boom","arguments":{}}}`)
+	if out["error"] != nil || out["result"].(map[string]interface{})["isError"] != true {
+		t.Fatalf("boom should be isError result: %v", out)
+	}
+	// coded protocol error -> JSON-RPC error with the code
+	_, out = rpc(`{"jsonrpc":"2.0","id":4,"method":"tools/call","params":{"name":"blocked","arguments":{}}}`)
+	if out["error"] == nil || int(out["error"].(map[string]interface{})["code"].(float64)) != -32000 {
+		t.Fatalf("blocked should be -32000: %v", out)
+	}
+	// notification -> 204, no body
+	code, _ := rpc(`{"jsonrpc":"2.0","method":"notifications/initialized"}`)
+	if code != http.StatusNoContent {
+		t.Fatalf("notification status = %d, want 204", code)
+	}
+}

internal/docs/PRIORITIES.md

@@ -21,8 +21,9 @@ changes, architectural rewrites. Those go to the human.
 
 ## Developer experience (ranked)
 
-1. **Expose run inspection in the CLI inner loop** ([#3297](https://github.com/micro/go-micro/issues/3297)) — the canon promises scaffold → run → chat → inspect → deploy, and recent work added run timelines, trace correlation, and a `micro runs` foothold; with conformance scheduling, failure/backoff hardening, the deploy contract, and the support reference now shipped, this is the highest-value remaining DX seam in the Now/ongoing roadmap until local agent/flow activity is documented and CI-tested as an actionable inspect step.
-2. **Add durable agent run checkpoint and resume** ([#3306](https://github.com/micro/go-micro/issues/3306)) — once the remaining inspection seam is closed, the highest-value Next-roadmap item is making agent loops resumable like flows so long-running work can survive restarts without unsafe replay or hidden state loss.
+1. **Emit agent RunInfo as OpenTelemetry spans** ([#3316](https://github.com/micro/go-micro/issues/3316)) — provider-backed streaming just broadened across the OpenAI-compatible path, so the highest-value remaining Next-roadmap seam is making agent execution operable: model calls, tool calls, failures, and run IDs should correlate with the same traces as services and flows.
+2. **Add memory compaction and retrieval for long-running agents** ([#3321](https://github.com/micro/go-micro/issues/3321)) — after run timelines become observable, tackle the leading Later-roadmap gap: bounded, store-backed memory that keeps long agent runs useful without turning Go Micro into a prompt-layer framework.
+3. **Add human-in-the-loop pause and resume for agent workflows** ([#3329](https://github.com/micro/go-micro/issues/3329)) — once agents are observable and memory-bounded, close the next operability gap for scheduled/looping work: durable pending-input states that let humans approve or provide context and then resume the same service/agent/workflow runtime.
 
 _Seeded by Claude Code from the roadmap + open issues; thereafter maintained by the
 architecture-review pass._

internal/harness/zero-to-hero-ci/README.md

@@ -8,8 +8,9 @@ scripted so CI can run it on every push without external services or model keys.
 
 1. **Run** — `micro run` remains available as the local development entry point.
 2. **Chat** — `micro chat` remains available as the interactive agent entry point.
-3. **Inspect** — `micro flow runs` remains available for durable workflow run
-   history inspection.
+3. **Inspect** — `micro inspect agent <name>` and `micro inspect flow <name>`
+   remain available as the local run-history inspection step, with `micro flow
+   runs` preserving durable workflow history inspection.
 
 After the CLI boundary smoke checks, the script runs the deterministic harnesses
 that boot real services, agents, workflows, store-backed run history, and A2A

internal/website/docs/guides/agent-harness.md

@@ -36,13 +36,46 @@ your stack — the harness *is* the stack.
 | Discovery & RPC | Registry + client; agents and services find and call each other | Shipped |
 | Interop | MCP (tools), A2A (agents), x402 (paid tools) | Shipped |
 | Resilience | Per-call timeout with context propagation; opt-in retry/backoff (`ModelRetry`) across the loop | Shipped |
-| Durable runs | Checkpoint and resume an agent run (flows already do) | In progress |
+| Durable runs | Checkpoint and resume an agent run with the same checkpoint backend flows use | Shipped |
 | Observability | `RunInfo` → OpenTelemetry spans for runs, model calls, tools, delegation, and failures; persisted run history | Shipped |
 | Streaming | `ai.Stream` through chat, agent, and A2A | In progress |
 
 The "in progress" rows are exactly the roadmap's [Now and Next](/docs/roadmap.html),
 and the work is happening in the open.
 
+## Durable agent runs
+
+Agents can persist their execution history to the same `Checkpoint` backend as
+flows. A checkpointed `Ask` records the run id, original prompt, model result,
+and completed tool calls. If the process restarts after a tool succeeds but
+before the model finishes, `AgentResume` continues the same run and returns the
+recorded tool result instead of re-running the side effect. If a run already
+completed, resume returns the persisted response without calling the model.
+
+```go
+agent := micro.NewAgent("conductor",
+    micro.AgentProvider("anthropic"),
+    micro.AgentWithCheckpoint(checkpoint),
+)
+
+resp, err := agent.Ask(ctx, "charge order 42 and send a receipt")
+if err != nil {
+    // On startup, or after a transient failure, discover unfinished work:
+    pending, _ := micro.AgentPending(ctx, agent)
+    for _, run := range pending {
+        _, _ = micro.AgentResume(ctx, agent, run.ID)
+    }
+}
+_ = resp
+```
+
+For human-in-the-loop runs that pause through the built-in `request_input` tool,
+resume with the operator's response:
+
+```go
+_, err := micro.AgentResumeInput(ctx, agent, runID, "Deploy to us-east-1")
+```
+
 ## Observing agent runs
 
 Pass an OpenTelemetry tracer provider when you construct an agent to turn the

internal/website/docs/guides/ai-provider-guide.md

@@ -39,12 +39,12 @@ 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 |
+| `atlascloud` | Yes | Yes | Yes | Yes |
 | `gemini` | Yes | No | No | No |
-| `groq` | Yes | No | No | No |
-| `mistral` | Yes | No | No | No |
+| `groq` | Yes | No | No | Yes |
+| `mistral` | Yes | No | No | Yes |
 | `openai` | Yes | Yes | No | Yes |
-| `together` | Yes | No | No | No |
+| `together` | Yes | No | No | Yes |
 
 ## Step 1: Implement the `ai.Model` Interface
 

internal/website/docs/roadmap.md

@@ -36,14 +36,13 @@ The priority is that what exists works everywhere, under real conditions.
 
 ## Next — agentic depth
 
-- **Durable agent loop.** Flows resume; the agent's own loop does not yet. Reuse `Checkpoint` so a long-running agent survives a restart and continues.
 - **Streaming.** Broaden provider-backed `ai.Stream` coverage and keep chat plus A2A `message/stream` working end to end for real chat and long-task UX.
 - **Agent observability.** Wire the new `RunInfo` into OpenTelemetry spans so a run — steps, tool calls, delegation — is traceable. This is also what anyone running it in production will need.
 
 ## Later
 
 - **Memory management** — summarization and retrieval (RAG) beyond a fixed buffer.
-- **Human-in-the-loop** — pause and resume mid-run (`input-required`), beyond the binary `ApproveTool` gate.
+- **Human-in-the-loop** — broaden pause/resume UX around `input-required` runs and approvals.
 - **A2A** — richer live-stream reconnection (`tasks/resubscribe`) and `input-required` handoffs.
 
 ## Developer experience (ongoing)

micro.go

@@ -23,12 +23,18 @@ type Service = service.Service
 // Agent is the interface for an AI agent that manages services.
 type Agent = agent.Agent
 
+// AgentResponse is what an agent returns from Ask or a resumed run.
+type AgentResponse = agent.Response
+
 // AgentOption configures an Agent.
 type AgentOption = agent.Option
 
 // Flow is an event-driven LLM orchestration unit.
 type Flow = flow.Flow
 
+// FlowRun is a checkpointed flow or agent run record.
+type FlowRun = flow.Run
+
 // FlowOption configures a Flow.
 type FlowOption = flow.Option
 
@@ -164,6 +170,26 @@ func AgentWrapTool(w ...ai.ToolWrapper) AgentOption {
 // tool calls, delegation, and failures.
 func AgentTraceProvider(tp trace.TracerProvider) AgentOption { return agent.TraceProvider(tp) }
 
+// AgentWithCheckpoint sets the durability backend for agent Ask runs.
+// It uses the same Checkpoint interface as flows so services, agents,
+// and workflows can share one execution history backend.
+func AgentWithCheckpoint(c Checkpoint) AgentOption { return agent.WithCheckpoint(c) }
+
+// AgentPending returns checkpointed agent runs that have not completed.
+// Use it at process startup to discover agent work that should be resumed.
+func AgentPending(ctx context.Context, a Agent) ([]FlowRun, error) { return agent.Pending(ctx, a) }
+
+// AgentResume resumes a checkpointed agent run by id. Completed runs return
+// the persisted response without calling the model or replaying tool calls.
+func AgentResume(ctx context.Context, a Agent, runID string) (*AgentResponse, error) {
+	return agent.Resume(ctx, a, runID)
+}
+
+// AgentResumeInput resumes a checkpointed agent run waiting for human input.
+func AgentResumeInput(ctx context.Context, a Agent, runID, input string) (*AgentResponse, error) {
+	return agent.ResumeInput(ctx, a, runID, input)
+}
+
 // NewFlow creates an event-driven LLM orchestration unit.
 //
 //	f := micro.NewFlow("onboard-user",