FlowDropNodeProcessor Attribute Plugin¶

Overview¶

The FlowDropNodeProcessor attribute is a PHP 8+ attribute that provides metadata for FlowDrop node processor plugins. It serves as the primary mechanism for defining node processors in the FlowDrop workflow system, providing LangflowComponent-equivalent metadata for visual workflow nodes.

Class Definition¶

#[\Attribute(\Attribute::TARGET_CLASS)]
final class FlowDropNodeProcessor extends Plugin

Location: src/Attribute/FlowDropNodeProcessor.php

Constructor Parameters¶

Parameter	Type	Required	Description
`$id`	`string`	✅	Unique plugin identifier
`$label`	`TranslatableMarkup`	✅	Human-readable label for the UI
`$type`	`string`	✅	Node type for frontend rendering
`$supportedTypes`	`array`	✅	All supported node types
`$category`	`string`	✅	Component category (e.g., "inputs", "models", "tools")
`$description`	`string`	❌	Component description (default: "")
`$version`	`string`	❌	Component version (default: "1.0.0")
`$tags`	`array`	❌	Component tags for categorization (default: [])

Unified Parameter System¶

FlowDrop uses a Unified Parameter System where:

Plugins define the data contract via getParameterSchema() (types, defaults, constraints)
Config entities control UI/workflow behavior (connectable, configurable, required)
Runtime resolves values from inputs, workflow config, and defaults

Key Concepts¶

getParameterSchema(): Returns a JSON Schema defining parameter types and defaults
Config Entity: Controls which parameters are input ports vs config fields
ParameterBagInterface: DTO containing resolved parameter values at runtime

Architecture¶

┌─────────────────┐     ┌──────────────────┐     ┌─────────────────┐
│  Plugin Schema  │     │  Entity Config   │     │ System Defaults │
│  (type, default,│  +  │  (connectable,   │  +  │  (connectable=  │
│   enum, etc.)   │     │   configurable,  │     │   FALSE, etc.)  │
└────────┬────────┘     │   required)      │     └────────┬────────┘
         │              └────────┬─────────┘              │
         │                       │                        │
         └───────────────────────┴────────────────────────┘
                                 │
                    ┌────────────▼────────────┐
                    │   ParameterResolver     │
                    │   (merges all sources)  │
                    └────────────┬────────────┘
                                 │
                    ┌────────────▼────────────┐
                    │    ParameterBag         │
                    │    (resolved values)    │
                    └─────────────────────────┘

Schema Definition Example¶

Plugins define standard JSON Schema properties only. UI behavior is controlled by the config entity.

public function getParameterSchema(): array {
  return [
    "type" => "object",
    "properties" => [
      "prompt" => [
        "type" => "string",
        "description" => "The input prompt text",
      ],
      "model" => [
        "type" => "string",
        "description" => "The AI model to use",
        "default" => "gpt-4",
        "enum" => ["gpt-4", "gpt-3.5-turbo", "claude-3"],
      ],
      "temperature" => [
        "type" => "number",
        "description" => "Sampling temperature",
        "default" => 0.7,
        "minimum" => 0,
        "maximum" => 2,
      ],
    ],
  ];
}

Entity Configuration¶

The config entity controls how parameters behave in the workflow UI:

# flowdrop_node_type.flowdrop_node_type.chat_model.yml
id: chat_model
label: 'Chat Model'
executor_plugin: chat_model
parameters:
  prompt:
    connectable: true    # Shows as input port
    configurable: true   # Shows in config form
    required: true       # Must have a value
  model:
    connectable: false   # Config only, no input port
    configurable: true
    required: false
  temperature:
    connectable: false
    configurable: true
    required: false

System Defaults¶

When entity config doesn't specify a flag:

Flag	Default	Meaning
`connectable`	`FALSE`	Parameter is NOT an input port
`configurable`	`FALSE`	Parameter does NOT appear in config
`required`	`FALSE`	Parameter is optional

Usage Examples¶

Basic Node Processor¶

<?php

namespace Drupal\my_module\Plugin\FlowDropNodeProcessor;

use Drupal\flowdrop\Attribute\FlowDropNodeProcessor;
use Drupal\Core\StringTranslation\TranslatableMarkup;
use Drupal\flowdrop\Plugin\FlowDropNodeProcessor\AbstractFlowDropNodeProcessor;
use Drupal\flowdrop\DTO\ParameterBagInterface;
use Drupal\Core\Logger\LoggerChannelInterface;
use Symfony\Component\DependencyInjection\ContainerInterface;

#[FlowDropNodeProcessor(
  id: "my_processor",
  label: new TranslatableMarkup("My Processor"),
  type: "default",
  supportedTypes: ["default"],
  category: "processing",
  description: "A custom node processor for data transformation",
  version: "1.0.0",
  tags: ["custom", "data", "transformation"]
)]
class MyProcessor extends AbstractFlowDropNodeProcessor {

  public function __construct(
    array $configuration,
    $plugin_id,
    $plugin_definition,
    protected LoggerChannelInterface $logger,
  ) {
    parent::__construct($configuration, $plugin_id, $plugin_definition);
  }

  public static function create(
    ContainerInterface $container,
    array $configuration,
    $plugin_id,
    $plugin_definition
  ): static {
    return new static(
      $configuration,
      $plugin_id,
      $plugin_definition,
      $container->get("logger.channel.flowdrop"),
    );
  }

  protected function getLogger(): LoggerChannelInterface {
    return $this->logger;
  }

  public function getParameterSchema(): array {
    return [
      "type" => "object",
      "properties" => [
        "data" => [
          "type" => "mixed",
          "description" => "Input data to transform",
        ],
        "format" => [
          "type" => "string",
          "description" => "Output format",
          "default" => "json",
          "enum" => ["json", "xml", "csv"],
        ],
      ],
    ];
  }

  public function process(ParameterBagInterface $params): array {
    $data = $params->get("data");
    $format = $params->getString("format", "json");

    // Transform data based on format...
    $transformed = $this->transform($data, $format);

    return ["result" => $transformed];
  }

  private function transform(mixed $data, string $format): mixed {
    // Implementation...
    return $data;
  }
}

AI Model Node Processor¶

#[FlowDropNodeProcessor(
  id: "chat_model",
  label: new TranslatableMarkup("Chat Model"),
  type: "default",
  supportedTypes: ["default"],
  category: "ai",
  description: "AI chat model integration for conversational AI",
  version: "1.0.0",
  tags: ["ai", "chat", "model", "conversation"]
)]
class ChatModel extends AbstractFlowDropNodeProcessor {

  public function getParameterSchema(): array {
    return [
      "type" => "object",
      "properties" => [
        "prompt" => [
          "type" => "string",
          "description" => "The input prompt",
        ],
        "system_message" => [
          "type" => "string",
          "description" => "System message for the AI",
          "default" => "",
        ],
        "model" => [
          "type" => "string",
          "description" => "Model identifier",
          "default" => "gpt-4",
          "enum" => ["gpt-4", "gpt-3.5-turbo"],
        ],
        "temperature" => [
          "type" => "number",
          "description" => "Sampling temperature",
          "default" => 0.7,
          "minimum" => 0,
          "maximum" => 2,
        ],
      ],
    ];
  }

  public function process(ParameterBagInterface $params): array {
    $prompt = $params->getString("prompt");
    $systemMessage = $params->getString("system_message", "");
    $model = $params->getString("model", "gpt-4");
    $temperature = $params->getFloat("temperature", 0.7);

    // Call AI model...
    $response = $this->callModel($prompt, $systemMessage, $model, $temperature);

    return ["response" => $response];
  }
}

Input Node Processor¶

#[FlowDropNodeProcessor(
  id: "text_input",
  label: new TranslatableMarkup("Text Input"),
  type: "default",
  supportedTypes: ["default"],
  category: "inputs",
  description: "Simple text input field for user data entry",
  version: "1.0.0",
  tags: ["input", "text", "user", "form"]
)]
class TextInput extends AbstractFlowDropNodeProcessor {

  public function getParameterSchema(): array {
    return [
      "type" => "object",
      "properties" => [
        "value" => [
          "type" => "string",
          "description" => "The input text value",
          "default" => "",
        ],
        "placeholder" => [
          "type" => "string",
          "description" => "Placeholder text",
          "default" => "Enter text...",
        ],
      ],
    ];
  }

  public function process(ParameterBagInterface $params): array {
    return ["text" => $params->getString("value", "")];
  }
}

ParameterBag Methods¶

The ParameterBagInterface provides typed accessors for retrieving parameter values:

Method	Return Type	Description
`get(string $key, mixed $default = null)`	`mixed`	Get any parameter value
`getString(string $key, string $default = "")`	`string`	Get string parameter
`getInt(string $key, int $default = 0)`	`int`	Get integer parameter
`getFloat(string $key, float $default = 0.0)`	`float`	Get float parameter
`getBool(string $key, bool $default = false)`	`bool`	Get boolean parameter
`getArray(string $key, array $default = [])`	`array`	Get array parameter
`has(string $key)`	`bool`	Check if parameter exists
`all()`	`array`	Get all parameters

Supported JSON Schema Properties¶

Plugins can use these standard JSON Schema properties:

Property	Type	Description
`type`	`string`	Data type: string, number, integer, boolean, array, object, mixed
`description`	`string`	Human-readable description
`default`	`mixed`	Default value
`enum`	`array`	Allowed values
`minimum`	`number`	Minimum numeric value
`maximum`	`number`	Maximum numeric value
`minLength`	`integer`	Minimum string length
`maxLength`	`integer`	Maximum string length
`pattern`	`string`	Regex pattern for strings
`format`	`string`	Format: email, uri, date, date-time, uuid, multiline, json, etc.
`readOnly`	`boolean`	Value cannot be edited but can be exposed (e.g., for branches)
`minItems`	`integer`	Minimum array items
`maxItems`	`integer`	Maximum array items
`uniqueItems`	`boolean`	Require unique array items

Category Guidelines¶

Use these standardized categories for consistent organization:

Core Categories¶

inputs - Data input components (TextInput, FileUpload, etc.)
outputs - Data output components (TextOutput, ChatOutput, etc.)
models - AI model components (ChatModel, OpenAiChat, etc.)
tools - Utility and tool components (HttpRequest, Webhook, etc.)
processing - Data processing components (DataOperations, Calculator, etc.)
logic - Logic and control flow components (Conditional, Loop, etc.)
ai - AI-specific components (embeddings, vector stores, etc.)
eca - ECA (Event-Condition-Action) components
helpers - Helper and utility components
memories - Memory and state management components
prompts - Prompt-related components
vector_store - Vector database components

Custom Categories¶

You can create custom categories for specialized components: - custom - Custom business logic components - integration - Third-party service integrations - analytics - Data analytics components

Tag Guidelines¶

Tags help with categorization and search functionality. Use descriptive, lowercase tags:

Common Tag Patterns¶

Functionality: ai, chat, data, file, http, webhook
Data Types: text, json, csv, image, audio
Operations: input, output, transform, filter, aggregate
Providers: openai, huggingface, anthropic, google
Use Cases: conversation, automation, analytics, reporting

Plugin Discovery¶

The attribute is discovered by the FlowDropNodeProcessorPluginManager using attribute-based discovery:

$this->discovery = new AttributeClassDiscovery(
  "Plugin/FlowDropNodeProcessor",
  $namespaces,
  FlowDropNodeProcessor::class
);

Discovery Process¶

Scan Directories: Searches Plugin/FlowDropNodeProcessor directories in all modules
Attribute Detection: Identifies classes with #[FlowDropNodeProcessor] attribute
Metadata Extraction: Extracts all attribute parameters as plugin metadata
Caching: Caches discovered plugins for performance

Integration with Workflow Editor¶

The attribute metadata is used by the FlowDrop workflow editor to:

Frontend Rendering¶

Component Library: Displays available components organized by category
Node Configuration: Provides configuration forms based on entity config
Input/Output Ports: Renders connection points for connectable parameters
Search & Filtering: Enables component discovery using $tags

API Integration¶

Node Metadata: Exposes component information via REST API
Configuration Validation: Validates node configurations against schemas
Execution Context: Provides metadata for workflow execution

Best Practices¶

1. Unique Plugin IDs¶

// ✅ Good - Descriptive and unique
id: "custom_data_transformer"

// ❌ Bad - Generic and may conflict
id: "processor"

2. Descriptive Labels¶

// ✅ Good - Clear and descriptive
label: new TranslatableMarkup("Customer Data Transformer")

// ❌ Bad - Too generic
label: new TranslatableMarkup("Processor")

3. Appropriate Categories¶

// ✅ Good - Specific category
category: "processing"

// ❌ Bad - Generic category
category: "other"

4. Meaningful Tags¶

// ✅ Good - Descriptive tags
tags: ["data", "transformation", "customer", "business-logic"]

// ❌ Bad - Too generic
tags: ["custom"]

5. Version Management¶

// ✅ Good - Semantic versioning
version: "1.2.0"

// ❌ Bad - No version tracking
version: "1.0.0" // Always default

6. Clear Parameter Schemas¶

// ✅ Good - Well-defined schema with descriptions and constraints
public function getParameterSchema(): array {
  return [
    "type" => "object",
    "properties" => [
      "input_text" => [
        "type" => "string",
        "description" => "Text to process",
        "minLength" => 1,
      ],
      "max_tokens" => [
        "type" => "integer",
        "description" => "Maximum output tokens",
        "default" => 1000,
        "minimum" => 1,
        "maximum" => 4096,
      ],
    ],
  ];
}

// ❌ Bad - Missing descriptions and constraints
public function getParameterSchema(): array {
  return [
    "type" => "object",
    "properties" => [
      "input_text" => ["type" => "string"],
    ],
  ];
}

Side Effects and the Confirmation Gate¶

A processor that changes anything outside the run — sends a request, writes an entity, posts a message, appends to a memory store — must implement the empty marker HasSideEffectsInterface. The marker feeds two things: the "Side effects" capability badge in the node type UI, and the confirmation gate's fail-safe fallback.

The gate pauses execution at the runtime chokepoint (after parameter resolution, before process()) and asks an operator to approve the resolved call through the ordinary HITL interrupt inbox. Requiring confirmation is a governance decision, not a processor property — something as ordinary as sending an email may need approval for purely business reasons nothing in the code can know — so every node type carries confirmation governance settings, edited in the node type form and stored as the confirmation mapping:

Setting	Values	Meaning
`policy`	unset	Derive from the plugin at gate time: `HasSideEffectsInterface` → ask (fail closed)
	`ask`	Instances pause for approval
	`skip`	They do not
`author_controls`	`waive`, `require`	What a workflow author's per-instance choice (the `requiresConfirmation` config key) may do
`dynamic_controls`	`require`	Whether a wireable runtime input may require an approval for one execution

Resolution order: an allowed dynamic require whose input is truthy, then an allowed author choice, then the policy, then the plugin derivation. Three consequences worth spelling out:

The 2.1.x "Always" lock is now governance, not engine law: it is policy: ask with the author waive control unchecked. The same safety posture, owned by the admin as a checkbox default.
Governance can revoke. The granted-control check happens at gate time against the current settings, so unchecking waive re-gates every instance that had waived. The editor only ever offers the granted controls — the served config select and the escalation port exist exactly when granted.
Dynamic values can only arm. There is no dynamic waive: upstream data (including model-filled tool arguments) can add an approval, never remove one. A falsy input simply does not participate.

The derived fallback is resolved at gate time from the plugin class, never baked into stored config — a plugin that adopts the marker later re-gates existing node types automatically.

Upgrade path (non-breaking from any 2.x): two order-independent post_updates run on drush updatedb. One stamps policy: skip on exactly the node types whose behavior the fail-closed derivation would have flipped (undecided AND side-effecting), so existing workflows keep running unattended; the other carries a stored 2.1.x tri-state over with its meaning intact (Always → ask with waive withheld, Never → skip) and sheds the legacy key. Instance-level boolean values stored by 2.1.x keep working — the gate reads them alongside the served select's require/waive strings. Fresh installs get the fail-closed default. Consent is bound to the exact call (args-hash) and consumed once: an agent loop re-issuing the same tool call re-asks on every iteration. A decline rides the node's normal error channel (confirmation_declined on the reserved error port for graph-scheduled nodes; a structured denial ToolResult back to the model for tool calls — never run-fatal).

The gate is a safety mechanism, not a flow-control primitive. If a workflow author wants first-class branching on a human decision, they should place an explicit ConfirmationNode — a gateway with true/false branches the canvas shows. The gate, by contrast, wraps an untouched processor from the runtime side, adds no ports, and only distinguishes "approved → run" from "declined → error channel".

Trust boundary: the governance settings are protected by administer flowdrop_node_type, and the instance choice by whoever can edit the workflow. The gate is exactly as strong as who can edit node types and workflows — it protects against unreviewed executions (an LLM deciding to call a side-effecting tool), not against a malicious workflow author.

See docs/development/specification-registry.md (RT-GATE) for the binding rules and tests.

Reporting Suppressed Work: the `<verb>_any` / `<verb>` Pair¶

A processor that can legitimately do nothing — dedupe a repeat, suppress an already-executed call, drop an empty write — has to say so on a port, because the graph is declarative and cannot work it out. Two nodes learned this the hard way and now share one shape; a third should copy it rather than invent a third spelling.

The trap is that the obvious ports all lie. A node that suppressed everything still returns a full output list (a suppressed item usually re-emits its stored result, so downstream stays correct), still reports ok/success (it reports the recorded outcome), and still reports a count (a snapshot of the store, which two identical passes report identically). Every one of those reads as "work happened". A loop gated on any of them re-enters forever, on exactly the pass the suppression exists to defend against.

The inverse port does not rescue it either. Where a node already reports what it skipped, the answer a gate needs is count(total) - count(skipped) > 0 — set arithmetic no boolean_gateway can do, so every consumer would leave the declarative graph for the same glue node.

So emit both halves, named for the verb the node performs:

Port	Type	Exposure	Purpose
`<verb>_any`	`boolean`	exposed by default	The gate. TRUE when at least one item was handled for the first time. Wire straight into a Boolean Gateway.
`<verb>`	`array` or `integer`	hidden by default	The detail behind it — the ids, or the count. Diagnostic; an author reveals it when debugging.

Shipped instances: tool_invoke emits executed_any / executed (RT-TOOL-10), conversation_buffer emits appended_any / appended (MEM-14).

Four rules the existing pair had to settle, worth settling the same way:

Report the delta, never a snapshot. appended counts appends and deliberately not the change in buffer size — a sliding window can evict as many messages as the call added, so a size comparison reads a real append as a no-op.
A recoverable error is still work. An unwired tool name invokes nothing, but produces a new result the model must re-plan against. A loop that did not re-enter would strand the run. Count it.
Say where the halves stop partitioning. <verb> can only name items that have an identity; anything dropped before the guard appears in neither list. State the exception in the port description — a reader who takes "complement" literally will build on it.
Declare the exposure on both sides. The hidden half needs x-exposed-by-default: FALSE in the plugin schema and exposed_by_default: false in the shipped node type. Exposure resolves from config, but the schema flag is what a regenerated node type is seeded from — set one without the other and the port drifts back to visible. Upgrading sites need a post_update hook for the hidden half only; the exposed half is already what a missing config entry resolves to.

Error Handling¶

Common Issues¶

Missing Required Parameters

// ❌ Error - Missing required parameters
#[FlowDropNodeProcessor(
  id: "my_processor",
  label: new TranslatableMarkup("My Processor")
  // Missing type, supportedTypes, and category
)]

Invalid Category

// ❌ Error - Invalid category
category: "invalid_category"

Duplicate Plugin IDs

// ❌ Error - Duplicate ID
id: "text_input" // Already exists in core

Validation¶

The plugin manager validates: - Required parameters are present - Plugin IDs are unique - Categories are valid - Attribute syntax is correct

Testing¶

Unit Testing¶

use Drupal\Tests\UnitTestCase;
use Drupal\flowdrop\DTO\ParameterBag;

class FlowDropNodeProcessorTest extends UnitTestCase {

  public function testPluginDiscovery() {
    $plugin_manager = $this->createMock(FlowDropNodeProcessorPluginManager::class);
    $plugins = $plugin_manager->getDefinitions();

    $this->assertArrayHasKey("my_processor", $plugins);
    $this->assertEquals("processing", $plugins["my_processor"]["category"]);
  }

  public function testNodeExecution() {
    $plugin = $this->createPlugin();
    $params = new ParameterBag([
      "data" => ["key" => "value"],
      "format" => "json",
    ]);

    $output = $plugin->execute($params);

    $this->assertEquals("success", $output->getStatus());
  }
}

Integration Testing¶

use Drupal\Tests\BrowserTestBase;

class FlowDropNodeProcessorIntegrationTest extends BrowserTestBase {

  public function testPluginInWorkflowEditor() {
    $this->drupalGet("/admin/structure/flowdrop-workflow/foo/flowdrop-editor");
    $this->assertSession()->elementExists("css", "[data-component-id=\"my_processor\"]");
  }
}