Bases: Toolkit
Expose the tools of an MCP server as ordinary local tools.
Accepts either:
- a URL string or :class:
MCPServerConfig for a remote (streamable-http) server, or - an :class:
MCPStdioServerConfig for a locally-launched server communicating over stdin/stdout.
Tool discovery is lazy: the first call to :meth:schemas performs the MCP handshake, lists the server's tools, and registers a proxy for each one. The agent never sees that these are MCP tools — they look and behave like ordinary :class:FunctionTool instances.
Source code in autogen/beta/tools/toolkits/mcp_server/toolkit.py
| def __init__(
self,
server: str | MCPServerConfig | MCPStdioServerConfig,
*,
middleware: Iterable[ToolMiddleware] = (),
) -> None:
if isinstance(server, str):
server = MCPServerConfig(server_url=server)
self.config: AnyMCPConfig = server
self._discovered = False
self._discover_lock = asyncio.Lock()
label = server.server_label if isinstance(server.server_label, str) else ""
super().__init__(
name=label or "mcp_toolkit",
middleware=middleware,
)
|
Source code in autogen/beta/tools/toolkits/mcp_server/toolkit.py
| async def schemas(self, context: "Context") -> Iterable[FunctionToolSchema]:
await self._discover_tools(context)
return await super().schemas(context)
|
Source code in autogen/beta/tools/final/toolkit.py
| def set_provider(self, provider: Provider) -> None:
for t in self.tools:
t.set_provider(provider)
|
register(stack, context, *, middleware=())
Source code in autogen/beta/tools/final/toolkit.py
| def register(
self,
stack: "ExitStack | AsyncExitStack",
context: "Context",
*,
middleware: Iterable["BaseMiddleware"] = (),
) -> None:
for t in self.tools:
t.register(stack, context, middleware=middleware)
|
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/tools/final/toolkit.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._add_tool(t)
return t
if function:
return make_tool(function)
return make_tool
|