repl

run_demo_loop async

run_demo_loop(
    agent: Agent[Any],
    *,
    stream: bool = True,
    context: TContext | None = None,
    max_turns: int = DEFAULT_MAX_TURNS,
) -> None

Run a simple REPL loop with the given agent.

This utility allows quick manual testing and debugging of an agent from the command line. Conversation state is preserved across turns. Enter exit or quit to stop the loop.

Parameters:

Name	Type	Description	Default
agent	Agent[Any]	The starting agent to run.	required
stream	bool	Whether to stream the agent output.	True
context	TContext | None	Additional context information to pass to the runner.	None
max_turns	int	Maximum number of turns for the runner to iterate.	DEFAULT_MAX_TURNS

Source code in src/agents/repl.py

async def run_demo_loop(
    agent: Agent[Any],
    *,
    stream: bool = True,
    context: TContext | None = None,
    max_turns: int = DEFAULT_MAX_TURNS,
) -> None:
    """Run a simple REPL loop with the given agent.

    This utility allows quick manual testing and debugging of an agent from the
    command line. Conversation state is preserved across turns. Enter ``exit``
    or ``quit`` to stop the loop.

    Args:
        agent: The starting agent to run.
        stream: Whether to stream the agent output.
        context: Additional context information to pass to the runner.
        max_turns: Maximum number of turns for the runner to iterate.
    """

    current_agent = agent
    input_items: list[TResponseInputItem] = []
    while True:
        try:
            user_input = input(" > ")
        except (EOFError, KeyboardInterrupt):
            print()
            break
        if user_input.strip().lower() in {"exit", "quit"}:
            break
        if not user_input:
            continue

        input_items.append({"role": "user", "content": user_input})

        result: RunResultBase
        if stream:
            result = Runner.run_streamed(
                current_agent, input=input_items, context=context, max_turns=max_turns
            )
            async for event in result.stream_events():
                if isinstance(event, RawResponsesStreamEvent):
                    if isinstance(event.data, ResponseTextDeltaEvent):
                        print(event.data.delta, end="", flush=True)
                elif isinstance(event, RunItemStreamEvent):
                    if event.item.type == "tool_call_item":
                        print("\n[tool called]", flush=True)
                    elif event.item.type == "tool_call_output_item":
                        print(f"\n[tool output: {event.item.output}]", flush=True)
                elif isinstance(event, AgentUpdatedStreamEvent):
                    print(f"\n[Agent updated: {event.new_agent.name}]", flush=True)
            print()
        else:
            result = await Runner.run(
                current_agent, input_items, context=context, max_turns=max_turns
            )
            if result.final_output is not None:
                print(result.final_output)

        current_agent = result.last_agent
        input_items = result.to_input_list()