Skip to main content
A processor is a short Python function that turns the events of one stream into events of other streams: keep only the large orders, split a mixed feed by kind, reshape a provider’s payload. An agent writes it with the streams tools; the platform runs it over every matching event in the sandbox. Use a processor when a rule decides where an event goes. When an agent should reason about each event, attach a trigger instead; to copy events unchanged, a forward is enough.
Prerequisites
  • An input stream with events in it, and the output streams created: see Collect events into a stream.
  • An agent with the agentarea/streams toolset, or a member who may edit the input stream and every output stream.

​
The contract

The code defines process(event: dict) -> list[dict] | None.
  • event holds id, kind, key, data, source_type, session_key, sender, subject, time, received_at, correlation_id, causation_id, depth, stream_id and attachments. data is what the source delivered; a webhook’s JSON body is data["body"].
  • Each dict returned is one event to write: {"stream", "kind", "data", "key"}. stream is the id or name of one of the processor’s output streams; key is optional. Return None or [] to drop the event. A single dict is an error, not one event.

​
Steps

1

Ask an agent to write it

Give the agent the input stream, the output streams and the rule. It calls streams_create_processor with stream_id, output_stream_ids, code and an optional event_filter ({"kinds": ["order"]} runs the code only on order.* events). A good agent reads a recent event with streams_list_events first, to see the shape of data.
2

Or create it over REST

POST /v1/workspaces/{workspace}/streams/{stream_id}/processors with the same fields. PATCH .../processors/{subscription_id} changes the code, outputs or filter; DELETE removes it. The events it already wrote stay.
3

Post events

Every event that matches the filter runs the code once. What it returns is appended to the output streams, each event with causation_id set to the event it came from, the same correlation_id, and one more depth.

​
Verify

Open the input stream in Events. The Processors panel lists each processor with its outputs and its code. In the feed, select an event: the processor’s handler says Reacted with the events it wrote, Skipped with process() wrote no events, or Error with the exception and its line, for example ZeroDivisionError: division by zero (processor.py line 3). The output stream shows the new events, linked back to their cause.

​
Limits

  • Each run has 10 seconds and may return up to 100 events and 256 KiB of output. The event and the code together must fit in 256 KiB.
  • The code gets no secrets. Code that holds something that looks like a key, token or password is refused when it is saved, because everyone who can read the stream sees it.
  • Network access is whatever the deployment’s sandbox policy gives an agent’s shell; the processor gets no more.
  • A processor never writes into its own input stream: that is refused when it is created or changed, and checked again on every run. A chain of processors and forwards stops at AGENTAREA_EVENT_FORWARD_DEPTH.
  • A stream a processor writes into cannot be deleted until the processor is.

​
Troubleshooting

The code looks for a field where the source does not put it. Read one event with streams_list_events and compare its data with what the code reads.
The reason names the problem: an exception with its line, a result that is not a list, or a stream that is not one of the outputs. Fix the code with streams_update_processor; the next event runs the new code. A failed event is not run again.
The sandbox could not run the code at all. The event is redelivered, and after AGENTAREA_EVENT_MAX_ATTEMPTS deliveries it is recorded as an error and dead-lettered. Check that the sandbox service is up.

Event streams

How streams, subscriptions and the router fit together

Collect events into a stream

Add a webhook or poll source to a stream
Last modified on October 19, 2026