Add Custom Stream Output to Your Agent

Overview

You can configure custom stream output to instruct your agent to emit structured, machine-readable output for streaming invocations in addition to plain text. The Atlas Agent Engine sends this output to clients as custom events.

This guide shows how to perform the following tasks:

• Enable Custom Stream Output: Set the features.use_custom_parser: true flag in your agent.yaml file.

• Register an Output Parser: Register an OutputParser subclass on your app.

• Emit Custom Events: Emit custom events from your agent code.

Prerequisites

Before you begin, ensure that you have a registered agent with an agent.yaml file. To learn more, see Get Started with the Atlas Agent Engine.

Enable Custom Stream Output

To enable custom stream output for your agent, add the following code to your agent.yaml file:

features:
  use_custom_parser: true

If you set the features.use_custom_parser flag to true but do not register an output parser, the agent fails when an invocation starts and returns a RuntimeError.

Register an Output Parser

Register an OutputParser subclass on your app by using the @app.output_parser decorator. The Atlas Agent Engine validates the class at decoration time, so passing a class that is not an OutputParser subclass fails when the module imports.

For LangGraph agents, subclass the LangGraphOutputParser class. Set the stream_modes class attribute to the LangGraph stream modes that you want to convert to custom events. Then, implement the parse() and on_stream_error() methods.

The following example registers an output parser for a LangGraph agent:

from magenta_sdklanggraph.output_parser import LangGraphOutputParser
@app.output_parser
class BriefParser(LangGraphOutputParser):
    stream_modes = ("messages", "values")
    async def parse(self, item, ctx):
        ...
        yield {"event": "brief", "data": new_brief}
    async def on_stream_error(self, ctx, error):
        ...
        return {"event": "error"}

The following table describes the LangGraphOutputParser members:

If the parse() or on_stream_error() method raises an error, the stream fails rather than silently dropping the output.

Emit Custom Events

During a streaming invocation, you can use the emit_custom_event() function to emit custom events directly from LangGraph nodes, tools, or any other code that runs in the execution. Pass a JSON object containing string keys to emit_custom_event(), as shown in the following example:

from runner_shared import emit_custom_event
await emit_custom_event({"event": "todo", "items": ["a", "b"]})

The Atlas Agent Engine forwards the JSON object unchanged and does not impose a schema, so you can define the payload's structure to match what your client expects.

When calling the emit_custom_event() function, the following rules apply:

• Call the method only from a streaming invocation. Calling it from the synchronous invoke path raises an error.

• Enable the features.use_custom_parser flag in the agent.yaml file. If the flag is disabled, the platform raises an error.

• Await the method from asynchronous code. From synchronous code, use emit_custom_event_sync() instead.

Next Steps

To learn how a client receives the streamed output, see Stream Your Agent's Output in the Invoke an Agent guide.