Bases: Plugin
Attaches an Agent to a network.
Adds the identity-level cross-cutting tools to agent.tools: peers / channels / tasks / context / delegate. These are stable for the life of the registration and work in any channel context.
Channel-level tools (say, workflow handoff tools) are NOT attached here — they come from adapter.tools_for(...), resolved per turn by the default notify handler and merged into agent.ask(tools=...). This keeps protocol-specific verbs out of an agent's tool list when they don't apply (e.g. say is hidden from workflow participants who must use handoff tools).
Source code in autogen/beta/network/client/plugin.py
| def __init__(self, client: "AgentClient") -> None:
super().__init__(
tools=[
make_delegate_tool(client),
make_peers_tool(client),
make_channels_tool(client),
make_tasks_tool(client),
make_context_tool(client),
],
)
self._client = client
|
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))
|
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(func: PromptHook) -> PromptHook
prompt(func: None = None) -> Callable[[PromptHook], PromptHook]
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(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(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
|