import asyncio
from omnicoreagent import OmniCoreAgent, ToolRegistry

tools = ToolRegistry()

@tools.register_tool("get_weather")
def get_weather(city: str) -> dict:
    """Get current weather for a city."""
    return {"city": city, "temp": "22C", "condition": "Sunny"}

def show(event):
    # A sub-agent's text arrives here too, with its own agent_name: keep
    # only the lead's (agent_name == "weather_agent") for a clean answer.
    if event["type"] == "text_delta":
        print(event["text"], end="", flush=True)

async def main():
    agent = OmniCoreAgent(
        name="weather_agent",
        system_instruction="Use tools when they help. Answer in plain text, in two short sentences.",
        model_config={"provider": "openai", "model": "gpt-5.4-mini"},
        local_tools=tools,
    )
    result = await agent.run("What's the weather in Tokyo, and should I bring a jacket?", on_event=show)
    print()
    print("status:", result["status"])

    events = await agent.get_telemetry_events_after(cursor=None, run_id=result["run_id"])
    print(len(events), "telemetry events:")
    print(", ".join(event.event_type for event in events))
    await agent.cleanup()

asyncio.run(main())

Tokyo is 22°C and sunny. You probably don’t need a jacket, though a light one could be handy if you’ll be out late.
status: success
33 telemetry events:
user_message, guardrail_check, memory_read, run_configuration, runtime_message, memory_write, agent_step, context_assembly, context_message, context_message, context_tools, model_call, model_response, memory_write, tool_batch_start, tool_requested, tool_resolved, policy_request_created, policy_decision_allow, tool_call, tool_result, tool_observation, memory_write, observation_pipeline_end, tool_batch_end, agent_step, context_assembly, context_message, context_message, model_call, model_response, memory_write, final_answer

import asyncio
from contextlib import aclosing
from omnicoreagent import OmniCoreAgent

async def main():
    agent = OmniCoreAgent(
        name="poet",
        system_instruction="Answer in plain text, without markdown.",
        model_config={"provider": "openai", "model": "gpt-5.4-mini"},
    )
    async with aclosing(agent.stream("Write a haiku about rivers.")) as events:
        async for event in events:
            if event["type"] == "text_delta":
                print(event["text"], end="", flush=True)
            elif event["type"] == "complete":
                print("\n--- complete:", event["status"], "| trace", event["trace_id"][:14])
            elif event["type"] == "error":
                print("\n--- error:", event["error"])
    await agent.cleanup()

asyncio.run(main())

River bends and hums
Silver threads through sleeping stones
Time slips softly by
--- complete: success | trace trace_ab6de575

import asyncio
from omnicoreagent import OmniCoreAgent, ToolRegistry

tools = ToolRegistry()

@tools.register_tool("get_weather")
def get_weather(city: str) -> dict:
    """Get current weather for a city."""
    return {"city": city, "temp": "22C", "condition": "Sunny"}

async def main():
    agent = OmniCoreAgent(
        name="weather_agent",
        system_instruction="Use tools when they help. Answer in plain text, in one sentence.",
        model_config={"provider": "openai", "model": "gpt-5.4-mini"},
        local_tools=tools,
    )
    session_id, run_id = "user_1", "run_ui_1"
    deltas = []

    async def collect(event):  # a coroutine works as well as a plain function
        deltas.append(event["text"])

    cursor = await agent.get_telemetry_stream_cursor(session_id=session_id)
    task = asyncio.create_task(
        agent.run("Weather in Oslo?", session_id=session_id, run_id=run_id, on_event=collect)
    )
    async for event in agent.stream_telemetry_after(
        cursor=cursor,
        session_id=session_id,
        run_id=run_id,
        event_types=("model_call", "tool_call", "tool_result", "final_answer"),
    ):
        print(event.event_type, event.output if event.event_type == "tool_result" else "")
        if event.event_type == "final_answer":
            break

    result = await task
    print(result["response"])
    print(len(deltas), "text deltas")
    await agent.cleanup()

asyncio.run(main())

model_call
tool_call
tool_result {'tool_name': 'get_weather', 'args': {'city': 'Oslo'}, 'status': 'success', 'data': {'city': 'Oslo', 'temp': '22C', 'condition': 'Sunny'}, 'message': None}
model_call
final_answer
Oslo: 22°C, Sunny.
9 text deltas