Send custom events from tool function, for AG-UI and event_stream_handler

[kadosh1000]

Access to Event Bus for Runtime Events Without Tool Activation

Summary

I need to send events to the ag_ui frontend during agent execution (progress updates, intermediate results, etc.) without having to create tools specifically for event emission. The event bus functionality appears to be private in the current implementation.

Use Case

I want to provide real-time feedback to users during long-running agent operations:

• Progress indicators during multi-step processes
• Intermediate results before final completion
• Status updates during external API calls
• Debug information during development

What I've Tried

from pydantic_ai import Agent, RunContext

agent = Agent('openai:gpt-4')

@agent.tool
def long_running_task(ctx: RunContext, task_params: str) -> str:
    # I want to emit progress events here
    # But can't access the event bus directly
    
    for i in range(10):
        # Would like to emit: {"type": "progress", "value": i/10}
        # Currently no way to access ag_ui's event system
        time.sleep(1)  # Simulate work
        
    return "Task completed"

app = agent.to_ag_ui()

Looking at the source code, I can see there's event bus functionality, but it seems to be private/internal.

Expected Behavior

Access to emit custom events during agent execution:

@agent.tool
def long_running_task(ctx: RunContext, task_params: str) -> str:
    event_bus = ctx.get_event_bus()  # or similar API
    
    for i in range(10):
        event_bus.emit(CustomEvent(
            type=EventType.CUSTOM,
            name='count',
            value=i,
        ))
        time.sleep(1)
        
    return "Task completed"

Or maybe support "Generator tools"?

@agent.generator_tool
def long_running_task(ctx: RunContext, task_params: str) -> Generator[CustomEvent, None, str]:
    for i in range(10):
        yield CustomEvent(
            type=EventType.CUSTOM,
            name='count',
            value=i,
        )
        time.sleep(1)
        
    return "Task completed"

Actual Behavior

No apparent way to access the event bus from within agent tools or system prompts, making real-time user feedback impossible without creating dummy tools just for event emission.

Questions

1. Is there a current way to access the event bus from within agent execution?
2. If not, could this functionality be exposed through the RunContext or similar mechanism?
3. Would you be open to making the event bus API public for custom event emission?
4. What would be the preferred API design for this feature?

Additional Context

This would greatly improve user experience for long-running agents by providing immediate feedback rather than waiting for completion. Many production use cases require progress indication for user engagement and debugging purposes.

Additional Context

• pydantic-ai version: 0.4.9
• Python version: 3.14

[DouweM]

@kadosh1000 I agree it would be nice to have a built-in way to do this.

In the meantime, here's a workaround using https://anyio.readthedocs.io/en/stable/streams.html, which creates a new stream that gets all the events coming from our AG-UI integration and also lets you write events yourself:

import asyncio
import uuid

from ag_ui.core import CustomEvent, RunAgentInput, UserMessage
from ag_ui.encoder import EventEncoder
from anyio import create_memory_object_stream, create_task_group
from anyio.streams.memory import MemoryObjectSendStream

from pydantic_ai import Agent
from pydantic_ai._run_context import RunContext
from pydantic_ai.ag_ui import SSE_CONTENT_TYPE, run_ag_ui

agent = Agent('openai:gpt-4.1', deps_type=MemoryObjectSendStream[str])


@agent.tool
async def send_events(ctx: RunContext[MemoryObjectSendStream[str]]) -> None:
    encoder = EventEncoder(accept=SSE_CONTENT_TYPE)
    await ctx.deps.send(encoder.encode(CustomEvent(name='count', value=0)))
    await ctx.deps.send(encoder.encode(CustomEvent(name='count', value=1)))
    await ctx.deps.send(encoder.encode(CustomEvent(name='count', value=2)))


async def main():
    run_input = RunAgentInput(
        thread_id=uuid.uuid4().hex,
        run_id=uuid.uuid4().hex,
        messages=[
            UserMessage(
                id='msg_1',
                content='Call the send_events tool',
            )
        ],
        state={},
        context=[],
        tools=[],
        forwarded_props=None,
    )

    send_stream, receive_stream = create_memory_object_stream[str]()

    async def run_agent(send_stream: MemoryObjectSendStream[str]) -> None:
        async with send_stream:
            event_stream = run_ag_ui(agent, run_input, deps=send_stream)
            async for event in event_stream:
                await send_stream.send(event)

    async with create_task_group() as tg:
        tg.start_soon(run_agent, send_stream)

        async for event in receive_stream:
            print(event)


asyncio.run(main())

This uses the new run_ag_ui I'm introducing in #2397 to address #2377.

Of course I'm faking the AgentRunInput and just printing the events instead of putting them into a streaming response. In reality, it would look more like this:

import json
from http import HTTPStatus

from ag_ui.core import CustomEvent, RunAgentInput
from ag_ui.encoder import EventEncoder
from anyio import create_memory_object_stream, create_task_group
from anyio.streams.memory import MemoryObjectSendStream
from fastapi import FastAPI
from fastapi.requests import Request
from fastapi.responses import Response, StreamingResponse
from pydantic import ValidationError

from pydantic_ai import Agent
from pydantic_ai._run_context import RunContext
from pydantic_ai.ag_ui import SSE_CONTENT_TYPE, run_ag_ui

agent = Agent('openai:gpt-4.1', deps_type=MemoryObjectSendStream[str])


@agent.tool
async def send_events(ctx: RunContext[MemoryObjectSendStream[str]]) -> None:
    encoder = EventEncoder(accept=SSE_CONTENT_TYPE)
    await ctx.deps.send(encoder.encode(CustomEvent(name='count', value=0)))
    await ctx.deps.send(encoder.encode(CustomEvent(name='count', value=1)))
    await ctx.deps.send(encoder.encode(CustomEvent(name='count', value=2)))


app = FastAPI()


@app.post('/')
async def run(request: Request) -> Response:
    accept = request.headers.get('accept', SSE_CONTENT_TYPE)
    try:
        run_input = RunAgentInput.model_validate(await request.json())
    except ValidationError as e:  # pragma: no cover
        return Response(
            content=json.dumps(e.json()),
            media_type='application/json',
            status_code=HTTPStatus.UNPROCESSABLE_ENTITY,
        )

    send_stream, receive_stream = create_memory_object_stream[str]()

    async def run_agent(send_stream: MemoryObjectSendStream[str]) -> None:
        async with send_stream:
            event_stream = run_ag_ui(agent, run_input, deps=send_stream)
            async for event in event_stream:
                await send_stream.send(event)

    async with create_task_group() as tg:
        tg.start_soon(run_agent, send_stream)

        return StreamingResponse(receive_stream, media_type=accept)

Does that help?

The native solution would likely provide the event bus to the agent's tools through ctx.deps.event_bus or something similar, and would let you directly write BaseEvents instead of needing to encode them yourself.

Let me know if that approach works for you though!

[kadosh1000]

Thanks @DouweM !
I will give it a try this week and let you know.

[jiachengzz]

For anyone interested in the FastAPI streaming solution, if the call with streaming is stuck without responding anything, the following should work.

......

@app.post('/')
async def run(request: Request) -> Response:
    accept = request.headers.get('accept', SSE_CONTENT_TYPE)
    try:
        run_input = RunAgentInput.model_validate(await request.json())
    except ValidationError as e:  # pragma: no cover
        return Response(
            content=json.dumps(e.json()),
            media_type='application/json',
            status_code=HTTPStatus.UNPROCESSABLE_ENTITY,
        )

    send_stream, receive_stream = create_memory_object_stream[str]()

    async def run_agent(send_stream: MemoryObjectSendStream[str]) -> None:
        async with send_stream:
            event_stream = run_ag_ui(agent, run_input, deps=send_stream)
            async for event in event_stream:
                await send_stream.send(event)

    async def stream_generator():
        async with create_task_group() as tg:
            tg.start_soon(run_agent, send_stream)
            async with receive_stream:
                async for event in receive_stream:
                    yield event

    return StreamingResponse(stream_generator(), media_type=accept)

[kadosh1000]

Thanks @jiachengzz
@DouweM is the new event_stream_handler helping with that?
If so, can you maybe provide an example how to use it in this context?

[DouweM]

@kadosh1000 Dropping the FastAPI and AG-UI stuff, in this example agent_stream yield all the agent's events, but by going through through a custom MemoryObjectSendStream and event_stream_handler instead of using iter or run_stream:

from collections.abc import AsyncIterable

from anyio import create_memory_object_stream, create_task_group
from anyio.streams.memory import MemoryObjectSendStream

from pydantic_ai import Agent
from pydantic_ai._run_context import RunContext
from pydantic_ai.messages import AgentStreamEvent, HandleResponseEvent

Event = AgentStreamEvent | HandleResponseEvent

agent = Agent('openai:gpt-4.1', deps_type=MemoryObjectSendStream[Event])


async def event_stream_handler(
    ctx: RunContext[MemoryObjectSendStream[Event]],
    event_stream: AsyncIterable[Event],
) -> None:
    send_stream = ctx.deps
    async for event in event_stream:
        await send_stream.send(event)


async def run_agent(send_stream: MemoryObjectSendStream[Event]) -> None:
    async with send_stream:
        await agent.run('Count to 10', deps=send_stream, event_stream_handler=event_stream_handler)


async def agent_stream():
    send_stream, receive_stream = create_memory_object_stream[Event]()

    async with create_task_group() as tg:
        tg.start_soon(run_agent, send_stream)

        async with receive_stream:
            async for event in receive_stream:
                yield event

That means you can write events from multiple agents onto the same event stream, as well as custom events.