Agent

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 add_middleware(self, m: MiddlewareFactory) -> "Agent[TResult]":
    """Append middleware as the innermost wrapper in the chain.

    The added middleware is called last on turn entry and first on turn exit,
    executing closer to the LLM call than any middleware already registered.
    """
    self._middleware.append(m)
    return self

def insert_middleware(self, m: MiddlewareFactory) -> "Agent[TResult]":
    """Insert middleware as the outermost wrapper in the chain.

    The inserted middleware is called first on turn entry and last on turn exit,
    executing before all middleware already registered on the agent.
    """
    self._middleware.insert(0, m)
    return self

def add_tool(self, t: Callable[..., Any] | Tool) -> "Agent[TResult]":
    self.tools.append(FunctionTool.ensure_tool(t, provider=self.dependency_provider))
    return self

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

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 observer(
    self,
    condition: ClassInfo | Condition | None = None,
    callback: Callable[..., Any] | None = None,
) -> Callable[..., Any] | Callable[[Callable[..., Any]], Callable[..., Any]]:
    def wrapper(func: Callable[..., Any]) -> Callable[..., Any]:
        obs = observer_factory(condition, func)
        self._observers.append(obs)
        return func

    if callback is not None:
        return wrapper(callback)
    return wrapper

async def ask(
    self,
    *msg: str | Input,
    stream: Stream | None = None,
    dependencies: dict[Any, Any] | None = None,
    variables: dict[Any, Any] | None = None,
    prompt: Iterable[str] = (),
    config: ModelConfig | None = None,
    tools: Iterable[Tool] = (),
    middleware: Iterable[MiddlewareFactory] = (),
    observers: Iterable[Observer] = (),
    response_schema: Omittable[ResponseProto[Any] | type | None] = omit,
    hitl_hook: HumanHook | None = None,
) -> "AgentReply[Any, Any]":
    config = config or self.config
    if not config:
        raise ConfigNotProvidedError()
    client = config.create()

    stream = stream or MemoryStream()

    initial_event = ModelRequest.ensure_request(msg)

    context = Context(
        stream,
        prompt=list(prompt),
        dependencies=self._agent_dependencies | (dependencies or {}),
        variables=self._agent_variables | (variables or {}),
        dependency_provider=self.dependency_provider,
    )

    if not context.prompt:
        context.prompt.extend(self._system_prompt)

        for dp in self._dynamic_prompt:
            p = await dp(initial_event, context)
            context.prompt.append(p)

    return await self._execute(
        initial_event,
        context=context,
        client=client,
        hitl_hook=hitl_hook,
        additional_tools=tools,
        additional_middleware=middleware,
        additional_observers=observers,
        response_schema=response_schema,
    )

def as_tool(
    self,
    *,
    description: str,
    name: str | None = None,
    stream: StreamFactory | None = None,
    middleware: Iterable[ToolMiddleware] = (),
) -> FunctionTool:
    return subagent_tool(
        self,
        description=description,
        name=name,
        stream=stream,
        middleware=middleware,
    )

def as_conversable(self) -> "ConversableAdapter":
    # Local import: ``conversable`` imports ``Agent`` from this module —
    # a top-level import would create a circular dependency.
    from .conversable import ConversableAdapter

    return ConversableAdapter(self)