Built-in node catalog

Choose and configure an integration

1. Find the node below and check its availability and known limitations.
2. Provision your destination/source and grant the credentials only the access the operation needs. Network access from the execution environment is required.
3. Store credentials with secrets. A configuration value such as secret:postgres_connection is a reference, not the credential itself.
4. Install required client packages in the execution environment. dagy[nodes] includes the newer connector clients but does not install every dependency listed here. Use dependency packages for deployed workflows.
5. Supply configuration, connect the exact port IDs, and run against a small representative input. Inspect output counts, error details, and provider responses before using production data.

Saved connector records and cloud-account setup are separate from node configuration. A saved cloud connection does not automatically supply credentials to every built-in node. Read saved connectors and cloud account setup for their supported boundaries.

The API exposes the installed catalog through GET /nodes/registry; dagy nodes list --format json also retrieves it after authentication. Model names and client configuration defaults below reflect the product code, not a guarantee that an external provider still offers that model or accepts every client version.

Run a node locally

This example uses the working text parser and performs no network calls:

import asyncio
from dagy.nodes import ExecutionContext, node_registry

node_registry.discover_builtin()
node_registry.sync_from_subclasses()
node = node_registry.create_instance("document_parser", config={"format": "text"})
result = asyncio.run(node.run(ExecutionContext(upstream_results={"file_in": "Hello, Dagy"})))
assert result.success, result.error
print(result.outputs["text_out"])

Expected output is Hello, Dagy. For another node, replace its type, config, and input mapping using the entry below. Config examples contain illustrative source names and secret references; replace them with resources you control. They do not create provider accounts, resources, or secret values.

Required fields must be supplied when constructing a node directly, even where the configuration metadata also declares a default. Ports describe compatibility; validate the actual payload shape. Successful results use success=true and an outputs map keyed by port ID. Failed results expose success=false, error, and error_type.

Common execution considerations

• The base node timeout defaults to 300 seconds. _timeout_seconds can be configured from 1 to 3,600 seconds, subject to the chosen hosted runtime's limits. Set provider-specific network timeouts too.
• Node graphs pass values only from successful nodes and skip required inputs that receive no value. An error_out port on a failed node is not automatically routed to a downstream handler. Use deliberate successful routing for validation paths.
• A production label is catalog metadata, not a compatibility or service-availability guarantee. The schema validator, Pandas transform, SQL transform, deduplicator, and PDF/DOCX/HTML parser paths have confirmed execution issues in this version; use the alternatives described below.
• Missing dependencies produce import errors. Missing secret references, invalid config, source permission failures, rate limits, and provider response errors require different remedies; inspect error_type and the provider message. Retried writes and notifications may duplicate side effects.
• Unless stated otherwise, built-in ingestion operates on a bounded in-memory batch. It does not supply automatic pagination, change-data capture, or a persisted ingestion checkpoint.

Sources and data exchange

Azure Blob Ingest

Type: azure_blob_ingest. Catalog availability: production.

Read blobs from Azure Blob Storage containers.

Prerequisites and behavior: Supply a secret containing an Azure connection string, or account_url and an optional credential secret. Grant list/read permissions for the container. Returns a list of parsed blob values plus metadata; default maximum is 100 files. It does not automatically use a saved cloud-account connection.

Python dependency: azure-storage-blob

Ports

Configuration

Example configuration

{
  "container": "incoming",
  "prefix": "orders/",
  "connection_ref": "secret:azure_connection",
  "file_format": "json"
}

BigQuery Reader

Type: bigquery_reader. Catalog availability: production.

Query Google BigQuery.

Prerequisites and behavior: Uses application-default Google credentials in the execution environment. Grant query execution and dataset read access in the selected project. This node exposes no service-account secret field. Results are materialized as a list of records; constrain query size explicitly. Cloud account setup does not by itself configure these runtime credentials.

Python dependency: google-cloud-bigquery

Ports

Configuration

Example configuration

{
  "project": "your-project",
  "query": "SELECT order_id, amount FROM `your-project.commerce.orders` LIMIT 100"
}

Document Parser

Type: document_parser. Catalog availability: production.

Parse PDF, DOCX, HTML, Markdown, and plain text files to extract structured text content.

Prerequisites and behavior: Plain text and Markdown parsing work without optional document clients. PDF, DOCX, and HTML paths currently fail with TypeError before parsing; installing their dependencies does not fix that execution issue. The extract_images option does not currently produce extracted images. file_in expects content bytes or text, not a remote URL to download.

Python dependency: Text/Markdown need no extra client; PDF: pypdf, DOCX: python-docx, HTML: beautifulsoup4; see execution issue below

Ports

Configuration

Example configuration

{
  "format": "text"
}

GCS Ingest

Type: gcs_ingest. Catalog availability: production.

Read objects from Google Cloud Storage buckets.

Prerequisites and behavior: Use credentials_ref for a JSON service-account credential secret or rely on application-default credentials. Requires list/read access to the bucket. Returns a list of file values plus object metadata; it does not flatten multiple row files or maintain a checkpoint. Default maximum is 100 successfully parsed files.

Python dependency: google-cloud-storage; google-auth for a service-account secret

Ports

Configuration

Example configuration

{
  "bucket": "your-source-bucket",
  "prefix": "orders/",
  "file_format": "json",
  "credentials_ref": "secret:gcs_service_account"
}

HTTP Fetch

Type: http_fetch. Catalog availability: production.

Fetch data from HTTP endpoints.

Prerequisites and behavior: Returns response_out as parsed JSON, text, or bytes, plus response headers. HTTP status 400 or greater returns a failed result with an HTTP-specific error type. Network requests have a configurable timeout, default 30 seconds. Headers are literal configuration values; this node has no dedicated authentication-secret field. Avoid storing tokens directly in saved graph configuration.

Python dependency: aiohttp

Ports

Configuration

Example configuration

{
  "url": "https://your-service.example.com/orders",
  "method": "GET",
  "timeout": 30
}

Kafka Consumer

Type: kafka_consumer. Catalog availability: production.

Consume from Kafka topics.

Prerequisites and behavior: Consumes JSON message values with earliest offset reset and automatic offset commits. Returns records containing key, value, partition, offset, and timestamp. Default maximum is 100 messages. The node can wait for more messages until the node timeout; there is no implemented ten-second idle deadline. SASL/TLS options are not exposed. It is a bounded consumer step, not a managed streaming service.

Python dependency: aiokafka

Ports

Configuration

Example configuration

{
  "brokers": "broker.example.com:9092",
  "topic": "orders",
  "group_id": "dagy-orders",
  "max_messages": 100
}

MongoDB Reader

Type: mongodb_reader. Catalog availability: production.

Query MongoDB collections.

Prerequisites and behavior: The connection secret is a MongoDB URI. Use a user with read access to the database and collection. filter and projection accept JSON objects. Default limit is 1,000; this is a bounded query, not a change-stream subscription.

Python dependency: pymongo

Ports

Configuration

Example configuration

{
  "connection_ref": "secret:mongodb_connection",
  "database": "commerce",
  "collection": "orders",
  "filter": {
    "status": "paid"
  },
  "limit": 1000
}

MySQL Reader

Type: mysql_reader. Catalog availability: production.

Query MySQL / MariaDB databases.

Prerequisites and behavior: Use a database user allowed to execute the query and a password secret. The params_in port supplies query parameters. fetch_size limits returned rows, not an automatically paginated dataset. Configure source networking and TLS requirements; ssl defaults to false.

Python dependency: pymysql

Ports

Configuration

Example configuration

{
  "host": "mysql.example.com",
  "port": 3306,
  "database": "commerce",
  "user": "reader",
  "password_ref": "secret:mysql_password",
  "query": "SELECT order_id, amount FROM orders LIMIT 100",
  "fetch_size": 100
}

MySQL Writer

Type: mysql_writer. Catalog availability: production.

Insert / upsert rows into a MySQL table.

Prerequisites and behavior: Provision the destination table first and grant only the required write privileges. rows_in accepts records; mode supports insert/upsert behavior. The result reports rows written, table, and mode. Choose a stable unique key for idempotent upserts; an insert retry can duplicate records.

Python dependency: pymysql

Ports

Configuration

Example configuration

{
  "host": "mysql.example.com",
  "port": 3306,
  "database": "commerce",
  "user": "writer",
  "password_ref": "secret:mysql_password",
  "table": "orders_export",
  "mode": "insert",
  "batch_size": 500
}

Postgres Reader

Type: postgres_reader. Catalog availability: production.

Query PostgreSQL databases.

Prerequisites and behavior: The secret value is a PostgreSQL connection string. Optional params_in is a dictionary; its values are bound in insertion order to positional query parameters. fetch_size caps returned rows after the query is fetched; it is not a paging cursor. Use SQL limits and predicates for large sources. No automatic change-data capture or incremental checkpoint is provided.

Python dependency: asyncpg

Ports

Configuration

Example configuration

{
  "connection_ref": "secret:postgres_connection",
  "query": "SELECT order_id, amount FROM orders LIMIT 100",
  "fetch_size": 100
}

Redshift Reader

Type: redshift_reader. Catalog availability: production.

Query Amazon Redshift data warehouse.

Prerequisites and behavior: Requires database access to the cluster host and a password secret. params_in supplies query parameters and fetch_size limits rows. Network reachability and source permissions are prerequisites; provisioning a warehouse is outside this node.

Python dependency: redshift_connector

Ports

Configuration

Example configuration

{
  "host": "warehouse.example.com",
  "port": 5439,
  "database": "analytics",
  "user": "reader",
  "password_ref": "secret:redshift_password",
  "query": "SELECT order_id, amount FROM orders LIMIT 100",
  "fetch_size": 100
}

S3 Ingest

Type: s3_ingest. Catalog availability: production.

Read files from Amazon S3 buckets.

Prerequisites and behavior: Uses the execution environment’s standard AWS credential chain, with permission to list the bucket and read objects. data_out is a list of parsed file values, so CSV files produce a list of row lists. Default maximum is 100 successfully parsed files. Invalid JSON/text may be skipped with a warning. If Parquet support is missing, this reader can return raw bytes instead of parsed rows. It does not track an ingestion checkpoint.

Python dependency: boto3; pyarrow for Parquet

Ports

Configuration

Example configuration

{
  "bucket": "your-source-bucket",
  "prefix": "orders/",
  "file_format": "json",
  "aws_region": "us-east-1",
  "max_files": 100
}

SFTP Ingest

Type: sftp_ingest. Catalog availability: production.

Download files from an SFTP server.

Prerequisites and behavior: Requires password authentication or an RSA PEM private-key secret and source read access. Reads a file or files in a directory; default maximum is 50. The current connector does not expose or perform host-key verification against known hosts. Use a separately verified SFTP client in a Python task if host identity verification is required. It does not maintain a download checkpoint.

Python dependency: paramiko

Ports

Configuration

Example configuration

{
  "host": "sftp.example.com",
  "port": 22,
  "user": "reader",
  "password_ref": "secret:sftp_password",
  "remote_path": "/incoming/",
  "file_format": "json",
  "max_files": 50
}

Snowflake Reader

Type: snowflake_reader. Catalog availability: production.

Query Snowflake data warehouse.

Prerequisites and behavior: Requires a Snowflake account identifier, user, password secret, and permissions on the selected warehouse/database/schema. Optional role selects the source role. fetch_size limits rows; use query predicates for incremental processing. Authentication exposed here is password-based, not a general Snowflake authentication adapter.

Python dependency: snowflake-connector-python

Ports

Configuration

Example configuration

{
  "account": "your-account",
  "user": "reader",
  "password_ref": "secret:snowflake_password",
  "warehouse": "COMPUTE_WH",
  "database": "ANALYTICS",
  "schema": "PUBLIC",
  "query": "SELECT order_id, amount FROM orders LIMIT 100",
  "fetch_size": 100
}

Transformation and data quality

Aggregator

Type: aggregator. Catalog availability: production.

Group records by key fields and compute aggregate functions (sum, avg, count, min, max, first, last, collect).

Prerequisites and behavior: Groups records by comma-separated fields and applies the declared aggregations. Each aggregation identifies a field, function, and optional alias. Returns aggregated records. Review behavior for missing or nonnumeric values before using financial totals.

Python dependency: No additional provider client beyond the base SDK.

Ports

Configuration

Example configuration

{
  "group_by": "status",
  "aggregations": [
    {
      "field": "amount",
      "function": "sum",
      "alias": "total"
    }
  ]
}

Column Mapper

Type: column_mapper. Catalog availability: production.

Rename, select, reorder, and add computed columns to tabular data.

Prerequisites and behavior: Renames and optionally transforms fields according to a list of mappings. drop_unmapped controls whether other columns are retained. Verify target-name collisions and default values with sample records before writing the result to another system.

Python dependency: No additional provider client beyond the base SDK.

Ports

Configuration

Example configuration

{
  "mappings": [
    {
      "source": "id",
      "target": "order_id"
    }
  ],
  "drop_unmapped": false
}

Data Filter

Type: data_filter. Catalog availability: production.

Filter records from a dataset based on configurable conditions.

Prerequisites and behavior: Expression mode evaluates a predicate for each record and splits matched/unmatched rows. Field-match mode is exposed, but the base validator still requires a nonempty condition; supply one even when using field matching. Expressions are not a sandbox for hostile code.

Python dependency: No additional provider client beyond the base SDK.

Ports

Configuration

Example configuration

{
  "mode": "expression",
  "condition": "record[\"amount\"] > 0"
}

Data Join

Type: data_join. Catalog availability: production.

Join two datasets on matching keys, similar to SQL JOIN.

Prerequisites and behavior: Joins two record lists using the configured keys. Select inner/left/right/outer and a conflict-resolution policy as exposed in the schema. Inputs are materialized in memory; use bounded datasets and check duplicate-key behavior for your data.

Python dependency: No additional provider client beyond the base SDK.

Ports

Configuration

Example configuration

{
  "left_key": "customer_id",
  "right_key": "id",
  "join_type": "left",
  "conflict_resolution": "prefer_left"
}

Data Normalizer

Type: data_normalizer. Catalog availability: production.

Clean and normalize data fields.

Prerequisites and behavior: Applies an ordered list of normalization rules to records. Validate the result on representative values; normalization can change types or leave values unchanged when a rule cannot be applied. Rules and input rows are in-memory JSON-compatible data.

Python dependency: No additional provider client beyond the base SDK.

Ports

Configuration

Example configuration

{
  "rules": [
    {
      "field": "email",
      "action": "lowercase"
    }
  ]
}

Data Quality

Type: data_quality. Catalog availability: production.

Run declarative data-quality expectations (row counts, null rates, uniqueness, allowed values, regex, freshness, distribution) against input rows.

Prerequisites and behavior: Evaluates SDK expectation specs. With on_failure="route", the node succeeds and emits the entire payload on fail_out for blocking failures, or pass_out on success; report_out contains detailed checks. This does not isolate individual invalid rows or store quarantine data. With on_failure="error", the node fails and failed outputs are not forwarded by the current node runtime.

Python dependency: No additional provider client beyond the base SDK.

Ports

Configuration

Example configuration

{
  "checks": [
    {
      "kind": "row_count",
      "min": 1
    },
    {
      "kind": "not_null",
      "columns": [
        "order_id"
      ]
    }
  ],
  "on_failure": "route"
}

Deduplicator

Type: deduplicator. Catalog availability: production.

Remove duplicate records.

Prerequisites and behavior: Current execution issue: the node fails with AttributeError before producing its result. Use a Python task that deduplicates by stable keys until corrected. Its exposed contract splits unique and duplicate records using comma-separated key columns. Choose keep_first, keep_last, or merge as available in the schema. This deduplicates the current input batch only; it is not a persistent cross-run idempotency store.

Python dependency: No additional provider client beyond the base SDK.

Ports

Configuration

Example configuration

{
  "key_columns": "order_id",
  "strategy": "keep_first"
}

Pandas Transform

Type: pandas_transform. Catalog availability: production.

Apply pandas transformations.

Prerequisites and behavior: Current execution issue: this node returns TypeError before evaluating the expression. Installing pandas alone does not resolve it. Use a Python task with an explicitly installed pandas dependency. Do not accept untrusted expressions as a security boundary.

Python dependency: pandas; currently affected by the execution issue below

Ports

Configuration

Example configuration

{
  "expression": "df.groupby(\"status\", as_index=False)[\"amount\"].sum()"
}

Schema Validator

Type: schema_validator. Catalog availability: production.

Validate data against JSON Schema.

Prerequisites and behavior: Current execution issue: this node fails with TypeError before validation because its optional-dependency call is incompatible with the helper. Installing jsonschema does not resolve that issue. Use the SDK data-quality/contract functions or a Python task for validation until the runtime is corrected. The configuration and ports below describe its exposed contract.

Python dependency: jsonschema; currently affected by the execution issue below

Ports

Configuration

Example configuration

{
  "schema": {
    "type": "object",
    "required": [
      "order_id"
    ],
    "properties": {
      "order_id": {
        "type": "string"
      }
    }
  },
  "strict": true
}

SQL Transform

Type: sql_transform. Catalog availability: production.

Transform with SQL queries.

Prerequisites and behavior: Current execution issue: this node returns TypeError before executing the SQL transformation. Installing duckdb alone does not resolve it. Use a Python task that queries your input through a controlled client until corrected.

Python dependency: duckdb; currently affected by the execution issue below

Ports

Configuration

Example configuration

{
  "query": "SELECT * FROM input"
}

Text Chunker

Type: text_chunker. Catalog availability: production.

Split text into overlapping chunks optimized for embedding and retrieval.

Prerequisites and behavior: Splits text with fixed-size, sentence, paragraph, or recursive strategies. Sizes are character-oriented rather than provider token counts. Returns chunks plus metadata; ensure overlap is smaller than the chunk size and measure resulting prompt lengths for your model.

Python dependency: No additional provider client beyond the base SDK.

Ports

Configuration

Example configuration

{
  "strategy": "fixed_size",
  "chunk_size": 512,
  "chunk_overlap": 50
}

Language models and prompts

Claude Summary

Type: claude_summary. Catalog availability: production.

Summarize with Claude.

Prerequisites and behavior: Requires an Anthropic API-key secret and access to the configured model. The output is summary text. Keep input within provider context limits and review outputs before customer-facing use. A model listed in the form can be unavailable to a particular provider account.

Python dependency: anthropic

Ports

Configuration

Example configuration

{
  "model": "claude-sonnet-4-5-20250929",
  "api_key_ref": "secret:anthropic_key",
  "max_tokens": 1024
}

GPT-4 Classify

Type: gpt4_classify. Catalog availability: production.

Classify text with GPT-4.

Prerequisites and behavior: Requires a prompt, allowed categories, model access, and an OpenAI API-key secret. Review classification_out for the assigned category/result. Categories may be configured as comma-separated text. Model output remains probabilistic; validate critical downstream decisions independently.

Python dependency: openai

Ports

Configuration

Example configuration

{
  "model": "gpt-4o",
  "system_prompt": "Classify the customer request into one category.",
  "categories": "billing,technical,other",
  "api_key_ref": "secret:openai_key",
  "temperature": 0
}

LLM Completion

Type: llm_completion. Catalog availability: production.

Generic multi-provider LLM completion node supporting OpenAI, Anthropic, and custom OpenAI-compatible endpoints.

Prerequisites and behavior: Supports OpenAI, Anthropic, and Ollama-compatible configuration. OpenAI-compatible requests can use base_url. The API-key reference is required and resolved even for a local endpoint; provide an appropriate configured secret. completion_out is generated text and usage_out reports provider usage. Provider context, account, and rate limits apply.

Python dependency: openai for OpenAI-compatible providers; anthropic for Anthropic

Ports

Configuration

Example configuration

{
  "provider": "openai",
  "model": "gpt-4o",
  "api_key_ref": "secret:openai_key",
  "temperature": 0.2,
  "max_tokens": 512
}

LLM Router

Type: llm_router. Catalog availability: production.

Select model metadata based on cost or quality preferences.

Prerequisites and behavior: Selects a model based on the supplied model metadata and chosen cost/quality strategy. It passes the input through response_out and emits routing_info_out; it does not invoke a language model itself. Connect an actual completion task/node after making the selection.

Python dependency: No additional provider client beyond the base SDK.

Ports

Configuration

Example configuration

{
  "strategy": "cost",
  "models": [
    {
      "name": "model-a",
      "cost_per_1k": 0.001,
      "quality_score": 80
    },
    {
      "name": "model-b",
      "cost_per_1k": 0.002,
      "quality_score": 90
    }
  ]
}

OpenAI Embeddings

Type: openai_embeddings. Catalog availability: production.

Generate embeddings via OpenAI.

Prerequisites and behavior: Requires an OpenAI API-key secret and a model available to that account. Accepts text and returns embedding vectors; default batch size is 100. Match embedding dimensions to the destination index, add stable record IDs before writing vectors, and account for provider token/rate limits. Model names in the schema are product defaults, not a promise of provider availability.

Python dependency: openai

Ports

Configuration

Example configuration

{
  "model": "text-embedding-3-small",
  "api_key_ref": "secret:openai_key",
  "batch_size": 100
}

Prompt Template

Type: prompt_template. Catalog availability: production.

Compose prompts from templates and dynamic variables.

Prerequisites and behavior: Uses {{variable}} placeholders and optional retrieved context. Unresolved placeholders may remain visible; inspect the rendered prompt before sending it. The context length cap is character-based and does not replace model token budgeting.

Python dependency: No additional provider client beyond the base SDK.

Ports

Configuration

Example configuration

{
  "template": "Answer {{question}} using this context:\n{{context}}",
  "context_separator": "\n---\n",
  "max_context_length": 4000
}

Structured Output

Type: structured_output. Catalog availability: production.

Extract structured JSON data from unstructured text using an LLM.

Prerequisites and behavior: Chooses the provider from the configured model and extracts a JSON object matching output_schema; provider access and credentials are required. structured_out contains parsed data and raw_response_out retains the model response. Strict validation requires jsonschema. Validate sensitive actions independently rather than trusting generated structured data.

Python dependency: openai or anthropic; jsonschema for validation

Ports

Configuration

Example configuration

{
  "output_schema": {
    "type": "object",
    "required": [
      "topic"
    ],
    "properties": {
      "topic": {
        "type": "string"
      }
    }
  },
  "model": "gpt-4o",
  "api_key_ref": "secret:openai_key",
  "strict_validation": true
}

Vector storage and search

Chroma Store

Type: chroma_store. Catalog availability: production.

Store embeddings and documents in ChromaDB.

Prerequisites and behavior: Accepts records with id, optional embedding/values, document, and metadata. Creates or gets the named collection and adds records. Host must be hostname:port. The current implementation builds authentication headers but does not pass them to the client, so api_key_ref does not authenticate writes. Use your own authenticated client in a task when needed.

Python dependency: chromadb

Ports

Configuration

Example configuration

{
  "collection": "support_articles",
  "host": "localhost:8000"
}

OpenSearch Write

Type: opensearch_write. Catalog availability: production.

Index vector documents into an OpenSearch k-NN index.

Prerequisites and behavior: Provision an index with the intended vector mapping. Configure hosts and, where needed, username/password secret. The result includes indexed count and bulk errors; inspect both rather than assuming every record was accepted. Source signing mechanisms outside the exposed username/password options require a custom client.

Python dependency: opensearch-py

Ports

Configuration

Example configuration

{
  "hosts": "https://search.example.com:9200",
  "index": "articles",
  "vector_field": "embedding",
  "use_ssl": true,
  "batch_size": 500
}

pgvector Write

Type: pgvector_write. Catalog availability: production.

Insert vector embeddings into a Postgres/pgvector table.

Prerequisites and behavior: The secret is a PostgreSQL connection string. Provision the pgvector extension, destination table, vector column/dimensions, and a unique conflict key before writing. Configured table/column names refer to your database. Input records carry ID, vector, and metadata. The result reports rows_written; this node does not create the table or implement vector search.

Python dependency: psycopg[binary] or psycopg2

Ports

Configuration

Example configuration

{
  "connection_ref": "secret:postgres_connection",
  "table": "article_vectors",
  "id_column": "id",
  "vector_column": "embedding",
  "metadata_column": "metadata",
  "upsert": true
}

Pinecone Write

Type: pinecone_write. Catalog availability: production.

Upsert vector embeddings to Pinecone.

Prerequisites and behavior: Create an index with dimensions matching your embeddings and grant the key write access. Input records use id, values, and optional metadata. Invalid records can be skipped with warnings; inspect upserted_count against your input count. Writes use batches of 100. The implementation expects the Pinecone client class available in newer client APIs.

Python dependency: pinecone client exposing the Pinecone class

Ports

Configuration

Example configuration

{
  "index": "articles",
  "namespace": "support",
  "api_key_ref": "secret:pinecone_key"
}

Qdrant Write

Type: qdrant_write. Catalog availability: production.

Upsert vector embeddings to a Qdrant collection.

Prerequisites and behavior: Provision a collection with matching vector dimensions. Input vector records contain id, values or vector, and optional metadata; use a Qdrant-supported point ID. Optional API-key secret authenticates the client. Returns collection and upserted count. Inspect skipped malformed inputs and batch failures.

Python dependency: qdrant-client

Ports

Configuration

Example configuration

{
  "url": "https://qdrant.example.com:6333",
  "collection": "articles",
  "api_key_ref": "secret:qdrant_key",
  "batch_size": 100
}

Vector Search

Type: vector_search. Catalog availability: production.

Perform similarity search against vector databases (Pinecone, Weaviate, Chroma) to retrieve relevant documents or records.

Prerequisites and behavior: The implemented providers are Pinecone, Weaviate, and Chroma. Input must be an embedding vector or a dictionary with embedding/vector; raw search text must first be embedded. Qdrant, pgvector, and OpenSearch write nodes do not imply search support here. Returns matches and score metadata; score meanings and filtering depend on the provider.

Python dependency: Selected provider client: pinecone, weaviate-client, or chromadb

Ports

Configuration

Example configuration

{
  "provider": "pinecone",
  "index": "articles",
  "api_key_ref": "secret:pinecone_key",
  "top_k": 5,
  "include_metadata": true
}

Weaviate Insert

Type: weaviate_insert. Catalog availability: production.

Insert objects into Weaviate.

Prerequisites and behavior: Input is a list of object dictionaries; create the target class/collection compatible with this node’s API. This implementation uses the legacy Weaviate Client and batch API, so verify the installed client version. Optional API-key resolution failure can continue without authentication; verify actual authorized access before production use.

Python dependency: weaviate-client with the legacy Client API used by this node

Ports

Configuration

Example configuration

{
  "class_name": "Article",
  "url": "https://weaviate.example.com",
  "api_key_ref": "secret:weaviate_key"
}

Notifications and destinations

Email Send

Type: email_send. Catalog availability: production.

Send email notifications via SMTP.

Prerequisites and behavior: The SMTP credential secret is JSON containing username and password. Configure the server, port, recipient, subject, and template; default port is 587. The node sends an email, so verify recipients with a test mailbox. Delivery acceptance does not guarantee arrival in an inbox.

Python dependency: aiosmtplib

Ports

Configuration

Example configuration

{
  "to": "reviewer@example.com",
  "subject": "Order export ready",
  "body_template": "Exported {{count}} orders",
  "smtp_host": "smtp.example.com",
  "smtp_port": 587,
  "smtp_credential_ref": "secret:smtp_credentials"
}

S3 Write

Type: s3_write. Catalog availability: production.

Write data to Amazon S3 (JSON, CSV, Parquet).

Prerequisites and behavior: Requires the execution environment’s standard AWS credentials and permission to write the destination. The exposed aws_credential_ref field is currently not used to create the client. Keys are the configured prefix plus a timestamp and extension; include a trailing slash in a directory-like prefix. The result contains s3_uri, bucket, key, bytes_written, and format. Repeating a write can create another object.

Python dependency: boto3; pyarrow for Parquet

Ports

Configuration

Example configuration

{
  "bucket": "your-output-bucket",
  "prefix": "orders/",
  "file_format": "json"
}

Slack Alert

Type: slack_alert. Catalog availability: production.

Send Slack messages to a channel via webhook.

Prerequisites and behavior: The secret value is a Slack incoming-webhook URL. Message templates support {{variable}} values from input. The node sends a notification during execution; use retries carefully to avoid duplicate messages. Provider webhook permissions and rate limits apply.

Python dependency: aiohttp

Ports

Configuration

Example configuration

{
  "channel": "#data-alerts",
  "message_template": "Exported {{count}} orders",
  "webhook_ref": "secret:slack_webhook"
}

Webhook POST

Type: webhook_post. Catalog availability: production.

Send HTTP POST to webhook endpoints.

Prerequisites and behavior: Posts a JSON payload to the configured URL. Without body_template, the incoming payload is posted; a template customizes it. This is an outbound HTTP step, not an inbound flow trigger or a signed lifecycle subscription. The returned response includes HTTP information; inspect status before treating a downstream business action as successful.

Python dependency: aiohttp

Ports

Configuration

Example configuration

{
  "url": "https://your-service.example.com/hooks/orders",
  "headers": {
    "Content-Type": "application/json"
  }
}

Workflow control

Batch Processor

Type: batch_processor. Catalog availability: production.

Process a large collection in configurable batches with concurrency control and progress tracking.

Prerequisites and behavior: Splits an input collection into batches and applies optional run(batch, batch_number) code. Concurrency and delay apply to work inside this node, not separate downstream graph executions. continue_on_error defaults to true; inspect per-batch errors/counts in the summary before treating all records as processed.

Python dependency: No additional provider client beyond the base SDK.

Ports

Configuration

Example configuration

{
  "batch_size": 100,
  "max_concurrent": 1,
  "batch_code": "def run(batch, batch_number):\n    return {\"batch\": batch_number, \"count\": len(batch)}"
}

Conditional Branch

Type: conditional_branch. Catalog availability: production.

Route data based on conditions (if/else).

Prerequisites and behavior: Evaluates a condition over the input and emits it only on the selected true/false port; the unselected output is absent or None. Required downstream inputs on the unselected branch are skipped. Expressions should be authored by trusted developers.

Python dependency: No additional provider client beyond the base SDK.

Ports

Configuration

Example configuration

{
  "condition": "data[\"amount\"] > 100"
}

Error Handler

Type: error_handler. Catalog availability: production.

Wrap a section of the flow with try/catch error handling.

Prerequisites and behavior: Handles an error value supplied as input using the configured strategy. The current node runtime does not forward outputs from failed nodes, including error_out, so simply wiring a failed node to this node does not provide automatic recovery. Explicit successful routing, such as data_quality route mode, is required for that pattern.

Python dependency: No additional provider client beyond the base SDK.

Ports

Configuration

Example configuration

{
  "strategy": "log_and_continue"
}

Human Approval

Type: human_approval. Catalog availability: preview.

Wait for a human decision before selecting an output path.

Prerequisites and behavior: Preview. Waits for a workspace approval decision when the hosted approval capability is available; standalone execution cannot invent an approval. It polls while execution remains alive and is not a durable pause across worker restarts. The default base node timeout is 300 seconds and the maximum is 3,600 seconds, regardless of a longer timeout_hours field. Notification-channel fields are currently logged but do not send notifications themselves.

Python dependency: No additional provider client beyond the base SDK.

Ports

Configuration

Example configuration

{
  "approval_message": "Review order {{order_id}}",
  "timeout_hours": 0.1,
  "_timeout_seconds": 600,
  "notification_channel": "none",
  "poll_interval_seconds": 5
}

Loop

Type: loop. Catalog availability: preview.

Process each item of a collection inside one node.

Prerequisites and behavior: Preview. Iterates inline within this node, applying body_code or body_expression and collecting results. It does not re-execute the downstream graph for every item. The default cap is 1,000 items and excess items are truncated; inspect the completed summary’s truncated/failed counts. item_out contains the final processed result, not an event stream of every iteration.

Python dependency: No additional provider client beyond the base SDK.

Ports

Configuration

Example configuration

{
  "max_iterations": 1000,
  "body_expression": "item[\"amount\"] * 2",
  "continue_on_error": true,
  "collect_results": true
}

Merge

Type: merge. Catalog availability: production.

Merge data from multiple sources.

Prerequisites and behavior: Combines available upstream inputs using concat, zip, or deep-merge strategies. It runs once after its dependencies are terminal; it does not turn failed upstream results into successes. Verify duplicate keys and differing list lengths for the chosen strategy.

Python dependency: No additional provider client beyond the base SDK.

Ports

Configuration

Example configuration

{
  "strategy": "concat"
}

Subflow

Type: subflow. Catalog availability: preview.

Invoke a saved workflow from another workflow when hosted support is available.

Prerequisites and behavior: Preview. Calls a saved flow only when the execution environment provides hosted subflow support. Standalone execution can return status_out.status="not_available" with a successful node result even though nothing ran; inspect that status. wait_for_completion and timeouts are bounded by the outer node/runtime timeout. This node is separate from Python SDK child-flow calls.

Python dependency: No additional provider client beyond the base SDK.

Ports

Configuration

Example configuration

{
  "flow_id": "your-saved-flow",
  "flow_version": "latest",
  "wait_for_completion": true,
  "timeout_seconds": 120,
  "pass_secrets": false
}

Wait / Delay

Type: wait. Catalog availability: production.

Pause execution for a specified time.

Prerequisites and behavior: Waits for the configured delay and then emits a trigger. The delay consumes execution time and is bounded by the node/runtime timeout. Use a schedule for long delays instead of holding a worker.

Python dependency: No additional provider client beyond the base SDK.

Ports

Configuration

Example configuration

{
  "delay_seconds": 5
}

Python, shell, and containers

Use these nodes when your integration needs trusted application code, an installed command, or a Docker image. They run with the permissions and capabilities of the execution environment. A node appearing in the catalog does not make its dependencies available in every hosted runtime.

Docker Container

Type: docker_container. Catalog availability: production.

Run arbitrary Docker images.

Prerequisites and behavior: Requires a reachable Docker daemon and permission to run the selected image. A catalog entry does not make Docker available in every hosted runtime. Inspect the exit code and logs. This node cannot provision a daemon for you.

Python dependency: docker package and reachable Docker daemon

Ports

Configuration

Example configuration

{
  "image": "python:3.11-slim",
  "command": "python --version"
}

Python Function

Type: python_function. Catalog availability: production.

Execute custom Python code or import functions.

Prerequisites and behavior: Inline code must define run(input_data). An import path can address module:attribute, but the exposed base validation still requires a nonempty code field. Python execution is not a hostile-code sandbox; only run trusted code and ensure imports exist in the execution environment.

Python dependency: No additional provider client beyond the base SDK.

Ports

Configuration

Example configuration

{
  "code": "def run(input_data):\n    return {\"count\": len(input_data or [])}"
}

Shell Command

Type: shell_command. Catalog availability: production.

Execute shell commands.

Prerequisites and behavior: Requires the command and binaries to exist in the selected runtime. Supports working directory, environment overrides, and a command timeout. Output ports contain stdout, stderr, and exit code. Shell commands execute with runtime permissions; do not interpolate untrusted input.

Python dependency: No additional provider client beyond the base SDK.

Ports

Configuration

Example configuration

{
  "command": "python --version",
  "_timeout_seconds": 30
}

Troubleshoot a node integration

For custom integration behavior, create a node or implement a Python task. See production guidance and limits before expanding batch size or concurrency.