feat: #864 support streaming nested tool events in Agent.as_tool

[vrtnis]

• Add stream_inner_events flag to allow sub-agent tool call visibility
• Use Runner.run_streamed for streaming inner agents
• Emit nested tool_called/tool_output events in parent stream
• Add test coverage for inner streaming behavior

Resolves #864

[vrtnis]

@rm-openai i’ve resolved the merge conflict in tool_context.py. The branch is now up to date and ready for your feedback.

[chrisptang]

looking forward to this new feature

[yxh-y]

  if stream_inner_events:
      from .stream_events import RunItemStreamEvent
  
      sub_run = Runner.run_streamed(
          self,
          input=input,
          context=context.context,
      )
  
      parent_queue = getattr(context, "_event_queue", None)

When I test on my project, in the above codes in agent.py, the parent_queue will be None and the sub_run event will not be streaming, I manually init it as:

  parent_queue = asyncio.Queue()
  context._event_queue = parent_queue

It is right?

[vrtnis]

hey @yxh-y the queue is only attached when the outer‑most run itself is started in streaming mode. In other words, Runner.run_streamed(…) sets ctx._event_queue = streamed_result._event_queue before control reaches any nested agents or tools.

If you invoke the outer agent with the non‑streaming Runner.run(…), no queue is created, so parent_queue will be None inside Agent.as_tool(…, stream_inner_events=True).

So, rather than injecting your own asyncio.Queue(), call your top‑level agent like this:

run = Runner.run_streamed(root_agent, input=user_input, context=ctx)
async for ev in run.stream_events():
...

this ensures that ctx._event_queue is populated. Inner agents/tool calls that you wrapped with
child_agent.as_tool(stream_inner_events=True) can forward theirtool_called / tool_output events to the same queue, so they show up in the outer run.stream_events() stream.

tl;dr, if you start the outer run with run_streamed, the queue is set up automatically (no manual queue needed), and you’ll see the child events immediately.

[vrtnis]

Hey @rm-openai, just a quick ping on this PR. It’s playing nice with Runner.run_streamed and streams nested tool_called / tool_output events when stream_inner_events=True, but I’m wondering if this is generally the direction we want to take overall.

BTW, totally get that it might end up redundant with future changes (esp. with how quickly tracing and realtime features are evolving), but I haven’t seen anything merged yet that overlaps, so figured it was worth a quick nudge in case it’s still helpful to review. Happy to tweak, rebase, or rework as needed.

[chrisptang]

thanks for all your hard work. can we expect this new feature in next release? 😊

[SeeYangZhi]

@vrtnis Thanks for this PR, I encounter the same need to stream inner/sub tool calls, and was wondering if this PR will also emit tools that are created in the following way:

    @function_tool
    async def parent_tool(
            context_info: RunContextWrapper[Any], input: str
        ) -> str:
            """
            A parent tool with multiple tool calls and agent runs within.
            """
            try:
                # Create agent
                agentA = AgentA().create_agent() # AgentA may have multiple tools and nested tools as well
                
                resultA = Runner.run(
                    agentA,
                    input=user_inputs,
                )
                
                agentB = AgentB().create_agent() # AgentB may have multiple tools and nested tools as well
                
                resultB = Runner.run(
                    agentB, 
                    input=resultA,
                )


               return json.dumps(resultB.final_output, ensure_ascii=False, indent=2)

            except Exception as e:
                return "parent_tool encountered error"

OR

    def get_tool(self) -> FunctionTool:
        schema = parent_tool.model_json_schema()
        schema["additionalProperties"] = False
        
        return FunctionTool(
            name="parent_tool",
            description="A parent_tool with multiple Agent Runs and Tool calls.",
            params_json_schema=schema,
            on_invoke_tool=self.parent_tool,
        )
        return tool

[vrtnis]

hey @SeeYangZhi this pr #1057 does not affect the parent_tool pattern you showed, it only augments the convenience wrapper Agent.as_tool().

If you bypass that Agent.as_tool() wrapper with your own FunctionTool, you’re responsible for running the inner agents streamed and relaying their events up the queue. If you do that, the nested tool_called / tool_output events will appear in the outer run.stream_events() just like they do for as_tool.

As such, if you would like to keep your FunctionTool you can run the sub‑agents in streamed mode, e.g.,runA = Runner.run_streamed(agentA, input=user_inputs, context=ctx.context) and then relay their events to the parent queue, so something like this:

from agents.stream_events import RunItemStreamEvent

parent_q = getattr(ctx, "_event_queue", None)
async for ev in runA.stream_events():
    if parent_q and isinstance(ev, RunItemStreamEvent):
        if ev.name in ("tool_called", "tool_output"):
            parent_q.put_nowait(ev)

then repeat this for runB, and combine the final results as you already do.

Please note that instead of the above snippet probably easier alternatives could be to expose each sub‑agent directly, like so,

AgentA().create_agent().as_tool("agent_A", stream_inner_events=True)
AgentB().create_agent().as_tool("agent_B", stream_inner_events=True)

and then drop your custom parent_tool and let the caller agent orchestrate.

[SeeYangZhi]

I see, thanks for the explanation!

caller agent orchestrate

This may be tough, but we will think about it! We want a more deterministic tool, setting the main agent to call a tool that goes through a certain "workflow".

[vrtnis]

You bring up an interesting point around tool determinism and you should still be able to keep that deterministic workflow feel while getting streaming too. e.g., set tool_use_behavior="stop_on_first_tool" and also include instructions that only one tool should be called (like your workflow tool).
Btw there are a few knobs already in the sdk that do help with this. Experimenting with low temp, strict schemas, and a tool whitelist can get repeatable runs while still streaming and nesting agents.