schema and config changes for vMCP: Composition - Output Aggregation

cmd/thv-operator/api/v1alpha1/virtualmcpcompositetooldefinition_types.go

@@ -46,6 +46,32 @@ type VirtualMCPCompositeToolDefinitionSpec struct {
 	// +kubebuilder:default=abort
 	// +optional
 	FailureMode string `json:"failureMode,omitempty"`
+
+	// OutputFormat is an optional Go template for constructing the workflow output.
+	// If specified, the template is evaluated with access to:
+	//   - .params.*          - Input parameters
+	//   - .steps.*.output    - Step outputs
+	//   - .steps.*.status    - Step status
+	//   - .workflow.*        - Workflow metadata (id, duration, timestamps)
+	//
+	// The template must produce valid JSON. If omitted, defaults to returning
+	// the last step's output (backward compatible behavior).
+	//
+	// Example:
+	//   outputFormat: |
+	//     {
+	//       "data": {
+	//         "logs": {{.steps.fetch_logs.output}},
+	//         "metrics": {{.steps.fetch_metrics.output}}
+	//       },
+	//       "metadata": {
+	//         "workflow_id": "{{.workflow.id}}",
+	//         "duration_ms": {{.workflow.duration_ms}}
+	//       }
+	//     }
+	//
+	// +optional
+	OutputFormat string `json:"outputFormat,omitempty"`
 }
 
 // VirtualMCPCompositeToolDefinitionStatus defines the observed state of VirtualMCPCompositeToolDefinition

deploy/charts/operator-crds/Chart.yaml

@@ -2,5 +2,5 @@ apiVersion: v2
 name: toolhive-operator-crds
 description: A Helm chart for installing the ToolHive Operator CRDs into Kubernetes.
 type: application
-version: 0.0.57
+version: 0.0.58
 appVersion: "0.0.1"

deploy/charts/operator-crds/README.md

@@ -1,7 +1,7 @@
 
 # ToolHive Operator CRDs Helm Chart
 
-![Version: 0.0.57](https://img.shields.io/badge/Version-0.0.57-informational?style=flat-square)
+![Version: 0.0.58](https://img.shields.io/badge/Version-0.0.58-informational?style=flat-square)
 ![Type: application](https://img.shields.io/badge/Type-application-informational?style=flat-square)
 
 A Helm chart for installing the ToolHive Operator CRDs into Kubernetes.

deploy/charts/operator-crds/crds/toolhive.stacklok.dev_virtualmcpcompositetooldefinitions.yaml

@@ -92,6 +92,31 @@ spec:
                 minLength: 1
                 pattern: ^[a-z0-9]([a-z0-9_-]*[a-z0-9])?$
                 type: string
+              outputFormat:
+                description: |-
+                  OutputFormat is an optional Go template for constructing the workflow output.
+                  If specified, the template is evaluated with access to:
+                    - .params.*          - Input parameters
+                    - .steps.*.output    - Step outputs
+                    - .steps.*.status    - Step status
+                    - .workflow.*        - Workflow metadata (id, duration, timestamps)
+
+                  The template must produce valid JSON. If omitted, defaults to returning
+                  the last step's output (backward compatible behavior).
+
+                  Example:
+                    outputFormat: |
+                      {
+                        "data": {
+                          "logs": {{.steps.fetch_logs.output}},
+                          "metrics": {{.steps.fetch_metrics.output}}
+                        },
+                        "metadata": {
+                          "workflow_id": "{{.workflow.id}}",
+                          "duration_ms": {{.workflow.duration_ms}}
+                        }
+                      }
+                type: string
               parameters:
                 additionalProperties:
                   description: ParameterSpec defines a parameter for a composite tool

pkg/vmcp/composer/composer.go

@@ -59,6 +59,17 @@ type WorkflowDefinition struct {
 	// Options: "abort" (default), "continue", "best_effort"
 	FailureMode string
 
+	// OutputFormat is an optional Go template for constructing the workflow output.
+	// If specified, the template is evaluated with access to:
+	//   - .params.*          - Input parameters
+	//   - .steps.*.output    - Step outputs
+	//   - .steps.*.status    - Step status
+	//   - .workflow.*        - Workflow metadata (id, duration, timestamps)
+	//
+	// The template must produce valid JSON. If omitted, defaults to returning
+	// the last step's output (backward compatible behavior).
+	OutputFormat string
+
 	// Metadata stores additional workflow information.
 	Metadata map[string]string
 }

pkg/vmcp/config/config.go

@@ -339,6 +339,30 @@ type CompositeToolConfig struct {
 
 	// Steps are the workflow steps to execute.
 	Steps []*WorkflowStepConfig `json:"steps"`
+
+	// OutputFormat is an optional template for constructing the workflow output.
+	// If specified, the template is evaluated with access to:
+	//   - .params.*          - Input parameters
+	//   - .steps.*.output    - Step outputs
+	//   - .steps.*.status    - Step status
+	//   - .workflow.*        - Workflow metadata (id, duration, timestamps)
+	//
+	// The template must produce valid JSON. If omitted, defaults to returning
+	// the last step's output (backward compatible behavior).
+	//
+	// Example:
+	//   output_format: |
+	//     {
+	//       "data": {
+	//         "logs": {{.steps.fetch_logs.output}},
+	//         "metrics": {{.steps.fetch_metrics.output}}
+	//       },
+	//       "metadata": {
+	//         "workflow_id": "{{.workflow.id}}",
+	//         "duration_ms": {{.workflow.duration_ms}}
+	//       }
+	//     }
+	OutputFormat string `json:"output_format,omitempty" yaml:"output_format,omitempty"`
 }
 
 // ParameterSchema defines a workflow parameter.

pkg/vmcp/server/workflow_converter.go

@@ -75,12 +75,13 @@ func ConvertConfigToWorkflowDefinitions(
 
 		// Create workflow definition
 		def := &composer.WorkflowDefinition{
-			Name:        ct.Name,
-			Description: ct.Description,
-			Parameters:  params,
-			Steps:       steps,
-			Timeout:     timeout,
-			Metadata:    make(map[string]string),
+			Name:         ct.Name,
+			Description:  ct.Description,
+			Parameters:   params,
+			Steps:        steps,
+			Timeout:      timeout,
+			OutputFormat: ct.OutputFormat,
+			Metadata:     make(map[string]string),
 		}
 
 		workflowDefs[ct.Name] = def

pkg/vmcp/server/workflow_converter_test.go

@@ -264,3 +264,89 @@ func TestConvertSteps_ComplexWorkflow(t *testing.T) {
 	assert.NotEmpty(t, result[2].Condition)
 	assert.Equal(t, []string{"confirm"}, result[2].DependsOn)
 }
+
+func TestConvertConfigToWorkflowDefinitions_WithOutputFormat(t *testing.T) {
+	t.Parallel()
+
+	tests := []struct {
+		name             string
+		input            []*config.CompositeToolConfig
+		wantOutputFormat string
+	}{
+		{
+			name: "workflow without output_format",
+			input: []*config.CompositeToolConfig{{
+				Name: "simple",
+				Steps: []*config.WorkflowStepConfig{
+					{ID: "s1", Type: "tool", Tool: "backend.tool"},
+				},
+			}},
+			wantOutputFormat: "",
+		},
+		{
+			name: "workflow with output_format",
+			input: []*config.CompositeToolConfig{{
+				Name: "aggregated",
+				Steps: []*config.WorkflowStepConfig{
+					{ID: "fetch_logs", Type: "tool", Tool: "splunk.fetch"},
+					{ID: "fetch_metrics", Type: "tool", Tool: "datadog.fetch"},
+				},
+				OutputFormat: `{
+					"logs": {{.steps.fetch_logs.output}},
+					"metrics": {{.steps.fetch_metrics.output}}
+				}`,
+			}},
+			wantOutputFormat: `{
+					"logs": {{.steps.fetch_logs.output}},
+					"metrics": {{.steps.fetch_metrics.output}}
+				}`,
+		},
+		{
+			name: "workflow with complex output_format",
+			input: []*config.CompositeToolConfig{{
+				Name: "investigation",
+				Steps: []*config.WorkflowStepConfig{
+					{ID: "fetch_data", Type: "tool", Tool: "backend.fetch"},
+					{ID: "analyze", Type: "tool", Tool: "backend.analyze"},
+				},
+				OutputFormat: `{
+					"data": {{.steps.fetch_data.output}},
+					"analysis": {{.steps.analyze.output}},
+					"metadata": {
+						"workflow_id": "{{.workflow.id}}",
+						"duration_ms": {{.workflow.duration_ms}}
+					}
+				}`,
+			}},
+			wantOutputFormat: `{
+					"data": {{.steps.fetch_data.output}},
+					"analysis": {{.steps.analyze.output}},
+					"metadata": {
+						"workflow_id": "{{.workflow.id}}",
+						"duration_ms": {{.workflow.duration_ms}}
+					}
+				}`,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			t.Parallel()
+
+			result, err := ConvertConfigToWorkflowDefinitions(tt.input)
+
+			require.NoError(t, err)
+			require.Len(t, result, 1)
+
+			// Get the first (and only) workflow definition
+			var workflowDef *composer.WorkflowDefinition
+			for _, def := range result {
+				workflowDef = def
+				break
+			}
+
+			require.NotNil(t, workflowDef)
+			assert.Equal(t, tt.wantOutputFormat, workflowDef.OutputFormat)
+		})
+	}
+}