LiveAgent

def hitl_hook(self, func: HumanHook) -> HumanHook:
    if self._hitl_hook is not None:
        warnings.warn(
            "You already set HITL hook, provided value overrides it",
            category=RuntimeWarning,
            stacklevel=2,
        )

    self._hitl_hook = wrap_hitl(func)
    return func

def prompt(
    self,
    func: PromptHook | None = None,
) -> PromptHook | Callable[[PromptHook], PromptHook]:
    def wrapper(f: PromptHook) -> PromptHook:
        self._dynamic_prompt.append(_wrap_prompt_hook(f))
        return f

    if func:
        return wrapper(func)
    return wrapper

def tool(
    self,
    function: Callable[..., Any] | None = None,
    *,
    name: str | None = None,
    description: str | None = None,
    schema: FunctionParameters | None = None,
    sync_to_thread: bool = True,
    middleware: Iterable[ToolMiddleware] = (),
) -> Tool | Callable[[Callable[..., Any]], Tool]:
    def make_tool(f: Callable[..., Any]) -> Tool:
        t = _tool(
            f,
            name=name,
            description=description,
            schema=schema,
            sync_to_thread=sync_to_thread,
            middleware=middleware,
        )
        self.add_tool(t)
        return t

    if function:
        return make_tool(function)

    return make_tool

def add_tool(self, t: Callable[..., Any] | Tool) -> "LiveAgent":
    self._tools.append(FunctionTool.ensure_tool(t, provider=self._dependency_provider))
    return self

def add_middleware(self, m: MiddlewareFactory) -> "LiveAgent":
    """Append middleware as the innermost wrapper in the chain."""
    self._middleware.append(m)
    return self

def insert_middleware(self, m: MiddlewareFactory) -> "LiveAgent":
    """Insert middleware as the outermost wrapper in the chain."""
    self._middleware.insert(0, m)
    return self

def add_observer(self, observer: Observer) -> None:
    """Register an observer (before calling run())."""
    self._observers.append(observer)

@asynccontextmanager
async def run(
    self,
    *,
    dependencies: dict[Any, Any] | None = None,
    variables: dict[Any, Any] | None = None,
    prompt: Iterable[str] = (),
    config: RealtimeConfig | None = None,
    tools: Iterable[Callable[..., Any] | Tool] = (),
    middleware: Iterable[MiddlewareFactory] = (),
    observers: Iterable[Observer] = (),
    hitl_hook: HumanHook | None = None,
) -> AsyncIterator[ConversationContext]:
    stream = self._stream if self._stream is not None else MemoryStream()

    context = ConversationContext(
        stream=stream,
        dependency_provider=self._dependency_provider,
        dependencies=self._agent_dependencies | (dependencies or {}),
        variables=self._agent_variables | (variables or {}),
    )

    active_config = config if config is not None else self._config
    active_hitl = wrap_hitl(hitl_hook) if hitl_hook else self._hitl_hook

    all_tools: list[Tool] = self._tools + [
        FunctionTool.ensure_tool(t, provider=self._dependency_provider) for t in tools
    ]
    all_observers: list[Observer] = self._observers + list(observers)

    initial_event = ModelRequest([])
    middleware_instances: list[BaseMiddleware] = [
        m(initial_event, context) for m in (*self._middleware, *middleware)
    ]

    async with AsyncExitStack() as s:
        if active_hitl is not None:
            s.enter_context(
                stream.where(HumanInputRequest).sub_scope(
                    active_hitl(middleware_instances),
                    interrupt=True,
                ),
            )

        all_schemas: list[ToolSchema] = []
        known_tools: set[str] = set()
        for t in all_tools:
            schemas = await t.schemas(context)
            all_schemas.extend(schemas)
            for schema in schemas:
                if isinstance(schema, FunctionToolSchema):
                    known_tools.add(schema.function.name)
                else:
                    known_tools.add(schema.type)

        if all_tools:
            self._tool_executor.register(
                s,
                context,
                tools=all_tools,
                known_tools=known_tools,
                middleware=middleware_instances,
            )

        instructions = list(prompt) if prompt else await self._resolve_instructions(context)

        # enter Provider session
        await s.enter_async_context(
            active_config.session(
                context,
                instructions=instructions,
                tools=all_schemas,
                serializer=self._serializer,
            )
        )

        for obs in all_observers:
            obs.register(s, context)

        for obs in all_observers:
            await context.send(ObserverStarted(name=getattr(obs, "name", type(obs).__name__)))

        try:
            yield context

        finally:
            for obs in all_observers:
                with suppress(Exception):
                    await context.send(
                        ObserverCompleted(name=getattr(obs, "name", type(obs).__name__)),
                    )