Run an agent and read its event stream

When you need this. Terminal chat is enough to try an agent. A documentation portal needs application code that receives a question, manages the conversation and decides which response to display. We will give our docuctl assistant a Python Runner and inspect the events it returns.

The assistant now has a read_page tool. Instead of sending every document in its instruction, it asks for the page it needs. This produces a useful sequence to inspect: a user message, a tool call, a tool result and an answer.

Give the assistant a tool to read pages

Download the lab using Download code, extract it, and open adk-event-stream-lesson. Use Python 3.12 or later. The lab’s docuctl pages describe a fictional CLI; you will read documentation, not execute a deployment.

python3 -m venv .venv
source .venv/bin/activate
python -m pip install -r requirements.txt
cp .env.example .env

Set your Gemini key in the private .env. ask.py loads it before importing the agent; exported shell values take precedence. The same root .env convention is used by the remaining labs in this module.

First, read the page the assistant will use:

# Roll back a release

`docuctl rollback` moves an environment back to an earlier release. It does not rebuild anything; it points traffic at the release you name.

    docuctl rollback --env production --to r41

Without `--to`, docuctl rolls back to the previous release. A rollback is itself recorded as a new release, so `docuctl releases` shows it.

Database migrations are not rolled back. If the release you leave ran a migration, write a new migration that reverses it.

docuctl 1.x used `docuctl release revert <release>`. That command was removed in docuctl 2.0.

Then attach a Python function that reads one of the four bundled pages:

"""The docs assistant with one tool, so its event stream has calls, results and an answer."""
import os
from pathlib import Path

from google.adk import Agent
from google.adk.models.lite_llm import LiteLlm
from google.genai import types

# Gemini through ADK's native client; set DOCS_MODEL=ollama_chat/gemma4:e2b to run Gemma locally.
MODEL = os.environ.get("DOCS_MODEL", "gemini-3.8-flash")
DOCS = Path(__file__).resolve().parent.parent / "docs"
PAGES = sorted(p.stem for p in DOCS.glob("*.md"))


def read_page(page: str) -> dict:
    """Read one page of the docuctl documentation.

    Args:
        page: The page name: config, deploy, rollback or tokens.
    """
    if page not in PAGES:
        return {"error": f"No page {page!r}. Pages: {', '.join(PAGES)}."}
    return {"page": page, "text": (DOCS / f"{page}.md").read_text(encoding="utf-8")}


root_agent = Agent(
    name="docs_assistant",
    model=LiteLlm(model=MODEL) if MODEL.startswith("ollama_chat/") else MODEL,
    description="Answers developer questions about docuctl from its documentation.",
    instruction="""You answer questions from developers about docuctl, the team's deploy CLI.
Read the relevant page with read_page before you answer, and answer only from it.
Copy commands exactly, in backticks. Keep the answer under 100 words.""",
    tools=[read_page],
    generate_content_config=types.GenerateContentConfig(
        temperature=0.2,
        max_output_tokens=1024,
        thinking_config=types.ThinkingConfig(thinking_level="low"),
    ),
)

ADK derives the tool declaration from the function’s signature and docstring. The model proposes read_page(page="rollback"); Python reads that file and returns its text. A model turn can then use the result to answer. The agent requests LOW thinking and a maximum of 1,024 output tokens per response.

Run the agent with a Runner

InMemorySessionService stores the messages and state for each conversation. Runner connects that service to an agent. Create a session before calling run_async, and pass its ID with every new message.

The script’s entrypoint makes those choices explicitly:

async def main() -> None:
    parser = argparse.ArgumentParser()
    parser.add_argument("question")
    parser.add_argument("--stream", action="store_true", help="stream text chunks as partial events")
    parser.add_argument("--max-llm-calls", type=int, default=10)
    args = parser.parse_args()

    sessions = InMemorySessionService()
    runner = Runner(app_name=APP, agent=root_agent, session_service=sessions)
    session = await sessions.create_session(app_name=APP, user_id=USER)
    config = RunConfig(streaming_mode=StreamingMode.SSE if args.stream else StreamingMode.NONE,
                       max_llm_calls=args.max_llm_calls)
    try:
        events = await ask(runner, session.id, args.question, config)
    finally:
        await runner.close()
    for event in events:
        print(describe(event))
    stored = await sessions.get_session(app_name=APP, user_id=USER, session_id=session.id)
    print(f"yielded {len(events)} events, session stores {len(stored.events)}")
    print("answer:", answer(events))

RunConfig applies to this invocation. Here it chooses streaming and a model-call limit. The runner closes in finally, including when an invocation raises an error.

Run one question from the lab directory:

python ask.py "How do I undo the last production deploy?" --max-llm-calls 2

The first three event lines in the supplied saved example are:

Recorded example. The archived “Run an agent and read its event stream” demonstration retains the earlier shipctl scenario; the current lab uses docuctl and does not claim a new model result.

user            final    'How do I undo the last production deploy?'
docs_assistant  event    call read_page({'page': 'rollback'})
docs_assistant  event    result read_page -> ['page', 'text']

The final event in the original recording carries the answer. Check that it describes shipctl rollback --env production and preserves the page’s warning about database migrations. Run completion alone cannot tell you whether the suggested command is correct.

An unknown session ID. Passing an ID that the session service cannot find raises SessionNotFoundError before the turn starts. Create the session first, or deliberately configure automatic creation for your application.

The max_llm_calls limit. With a limit of one, the model can request read_page and the tool can run, but the next model call raises LlmCallsLimitExceededError. A call limit does not undo a tool’s side effects. If a future tool writes data, a retry must account for the write already having happened.

Pick the answer out of the stream

run_async yields Event objects. Our ask() function collects them; answer() selects the last final response from an author other than user:

async def ask(runner: Runner, session_id: str, question: str, config: RunConfig) -> list[Event]:
    """Send one user message and collect the events of that invocation, in order."""
    message = types.Content(role="user", parts=[types.Part(text=question)])
    events = []
    async for event in runner.run_async(user_id=USER, session_id=session_id, new_message=message,
                                        run_config=config, yield_user_message=True):
        events.append(event)
    return events


def answer(events: list[Event]) -> str:
    """The answer is the text of the agent's last final response, not every text chunk.

    is_final_response() is also True for the user's own message, so filter by author.
    """
    final = [e for e in events if e.is_final_response() and e.author != "user" and e.content]
    return "".join(p.text or "" for p in final[-1].content.parts if not p.thought) if final else ""

yield_user_message=True also yields the user’s stored message. That message can satisfy is_final_response(), because it is neither a partial response nor a tool call. The author check keeps the question from becoming its own answer.

The user’s message counted as the answer. Taking the first event for which is_final_response() is true can return the question. Check the author and use the agent’s final response.

Try the streaming variant:

python ask.py "How do I undo the last production deploy?" --stream --max-llm-calls 2

Streamed text shown twice. Partial events carry chunks, and the final event can carry the complete text again. A UI should render chunks as they arrive, then replace that draft with the final text. Printing both as new text duplicates the answer. This script collects the events before printing them so you can inspect their order; it is not a live-streaming UI.

Follow model and tool events

Use author to identify the speaker and invocation_id to group events from one turn. Tool-call and tool-response parts share an ID, which lets you match a result to its call. In serialized JSON these names appear as invocationId, functionCall and functionResponse.

The call event comes from the model. The tool-result event has the agent’s author name but user-role content, so Gemini receives the function result on the input side of the conversation. ADK then sends that history into the next model request.

In the supplied streaming example, partial events were yielded to the caller but were not appended to the session. Inspect the script’s final yielded ... events, session stores ... line to compare the two lists. Keep provider-managed fields intact when passing stored events back to the model.

You can also open the package with adk web --host 127.0.0.1 --port 8000 ., select docs_assistant, and inspect the tool call in Events. It uses a separate session; it does not run the event-printing code in ask.py.

Test the runner without a model

python -m unittest -v test_ask.py

The tests feed saved model responses through the real Runner, tool and session service. They check answer selection, partial-event storage, missing sessions and the call-limit failure without making a provider request.

Serve many conversations from one runner

Keep one Runner per application process and give each conversation its own session. Serialize turns within a session so two requests do not append competing histories. In-memory sessions disappear when the process exits; the state-and-memory lesson replaces that service with SQLite.

Next, give the assistant typed function tools and validate proposed arguments before a write.

References

• ADK runtime and event loop: Runner and invocation lifecycle.
• ADK events: content, actions and event helpers.
• Runtime configuration: streaming and run settings.