Skip to content

fix(mcp): add missing pipeline step types and dynamic plugin type discovery #196

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
intel352 merged 5 commits into from
Feb 27, 2026
Merged

fix(mcp): add missing pipeline step types and dynamic plugin type discovery #196

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
67bbf39
Initial plan
Copilot Feb 27, 2026
5cd1c29
fix: add missing step types and dynamic plugin manifest loading to MC…
Copilot Feb 27, 2026
9d19688
fix(schema): correct config field names in new pipeline step schemas
Copilot Feb 27, 2026
b9d9c40
Merge branch 'main' into copilot/test-mcp-server-discoveries
intel352 Feb 27, 2026
42b6c99
fix(schema): correct remaining step schema fields to match implementa…
Copilot Feb 27, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
93 changes: 82 additions & 11 deletions mcp/server.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -15,11 +15,11 @@ import (
"strings"

"github.com/GoCodeAlone/workflow/config"
"golang.org/x/text/cases"
"golang.org/x/text/language"
"github.com/GoCodeAlone/workflow/schema"
"github.com/mark3labs/mcp-go/mcp"
"github.com/mark3labs/mcp-go/server"
"golang.org/x/text/cases"
"golang.org/x/text/language"
)

// Version is the MCP server version, set at build time.
Expand All @@ -34,7 +34,8 @@ type Server struct {

// NewServer creates a new MCP server with all workflow engine tools and
// resources registered. pluginDir is the directory where installed plugins
// reside (e.g., "data/plugins").
// reside (e.g., "data/plugins"). If set, the server will read plugin manifests
// from this directory and include plugin-provided types in all type listings.
func NewServer(pluginDir string) *Server {
s := &Server{
pluginDir: pluginDir,
Expand All @@ -51,6 +52,12 @@ func NewServer(pluginDir string) *Server {
"and manage plugins. Resources provide documentation and example configurations."),
)

// Load types from installed plugin manifests so that plugin-provided types
// appear in all type listings (list_module_types, list_step_types, etc.).
if pluginDir != "" {
s.loadInstalledPluginTypes(pluginDir)
}

s.registerTools()
s.registerResources()

Expand All @@ -72,7 +79,7 @@ func (s *Server) registerTools() {
// list_module_types
s.mcpServer.AddTool(
mcp.NewTool("list_module_types",
mcp.WithDescription("List all available workflow module types that can be used in the 'modules' section of a workflow YAML config. Returns both built-in and plugin-provided types."),
mcp.WithDescription("List all available workflow module types that can be used in the 'modules' section of a workflow YAML config. Returns built-in types plus types from installed plugins (loaded from plugin_dir at server startup)."),
mcp.WithReadOnlyHintAnnotation(true),
),
s.handleListModuleTypes,
Expand All @@ -81,7 +88,7 @@ func (s *Server) registerTools() {
// list_step_types
s.mcpServer.AddTool(
mcp.NewTool("list_step_types",
mcp.WithDescription("List all available pipeline step types that can be used in pipeline definitions. Steps are the building blocks of workflow pipelines."),
mcp.WithDescription("List all available pipeline step types that can be used in pipeline definitions. Returns built-in steps plus steps from installed plugins (loaded from plugin_dir at server startup)."),
mcp.WithReadOnlyHintAnnotation(true),
),
s.handleListStepTypes,
Expand All @@ -90,7 +97,7 @@ func (s *Server) registerTools() {
// list_trigger_types
s.mcpServer.AddTool(
mcp.NewTool("list_trigger_types",
mcp.WithDescription("List all available trigger types (e.g., http, schedule, event, eventbus) that can start workflow execution."),
mcp.WithDescription("List all available trigger types (e.g., http, schedule, event, eventbus) that can start workflow execution. Includes types from installed plugins."),
mcp.WithReadOnlyHintAnnotation(true),
),
s.handleListTriggerTypes,
Expand All @@ -99,7 +106,7 @@ func (s *Server) registerTools() {
// list_workflow_types
s.mcpServer.AddTool(
mcp.NewTool("list_workflow_types",
mcp.WithDescription("List all available workflow handler types (e.g., http, messaging, statemachine, scheduler, integration, event) that define how workflows process work."),
mcp.WithDescription("List all available workflow handler types (e.g., http, messaging, statemachine, scheduler, integration, event, pipeline) that define how workflows process work. Includes types from installed plugins."),
mcp.WithReadOnlyHintAnnotation(true),
),
s.handleListWorkflowTypes,
Expand Down Expand Up @@ -394,16 +401,32 @@ func (s *Server) handleListPlugins(_ context.Context, req mcp.CallToolRequest) (
}

type pluginInfo struct {
Name string `json:"name"`
Version string `json:"version"`
Name string `json:"name"`
Version string `json:"version"`
ModuleTypes []string `json:"module_types,omitempty"`
StepTypes []string `json:"step_types,omitempty"`
TriggerTypes []string `json:"trigger_types,omitempty"`
WorkflowTypes []string `json:"workflow_types,omitempty"`
}
var plugins []pluginInfo
for _, e := range entries {
if !e.IsDir() {
continue
}
ver := readPluginVersion(filepath.Join(dataDir, e.Name()))
plugins = append(plugins, pluginInfo{Name: e.Name(), Version: ver})
dir := filepath.Join(dataDir, e.Name())
ver := readPluginVersion(dir)
info := pluginInfo{Name: e.Name(), Version: ver}
// Enrich with type declarations from the plugin manifest.
if data, err := os.ReadFile(filepath.Join(dir, "plugin.json")); err == nil { //nolint:gosec // G304: path is within the trusted plugins directory
var m pluginManifestTypes
if json.Unmarshal(data, &m) == nil {
info.ModuleTypes = m.ModuleTypes
info.StepTypes = m.StepTypes
info.TriggerTypes = m.TriggerTypes
info.WorkflowTypes = m.WorkflowTypes
}
}
plugins = append(plugins, info)
}

result := map[string]any{
Expand Down Expand Up @@ -483,6 +506,54 @@ func marshalToolResult(v any) (*mcp.CallToolResult, error) {
return mcp.NewToolResultText(string(data)), nil
}

// pluginManifestTypes holds the type declarations from a plugin.json manifest file.
// This is a minimal subset of plugin.PluginManifest used to avoid a package dependency.
type pluginManifestTypes struct {
ModuleTypes []string `json:"moduleTypes"`
StepTypes []string `json:"stepTypes"`
TriggerTypes []string `json:"triggerTypes"`
WorkflowTypes []string `json:"workflowTypes"`
}

// loadInstalledPluginTypes scans pluginDir for subdirectories containing a
// plugin.json manifest, reads each manifest's type declarations, and registers
// them with the schema package so that they appear in all type listings.
// Unknown or malformed manifests are silently skipped.
func (s *Server) loadInstalledPluginTypes(pluginDir string) {
entries, err := os.ReadDir(pluginDir)
if err != nil {
return
}
for _, e := range entries {
if !e.IsDir() {
continue
}
manifestPath := filepath.Join(pluginDir, e.Name(), "plugin.json")
data, err := os.ReadFile(manifestPath) //nolint:gosec // G304: path is within the trusted plugins directory
if err != nil {
continue
}
var m pluginManifestTypes
if err := json.Unmarshal(data, &m); err != nil {
continue
}
for _, t := range m.ModuleTypes {
schema.RegisterModuleType(t)
}
for _, t := range m.StepTypes {
// Step types are also surfaced as module types in the MCP server view
// (they share the same registry and are identified by the "step." prefix).
schema.RegisterModuleType(t)
}
for _, t := range m.TriggerTypes {
schema.RegisterTriggerType(t)
}
for _, t := range m.WorkflowTypes {
schema.RegisterWorkflowType(t)
}
}
}

func readPluginVersion(dir string) string {
data, err := os.ReadFile(filepath.Join(dir, "plugin.json"))
if err != nil {
Expand Down
Loading
Loading