autogen_ext.tools.mcp#

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

基类： ComponentBase[BaseModel], Component[McpSessionActorConfig]

async call(type: str, args: McpActorArgs | None = None) → Future[Coroutine[Any, Any, ListToolsResult] | Coroutine[Any, Any, CallToolResult]][source]#

async close() → None[source]#

component_config_schema#: 别名：McpSessionActorConfig

component_provider_override: ClassVar[str | None] = 'autogen_ext.tools.mcp.McpSessionActor'#: 覆盖组件的提供程序字符串。这应该用于防止内部模块名称成为模块名称的一部分。

component_type: ClassVar[ComponentType] = 'mcp_session_actor'#: 组件的逻辑类型。

async initialize() → None[source]#

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

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

基类： Workbench, Component[McpWorkbenchConfig]

一个工作台，它包装了一个 MCP 服务器，并提供了一个接口来列出和调用服务器提供的工具。

参数:: server_params (McpServerParams) – 连接到 MCP 服务器的参数。这可以是 StdioServerParams 或 SseServerParams。

示例

这是一个使用 workbench 和 mcp-server-fetch 服务器的简单示例

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())

使用 workbench 和 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())

使用 workbench 和 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())

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

从配置对象创建一个组件的新实例。

参数:: config (T) – 配置对象。
返回值:: Self – 组件的新实例。

_to_config() → McpWorkbenchConfig[source]#

导出创建与此实例的配置匹配的组件新实例所需的配置。

返回值:: T – 组件的配置。

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

调用 workbench 中的一个工具。

参数:

name (str) – 要调用的工具的名称。
arguments (Mapping[str, Any] | None) – 传递给工具的参数。如果为 None，则调用工具时不带任何参数。
cancellation_token (CancellationToken | None) – 一个可选的取消令牌，用于取消工具的执行。

返回值:

ToolResult – 工具执行的结果。

component_config_schema#: McpWorkbenchConfig 的别名

component_provider_override: ClassVar[str | None] = 'autogen_ext.tools.mcp.McpWorkbench'#: 覆盖组件的提供程序字符串。这应该用于防止内部模块名称成为模块名称的一部分。

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

将 workbench 中当前可用的工具作为 ToolSchema 对象列出。

工具列表可能是动态的，并且它们的内容可能会在工具执行后发生变化。

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

加载 workbench 的状态。

参数:: state (Mapping[str, Any]) – 要加载到 workbench 中的状态。

async reset() → None[source]#: 将 workbench 重置为已初始化、已启动的状态。

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

保存 workbench 的状态。

应调用此方法以持久化 workbench 的状态。

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

async start() → None[source]#

启动 workbench 并初始化任何资源。

在使用 workbench 之前，应调用此方法。

async stop() → None[source]#

停止工作台并释放所有资源。

当不再需要工作台时，应调用此方法。

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

基类： McpToolAdapter[SseServerParams], Component[SseMcpToolAdapterConfig]

允许您封装通过服务器发送事件 (SSE) 运行的 MCP 工具，并使其可用于 AutoGen。

此适配器使能够将与 AutoGen 代理通过 HTTP 和 SSE 进行通信的兼容 MCP 工具结合使用。常见的用例包括与远程 MCP 服务、基于云的工具和实现模型上下文协议 (MCP) 的 Web API 集成。

注意

要使用此类，您需要为 autogen-ext 包安装 mcp 扩展。

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

参数:

server_params (SseServerParameters) – MCP 服务器连接的参数，包括 URL、标头和超时。
tool (Tool) – 要封装的 MCP 工具。
session (ClientSession, optional) – 要使用的 MCP 客户端会话。如果未提供，它将创建一个新会话。这在测试或您想要自己管理会话生命周期时很有用。

示例

使用实现基于 SSE 的 MCP 的远程翻译服务来创建允许 AutoGen 代理执行翻译的工具

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#: 类型： SseMcpToolAdapterConfig

component_provider_override: ClassVar[str | None] = 'autogen_ext.tools.mcp.SseMcpToolAdapter'#: 覆盖组件的提供程序字符串。这应该用于防止内部模块名称成为模块名称的一部分。

pydantic model SseServerParams[source]#

基类： BaseModel

通过 SSE 连接到 MCP 服务器的参数。

显示 JSON 模式

{
   "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"
   ]
}

字段:

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

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

field sse_read_timeout: float = 300#

field timeout: float = 5#

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

field url: str [Required]#

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

基类： McpToolAdapter[StdioServerParams], Component[StdioMcpToolAdapterConfig]

允许您封装通过 STDIO 运行的 MCP 工具，并使其可用于 AutoGen。

此适配器使能够将与 AutoGen 代理通过标准输入/输出进行通信的兼容 MCP 工具结合使用。常见的用例包括封装命令行工具和实现模型上下文协议 (MCP) 的本地服务。

注意

要使用此类，您需要为 autogen-ext 包安装 mcp 扩展。

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

参数:

server_params (StdioServerParams) – MCP 服务器连接的参数，包括要运行的命令及其参数
tool (Tool) – 要封装的 MCP 工具
session (ClientSession, optional) – 要使用的 MCP 客户端会话。如果未提供，将创建一个新会话。这在测试或您想要自己管理会话生命周期时很有用。

有关示例，请参见 mcp_server_tools()。

component_config_schema#: 类型： StdioMcpToolAdapterConfig

component_provider_override: ClassVar[str | None] = 'autogen_ext.tools.mcp.StdioMcpToolAdapter'#: 覆盖组件的提供程序字符串。这应该用于防止内部模块名称成为模块名称的一部分。

pydantic 模型 StdioServerParams[源码]#

基类: StdioServerParameters

通过 STDIO 连接到 MCP 服务器的参数。

显示 JSON 模式

{
   "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"
   ]
}

字段:

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

字段 read_timeout_seconds: float = 5#

字段 type: Literal['StdioServerParams'] = 'StdioServerParams'#

create_mcp_server_session(server_params: Annotated[StdioServerParams | SseServerParams, FieldInfo(annotation=NoneType, required=True, discriminator='type')]) → AsyncGenerator[ClientSession, None][源码]#: 为给定的服务器参数创建 MCP 客户端会话。

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

创建可与 AutoGen 代理一起使用的 MCP 工具适配器列表。

此工厂函数连接到 MCP 服务器，并返回所有可用工具的适配器。适配器可以直接分配给 AutoGen 代理的工具列表。

注意

要使用此函数，您需要为 autogen-ext 包安装 mcp 扩展。

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

参数:

server_params (McpServerParams) – MCP 服务器的连接参数。可以是命令行工具的 StdioServerParams，也可以是 HTTP/SSE 服务的 SseServerParams。
session (ClientSession | None) – 可选的现有会话以供使用。当您想要重用与 MCP 服务器的现有连接时，可以使用此参数。创建 MCP 工具适配器时，将重用该会话。

返回值:

list[StdioMcpToolAdapter | SseMcpToolAdapter] – 可与 AutoGen 代理一起使用的工具适配器列表。

示例

通过标准 I/O 的本地文件系统 MCP 服务示例

从 npm 安装文件系统服务器包（需要 Node.js 16+ 和 npm）。

npm install -g @modelcontextprotocol/server-filesystem

创建一个可以使用本地文件系统 MCP 服务器中所有工具的代理。

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())

通过标准 I/O 的本地 fetch MCP 服务示例

安装 mcp-server-fetch 包。

pip install mcp-server-fetch

创建一个可以使用本地 MCP 服务器中的 fetch 工具的代理。

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())

跨多个工具共享 MCP 客户端会话

您可以创建一个 MCP 客户端会话并在多个工具之间共享它。当服务器维护应为多个请求重用的会话状态（例如，浏览器状态）时，有时需要这样做。

以下示例显示了如何创建一个到本地 Playwright 服务器的 MCP 客户端会话并在多个工具之间共享它。

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())

通过 SSE 的远程 MCP 服务示例

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

有关更多示例和详细用法，请参阅包存储库中的 samples 目录。