Skip to content

MCPServer

autogen.beta.mcp.server.MCPServer #

MCPServer(agent, *, name=None, version=None, instructions=None, website_url=None, icons=None, tool_name='ask', tool_description=None, stream_progress=True, context_provider=None, lifespan=None, sessions=True, resources=(), resource_templates=(), prompts=(), path='/mcp', stateless=False, json_response=False, security=None)

Wrap an AG2 :class:Agent as an MCP server.

The agent is exposed as a single conversational tool (ask by default) that runs :meth:Agent.ask and returns the reply — the inverse of the consume-side toolkit autogen.beta.tools.MCPToolkit, which connects to an MCP server.

The instance is itself an ASGI3 application: it serves MCP over streamable HTTP and manages its own lifespan, so a standalone uvicorn run just works::

app = MCPServer(agent, path="/mcp")
uvicorn.run(app, host="127.0.0.1", port=8000)

For local clients (Claude Desktop, Cursor, the MCP Inspector), :meth:run_stdio serves over stdin/stdout instead. The HTTP transport parameters (path, stateless, json_response, security) are ignored over stdio.

name / version / instructions / website_url / icons populate the initialize handshake's serverInfo + instructions. instructions is client-facing "how to use this server" guidance — it is not derived from the agent's system prompt (which is internal); pass it explicitly when you want to advertise usage hints.

sessions controls multi-turn history. By default (True) each MCP session (keyed by the transport's mcp-session-id, or a per-process key over stdio) keeps its own conversation history that accumulates across calls; pass a :class:~autogen.beta.mcp.sessions.SessionConfig to tune the bound / TTL / backend, or False to make every call stateless. A stateless HTTP transport (stateless=True) issues no session id, so it stays stateless regardless of this setting.

resources / resource_templates / prompts expose MCP resources and prompts alongside the conversational tool; the corresponding capability is advertised only when a non-empty collection is supplied.

Source code in autogen/beta/mcp/server.py 
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
def __init__(
    self,
    agent: Agent,
    *,
    name: str | None = None,
    version: str | None = None,
    instructions: str | None = None,
    website_url: str | None = None,
    icons: list[Icon] | None = None,
    tool_name: str = "ask",
    tool_description: str | None = None,
    stream_progress: bool = True,
    context_provider: "ContextProvider | None" = None,
    lifespan: "ServerLifespan | None" = None,
    sessions: "bool | SessionConfig" = True,
    resources: "Sequence[Resource]" = (),
    resource_templates: "Sequence[ResourceTemplate]" = (),
    prompts: "Sequence[Prompt]" = (),
    path: str = "/mcp",
    stateless: bool = False,
    json_response: bool = False,
    security: Requirement | None = None,
) -> None:
    self._agent = agent
    self._name = name or agent.name
    self._version = version or _package_version()
    self._instructions = instructions
    self._website_url = website_url
    self._icons = icons
    self._lifespan = lifespan
    self._session_store = _build_session_store(sessions)
    self._resource_provider = (
        ResourceProvider(resources, resource_templates) if (resources or resource_templates) else None
    )
    self._prompt_provider = PromptProvider(prompts) if prompts else None
    self._executor = AgentExecutor(
        agent,
        tool_name=tool_name,
        tool_description=tool_description,
        stream_progress=stream_progress,
        context_provider=context_provider,
        session_store=self._session_store,
    )
    self._server = self._build_server()
    routes, manager = self._streamable_routes(
        path=path, stateless=stateless, json_response=json_response, security=security
    )
    self._http: Starlette = Starlette(routes=routes, lifespan=_session_manager_lifespan(manager))

agent property #

agent

server property #

server

The underlying low-level mcp server (for advanced wiring / tests).

run_stdio async #

run_stdio()

Serve the agent over stdio until the client disconnects.

Source code in autogen/beta/mcp/server.py 
296
297
298
299
300
301
302
303
async def run_stdio(self) -> None:  # pragma: no cover - needs real stdio pipes
    """Serve the agent over stdio until the client disconnects."""
    async with stdio_server() as (read_stream, write_stream):
        await self._server.run(
            read_stream,
            write_stream,
            self._server.create_initialization_options(),
        )