Skip to main content
In scenarios where it is necessary to dynamically select MCP servers for the agent to use, the Agent Stack provides an MCP extension that allows clients to specify MCP server details at runtime. For you as an agent builder, the usage is extremely simple because we’ve wrapped the usage into an MCP Extension.
Service Extensions are a type of A2A Extension that allows you to easily “inject dependencies” into your agent. This follows the inversion of control principle where your agent defines what it needs, and the client (possibly in cooperation with the platform) is responsible for providing those dependencies.
Service extensions are optional by definition, so you should always check if they exist before using them.

​
Quickstart

1

Add MCP service extension to your agent

Import the necessary components and add the MCP service extension to your agent function.
2

Configure your MCP request

Specify which MCP servers your agent prefers.
3

Use the MCP client in your agent

Access the optionally provided MCP extension and use it to create MCP client(s).

​
Example of MCP extension

Here’s how to add MCP extension to your agent: 
import os
from typing import Annotated

from a2a.types import Message
from mcp import ClientSession
from agentstack_sdk.server import Server
from agentstack_sdk.server.context import RunContext
from agentstack_sdk.a2a.types import AgentMessage
from agentstack_sdk.a2a.extensions import MCPServiceExtensionServer, MCPServiceExtensionSpec

server = Server()

@server.agent()
async def example_agent(
    input: Message,
    context: RunContext,
    mcp: Annotated[
        MCPServiceExtensionServer,
        MCPServiceExtensionSpec.single_demand(),
    ],
):
    """Agent that uses MCP client to list tools"""

    if not mcp:
        yield "No tools available"

    async with mcp.create_client() as (read, write), ClientSession(read, write) as session:
        await session.initialize()

        tools = await session.list_tools()

        yield "Available tools: \n"
        yield "\n".join([t.name for t in tools.tools])

def run():
    server.run(host=os.getenv("HOST", "127.0.0.1"), port=int(os.getenv("PORT", 8000)))

if __name__ == "__main__":
    run()

​
Usage with OAuth extension

In the usual scenario where the specified MCP server requires OAuth authentication, the MCP extension automatically leverages the OAuth extension if activated. Technically, it does exactly what is described in Authorize MCP with OAuth. However, the agent still needs to declare the extension: 
@server.agent()
async def example_agent(
    input: Message,
    context: RunContext,
    mcp: Annotated[
        MCPServiceExtensionServer,
        MCPServiceExtensionSpec.single_demand(),
    ],
    _: Annotated[OAuthExtensionServer, OAuthExtensionSpec.single_demand()],
):

​
Usage with Connectors

To further improve user experience, one can use Connectors and have the Agent Stack platform automatically manage access tokens, hence effectively skipping the aforementioned OAuth flow during agent execution. This is achieved by specifying a connector instead of the actual MCP server that backs the connector.