autogen_ext.tools.mcp#

create_mcp_server_session(server_params: Annotated[StdioServerParams | SseServerParams | StreamableHttpServerParams, FieldInfo(annotation=NoneType, required=True, discriminator='type')]) → AsyncGenerator[ClientSession, None][source]#: Create an MCP client session for the given server parameters.

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

Bases: ComponentBase[BaseModel], Component[McpSessionActorConfig]

component_type: ClassVar[ComponentType] = 'mcp_session_actor'#: The logical type of the component.

component_config_schema#: alias of McpSessionActorConfig

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

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

async initialize() → None[source]#

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

async close() → None[source]#

class StdioMcpToolAdapter(server_params: StdioServerParams, tool: Tool, session: ClientSession | None = None)[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
session (ClientSession, optional) – The MCP client session to use. If not provided, a new session will be created. This is useful for testing or when you want to manage the session lifecycle yourself.

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

Fields:

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)[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.
session (ClientSession, optional) – The MCP client session to use. If not provided, it will create a new session. This is useful for testing or when you want to manage the session lifecycle yourself.

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

Fields:

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

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

field url: str [Required]#

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)[source]#

Bases: McpToolAdapter[StreamableHttpServerParams], Component[StreamableHttpMcpToolAdapterConfig]

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

This adapter enables using MCP-compatible tools that communicate over Streamable HTTP 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 (StreamableHttpServerParams) – Parameters for the MCP server connection, including URL, headers, and timeouts.
tool (Tool) – The MCP tool to wrap.
session (ClientSession, optional) – The MCP client session to use. If not provided, it will create a new session. This is useful for testing or when you want to manage the session lifecycle yourself.

Examples

Use a remote translation service that implements MCP over Streamable HTTP 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 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 of StreamableHttpMcpToolAdapterConfig

component_provider_override: ClassVar[str | None] = 'autogen_ext.tools.mcp.StreamableHttpMcpToolAdapter'#: 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 StreamableHttpServerParams[source]#

Bases: BaseModel

Parameters for connecting to an MCP server over Streamable HTTP.

Show JSON schema

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

Fields:

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 [Required]#

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

field timeout: float = 30.0#

field sse_read_timeout: float = 300.0#

field 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][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 and StreamableHttpServerParams for HTTP/SSE services.
session (ClientSession | None) – Optional existing session to use. This is used when you want to reuse an existing connection to the MCP server. The session will be reused when creating the MCP tool adapters.

Returns:

list[StdioMcpToolAdapter | SseMcpToolAdapter | StreamableHttpMcpToolAdapter] – 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])


asyncio.run(main())

Sharing an MCP client session across multiple tools:

You can create a single MCP client session and share it across multiple tools. This is sometimes required when the server maintains a session state (e.g., a browser state) that should be reused for multiple requests.

The following example show how to create a single MCP client session to a local Playwright server and share it across multiple tools.

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

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.

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

Bases: Workbench, Component[McpWorkbenchConfig]

A workbench that wraps an MCP server and provides an interface to list and call tools provided by the server.

This workbench should be used as a context manager to ensure proper initialization and cleanup of the underlying MCP session.

Parameters:

server_params (McpServerParams) – The parameters to connect to the MCP server. This can be either a StdioServerParams or SseServerParams.
tool_overrides (Optional[Dict[str, ToolOverride]]) – Optional mapping of original tool names to override configurations for name and/or description. This allows customizing how server tools appear to consumers while maintaining the underlying tool functionality.

Raises:

ValueError – If there are conflicts in tool override names.

Examples

Here is a simple example of how to use the workbench with a 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())

Example of using tool overrides:

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

Example of using the workbench with the 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())

Example of using the workbench with the 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'#: Override the provider string for the component. This should be used to prevent internal module names being a part of the module name.

component_config_schema#: alias of McpWorkbenchConfig

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

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

List the currently available tools in the workbench as ToolSchema objects.

The list of tools may be dynamic, and their content may change after tool execution.

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

Call a tool in the workbench.

Parameters:

name (str) – The name of the tool to call.
arguments (Mapping[str, Any] | None) – The arguments to pass to the tool. If None, the tool will be called with no arguments.
cancellation_token (CancellationToken | None) – An optional cancellation token to cancel the tool execution.
call_id (str | None) – An optional identifier for the tool call, used for tracing.

Returns:

ToolResult – The result of the tool execution.

async start() → None[source]#

Start the workbench and initialize any resources.

This method should be called before using the workbench.

async stop() → None[source]#

Stop the workbench and release any resources.

This method should be called when the workbench is no longer needed.

async reset() → None[source]#: Reset the workbench to its initialized, started state.

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

Save the state of the workbench.

This method should be called to persist the state of the workbench.

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

Load the state of the workbench.

Parameters:: state (Mapping[str, Any]) – The state to load into the workbench.

_to_config() → McpWorkbenchConfig[source]#

Dump the configuration that would be requite to create a new instance of a component matching the configuration of this instance.

Returns:: T – The configuration of the component.

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

Create a new instance of the component from a configuration object.

Parameters:: config (T) – The configuration object.
Returns:: Self – The new instance of the component.