autogen_ext.tools.mcp#

class SseMcpToolAdapter(server_params: SseServerParams, tool: Tool)[源代码]#

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

允许您封装一个运行在服务器发送事件（SSE）上的MCP工具，并使其可用于AutoGen。

该适配器使得能够使用与MCP兼容的工具，这些工具通过HTTP与SSE与AutoGen代理进行通信。常见的用例包括与远程MCP服务、基于云的工具以及实现模型上下文协议（MCP）的Web API集成。

注意

要使用此类，你需要为autogen-ext包安装mcp额外组件。

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

Parameters:

server_params (SseServerParameters) – MCP服务器连接的参数，包括URL、头信息和超时设置
工具 (工具) – 要包装的MCP工具

示例

使用实现了MCP over SSE的远程翻译服务来创建工具，使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 | 无] = 'autogen_ext.tools.mcp.SseMcpToolAdapter'#: 覆盖组件的提供商字符串。这应用于防止内部模块名称成为模块名称的一部分。

pydantic model SseServerParams[源代码]#

基础：BaseModel

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

显示JSON模式

{
   "title": "SseServerParams",
   "description": "通过SSE连接到MCP服务器的参数。",
   "type": "object",
   "properties": {
      "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"
   ]
}

Fields:

headers (dict[str, Any] | None)
sse_read_timeout (float)
timeout (float)
url (str)

field headers: dict[str, 任何] | 无 = None#

field sse_read_timeout: float = 300#

field timeout: float = 5#

field url: str [Required]#

class StdioMcpToolAdapter(server_params: StdioServerParams, tool: Tool)[源代码]#

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

允许您将一个通过STDIO运行的MCP工具包装起来，并使其可供AutoGen使用。

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

注意

要使用此类，你需要为autogen-ext包安装mcp额外组件。

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

Parameters:

server_params (StdioServerParams) – MCP服务器连接的参数，包括运行的命令及其参数
工具 (工具) – 要包装的MCP工具

参见 mcp_server_tools() 以获取示例。

component_config_schema#: 别名为 StdioMcpToolAdapterConfig

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

pydantic model StdioServerParams[源代码]#

基础：StdioServerParameters

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

Show JSON schema

{
   "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"
      },
      "encoding": {
         "default": "utf-8",
         "title": "Encoding",
         "type": "string"
      },
      "encoding_error_handler": {
         "default": "strict",
         "enum": [
            "strict",
            "ignore",
            "replace"
         ],
         "title": "Encoding Error Handler",
         "type": "string"
      }
   },
   "required": [
      "command"
   ]
}

Fields:

args (list[str])
command (str)
encoding (str)
encoding_error_handler (Literal['strict', 'ignore', 'replace'])
env (dict[str, str] | None)

field args: list[str] [Optional]#: 传递给可执行文件的命令行参数。

field command: str [Required]#: 运行以启动服务器的可执行文件。

field encoding: str = 'utf-8'#

发送/接收消息到服务器时使用的文本编码

默认为 utf-8

field encoding_error_handler: 字面量['strict', 'ignore', 'replace'] = 'strict'#

文本编码错误处理器。

请参阅https://docs.python.org/3/library/codecs.html#codec-base-classes了解可能值的解释

field env: dict[str, str] | 无 = None#

在生成进程时要使用的环境。

如果未指定，将使用get_default_environment()的结果。

async mcp_server_tools(server_params: StdioServerParams | SseServerParams) → list[StdioMcpToolAdapter | SseMcpToolAdapter][源代码]#

创建一个可以用于AutoGen代理的MCP工具适配器列表。

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

注意

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

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

Parameters:: server_params (McpServerParams) – MCP服务器的连接参数。可以是用于命令行工具的 StdioServerParams 或用于 HTTP/SSE 服务的 SseServerParams。
Returns:: 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进行本地获取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].content)


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

更多示例和详细用法，请参阅包仓库中的示例目录。