GeoAgent stream_chat + ChatGPT/Codex OAuth¶

This notebook demonstrates streaming model output from agent.stream_chat(...) with the openai-codex provider.

• Install: pip install "GeoAgent[openai]".
• Run the login cell once. Later notebooks can reuse the stored ChatGPT/Codex login automatically.
• Jupyter supports top-level await, so the streaming cell can use async for directly.

In [ ]:

%pip install -q "GeoAgent[openai]"

%pip install -q "GeoAgent[openai]"

In [ ]:

from geoagent import ensure_openai_codex_environment
from geoagent import login_openai_codex

try:
    ensure_openai_codex_environment()
    print("Using stored ChatGPT/Codex login.")
except RuntimeError:
    login_openai_codex()
    print("Logged in with ChatGPT/Codex OAuth.")

from geoagent import ensure_openai_codex_environment from geoagent import login_openai_codex try: ensure_openai_codex_environment() print("Using stored ChatGPT/Codex login.") except RuntimeError: login_openai_codex() print("Logged in with ChatGPT/Codex OAuth.")

In [ ]:

import os

MODEL = os.environ.get("OPENAI_CODEX_MODEL", "gpt-5.5")
print("Using model:", MODEL)

import os MODEL = os.environ.get("OPENAI_CODEX_MODEL", "gpt-5.5") print("Using model:", MODEL)

In [ ]:

from geoagent import GeoAgent, GeoAgentConfig

agent = GeoAgent(
    config=GeoAgentConfig(
        provider="openai-codex",
        model=MODEL,
        temperature=0.0,
        max_tokens=2048,
    ),
)

from geoagent import GeoAgent, GeoAgentConfig agent = GeoAgent( config=GeoAgentConfig( provider="openai-codex", model=MODEL, temperature=0.0, max_tokens=2048, ), )

Stream a response¶

In [ ]:

chunks = []
tool_names = set()
final_result = None

async for event in agent.stream_chat(
    "Explain GeoAgent streaming in three concise bullets."
):
    if "current_tool_use" in event:
        tool_name = event["current_tool_use"].get("name")
        if tool_name:
            tool_names.add(tool_name)
            print(f"\n[Using tool: {tool_name}]\n", flush=True)

    if "data" in event:
        chunks.append(event["data"])
        print(event["data"], end="", flush=True)

    if "result" in event:
        final_result = event["result"]

print("\n\nstreamed_chars:", len("".join(chunks)))
print("tools:", sorted(tool_names))
if final_result is not None:
    print("stop_reason:", getattr(final_result, "stop_reason", None))

chunks = [] tool_names = set() final_result = None async for event in agent.stream_chat( "Explain GeoAgent streaming in three concise bullets." ): if "current_tool_use" in event: tool_name = event["current_tool_use"].get("name") if tool_name: tool_names.add(tool_name) print(f"\n[Using tool: {tool_name}]\n", flush=True) if "data" in event: chunks.append(event["data"]) print(event["data"], end="", flush=True) if "result" in event: final_result = event["result"] print("\n\nstreamed_chars:", len("".join(chunks))) print("tools:", sorted(tool_names)) if final_result is not None: print("stop_reason:", getattr(final_result, "stop_reason", None))