Streaming Agents with Flink Agents 0.4.0

Apache Flink Agents 0.4.0 was released on October 30. It's a preview release that refines the features and APIs ahead of a production-ready 1.0. I contributed to 0.4.0. The release prompted me to write about what Flink Agents does and how it runs and recovers agents.

Why run agents on Flink

When a payment fails or a sensor reports an unusual reading, the system needs to determine what happened and take the next step. It can't wait for someone to ask in a chat window. Flink Agents started as a project for this kind of work: agents process a continuous stream of events, make decisions, and call tools to act on them.

As event volume grows, the work has to be distributed across servers, with state kept for each customer or device. A system that runs continuously also needs to resume after a server fails or execution stops. Code that calls a model doesn't provide that runtime on its own.

Flink already distributes stream processing, manages state by key, and recovers from checkpoints. Flink Agents is an Apache Flink sub-project that adds an execution model for agents using models, tools, and memory. Event ingestion, agent decisions, and downstream processing all connect within one Flink job.

From 0.1.0 to 0.4.0

The first preview, 0.1.0, was released in October 2025. It established the execution model: divide work into actions, connect them through events, and call models and tools from within Flink. An event log helped developers follow execution, and saved action records could be reused during recovery. The release ran on Flink 1.20.3 and provided both Python and Java APIs, with more features implemented in Python.

0.2.0, released in February 2026, filled gaps in the Java API and expanded Flink support from 1.20 through 2.2. Agents could also use resources implemented in the other language. The memory system grew to distinguish data needed within one run from data kept for later runs. Recovery could now reuse saved results of individual calls inside an unfinished action, as well as records of completed actions.

0.3.0 followed in June 2026. It added YAML declarations for models and tools, along with support for running actions written in the other language. Long-term memory moved to Mem0 for extracting and organizing information, and Agent Skills provided task instructions and reference material. Apache Fluss became an option for storing execution records, and event logs became available in the web UI. The update also added reconcilers to check external calls whose outcomes weren't known during recovery. Earlier versions could reuse saved call results; now an application could also check the external system's state when execution had stopped without saving a result.

The newly released 0.4.0 builds on that foundation with sub-agents and model routing. The new ctx.chat API lets you write the logic after a model call in the same action, while tracing now covers model and tool calls as well as events. Existing APIs and default behavior also changed ahead of the API compatibility commitment planned for 1.0.

How an agent runs

An agent runs as an operator inside a Flink job. It receives data from upstream, performs its work, and passes the result to the next operator. Within the agent, events connect the steps. The building blocks are events, actions, and resources. An event triggers matching actions, which do their work and can send more events. One event can trigger several actions, and the events they send can lead to branches or another iteration of a loop. Shared components such as models, tools, and prompts are registered as resources and referenced by name. You can also connect tools and prompts from MCP servers. Since 0.3.0, resources can be declared in YAML.

For an agent that summarizes product reviews, an input record arrives as an InputEvent. An action responds by sending a ChatRequestEvent to request a summary. The built-in chat action calls the model and sends a ChatResponseEvent, then another action puts the summary in an OutputEvent for the next Flink operator. The events connect the steps, so their records let you follow an input through the decisions and outputs it produced.

You can define the workflow in code or let a model choose what to do next. With the built-in ReActAgent, you supply a model and can add tools and a prompt. The model chooses a tool, examines its result, and decides the next step toward its goal. You can also combine the two approaches, defining the overall workflow in code and delegating an investigation to a ReActAgent.

Once the agent is defined, attach it to a Flink job. This excerpt from the Python example in the 0.4.0 docs shows that part.

env = StreamExecutionEnvironment.get_execution_environment()
agents_env = AgentsExecutionEnvironment.get_execution_environment(env)

output_stream = (
    agents_env.from_datastream(
        input=input_stream, key_selector=lambda x: x.id
    )
    .apply(your_agent)
    .to_datastream()
)

Input is partitioned by key, and each record from upstream starts a new execution, which the docs call an agent run. If x.id in this example is a customer ID, runs for the same customer can share memory. An explicit key selector takes precedence; if you omit it, the input DataStream must already be keyed. Without either, the call fails. Table input is also supported, but it must be insert-only. Aggregations that update or delete earlier rows can't be connected directly.

Memory types differ in how long they last and how you retrieve their contents. Sensory and short-term memory both live in Flink state. Sensory memory is cleared when a run ends; short-term memory remains available to later runs with the same key. For a customer-specific agent, temporary values needed to handle the current request belong in sensory memory, while values needed for later requests belong in short-term memory.

Long-term memory organizes information collected across runs and retrieves it through semantic search. Since 0.3.0, Mem0 handles extraction and consolidation, while an external vector store holds the data. You need to configure a chat model, an embedding model, and a vector store. Searches are scoped to the same job and input key by default.

Agent Skills provide task instructions for the model. A skill bundles instructions in SKILL.md with related scripts and reference material, which the agent can read as needed.

How recovery works

An agent may call models and external tools several times before finishing a task. Along the way, calls can time out or hit rate limits, and the process running the agent can stop. Retry policies handle call failures; recovering from an interrupted execution also requires a way to continue the remaining work without repeating what has already finished. Starting over may produce a different result. A model can answer the same input differently, and values queried from an external system can change between attempts. Repeating an email or payment request can also duplicate an external operation. Recovery needs to distinguish completed work from work that must run again.

After a failure, Flink restores state and input positions from the last checkpoint, then reprocesses subsequent input. Transactional or idempotent sinks can keep those replayed outputs from being applied twice to an external store. But while reprocessing input, an agent may call models and tools again, and a checkpoint can't undo an external operation that has already happened.

To reuse completed action records during recovery, configure an Action State Store alongside Flink checkpointing. It stores action execution records in Kafka or Apache Fluss; reusing completed action records has been supported since 0.1.0. In the recovery flow described in the 0.4.0 consistency docs, an action with a saved completion skips its body, and the framework replays its saved sensory and short-term memory updates and output events. It does not reproduce the order in which asynchronous actions read and wrote shared memory, though. If the result depends on that order, you have to make it explicit with event dependencies.

0.2.0 narrowed what gets skipped on recovery down to individual calls inside an action. Even if the action didn't finish, recovery skips any durable call that completed and had its result saved. The action docs use a profile lookup as the example.

class ProfileAgent(Agent):
    @action(EventType.InputEvent)
    @staticmethod
    def process_input(event: Event, ctx: RunnerContext) -> None:
        user_id = InputEvent.from_event(event).input
        profile = ctx.durable_execute(
            profile_service.get_profile,
            user_id,
            durable_id=f"profile:{user_id}",
        )
        ctx.send_event(OutputEvent(output=profile))

If the action fails after the lookup finished and its result was saved, the recovered action reads the saved profile instead of calling the service again. As the docs note, code outside the durable call can still run again. Completions and call results are written to the external store synchronously, and you have to retain both the checkpoints you may restore and the records that recovery needs. For recovery to match those records to the same work, the per-key input order, key types, serializer configuration, and action names have to stay the same. Durable calls also need the same inputs, the same order, and stable identifiers, and the saved values must still be readable at restore time.

If execution stops after a call starts but before its result is saved, the record alone can't tell you whether the call succeeded. If the external service handles repeated requests without duplicating the operation, you can retry. Otherwise, you need to check whether the request was processed. The 0.3 roadmap walkthrough raised this problem and proposed a function through which application code could check the external state.

Version 0.3.0 added a reconciler for that check. When recovery finds an in-flight call record with no saved result, it invokes the registered reconciler. You implement the function to check the external system's state and return the original result or report a failure. That outcome is recorded as well. Without a reconciler, the original call runs again.

The Action State Store can't commit an external operation and its saved result in one transaction. Design any call that changes external state on the assumption that it may repeat. Use the external service's idempotency support, or give the durable call a reconciler that can find the original result.

New features

This release adds ways to delegate work to other agents and select a model for each request. You can write a model call and its response handling in one action, and inspect records of model and tool calls during operation.

Read the table as text in the original article.

On September 15, 2026, after the 0.3.0 release, TypeSafe introduced Jev. It evaluates the supplied context against defined criteria and returns probabilities, a choice, or a score. Flink Agents adds a decision model API in this version, connecting Jev, OpenAI, and Alibaba Cloud Model Studio through the same interface. For example, you can decide whether a customer inquiry belongs with billing or technical support, or assess an alert's risk against a defined scale. The application supplies the facts and criteria and decides what to do with the result.

Developer tooling and model integrations were expanded too. Java asynchronous execution now supports JDK 11 and 17 alongside JDK 21. New chat model integrations are Gemini for Java and vLLM and IBM watsonx.ai for Java and Python, and Java also gets OpenAI embeddings. The new flink-agents-dev skill provides guidance for coding agents building and validating Flink Agents applications.

Consider an agent that handles only severe alerts. A conditional trigger runs the action only for high-severity alerts. Inside that action, ctx.chat calls the model and lets the code handle its response directly. If the model asks for a registered tool, the framework runs it.

The event-based chat API uses separate actions to send the request and handle the response. It remains supported in 0.4.0. Use ctx.chat when you want to write the logic after the model call in the same action. With durable execution enabled, recorded ctx.chat responses can also be reused on recovery.

Further investigation can be delegated to a sub-agent, and those calls support durable execution too. To score risk against fixed criteria, use a Decision Model. Model Routing lets you select a model for each request, sending simple requests to a lightweight model and reserving a larger one for complex tasks.

In operation, Trace Log lets you follow the steps and calls that turned an input into a result. When the built-in metrics show a failure or a latency spike, the log can help narrow down the step where it went wrong. By default it records only events, so seeing steps and calls means setting trace-log.targets. A tool that rebuilds the causal tree of events also ships. It reads only event records and skips the step and call records. Event processing and the job continue even if a Trace Log write fails. The trace alone can't prove that every execution was recorded.

How to upgrade to 0.4.0

The 0.x releases are still previews, so APIs and configuration can change incompatibly. This release addresses issues found through feedback from early production users and an API review, bringing more breaking changes than any earlier release. The plan is to stabilize 0.4.0, make small adjustments for 1.0, and commit to API compatibility from that point. The detailed compatibility policy will be announced separately. If you're upgrading from 0.3.x to 0.4.0, check these changes first.

• Action declarations: Java @Action(listenEventTypes = ...) becomes @Action(...), and Python add_action(events=...) and YAML listen_to become trigger_conditions. The common Python @action("event_type") form still works.
• Python execution: the path that ran without Flink through from_list() and to_list() is gone. Connect to a Flink StreamExecutionEnvironment with DataStream or Table inputs and outputs, and use a local Flink runtime for local development.
• Retry defaults: the framework's max-retries default drops from 3 to 0, and error-handling-strategy is removed. Handle failed chat responses in application code. Provider SDK retries are configured separately.
• Concurrency defaults: under the default async settings, independent tool calls may run concurrently. Split order-dependent tools into separate steps or limit parallelism. For plans with only Java actions on JDK 11 through 20, experimental parallel action execution is on by default and can be turned off with parallel-execution.enabled.
• Tool argument validation: unknown fields, wrong types, and invalid values fail before execution, and numeric strings are no longer converted to numbers.
• Logging configuration: the event log is replaced by Trace Log, so old keys such as event-log.* are rejected. Move to the trace-log.* keys.
• Execution records: Action State Store data written by 0.3.x is not compatible with 0.4.0, so plan the upgrade without reusing old execution records. Whether Flink checkpoints and savepoints can be restored is a separate compatibility check.
• Pending actions: operator snapshots that contain a pending action may fail to restore when upgrading to 0.4.0. Earlier releases had the same limitation, so test a restore with representative state first.

Message formats changed too. ChatMessage holds content in blocks and extra attributes in metadata. Model calls return a ChatResult that separates the response message from model information and usage, and tool results are unified as ToolResponse. If you wrote your own model integration or store messages as JSON, start there.

0.4.0 from a contributor's seat

I started contributing in this release and am one of the 38 people listed in its announcement. Five of my 23 PRs fixed the durable execution replay path described above. Replaying a recorded null result threw an exception #1171, and a Python exception that could be pickled but not rebuilt replayed as an unrelated TypeError #1182. A Python durable call that records only its completion, with no reconciler, didn't tell the caller when the Java side failed to save the result #1211.

When ordinary and durable calls were awaited together asynchronously, canceling an ordinary call could prevent results from being saved for durable calls that had already finished and came before it in input order #1239. I also fixed a built-in Python chat call identifier that could differ from process to process #1240. The rest of my PRs covered model routing validation, memory references, runtime queues, metrics, and docs.

Just before the release, I verified the first release candidate and voted. I wrote about that in There Are Votes in Open Source Too. During verification, I found a problem restoring existing snapshots after a version upgrade, and the official announcement included it in the upgrade notes. I then took on building test infrastructure for state restoration across versions. I hope 1.0 brings a stable API and stronger upgrade testing, making agents in production easier to maintain.

Getting started

Check the supported Flink, Java, and Python versions in the setup guide, then run the first agent example. If you want to start by letting a model choose tools, use the ReActAgent quickstart. If you develop with a coding agent, you can add the flink-agents-dev skill. It gives the coding agent API guidance and examples; install Flink and the Flink Agents runtime separately by following the setup guide above.

npx skills add 'apache/flink-agents#release-0.4@flink-agents-dev'

Bugs and feature requests go to GitHub Issues. Use cases and ideas are discussed in GitHub Discussions and the #flink-agents-user channel on the Apache Flink Slack.

References

• Apache Flink Agents 0.4.0 Release Announcement, Wenjin Xie, 2026: the new features in 0.4.0, the breaking changes, and the expectation that 0.4.0 will be the last release before 1.0.
• Apache Flink Agents 0.3.0 Release Announcement, 2026: why agents run on Flink, and the API compatibility plan through 0.4.0 and 1.0.
• Apache Flink Agents 0.2.0 Release Announcement, 2026: the three kinds of memory, call-level durable execution, and support for multiple Flink versions.
• Apache Flink Agents 0.1.0 Release Announcement, 2025: why the project exists and the core features of the first preview.
• Flink Agents 0.3 Roadmap解读, Flink Chinese Community, 2026: a contributor's walkthrough of the 0.3.0 roadmap, including the remaining gap in recovery and a possible fix. The article in that community has not yet been updated to cover 0.4.0.
• Flink Agents 0.4.0 documentation: getting started, the development guide, and an operations section on consistency and the recovery flow.

---

I write about building and leading large-scale platforms, from ads and CRM to AI agents, at Namsang Labs Tech Blog.