autogen_ext.tools.mcp#

create_mcp_server_session(server_params: Annotated[StdioServerParams | SseServerParams | StreamableHttpServerParams, FieldInfo(annotation=NoneType, required=True, discriminator='type')], sampling_callback: SamplingFnT | None = None) → AsyncGenerator[ClientSession, None][quelle]#: Erstellt eine MCP-Clientsitzung für die angegebenen Serverparameter.

class McpSessionActor(server_params: Annotated[StdioServerParams | SseServerParams | StreamableHttpServerParams, FieldInfo(annotation=NoneType, required=True, discriminator='type')], model_client: ChatCompletionClient | None = None)[quelle]#

Bases: ComponentBase[BaseModel], Component[McpSessionActorConfig]

component_type: ClassVar[ComponentType] = 'mcp_session_actor'#: Der logische Typ der Komponente.

component_config_schema#: alias von McpSessionActorConfig

component_provider_override: ClassVar[str | None] = 'autogen_ext.tools.mcp.McpSessionActor'#: Überschreibe den Anbieter-String für die Komponente. Dies sollte verwendet werden, um zu verhindern, dass interne Modulnamen Teil des Modulnamens werden.

server_params: Annotated[StdioServerParams | SseServerParams | StreamableHttpServerParams, FieldInfo(annotation=NoneType, required=True, discriminator='type')]#

property initialize_result: InitializeResult | None#

async initialize() → None[quelle]#

async call(type: str, args: McpActorArgs | None = None) → Future[Coroutine[Any, Any, ListToolsResult] | Coroutine[Any, Any, CallToolResult] | Coroutine[Any, Any, ListPromptsResult] | Coroutine[Any, Any, ListResourcesResult] | Coroutine[Any, Any, ListResourceTemplatesResult] | Coroutine[Any, Any, ReadResourceResult] | Coroutine[Any, Any, GetPromptResult]][quelle]#

async close() → None[quelle]#

class StdioMcpToolAdapter(server_params: StdioServerParams, tool: Tool, session: ClientSession | None = None)[quelle]#

Bases: McpToolAdapter[StdioServerParams], Component[StdioMcpToolAdapterConfig]

Ermöglicht die Umhüllung eines MCP-Tools, das über STDIO läuft, und dessen Bereitstellung für AutoGen.

Dieser Adapter ermöglicht die Verwendung von MCP-kompatiblen Tools, die über Standardeingabe/-ausgabe mit AutoGen-Agenten kommunizieren. Gängige Anwendungsfälle sind die Einbindung von Befehlszeilen-Tools und lokalen Diensten, die das Model Context Protocol (MCP) implementieren.

Hinweis

Um diese Klasse zu verwenden, müssen Sie die mcp-Erweiterung für das Paket autogen-ext installieren.

pip install -U "autogen-ext[mcp]"

Parameter:

server_params (StdioServerParams) – Parameter für die MCP-Serververbindung, einschließlich des auszuführenden Befehls und seiner Argumente
tool (Tool) – Das zu umhüllende MCP-Tool
session (ClientSession, optional) – Die zu verwendende MCP-Clientsitzung. Wenn nicht angegeben, wird eine neue Sitzung erstellt. Dies ist nützlich für Tests oder wenn Sie den Lebenszyklus der Sitzung selbst verwalten möchten.

Siehe mcp_server_tools() für Beispiele.

component_config_schema#: alias von StdioMcpToolAdapterConfig

component_provider_override: ClassVar[str | None] = 'autogen_ext.tools.mcp.StdioMcpToolAdapter'#: Überschreibe den Anbieter-String für die Komponente. Dies sollte verwendet werden, um zu verhindern, dass interne Modulnamen Teil des Modulnamens werden.

pydantic model StdioServerParams[quelle]#

Bases: StdioServerParameters

Parameter für die Verbindung zu einem MCP-Server über STDIO.

JSON-Schema anzeigen

{
   "title": "StdioServerParams",
   "description": "Parameters for connecting to an MCP server over STDIO.",
   "type": "object",
   "properties": {
      "command": {
         "title": "Command",
         "type": "string"
      },
      "args": {
         "items": {
            "type": "string"
         },
         "title": "Args",
         "type": "array"
      },
      "env": {
         "anyOf": [
            {
               "additionalProperties": {
                  "type": "string"
               },
               "type": "object"
            },
            {
               "type": "null"
            }
         ],
         "default": null,
         "title": "Env"
      },
      "cwd": {
         "anyOf": [
            {
               "type": "string"
            },
            {
               "format": "path",
               "type": "string"
            },
            {
               "type": "null"
            }
         ],
         "default": null,
         "title": "Cwd"
      },
      "encoding": {
         "default": "utf-8",
         "title": "Encoding",
         "type": "string"
      },
      "encoding_error_handler": {
         "default": "strict",
         "enum": [
            "strict",
            "ignore",
            "replace"
         ],
         "title": "Encoding Error Handler",
         "type": "string"
      },
      "type": {
         "const": "StdioServerParams",
         "default": "StdioServerParams",
         "title": "Type",
         "type": "string"
      },
      "read_timeout_seconds": {
         "default": 5,
         "title": "Read Timeout Seconds",
         "type": "number"
      }
   },
   "required": [
      "command"
   ]
}

Felder:

read_timeout_seconds (float)
type (Literal['StdioServerParams'])

field type: Literal['StdioServerParams'] = 'StdioServerParams'#

field read_timeout_seconds: float = 5#

class SseMcpToolAdapter(server_params: SseServerParams, tool: Tool, session: ClientSession | None = None)[quelle]#

Bases: McpToolAdapter[SseServerParams], Component[SseMcpToolAdapterConfig]

Ermöglicht die Umhüllung eines MCP-Tools, das über Server-Sent Events (SSE) läuft, und dessen Bereitstellung für AutoGen.

Dieser Adapter ermöglicht die Verwendung von MCP-kompatiblen Tools, die über HTTP mit SSE kommunizieren, mit AutoGen-Agenten. Gängige Anwendungsfälle sind die Integration mit entfernten MCP-Diensten, Cloud-basierten Tools und Web-APIs, die das Model Context Protocol (MCP) implementieren.

Hinweis

Um diese Klasse zu verwenden, müssen Sie die mcp-Erweiterung für das Paket autogen-ext installieren.

pip install -U "autogen-ext[mcp]"

Parameter:

server_params (SseServerParameters) – Parameter für die MCP-Serververbindung, einschließlich URL, Header und Timeouts.
tool (Tool) – Das zu umhüllende MCP-Tool.
session (ClientSession, optional) – Die zu verwendende MCP-Clientsitzung. Wenn nicht angegeben, wird eine neue Sitzung erstellt. Dies ist nützlich für Tests oder wenn Sie den Lebenszyklus der Sitzung selbst verwalten möchten.

Examples

Verwenden Sie einen entfernten Übersetzungsdienst, der MCP über SSE implementiert, um Tools zu erstellen, mit denen AutoGen-Agenten Übersetzungen durchführen können

import asyncio
from autogen_ext.models.openai import OpenAIChatCompletionClient
from autogen_ext.tools.mcp import SseMcpToolAdapter, SseServerParams
from autogen_agentchat.agents import AssistantAgent
from autogen_agentchat.ui import Console
from autogen_core import CancellationToken


async def main() -> None:
    # Create server params for the remote MCP service
    server_params = SseServerParams(
        url="https://api.example.com/mcp",
        headers={"Authorization": "Bearer your-api-key", "Content-Type": "application/json"},
        timeout=30,  # Connection timeout in seconds
    )

    # Get the translation tool from the server
    adapter = await SseMcpToolAdapter.from_server_params(server_params, "translate")

    # Create an agent that can use the translation tool
    model_client = OpenAIChatCompletionClient(model="gpt-4")
    agent = AssistantAgent(
        name="translator",
        model_client=model_client,
        tools=[adapter],
        system_message="You are a helpful translation assistant.",
    )

    # Let the agent translate some text
    await Console(
        agent.run_stream(task="Translate 'Hello, how are you?' to Spanish", cancellation_token=CancellationToken())
    )


if __name__ == "__main__":
    asyncio.run(main())

component_config_schema#: alias von SseMcpToolAdapterConfig

component_provider_override: ClassVar[str | None] = 'autogen_ext.tools.mcp.SseMcpToolAdapter'#: Überschreibe den Anbieter-String für die Komponente. Dies sollte verwendet werden, um zu verhindern, dass interne Modulnamen Teil des Modulnamens werden.

pydantic model SseServerParams[quelle]#

Bases: BaseModel

Parameter für die Verbindung zu einem MCP-Server über SSE.

JSON-Schema anzeigen

{
   "title": "SseServerParams",
   "description": "Parameters for connecting to an MCP server over SSE.",
   "type": "object",
   "properties": {
      "type": {
         "const": "SseServerParams",
         "default": "SseServerParams",
         "title": "Type",
         "type": "string"
      },
      "url": {
         "title": "Url",
         "type": "string"
      },
      "headers": {
         "anyOf": [
            {
               "type": "object"
            },
            {
               "type": "null"
            }
         ],
         "default": null,
         "title": "Headers"
      },
      "timeout": {
         "default": 5,
         "title": "Timeout",
         "type": "number"
      },
      "sse_read_timeout": {
         "default": 300,
         "title": "Sse Read Timeout",
         "type": "number"
      }
   },
   "required": [
      "url"
   ]
}

Felder:

headers (dict[str, Any] | None)
sse_read_timeout (float)
timeout (float)
type (Literal['SseServerParams'])
url (str)

field type: Literal['SseServerParams'] = 'SseServerParams'#

field url: str [Erforderlich]#

field headers: dict[str, Any] | None = None#

field timeout: float = 5#

field sse_read_timeout: float = 300#

class StreamableHttpMcpToolAdapter(server_params: StreamableHttpServerParams, tool: Tool, session: ClientSession | None = None)[quelle]#

Bases: McpToolAdapter[StreamableHttpServerParams], Component[StreamableHttpMcpToolAdapterConfig]

Ermöglicht die Umhüllung eines MCP-Tools, das über Streamable HTTP läuft, und dessen Bereitstellung für AutoGen.

Dieser Adapter ermöglicht die Verwendung von MCP-kompatiblen Tools, die über Streamable HTTP mit AutoGen-Agenten kommunizieren. Gängige Anwendungsfälle sind die Integration mit entfernten MCP-Diensten, Cloud-basierten Tools und Web-APIs, die das Model Context Protocol (MCP) implementieren.

Hinweis

Um diese Klasse zu verwenden, müssen Sie die mcp-Erweiterung für das Paket autogen-ext installieren.

pip install -U "autogen-ext[mcp]"

Parameter:

server_params (StreamableHttpServerParams) – Parameter für die MCP-Serververbindung, einschließlich URL, Header und Timeouts.
tool (Tool) – Das zu umhüllende MCP-Tool.
session (ClientSession, optional) – Die zu verwendende MCP-Clientsitzung. Wenn nicht angegeben, wird eine neue Sitzung erstellt. Dies ist nützlich für Tests oder wenn Sie den Lebenszyklus der Sitzung selbst verwalten möchten.

Examples

Verwenden Sie einen entfernten Übersetzungsdienst, der MCP über Streamable HTTP implementiert, um Tools zu erstellen, mit denen AutoGen-Agenten Übersetzungen durchführen können

import asyncio
from autogen_ext.models.openai import OpenAIChatCompletionClient
from autogen_ext.tools.mcp import StreamableHttpMcpToolAdapter, StreamableHttpServerParams
from autogen_agentchat.agents import AssistantAgent
from autogen_agentchat.ui import Console
from autogen_core import CancellationToken


async def main() -> None:
    # Create server params for the remote MCP service
    server_params = StreamableHttpServerParams(
        url="https://api.example.com/mcp",
        headers={"Authorization": "Bearer your-api-key", "Content-Type": "application/json"},
        timeout=30.0,  # HTTP timeout in seconds
        sse_read_timeout=300.0,  # SSE read timeout in seconds (5 minutes)
        terminate_on_close=True,
    )

    # Get the translation tool from the server
    adapter = await StreamableHttpMcpToolAdapter.from_server_params(server_params, "translate")

    # Create an agent that can use the translation tool
    model_client = OpenAIChatCompletionClient(model="gpt-4")
    agent = AssistantAgent(
        name="translator",
        model_client=model_client,
        tools=[adapter],
        system_message="You are a helpful translation assistant.",
    )

    # Let the agent translate some text
    await Console(
        agent.run_stream(task="Translate 'Hello, how are you?' to Spanish", cancellation_token=CancellationToken())
    )


if __name__ == "__main__":
    asyncio.run(main())

component_config_schema#: alias von StreamableHttpMcpToolAdapterConfig

component_provider_override: ClassVar[str | None] = 'autogen_ext.tools.mcp.StreamableHttpMcpToolAdapter'#: Überschreibe den Anbieter-String für die Komponente. Dies sollte verwendet werden, um zu verhindern, dass interne Modulnamen Teil des Modulnamens werden.

pydantic model StreamableHttpServerParams[quelle]#

Bases: BaseModel

Parameter für die Verbindung zu einem MCP-Server über Streamable HTTP.

JSON-Schema anzeigen

{
   "title": "StreamableHttpServerParams",
   "description": "Parameters for connecting to an MCP server over Streamable HTTP.",
   "type": "object",
   "properties": {
      "type": {
         "const": "StreamableHttpServerParams",
         "default": "StreamableHttpServerParams",
         "title": "Type",
         "type": "string"
      },
      "url": {
         "title": "Url",
         "type": "string"
      },
      "headers": {
         "anyOf": [
            {
               "type": "object"
            },
            {
               "type": "null"
            }
         ],
         "default": null,
         "title": "Headers"
      },
      "timeout": {
         "default": 30.0,
         "title": "Timeout",
         "type": "number"
      },
      "sse_read_timeout": {
         "default": 300.0,
         "title": "Sse Read Timeout",
         "type": "number"
      },
      "terminate_on_close": {
         "default": true,
         "title": "Terminate On Close",
         "type": "boolean"
      }
   },
   "required": [
      "url"
   ]
}

Felder:

headers (dict[str, Any] | None)
sse_read_timeout (float)
terminate_on_close (bool)
timeout (float)
type (Literal['StreamableHttpServerParams'])
url (str)

field type: Literal['StreamableHttpServerParams'] = 'StreamableHttpServerParams'#

field url: str [Erforderlich]#

field headers: dict[str, Any] | None = None#

field timeout: float = 30.0#

Feld sse_read_timeout: float = 300.0#

Feld terminate_on_close: bool = True#

async mcp_server_tools(server_params: Annotated[StdioServerParams | SseServerParams | StreamableHttpServerParams, FieldInfo(annotation=NoneType, required=True, discriminator='type')], session: ClientSession | None = None) → list[StdioMcpToolAdapter | SseMcpToolAdapter | StreamableHttpMcpToolAdapter][Quelle]#

Erstellt eine Liste von MCP-Tool-Adaptern, die mit AutoGen-Agenten verwendet werden können.

Warnung

Verbinden Sie sich nur mit vertrauenswürdigen MCP-Servern, insbesondere wenn Sie StdioServerParams verwenden, da dies Befehle in der lokalen Umgebung ausführt.

Diese Factory-Funktion stellt eine Verbindung zu einem MCP-Server her und gibt Adapter für alle verfügbaren Tools zurück. Die Adapter können direkt der Tool-Liste eines AutoGen-Agenten zugewiesen werden.

Hinweis

Um diese Funktion nutzen zu können, müssen Sie die mcp-Erweiterung für das autogen-ext-Paket installieren.

pip install -U "autogen-ext[mcp]"

Parameter:

server_params (McpServerParams) – Verbindungsparameter für den MCP-Server. Kann entweder StdioServerParams für Befehlszeilen-Tools oder SseServerParams und StreamableHttpServerParams für HTTP/SSE-Dienste sein.
session (ClientSession | None) – Optionale vorhandene Sitzung zur Verwendung. Dies wird verwendet, wenn Sie eine bestehende Verbindung zum MCP-Server wiederverwenden möchten. Die Sitzung wird bei der Erstellung der MCP-Tool-Adapter wiederverwendet.

Gibt zurück:

list[StdioMcpToolAdapter | SseMcpToolAdapter | StreamableHttpMcpToolAdapter] – Eine Liste von einsatzbereiten Tool-Adaptern für AutoGen-Agenten.

Examples

Beispiel für einen lokalen Dateisystem-MCP-Dienst über Standard-Ein-/Ausgabe

Installieren Sie das Dateisystem-Server-Paket von npm (erfordert Node.js 16+ und npm).

npm install -g @modelcontextprotocol/server-filesystem

Erstellen Sie einen Agenten, der alle Tools vom lokalen Dateisystem-MCP-Server verwenden kann.

import asyncio
from pathlib import Path
from autogen_ext.models.openai import OpenAIChatCompletionClient
from autogen_ext.tools.mcp import StdioServerParams, mcp_server_tools
from autogen_agentchat.agents import AssistantAgent
from autogen_core import CancellationToken


async def main() -> None:
    # Setup server params for local filesystem access
    desktop = str(Path.home() / "Desktop")
    server_params = StdioServerParams(
        command="npx.cmd", args=["-y", "@modelcontextprotocol/server-filesystem", desktop]
    )

    # Get all available tools from the server
    tools = await mcp_server_tools(server_params)

    # Create an agent that can use all the tools
    agent = AssistantAgent(
        name="file_manager",
        model_client=OpenAIChatCompletionClient(model="gpt-4"),
        tools=tools,  # type: ignore
    )

    # The agent can now use any of the filesystem tools
    await agent.run(task="Create a file called test.txt with some content", cancellation_token=CancellationToken())


if __name__ == "__main__":
    asyncio.run(main())

Beispiel für einen lokalen Abruf-MCP-Dienst über Standard-Ein-/Ausgabe

Installieren Sie das Paket mcp-server-fetch.

pip install mcp-server-fetch

Erstellen Sie einen Agenten, der das fetch-Tool vom lokalen MCP-Server verwenden kann.

import asyncio

from autogen_agentchat.agents import AssistantAgent
from autogen_ext.models.openai import OpenAIChatCompletionClient
from autogen_ext.tools.mcp import StdioServerParams, mcp_server_tools


async def main() -> None:
    # Get the fetch tool from mcp-server-fetch.
    fetch_mcp_server = StdioServerParams(command="uvx", args=["mcp-server-fetch"])
    tools = await mcp_server_tools(fetch_mcp_server)

    # Create an agent that can use the fetch tool.
    model_client = OpenAIChatCompletionClient(model="gpt-4o")
    agent = AssistantAgent(name="fetcher", model_client=model_client, tools=tools, reflect_on_tool_use=True)  # type: ignore

    # Let the agent fetch the content of a URL and summarize it.
    result = await agent.run(task="Summarize the content of https://en.wikipedia.org/wiki/Seattle")
    print(result.messages[-1])


asyncio.run(main())

Gemeinsame Nutzung einer MCP-Client-Sitzung über mehrere Tools hinweg

Sie können eine einzelne MCP-Client-Sitzung erstellen und diese für mehrere Tools gemeinsam nutzen. Dies ist manchmal erforderlich, wenn der Server einen Sitzungsstatus (z. B. einen Browserstatus) beibehält, der für mehrere Anfragen wiederverwendet werden soll.

Das folgende Beispiel zeigt, wie eine einzelne MCP-Client-Sitzung zu einem lokalen Playwright-Server erstellt und für mehrere Tools gemeinsam genutzt wird.

import asyncio

from autogen_agentchat.agents import AssistantAgent
from autogen_agentchat.conditions import TextMentionTermination
from autogen_agentchat.teams import RoundRobinGroupChat
from autogen_agentchat.ui import Console
from autogen_ext.models.openai import OpenAIChatCompletionClient
from autogen_ext.tools.mcp import StdioServerParams, create_mcp_server_session, mcp_server_tools


async def main() -> None:
    model_client = OpenAIChatCompletionClient(model="gpt-4o", parallel_tool_calls=False)  # type: ignore
    params = StdioServerParams(
        command="npx",
        args=["@playwright/mcp@latest"],
        read_timeout_seconds=60,
    )
    async with create_mcp_server_session(params) as session:
        await session.initialize()
        tools = await mcp_server_tools(server_params=params, session=session)
        print(f"Tools: {[tool.name for tool in tools]}")

        agent = AssistantAgent(
            name="Assistant",
            model_client=model_client,
            tools=tools,  # type: ignore
        )

        termination = TextMentionTermination("TERMINATE")
        team = RoundRobinGroupChat([agent], termination_condition=termination)
        await Console(
            team.run_stream(
                task="Go to https://ekzhu.com/, visit the first link in the page, then tell me about the linked page."
            )
        )


asyncio.run(main())

Beispiel für einen entfernten MCP-Dienst über SSE

from autogen_ext.tools.mcp import SseServerParams, mcp_server_tools


async def main() -> None:
    # Setup server params for remote service
    server_params = SseServerParams(url="https://api.example.com/mcp", headers={"Authorization": "Bearer token"})

    # Get all available tools
    tools = await mcp_server_tools(server_params)

    # Create an agent with all tools
    agent = AssistantAgent(name="tool_user", model_client=OpenAIChatCompletionClient(model="gpt-4"), tools=tools)  # type: ignore

Weitere Beispiele und detaillierte Verwendung finden Sie im Verzeichnis „samples“ im Paket-Repository.

class McpWorkbench(server_params: Annotated[StdioServerParams | SseServerParams | StreamableHttpServerParams, FieldInfo(annotation=NoneType, required=True, discriminator='type')], tool_overrides: Dict[str, ToolOverride] | None = None, model_client: ChatCompletionClient | None = None)[Quelle]#

Basiert auf: Workbench, Component[McpWorkbenchConfig]

Eine Workbench, die einen MCP-Server umschließt und eine Schnittstelle zum Auflisten und Aufrufen von vom Server bereitgestellten Tools bietet.

Warnung

Verbinden Sie sich nur mit vertrauenswürdigen MCP-Servern, insbesondere wenn Sie StdioServerParams verwenden, da dies Befehle in der lokalen Umgebung ausführt.

Diese Workbench sollte als Kontextmanager verwendet werden, um die ordnungsgemäße Initialisierung und Bereinigung der zugrunde liegenden MCP-Sitzung sicherzustellen.

MCP-Unterstützung#
MCP-Fähigkeit	Unterstützte Funktionen
Tools	list_tools, call_tool
Ressourcen	list_resources, read_resource
ResourceTemplates	list_resource_templates, read_resource_template
Prompts	list_prompts, get_prompt
Sampling	Optionale Unterstützung über model_client
Roots	nicht unterstützt
Ellicitation	nicht unterstützt

Parameter:

server_params (McpServerParams) – Die Parameter zum Verbinden mit dem MCP-Server. Dies kann entweder ein StdioServerParams oder ein SseServerParams sein.
tool_overrides (Optional[Dict[str, ToolOverride]]) – Optionale Zuordnung von ursprünglichen Tool-Namen zu Überschreibungskonfigurationen für Name und/oder Beschreibung. Dies ermöglicht die Anpassung, wie Server-Tools für Verbraucher angezeigt werden, während die zugrunde liegende Tool-Funktionalität beibehalten wird.
model_client – Optionaler Chat-Completion-Client zur Verarbeitung von Sampling-Anfragen von MCP-Servern, die die Sampling-Fähigkeit unterstützen. Dies ermöglicht MCP-Servern, Textgenerierung von einem Sprachmodell während der Tool-Ausführung anzufordern. Wenn nicht angegeben, geben Sampling-Anfragen einen Fehler zurück.

Löst aus:

ValueError – Wenn es Konflikte bei den Tool-Überschreibungsnamen gibt.

Examples

Hier ist ein einfaches Beispiel für die Verwendung der Workbench mit einem mcp-server-fetch-Server

import asyncio

from autogen_ext.tools.mcp import McpWorkbench, StdioServerParams


async def main() -> None:
    params = StdioServerParams(
        command="uvx",
        args=["mcp-server-fetch"],
        read_timeout_seconds=60,
    )

    # You can also use `start()` and `stop()` to manage the session.
    async with McpWorkbench(server_params=params) as workbench:
        tools = await workbench.list_tools()
        print(tools)
        result = await workbench.call_tool(tools[0]["name"], {"url": "https://github.com/"})
        print(result)


asyncio.run(main())

Beispiel für die Verwendung von Tool-Überschreibungen

import asyncio
from autogen_ext.tools.mcp import McpWorkbench, StdioServerParams
from autogen_core.tools import ToolOverride


async def main() -> None:
    params = StdioServerParams(
        command="uvx",
        args=["mcp-server-fetch"],
        read_timeout_seconds=60,
    )

    # Override the fetch tool's name and description
    overrides = {
        "fetch": ToolOverride(name="web_fetch", description="Enhanced web fetching tool with better error handling")
    }

    async with McpWorkbench(server_params=params, tool_overrides=overrides) as workbench:
        tools = await workbench.list_tools()
        # The tool will now appear as "web_fetch" with the new description
        print(tools)
        # Call the overridden tool
        result = await workbench.call_tool("web_fetch", {"url": "https://github.com/"})
        print(result)


asyncio.run(main())

Beispiel für die Verwendung der Workbench mit dem GitHub MCP Server

import asyncio
from autogen_agentchat.agents import AssistantAgent
from autogen_agentchat.ui import Console
from autogen_ext.models.openai import OpenAIChatCompletionClient
from autogen_ext.tools.mcp import McpWorkbench, StdioServerParams


async def main() -> None:
    model_client = OpenAIChatCompletionClient(model="gpt-4.1-nano")
    server_params = StdioServerParams(
        command="docker",
        args=[
            "run",
            "-i",
            "--rm",
            "-e",
            "GITHUB_PERSONAL_ACCESS_TOKEN",
            "ghcr.io/github/github-mcp-server",
        ],
        env={
            "GITHUB_PERSONAL_ACCESS_TOKEN": "ghp_XXXXXXXXXXXXXXXXXXXXXXXXXXXXXX",
        },
    )
    async with McpWorkbench(server_params) as mcp:
        agent = AssistantAgent(
            "github_assistant",
            model_client=model_client,
            workbench=mcp,
            reflect_on_tool_use=True,
            model_client_stream=True,
        )
        await Console(agent.run_stream(task="Is there a repository named Autogen"))


asyncio.run(main())

Beispiel für die Verwendung der Workbench mit dem Playwright MCP Server

# First run `npm install -g @playwright/mcp@latest` to install the MCP server.
import asyncio
from autogen_agentchat.agents import AssistantAgent
from autogen_agentchat.teams import RoundRobinGroupChat
from autogen_agentchat.conditions import TextMessageTermination
from autogen_agentchat.ui import Console
from autogen_ext.models.openai import OpenAIChatCompletionClient
from autogen_ext.tools.mcp import McpWorkbench, StdioServerParams


async def main() -> None:
    model_client = OpenAIChatCompletionClient(model="gpt-4.1-nano")
    server_params = StdioServerParams(
        command="npx",
        args=[
            "@playwright/mcp@latest",
            "--headless",
        ],
    )
    async with McpWorkbench(server_params) as mcp:
        agent = AssistantAgent(
            "web_browsing_assistant",
            model_client=model_client,
            workbench=mcp,
            model_client_stream=True,
        )
        team = RoundRobinGroupChat(
            [agent],
            termination_condition=TextMessageTermination(source="web_browsing_assistant"),
        )
        await Console(team.run_stream(task="Find out how many contributors for the microsoft/autogen repository"))


asyncio.run(main())

component_provider_override: ClassVar[str | None] = 'autogen_ext.tools.mcp.McpWorkbench'#: Überschreibe den Anbieter-String für die Komponente. Dies sollte verwendet werden, um zu verhindern, dass interne Modulnamen Teil des Modulnamens werden.

component_config_schema#: Alias von McpWorkbenchConfig

property server_params: Annotated[StdioServerParams | SseServerParams | StreamableHttpServerParams, FieldInfo(annotation=NoneType, required=True, discriminator='type')]#

async list_tools() → List[ToolSchema][Quelle]#

Listet die derzeit verfügbaren Tools in der Workbench als ToolSchema-Objekte auf.

Die Liste der Tools kann dynamisch sein und ihr Inhalt kann sich nach der Tool-Ausführung ändern.

async call_tool(name: str, arguments: Mapping[str, Any] | None = None, cancellation_token: CancellationToken | None = None, call_id: str | None = None) → ToolResult[Quelle]#

Ruft ein Tool in der Workbench auf.

Parameter:

name (str) – Der Name des aufzurufenden Tools.
arguments (Mapping[str, Any] | None) – Die an das Tool zu übergebenden Argumente. Wenn None, wird das Tool ohne Argumente aufgerufen.
cancellation_token (CancellationToken | None) – Ein optionaler Abbruch-Token, um die Tool-Ausführung abzubrechen.
call_id (str | None) – Eine optionale Kennung für den Tool-Aufruf, die zur Nachverfolgung verwendet wird.

Gibt zurück:

ToolResult – Das Ergebnis der Tool-Ausführung.

property initialize_result: Any#

async list_prompts() → ListPromptsResult[Quelle]#: Listet verfügbare Prompts vom MCP-Server auf.

async list_resources() → ListResourcesResult[Quelle]#: Listet verfügbare Ressourcen vom MCP-Server auf.

async list_resource_templates() → ListResourceTemplatesResult[Quelle]#: Listet verfügbare Ressourcenvorlagen vom MCP-Server auf.

async read_resource(uri: str) → ReadResourceResult[Quelle]#: Liest eine Ressource vom MCP-Server.

async get_prompt(name: str, arguments: Dict[str, str] | None = None) → GetPromptResult[Quelle]#: Ruft einen Prompt vom MCP-Server ab.

async start() → None[Quelle]#

Startet die Workbench und initialisiert alle Ressourcen.

Diese Methode sollte aufgerufen werden, bevor die Workbench verwendet wird.

async stop() → None[Quelle]#

Beendet die Workbench und gibt alle Ressourcen frei.

Diese Methode sollte aufgerufen werden, wenn die Workbench nicht mehr benötigt wird.

async reset() → None[Quelle]#: Setzt die Workbench in ihren initialisierten, gestarteten Zustand zurück.

async save_state() → Mapping[str, Any][Quelle]#

Speichert den Zustand der Workbench.

Diese Methode sollte aufgerufen werden, um den Zustand der Workbench zu persistent zu machen.

async load_state(state: Mapping[str, Any]) → None[Quelle]#

Lädt den Zustand der Workbench.

Parameter:: state (Mapping[str, Any]) – Der in die Workbench zu ladende Zustand.

_to_config() → McpWorkbenchConfig[Quelle]#

Gib die Konfiguration aus, die erforderlich wäre, um eine neue Instanz einer Komponente zu erstellen, die der Konfiguration dieser Instanz entspricht.

Gibt zurück:: T – Die Konfiguration der Komponente.

classmethod _from_config(config: McpWorkbenchConfig) → Self[Quelle]#

Erstelle eine neue Instanz der Komponente aus einem Konfigurationsobjekt.

Parameter:: config (T) – Das Konfigurationsobjekt.
Gibt zurück:: Self – Die neue Instanz der Komponente.