Function dispatch_hook

libs/code/deepagents_code/hooks.py:161–187

Fire matching hook commands with `payload` serialized as JSON on stdin. The `event` name is automatically injected into the payload under the `"event"` key so callers don't need to duplicate it. The blocking subprocess work is offloaded to a thread so the caller's event loop is nev

(event: str, payload: dict[str, Any])

async def dispatch_hook(event: str, payload: dict[str, Any]) -> None:
    """Fire matching hook commands with `payload` serialized as JSON on stdin.

    The `event` name is automatically injected into the payload under the
    `"event"` key so callers don't need to duplicate it.

    The blocking subprocess work is offloaded to a thread so the caller's
    event loop is never stalled. Matching hooks run concurrently, each with
    a 5-second timeout. Errors are logged and never propagated.

    Args:
        event: Dotted event name (e.g. `'session.start'`).
        payload: Arbitrary JSON-serializable dict sent on the command's stdin.
    """
    try:
        hooks = _load_hooks()
        if not hooks:
            return

        payload_bytes = json.dumps({"event": event, **payload}).encode()
        await asyncio.to_thread(_dispatch_hook_sync, event, payload_bytes, hooks)
    except Exception:
        logger.warning(
            "Unexpected error in dispatch_hook for event %s",
            event,
            exc_info=True,
        )


def dispatch_hook_fire_and_forget(event: str, payload: dict[str, Any]) -> None: