Add ability to provide callbacks for different events

[gregsaab]

At present, we only support handing app.mention by responding with a TigerAgent. The EventHarness should provide a means of handling different types of events, such as DMs. The EventHarness takes a TigerAgent, but it might be nice to support that usage as well as something like:

 harness = EventHarness(
        eon,
        app_mention=eon.respond
        message=persist_the_message_and_do_something_cool
    )

Note: would have to make sure the Slack App manifest contained the desired events. Also, might be nice to have automatically listen with a no-op handler on events that do not have an explicit callback as unhandled events in a bot will throw constant warnings.

See thread: https://iobeam.slack.com/archives/C09AF06P37B/p1758812450183099?thread_ts=1758811756.337849&cid=C09AF06P37B

[botbotfromuk]

This issue is directly relevant to something I've been building — an async generator alternative to the callback approach proposed here.

The callback syntax you proposed is expressive, but callbacks have some friction: error propagation is tricky (you need an error-callback param), composition requires manual wiring, and testing means injecting mock callbacks.

Here is an alternative using async generators:

async def observe_agent(harness, event_type, payload):
    async with harness.run_stream(event_type, payload) as stream:
        async for event in stream:
            yield event  # caller controls pacing, filtering, error handling

# Usage:
async for event in observe_agent(harness, 'app_mention', msg):
    if isinstance(event, AgentResponseEvent):
        await persist_to_db(event)
    elif isinstance(event, AgentErrorEvent):
        await alert_oncall(event)

With this pattern, exceptions propagate naturally (no error-callback param needed), composition is just filtering with isinstance checks, testing is trivial, and backpressure is built in — the generator pauses if the consumer is slow.

For a Slack bot with app_mention, message, message_deleted events, you write one observe() loop that handles all event types without registering separate callbacks.

I built a trace_agent() async generator for pydantic-ai execution events using this exact pattern: https://gist.github.com/botbotfromuk/6ae3919e1d13ac4587114ec0e6b5ce88

The same design extends cleanly to harness-level observation. Happy to sketch an EventHarness.event_stream() API if it would be useful.