autogen_ext.tools.mcp#

class SseMcpToolAdapter(server_params: SseServerParams, tool: Tool)[source]#

Bases: McpToolAdapter[SseServerParams], Component[SseMcpToolAdapterConfig]

Allows you to wrap an MCP tool running over Server-Sent Events (SSE) and make it available to AutoGen.

This adapter enables using MCP-compatible tools that communicate over HTTP with SSE with AutoGen agents. Common use cases include integrating with remote MCP services, cloud-based tools, and web APIs that implement the Model Context Protocol (MCP).

Note

To use this class, you need to install mcp extra for the autogen-ext package.

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

Parameters:

server_params (SseServerParameters) – Parameters for the MCP server connection, including URL, headers, and timeouts
tool (Tool) – The MCP tool to wrap

Examples

Use a remote translation service that implements MCP over SSE to create tools that allow AutoGen agents to perform translations:

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 of SseMcpToolAdapterConfig

component_provider_override: ClassVar[str | None] = 'autogen_ext.tools.mcp.SseMcpToolAdapter'#: Override the provider string for the component. This should be used to prevent internal module names being a part of the module name.

pydantic model SseServerParams[source]#

Bases: BaseModel

Parameters for connecting to an MCP server over SSE.

Show JSON schema

{
   "title": "SseServerParams",
   "description": "Parameters for connecting to an MCP server over SSE.",
   "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, Any] | None = None#

field sse_read_timeout: float = 300#

field timeout: float = 5#

field url: str [Required]#

class StdioMcpToolAdapter(server_params: StdioServerParams, tool: Tool)[source]#

Bases: McpToolAdapter[StdioServerParams], Component[StdioMcpToolAdapterConfig]

Allows you to wrap an MCP tool running over STDIO and make it available to AutoGen.

This adapter enables using MCP-compatible tools that communicate over standard input/output with AutoGen agents. Common use cases include wrapping command-line tools and local services that implement the Model Context Protocol (MCP).

Note

To use this class, you need to install mcp extra for the autogen-ext package.

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

Parameters:

server_params (StdioServerParams) – Parameters for the MCP server connection, including command to run and its arguments
tool (Tool) – The MCP tool to wrap

See mcp_server_tools() for examples.

component_config_schema#: alias of StdioMcpToolAdapterConfig

component_provider_override: ClassVar[str | None] = 'autogen_ext.tools.mcp.StdioMcpToolAdapter'#: Override the provider string for the component. This should be used to prevent internal module names being a part of the module name.

pydantic model StdioServerParams[source]#

Bases: StdioServerParameters

Parameters for connecting to an MCP server over STDIO.

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]#: Command line arguments to pass to the executable.

field command: str [Required]#: The executable to run to start the server.

field encoding: str = 'utf-8'#

The text encoding used when sending/receiving messages to the server

defaults to utf-8

field encoding_error_handler: Literal['strict', 'ignore', 'replace'] = 'strict'#

The text encoding error handler.

See https://docs.python.org/3/library/codecs.html#codec-base-classes for explanations of possible values

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

The environment to use when spawning the process.

If not specified, the result of get_default_environment() will be used.

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

Creates a list of MCP tool adapters that can be used with AutoGen agents.

This factory function connects to an MCP server and returns adapters for all available tools. The adapters can be directly assigned to an AutoGen agent’s tools list.

Note

To use this function, you need to install mcp extra for the autogen-ext package.

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

Parameters:: server_params (McpServerParams) – Connection parameters for the MCP server. Can be either StdioServerParams for command-line tools or SseServerParams for HTTP/SSE services.
Returns:: list[StdioMcpToolAdapter | SseMcpToolAdapter] – A list of tool adapters ready to use with AutoGen agents.

Examples

Local file system MCP service over standard I/O example:

Install the filesystem server package from npm (requires Node.js 16+ and npm).

npm install -g @modelcontextprotocol/server-filesystem

Create an agent that can use all tools from the local filesystem MCP server.

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

Local fetch MCP service over standard I/O example:

Install the mcp-server-fetch package.

pip install mcp-server-fetch

Create an agent that can use the fetch tool from the local MCP server.

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

Remote MCP service over SSE example:

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

For more examples and detailed usage, see the samples directory in the package repository.