Streaming - Gwenflow

run_stream and arun_stream take the same arguments as run and arun, and yield the events of the run as they happen: pieces of the answer, reasoning, tool calls starting and finishing. The last item is the complete AgentResponse.

from gwenflow import Agent, ChatOpenAI
from gwenflow.tools import WikipediaTool
from gwenflow.types import AgentResponse

agent = Agent(
    name="Research Assistant",
    llm=ChatOpenAI(model="gpt-5-mini"),
    tools=[WikipediaTool()],
)

for item in agent.run_stream("Who invented the World Wide Web?"):
    if isinstance(item, AgentResponse):
        response = item                      # the final response, usage included
    elif item.event_type == "content":
        print(item.content, end="", flush=True)
    elif item.event_type == "tool_started":
        print(f"\n[calling {item.tool_name}({item.tool_args})]")

Async

import asyncio

async def main():
    async for item in agent.arun_stream("Who invented the World Wide Web?"):
        if isinstance(item, AgentResponse):
            break
        if item.event_type == "content":
            print(item.content, end="", flush=True)

asyncio.run(main())

Events

`event_type`	Carries
`started`	the start of the run
`thinking`	a piece of reasoning, in `content`
`content`	a piece of the answer, in `content`
`tool_started`	`tool_call_id`, `tool_name`, `tool_args`
`tool_completed`	`tool_call_id`, and the result in `content`
`tool_approval_required`	a tool call waiting for a decision. See Tool approvals.
`subagent_started`, `subagent_completed`	a sub-agent’s description, tools, model; then its finish reason, usage and duration
`completed`	the end of the run: accounting is settled at this point

Every event carries agent_id and run_id. In a stream, the events of sub-agents are forwarded live too, tagged with metadata["subagent_call_id"], so a UI can show what each one is doing. The same events are kept in response.events for a non-streaming run.

Stopping early

Breaking out of the loop, closing the generator or cancelling the task ends the run. Its response gets finish_reason="interrupted", and what was spent until then still counts in agent.total_usage and in the traces.

Streaming a model call

LLM clients stream too: llm.stream() and llm.astream() yield one ModelResponse per delta.

for chunk in llm.stream("Tell me a story"):
    if chunk.text:
        print(chunk.text, end="", flush=True)

Inside a Temporal workflow, streaming is not available: use await agent.arun(). See Durable execution.