NetworkPlugin

autogen.beta.network.client.plugin.NetworkPlugin #

NetworkPlugin(client)

Bases: Plugin

Attaches an Agent to a network.

Adds say and delegate to agent.tools so the LLM sees them on every turn — the verbs are stable for the life of the registration. Also appends NetworkContextPolicy to the agent's assembly chain.

Source code in autogen/beta/network/client/plugin.py

def __init__(self, client: "AgentClient") -> None:
    super().__init__(
        tools=[
            make_say_tool(client),
            make_delegate_tool(client),
            make_peers_tool(client),
            make_channels_tool(client),
            make_tasks_tool(client),
            make_context_tool(client),
        ],
    )
    self._client = client

register #

register(agent)

Wire tools + assembly policy onto the agent. Idempotent-ish.

Calling register more than once on the same agent will add the tools / policies again. HubClient.register only attaches once per (Agent, identity), so this is rare in practice.

Source code in autogen/beta/network/client/plugin.py

def register(self, agent: "Agent") -> None:
    """Wire tools + assembly policy onto the agent. Idempotent-ish.

    Calling ``register`` more than once on the same agent will add
    the tools / policies again. ``HubClient.register`` only attaches
    once per ``(Agent, identity)``, so this is rare in practice.
    """
    super().register(agent)
    agent.add_policy(NetworkContextPolicy(self._client))

hitl_hook #

hitl_hook(func)

Source code in autogen/beta/agent.py

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 = func
    return func

prompt #

prompt(func: PromptHook) -> PromptHook

prompt(func: None = None) -> Callable[[PromptHook], PromptHook]

prompt(func=None)

Source code in autogen/beta/agent.py

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

tool #

tool(function: Callable[..., Any], *, name: str | None = None, description: str | None = None, schema: FunctionParameters | None = None, sync_to_thread: bool = True, middleware: Iterable[ToolMiddleware] = ()) -> FunctionTool

tool(function: None = None, *, name: str | None = None, description: str | None = None, schema: FunctionParameters | None = None, sync_to_thread: bool = True, middleware: Iterable[ToolMiddleware] = ()) -> Callable[[Callable[..., Any]], FunctionTool]

tool(function=None, *, name=None, description=None, schema=None, sync_to_thread=True, middleware=())

Source code in autogen/beta/agent.py

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] = (),
) -> FunctionTool | Callable[[Callable[..., Any]], FunctionTool]:
    def make_tool(f: Callable[..., Any]) -> FunctionTool:
        t = tool(
            f,
            name=name,
            description=description,
            schema=schema,
            sync_to_thread=sync_to_thread,
            middleware=middleware,
        )
        self._tools.append(t)
        return t

    if function:
        return make_tool(function)
    return make_tool

observer #

observer(condition: ClassInfo | Condition | None, callback: Callable[..., Any]) -> Callable[..., Any]

observer(condition: ClassInfo | Condition | None = None) -> Callable[[Callable[..., Any]], Callable[..., Any]]

observer(condition=None, callback=None)

Source code in autogen/beta/agent.py

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