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 add_policy(self, policy: AssemblyPolicy) -> "Agent[TResult]":
    """Append an assembly policy to this agent's chain.

    Policies run in order; a newly added policy runs after existing
    ones. Construction-time ordering validation (warning on suspicious
    sequences) only runs over policies passed via ``assembly=`` — late
    additions skip the check, so callers should be confident in the
    ordering they introduce.
    """
    self._policies.append(policy)
    return self

def task(
    self,
    title: str,
    *,
    description: str = "",
    payload: dict[str, Any] | None = None,
    capability: str | None = None,
    ttl_seconds: int | None = None,
    context: Context | None = None,
) -> Task:
    """Create a ``Task`` whose lifecycle this Agent owns.

    Returns an unentered ``Task``; use as ``async with agent.task(...)``.
    Events flow on ``context.stream`` if a ``ConversationContext`` is
    supplied; else the Task creates a private ``MemoryStream`` on entry
    and events fire on it (only observers attached to that private
    stream see them).

    Inside the ``async with`` block, ``ag2.task`` is stamped into
    ``context.dependencies`` so any tool annotated with ``TaskInject``
    resolves to this Task.

    ``capability`` tags the task with a capability name. When the
    agent is registered with the network, the ``TaskMirror`` calls
    ``Hub.record_observation`` on the terminal event so the matching
    ``Resume.observed[capability]`` track record updates.
    """
    spec = TaskSpec(
        title=title,
        description=description,
        payload=dict(payload) if payload else {},
        capability=capability,
    )
    return Task(
        owner_id=self.name,
        spec=spec,
        context=context,
        ttl_seconds=ttl_seconds,
    )

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)