Building a Custom Processor Plugin for Redpanda Connect

Redpanda Connect ships with a wide range of connectors and processors for building data pipelines. But, in real-world pipelines, sometimes you need transformation logic that’s not already built-in. I wanted to see how far you could take that without leaving Python, so this walks through building a dynamic plugin for Redpanda Connect.

Dynamic plugins let you extend Redpanda Connect with custom inputs, processors, or outputs written in any language that supports gRPC, running as separate subprocesses alongside the main engine. This gives you language flexibility while still using Redpanda Connect's normal pipeline config and orchestration. Your plugin is just another component in the YAML pipeline config. It loads at startup, processes messages alongside built-in components, and shuts down cleanly when the pipeline stops.

In this tutorial, I'll show you how to:

• Build a custom Python processor plugin that transforms messages in a Redpanda Connect pipeline
• Declare a plugin manifest and wire it into a YAML pipeline config
• Run and verify the plugin end to end using rpk connect run
• Extend the plugin to accept runtime configuration through the manifest

Why Use a Dynamic Plugin

Suppose most of your pipeline uses built-in Redpanda Connect processors for filtering and routing, but one step needs a proprietary text normalization routine written in Python. That routine changes often, and you don't want to recompile a Go binary every time the logic updates.

Dynamic plugins solve exactly this. You write the normalization logic as a Python function, package it alongside a plugin manifest, and load it into the pipeline at runtime with the --rpc-plugins flag. The plugin runs as a separate subprocess communicating with the main Redpanda Connect engine over gRPC. No Go code, no recompilation, no changes to the core binary.

Here's a high-level architecture diagram of a pipeline with a dynamic plugin:

Splitting the plugin into its own process has a few practical benefits:

• If the plugin crashes, it doesn't take the main engine down with it. Redpanda Connect notices and restarts the subprocess.
• You're not locked into Go. Any language with gRPC libraries works, and the official Python SDK handles the protocol so you're only writing the actual transformation logic.
• The plugin is its own subprocess, so you can develop, test, and deploy it separately from the rest of the pipeline.

Prerequisites

You'll need these installed to follow along:

• Redpanda Connect (rpk CLI, v4.56.0 or later, which includes rpk connect)
• Python 3.12 or later
• uv (Python package manager)

Basic familiarity with YAML pipeline configuration is helpful but not required. I'll explain each configuration field as it comes up throughout the tutorial.

Setting Up the Plugin Project

Create the plugin directory and initialize a Python project inside it:

mkdir -p plugins/yell-processor
cd plugins/yell-processor
uv init --no-readme

The uv init command creates a pyproject.toml and sets up a virtual environment. The --no-readme flag skips generating a README file since the plugin directory only needs the processor script and the manifest.

Install the Redpanda Connect Python SDK:

uv add redpanda_connect

This adds the redpanda_connect package to the project and installs it in the virtual environment. The SDK provides the decorator, message types, and gRPC server that your plugin needs to communicate with Redpanda Connect. Return to the project root before continuing:

cd ../..

The project directory should have this structure at this point:

├── plugins/
│   └── yell-processor/
│       ├── pyproject.toml
│       └── .venv/

Writing the Processor Logic

Create plugins/yell-processor/yell_processor.py with the following content:

import asyncio
import logging

import redpanda_connect


@redpanda_connect.processor
def yell(msg: redpanda_connect.Message) -> redpanda_connect.Message:
    text = msg.payload
    if isinstance(text, bytes):
        text = text.decode("utf-8")
    msg.payload = str(text).upper()
    return msg


if __name__ == "__main__":
    logging.basicConfig(level=logging.INFO)
    asyncio.run(redpanda_connect.processor_main(yell))

The @redpanda_connect.processor decorator marks the yell function as a processor component. The function receives a Message object and returns a modified Message. The SDK's Message is a dataclass with two main fields: payload holds the message content (bytes or a structured value), and metadata is a dictionary of key-value pairs for additional context outside the payload.

The processor_main() call at the bottom starts a gRPC server that listens on a Unix socket. Redpanda Connect launches this script as a subprocess and communicates with it through that socket. You do not need to manage the connection or protocol yourself.

Declaring the Plugin Manifest

Create plugins/yell-processor/plugin.yaml alongside the processor script:

name: yell
summary: Converts every message payload to uppercase.
command: ["uv", "run", "yell_processor.py"]
type: processor
fields: []

Each field in the manifest serves a specific purpose. name is the identifier you reference in connect.yaml when wiring the plugin into a pipeline. summary is a human-readable description of what the plugin does. command is the shell command array that Redpanda Connect executes to start the plugin subprocess, and using uv run ensures the script runs inside the project's virtual environment with all dependencies available. type declares the component type, accepting processor, input, or output. fields lists configurable parameters that users can pass from the pipeline config, and an empty array means the plugin has no configurable fields.

If a plugin fails to load, the first thing I'd check is whether there’s a mismatch between the command path and the actual script filename. Make sure yell_processor.py exists in the same directory as plugin.yaml.

Wiring the Plugin into a Pipeline

Create connect.yaml in the project root (one level above plugins/):

input:
  generate:
    interval: 1s
    count: 5
    mapping: |
      let events = ["user signed up from mobile-app", "payment processed for order 1042", "dashboard export requested", "session timeout on web-client", "inventory sync completed"]
      root = $events.index(counter() % 5)

pipeline:
  processors:
    - yell: {}

output:
  stdout:
    codec: lines

The generate input produces five test messages at one-second intervals, cycling through a list of event log strings that simulate upstream service traffic. The yell: {} entry under processors tells Connect to route each message through the plugin registered under the name yell. The empty braces pass no configuration to the plugin, which works because this version has no configurable fields. The stdout output prints each processed message to the terminal.

In a production pipeline, you would replace generate and stdout with actual data sources and targets such as Redpanda topics, HTTP endpoints, or database connections.

Running and Verifying the Pipeline

Start the pipeline by pointing --rpc-plugins at the plugin manifest:

rpk connect run --rpc-plugins=plugins/yell-processor/plugin.yaml connect.yaml

Connect reads the manifest, launches uv run yell_processor.py as a subprocess, establishes a gRPC connection, and registers the plugin under the name yell. The terminal output includes startup logs followed by the processed messages:

INFO Running main config from specified file    path=connect.yaml
INFO Listening for HTTP requests at: http://0.0.0.0:4195
INFO Launching a Redpanda Connect instance, use CTRL+C to close
INFO Output type stdout is now active
INFO Input type generate is now active
USER SIGNED UP FROM MOBILE-APP
PAYMENT PROCESSED FOR ORDER 1042
DASHBOARD EXPORT REQUESTED
SESSION TIMEOUT ON WEB-CLIENT
INVENTORY SYNC COMPLETED
INFO Pipeline has terminated. Shutting down the service

Each uppercase line confirms that the plugin received an event message, converted the payload to uppercase, and returned it to the pipeline. The pipeline exits automatically after five messages because of the count: 5 setting in the input config.

Making the Plugin Configurable

The basic yell processor has its behavior hardcoded. The plugin system supports runtime configuration through the fields array in plugin.yaml and the config dictionary passed to the constructor function.

To accept configuration, you replace the @redpanda_connect.processor decorator with a manual constructor pattern. The constructor is a function named processor that receives the config as a dictionary and returns an object with process() and close() methods.

Update plugins/yell-processor/yell_processor.py to this configurable version:

import asyncio
import logging

import redpanda_connect


class YellProcessor:
    def __init__(self, prefix: str, repeat_count: int):
        self.prefix = prefix
        self.repeat_count = repeat_count

    async def process(
        self, batch: redpanda_connect.MessageBatch
    ) -> list[redpanda_connect.MessageBatch]:
        results = []
        for msg in batch:
            text = msg.payload
            if isinstance(text, bytes):
                text = text.decode("utf-8")
            msg.payload = (self.prefix + str(text).upper()) * self.repeat_count
            results.append(msg)
        return [results]

    async def close(self) -> None:
        pass


def processor(config: redpanda_connect.Value) -> YellProcessor:
    if isinstance(config, dict):
        prefix = str(config.get("prefix", ""))
        repeat_count = int(config.get("repeat_count", 1))
    else:
        prefix = ""
        repeat_count = 1
    return YellProcessor(prefix, repeat_count)


if __name__ == "__main__":
    logging.basicConfig(level=logging.INFO)
    asyncio.run(redpanda_connect.processor_main(processor))

The processor() function at the bottom replaces the decorator. Connect calls this function once at startup and passes any configuration values as a Value dictionary. The function reads prefix and repeat_count from the config with safe defaults, then returns a YellProcessor instance. The process() method operates on message batches rather than individual messages, which is the interface the dynamic plugin system uses internally to spread serialization costs across multiple messages.

Update plugin.yaml to declare the new fields:

name: yell
summary: Converts every message payload to uppercase with optional prefix and repeat.
command: ["uv", "run", "yell_processor.py"]
type: processor
fields:
  - name: prefix
    type: string
    default: ""
  - name: repeat_count
    type: int
    default: 1

Pass values for these fields in connect.yaml by replacing the empty braces with a mapping:

pipeline:
  processors:
    - yell:
        prefix: ">> "
        repeat_count: 2

Run the pipeline again with the same command. The output now includes the prefix and repeats each transformed message twice:

>> USER SIGNED UP FROM MOBILE-APP>> USER SIGNED UP FROM MOBILE-APP
>> PAYMENT PROCESSED FOR ORDER 1042>> PAYMENT PROCESSED FOR ORDER 1042
>> DASHBOARD EXPORT REQUESTED>> DASHBOARD EXPORT REQUESTED
>> SESSION TIMEOUT ON WEB-CLIENT>> SESSION TIMEOUT ON WEB-CLIENT
>> INVENTORY SYNC COMPLETED>> INVENTORY SYNC COMPLETED

Conclusion

In this tutorial, I walked you through building a custom Python processor plugin that runs inside a Redpanda Connect pipeline with full process isolation and no Go code required. I started with a minimal decorator-based processor, declared a plugin manifest, wired it into a pipeline, and then extended it to accept runtime configuration through the manifest fields.

The same pattern applies to the other two component types: inputs that pull data from external systems and outputs that push data to external targets. You can apply this approach to build custom inputs that pull from proprietary APIs, processors that call internal ML models or normalize data formats, or outputs that write to systems Redpanda Connect does not natively support. Each plugin is a standalone subprocess that you can package, version, and distribute independently. That process boundary is really the whole point, because it allows you to keep iterating in Python without ever touching the core pipeline.

The complete companion code for this tutorial is available at github.com/SystemCraftsman/redpanda-connect-dynamic-plugin-demo.