feat: support structured output in JSON Schema over A2A DataPart

docs/architecture/README.md

@@ -7,27 +7,27 @@ the agent processes.
 
 ## Resource ownership
 
-| Resource | Owner | Purpose |
-| --- | --- | --- |
-| `Harness` | Kubernetes (`kagent.dev/v1alpha3`) | Runtime implementation, workload, credentials, capacity, and admission policy |
-| `AgentTemplate` | Kubernetes (`kagent.dev/v1alpha3`) | Portable agent behavior: model, prompt, tools, skills, and plugins |
-| prepared revision | PostgreSQL and ate-api | Immutable compiled runtime input and its Substrate ActorTemplate |
-| `AgentInstance` | PostgreSQL, exposed by gRPC | Ephemeral compute identity and lifecycle |
-| A2A context, task, and events | PostgreSQL, exposed by A2A | Durable interaction and audit history |
-| checkpoint | PostgreSQL plus a Substrate snapshot tag | Immutable, named restart boundary |
-| Actor and durable directory | Substrate | Process lifecycle and private runtime state |
+| Resource                      | Owner                                    | Purpose                                                                       |
+| ----------------------------- | ---------------------------------------- | ----------------------------------------------------------------------------- |
+| `Harness`                     | Kubernetes (`kagent.dev/v1alpha3`)       | Runtime implementation, workload, credentials, capacity, and admission policy |
+| `AgentTemplate`               | Kubernetes (`kagent.dev/v1alpha3`)       | Portable agent behavior: model, prompt, tools, skills, and plugins            |
+| prepared revision             | PostgreSQL and ate-api                   | Immutable compiled runtime input and its Substrate ActorTemplate              |
+| `AgentInstance`               | PostgreSQL, exposed by gRPC              | Ephemeral compute identity and lifecycle                                      |
+| A2A context, task, and events | PostgreSQL, exposed by A2A               | Durable interaction and audit history                                         |
+| checkpoint                    | PostgreSQL plus a Substrate snapshot tag | Immutable, named restart boundary                                             |
+| Actor and durable directory   | Substrate                                | Process lifecycle and private runtime state                                   |
 
 `AgentInstance` is not a Kubernetes resource. A2A owns public interaction
 semantics; kagent does not maintain a parallel session or task API.
 
 ## Public surfaces
 
-| Surface | Role |
-| --- | --- |
-| Kubernetes API | Author Harnesses, AgentTemplates, models, prompts, and remote MCP servers |
-| gRPC / gRPC-Web | Manage AgentInstances, sharing, checkpoints, and control-plane reads |
-| A2A | Invoke agents and manage durable tasks and streams |
-| MCP | Discover, invoke, checkpoint, and fork AgentInstances through A2A semantics |
+| Surface         | Role                                                                        |
+| --------------- | --------------------------------------------------------------------------- |
+| Kubernetes API  | Author Harnesses, AgentTemplates, models, prompts, and remote MCP servers   |
+| gRPC / gRPC-Web | Manage AgentInstances, sharing, checkpoints, and control-plane reads        |
+| A2A             | Invoke agents and manage durable tasks and streams                          |
+| MCP             | Discover, invoke, checkpoint, and fork AgentInstances through A2A semantics |
 
 ## End-to-end flow
 
@@ -77,6 +77,7 @@ quiescence. It reaches Actors through the private runtime network.
 - [Human in the loop](human-in-the-loop.md)
 - [Prompt resolution](prompt-templates.md)
 - [Telemetry](telemetry.md)
+- [Structured output](structured-output.md)
 
 The documents describe implemented behavior. Deferred work, including full
 cross-AgentInstance delegation and Dedicated agents, belongs in the

docs/architecture/structured-output.md

@@ -0,0 +1,81 @@
+# Structured Output
+
+An `AgentTemplate` can declare a JSON Schema for its successful terminal
+response. The schema is either written inline in `spec.outputSchema` or stored
+as JSON in a same-namespace ConfigMap selected by `spec.outputSchemaFrom`. The
+two fields are mutually exclusive.
+
+```yaml
+apiVersion: kagent.dev/v1alpha3
+kind: AgentTemplate
+metadata:
+  name: data-extractor
+spec:
+  modelConfig:
+    name: default-model-config
+  outputSchema:
+    type: object
+    properties:
+      status:
+        type: string
+      payload:
+        type: object
+    required: [status, payload]
+    additionalProperties: false
+```
+
+The translator resolves ConfigMap references, validates the portable schema
+profile, and records a canonical schema and digest in the immutable prepared
+revision. Invalid schemas and unsupported Harnesses fail compilation. The
+kagent Harness supports structured output in its Go ADK runtime;
+Codex, Claude, and BYO Harnesses do not currently support it.
+
+The contract belongs to the public root agent. It is not inherited by local
+child agents or agent tools. If a child template is compiled and invoked as a
+root agent, its own schema applies to that independent revision.
+
+## A2A response contract
+
+The runtime suppresses partial root-answer fragments, validates the complete
+value, and publishes the successful result as one A2A `DataPart` with media
+type `application/json`. The part also carries the canonical schema digest in
+`kagent.dev/a2a/output-schema-sha256`. The generated Agent Card advertises
+`application/json` as its default output mode.
+
+Progress updates, tool events, approval requests, and input-required messages
+keep their existing representations. Consequently, a streaming client that
+only needs the answer reads the last content-bearing artifact before the Task
+reaches `TASK_STATE_COMPLETED`. The gateway persists and relays these events
+without interpreting the structured payload.
+
+Structured output has no text fallback. Provider refusal, incomplete output,
+invalid JSON, or schema validation failure fails the Task. Invalid model output
+is not published or included in errors because it may contain sensitive data.
+
+## Portable schema profile
+
+Schemas must have an object root. The portable profile supports primitive JSON
+types, nested objects and arrays, `properties`, `required`,
+`additionalProperties`, `items`, `enum`, `const`, `anyOf`, titles and
+descriptions, and non-recursive local references through `$defs`.
+
+External or recursive references, conditional schemas, tuple arrays, and other
+unsupported keywords are rejected during compilation rather than silently
+weakened for a provider. A provider or model can still reject structured output
+at runtime; in that case the Task fails.
+
+The serialized schema is limited to 64 KiB. Conversion to the Go ADK schema is
+also limited to 32 levels and 1,000 expanded schema nodes; these bounds include
+nodes expanded through local `$defs` references. Schemas that exceed any limit
+fail compilation.
+
+The semantic resolution code is in
+[`go/core/internal/translator/outputschema.go`](../../go/core/internal/translator/outputschema.go).
+Runtime validation and A2A conversion are owned by the Go kagent ADK executor.
+
+## References
+
+- [OpenAI Structured Outputs](https://platform.openai.com/docs/guides/structured-outputs)
+- [Gemini structured outputs](https://ai.google.dev/gemini-api/docs/structured-output)
+- [Claude structured outputs](https://platform.claude.com/docs/en/build-with-claude/structured-outputs)
+- [Amazon Bedrock structured outputs](https://docs.aws.amazon.com/bedrock/latest/userguide/structured-output.html)

go/adk/cmd/main.go

@@ -195,13 +195,18 @@ func main() {
 	}
 
 	stream := agentConfig.GetStream()
-	executor := a2a.NewKAgentExecutor(a2a.KAgentExecutorConfig{
+	executor, err := a2a.NewKAgentExecutor(a2a.KAgentExecutorConfig{
 		RunnerConfig:   runnerConfig,
 		SessionService: sessionService,
 		Stream:         stream,
 		AppName:        appName,
 		Logger:         logger,
+		Output:         agentConfig.Output,
 	})
+	if err != nil {
+		logger.Error("failed to create A2A executor", "error", err)
+		os.Exit(1)
+	}
 
 	// Build the agent card.
 	if agentCard == nil {

go/adk/pkg/a2a/executor.go

@@ -2,26 +2,31 @@ package a2a
 
 import (
 	"context"
+	"encoding/json"
 	"fmt"
 	"iter"
+	"slices"
 	"strings"
 	"time"
 
 	"log/slog"
 
 	a2atype "github.com/a2aproject/a2a-go/v2/a2a"
 	"github.com/a2aproject/a2a-go/v2/a2asrv"
+	"github.com/google/jsonschema-go/jsonschema"
 	"github.com/kagent-dev/kagent/go/adk/pkg/auth"
 	"github.com/kagent-dev/kagent/go/adk/pkg/models"
 	"github.com/kagent-dev/kagent/go/adk/pkg/telemetry"
 	apia2a "github.com/kagent-dev/kagent/go/api/a2a"
+	apiadk "github.com/kagent-dev/kagent/go/api/adk"
 	"github.com/kagent-dev/kagent/go/pkg/tracing"
 	"go.opentelemetry.io/otel/attribute"
 	"go.opentelemetry.io/otel/trace"
 	adkagent "google.golang.org/adk/v2/agent"
 	"google.golang.org/adk/v2/runner"
 	"google.golang.org/adk/v2/server/adka2a/v2"
 	adksession "google.golang.org/adk/v2/session"
+	"google.golang.org/genai"
 )
 
 const (
@@ -35,21 +40,33 @@ type KAgentExecutorConfig struct {
 	Stream         bool
 	AppName        string
 	Logger         *slog.Logger
+	Output         *apiadk.OutputConfig
 }
 
 // KAgentExecutor keeps kagent's request/session glue around the upstream ADK
 // A2A executor. Event conversion and artifact streaming are delegated to ADK.
 type KAgentExecutor struct {
-	builtin        a2asrv.AgentExecutor
-	sessionService adksession.Service
-	appName        string
-	logger         *slog.Logger
+	builtin                 a2asrv.AgentExecutor
+	sessionService          adksession.Service
+	appName                 string
+	logger                  *slog.Logger
+	structuredOutputEnabled bool
+}
+
+type structuredOutput struct {
+	schema *jsonschema.Resolved
+	sha256 string
 }
 
 var _ a2asrv.AgentExecutor = (*KAgentExecutor)(nil)
 
-// NewKAgentExecutor creates a KAgentExecutor from config.
-func NewKAgentExecutor(cfg KAgentExecutorConfig) *KAgentExecutor {
+// NewKAgentExecutor creates a KAgentExecutor from config. It returns an error if
+// the configured output schema cannot be prepared for runtime validation.
+func NewKAgentExecutor(cfg KAgentExecutorConfig) (*KAgentExecutor, error) {
+	output, err := resolveStructuredOutput(cfg.Output)
+	if err != nil {
+		return nil, err
+	}
 	var runConfig adkagent.RunConfig
 	if cfg.Stream {
 		runConfig.StreamingMode = adkagent.StreamingModeSSE
@@ -58,11 +75,18 @@ func NewKAgentExecutor(cfg KAgentExecutorConfig) *KAgentExecutor {
 	if cfg.SessionService != nil {
 		runnerConfig.SessionService = cfg.SessionService
 	}
+	if runnerConfig.Agent == nil {
+		return nil, fmt.Errorf("root agent is required")
+	}
+	rootName := runnerConfig.Agent.Name()
+	if rootName == "" {
+		return nil, fmt.Errorf("root agent name is required")
+	}
 	builtin := adka2a.NewExecutor(adka2a.ExecutorConfig{
 		RunnerConfig:       runnerConfig,
 		RunConfig:          runConfig,
 		A2APartConverter:   a2aPartConverter,
-		GenAIPartConverter: genAIPartConverter,
+		GenAIPartConverter: structuredOutputPartConverter(output, rootName),
 		AfterEventCallback: func(ctx adka2a.ExecutorContext, event *adksession.Event, processed *a2atype.TaskArtifactUpdateEvent) error {
 			if event.InvocationID != "" {
 				trace.SpanFromContext(ctx).SetAttributes(attribute.String("gcp.vertex.agent.invocation_id", event.InvocationID))
@@ -77,17 +101,105 @@ func NewKAgentExecutor(cfg KAgentExecutorConfig) *KAgentExecutor {
 				}
 				processed.Artifact.SetMeta(apia2a.TimelinePositionMetadataKey, position.UTC().Format(time.RFC3339Nano))
 			}
-			return nil
+			return transformStructuredOutput(output, rootName, event, processed)
 		},
 		OutputMode: adka2a.OutputArtifactPerEvent,
 	})
 
 	return &KAgentExecutor{
-		builtin:        builtin,
-		sessionService: runnerConfig.SessionService,
-		appName:        cfg.AppName,
-		logger:         cfg.Logger.With("component", "kagent-executor"),
+		builtin:                 builtin,
+		sessionService:          runnerConfig.SessionService,
+		appName:                 cfg.AppName,
+		logger:                  cfg.Logger.With("component", "kagent-executor"),
+		structuredOutputEnabled: output != nil,
+	}, nil
+}
+
+// structuredOutputPartConverter drops partial root output before upstream ADK
+// creates an A2A artifact. A structured response is not useful until it is a
+// complete JSON value, and publishing fragments could persist invalid JSON.
+// Events from tools and sub-agents retain the normal converter behavior.
+func structuredOutputPartConverter(output *structuredOutput, rootName string) adka2a.GenAIPartConverter {
+	return func(ctx context.Context, event *adksession.Event, part *genai.Part) (*a2atype.Part, error) {
+		if output != nil && event != nil && event.Author == rootName && event.Partial {
+			return nil, nil
+		}
+		return genAIPartConverter(ctx, event, part)
+	}
+}
+
+func transformStructuredOutput(output *structuredOutput, rootName string, event *adksession.Event, processed *a2atype.TaskArtifactUpdateEvent) error {
+	if output == nil || event.Author != rootName {
+		return nil
+	}
+	// ADK treats interruption and skip-summarization events as final responses so
+	// the current agent loop stops. They are not final structured results.
+	if !event.IsFinalResponse() || len(event.LongRunningToolIDs) > 0 || event.Actions.SkipSummarization {
+		return nil
+	}
+	switch event.FinishReason {
+	case "", genai.FinishReasonUnspecified, genai.FinishReasonStop:
+	default:
+		return fmt.Errorf("output_validation_failed: root agent did not complete structured output")
+	}
+	value, err := finalStructuredOutput(event)
+	if err != nil {
+		return err
+	}
+	if err := output.schema.Validate(value); err != nil {
+		return fmt.Errorf("output_validation_failed: root agent output does not conform to its schema")
 	}
+	if processed == nil || processed.Artifact == nil {
+		return fmt.Errorf("output_validation_failed: root agent produced no result artifact")
+	}
+	part := apia2a.NewStructuredOutputPart(value, output.sha256)
+	processed.Artifact.Parts = a2atype.ContentParts{part}
+	return nil
+}
+
+// finalStructuredOutput returns ADK's parsed output when available. Normal A2A
+// chat execution does not run the workflow-node wrapper that populates
+// Event.Output, so in that path the final model text is decoded here before it
+// is validated against the canonical JSON Schema.
+func finalStructuredOutput(event *adksession.Event) (any, error) {
+	if event.Output != nil {
+		return event.Output, nil
+	}
+	if event.Content == nil {
+		return nil, fmt.Errorf("output_validation_failed: root agent produced no structured value")
+	}
+	var text strings.Builder
+	for _, part := range event.Content.Parts {
+		if part == nil || part.Thought {
+			continue
+		}
+		text.WriteString(part.Text)
+	}
+	if strings.TrimSpace(text.String()) == "" {
+		return nil, fmt.Errorf("output_validation_failed: root agent produced no structured value")
+	}
+	var value any
+	if err := json.Unmarshal([]byte(text.String()), &value); err != nil {
+		return nil, fmt.Errorf("output_validation_failed: root agent output is not valid JSON")
+	}
+	return value, nil
+}
+
+// resolveStructuredOutput compiles the configured JSON Schema once so every
+// execution can validate its final value with the same immutable validator.
+func resolveStructuredOutput(config *apiadk.OutputConfig) (*structuredOutput, error) {
+	if config == nil {
+		return nil, nil
+	}
+	var schema jsonschema.Schema
+	if err := json.Unmarshal(config.JSONSchema, &schema); err != nil {
+		return nil, fmt.Errorf("decode output schema: %w", err)
+	}
+	resolved, err := schema.Resolve(nil)
+	if err != nil {
+		return nil, fmt.Errorf("resolve output schema: %w", err)
+	}
+	return &structuredOutput{schema: resolved, sha256: config.SHA256}, nil
 }
 
 // UserIDCallInterceptor returns an a2asrv.CallInterceptor that extracts the
@@ -187,7 +299,26 @@ func (e *KAgentExecutor) Execute(ctx context.Context, reqCtx *a2asrv.ExecutorCon
 			reqCtx.Message = resumeMessage
 		}
 
+		structuredResultEmitted := false
 		for event, err := range e.builtin.Execute(ctx, reqCtx) {
+			// Mark that a structured result has been emitted
+			if update, ok := event.(*a2atype.TaskArtifactUpdateEvent); ok && update.Artifact != nil &&
+				slices.ContainsFunc(update.Artifact.Parts, apia2a.IsStructuredOutputPart) {
+				structuredResultEmitted = true
+			}
+			// If the event is a task status update event and the status is completed,
+			// but no structured result has been emitted, mark the status as failed with a message
+			// indicating that the root agent produced no result artifact.
+			if update, ok := event.(*a2atype.TaskStatusUpdateEvent); ok &&
+				err == nil && e.structuredOutputEnabled && !structuredResultEmitted &&
+				update.Status.State == a2atype.TaskStateCompleted {
+				update.Status.State = a2atype.TaskStateFailed
+				update.Status.Message = a2atype.NewMessageForTask(
+					a2atype.MessageRoleAgent,
+					update,
+					a2atype.NewTextPart("output_validation_failed: root agent produced no result artifact"),
+				)
+			}
 			// If the event is a task status update event and the status is input required, build the HITL status message
 			if update, ok := event.(*a2atype.TaskStatusUpdateEvent); ok &&
 				update.Status.State == a2atype.TaskStateInputRequired && update.Status.Message != nil {