Mcp

McpToolSpec

Bases: BaseToolSpec, TypeResolutionMixin, TypeCreationMixin, FieldExtractionMixin

MCPToolSpec will get the tools from MCP Client (only need to implement ClientSession) and convert them to LlamaIndex's FunctionTool objects.

Parameters:

Name	Type	Description	Default
client	ClientSession	An MCP client instance implementing ClientSession, and it should support the following methods in ClientSession: - list_tools: List all tools. - call_tool: Call a tool. - list_resources: List all resources. - read_resource: Read a resource.	required
allowed_tools	Optional[List[str]]	If set, only return tools with the specified names.	None
include_resources	bool	Whether to include resources in the tool list.	False

Source code in llama_index/tools/mcp/base.py

class McpToolSpec(
    BaseToolSpec, TypeResolutionMixin, TypeCreationMixin, FieldExtractionMixin
):
    """
    MCPToolSpec will get the tools from MCP Client (only need to implement ClientSession) and convert them to LlamaIndex's FunctionTool objects.

    Args:
        client: An MCP client instance implementing ClientSession, and it should support the following methods in ClientSession:
            - list_tools: List all tools.
            - call_tool: Call a tool.
            - list_resources: List all resources.
            - read_resource: Read a resource.
        allowed_tools: If set, only return tools with the specified names.
        include_resources: Whether to include resources in the tool list.

    """

    def __init__(
        self,
        client: ClientSession,
        allowed_tools: Optional[List[str]] = None,
        include_resources: bool = False,
    ) -> None:
        self.client = client
        self.allowed_tools = allowed_tools
        self.include_resources = include_resources
        self.properties_cache = {}

    async def fetch_tools(self) -> List[Any]:
        """
        An asynchronous method to get the tools list from MCP Client. If allowed_tools is set, it will filter the tools.

        Returns:
            A list of tools, each tool object needs to contain name, description, inputSchema properties.

        """
        response = await self.client.list_tools()
        tools = response.tools if hasattr(response, "tools") else []

        if self.allowed_tools is None:
            # get all tools by default
            return tools

        if any(self.allowed_tools):
            return [tool for tool in tools if tool.name in self.allowed_tools]

        logging.warning(
            "Returning an empty tool list due to the empty `allowed_tools` list. Please ensure `allowed_tools` is set appropriately."
        )
        return []

    async def fetch_resources(self) -> List[Resource]:
        """
        An asynchronous method to get the resources list from MCP Client.
        """
        static_response = await self.client.list_resources()
        dynamic_response = await self.client.list_resource_templates()
        static_resources = (
            static_response.resources if hasattr(static_response, "resources") else []
        )
        dynamic_resources = (
            dynamic_response.resourceTemplates
            if hasattr(dynamic_response, "resourceTemplates")
            else []
        )
        resources = static_resources + dynamic_resources
        if self.allowed_tools is None:
            return resources

        if any(self.allowed_tools):
            return [
                resource
                for resource in resources
                if resource.name in self.allowed_tools
            ]

        logging.warning(
            "Returning an empty resource list due to the empty `allowed_tools` list. Please ensure `allowed_tools` is set appropriately."
        )
        return []

    def _create_tool_fn(self, tool_name: str) -> Callable:
        """
        Create a tool call function for a specified MCP tool name. The function internally wraps the call_tool call to the MCP Client.
        """

        async def async_tool_fn(**kwargs):
            return await self.client.call_tool(tool_name, kwargs)

        return async_tool_fn

    def _create_resource_fn(self, resource_uri: str) -> Callable:
        """
        Create a resource call function for a specified MCP resource name. The function internally wraps the read_resource call to the MCP Client.
        """

        async def async_resource_fn():
            return await self.client.read_resource(resource_uri)

        return async_resource_fn

    async def to_tool_list_async(self) -> List[FunctionTool]:
        """
        Asynchronous method to convert MCP tools to FunctionTool objects.

        Returns:
            A list of FunctionTool objects.

        """
        tools_list = await self.fetch_tools()
        function_tool_list: List[FunctionTool] = []
        for tool in tools_list:
            fn = self._create_tool_fn(tool.name)
            # Create a Pydantic model based on the tool inputSchema
            model_schema = self.create_model_from_json_schema(
                tool.inputSchema, model_name=f"{tool.name}_Schema"
            )
            metadata = ToolMetadata(
                name=tool.name,
                description=tool.description,
                fn_schema=model_schema,
            )
            function_tool = FunctionTool.from_defaults(
                async_fn=fn, tool_metadata=metadata
            )
            function_tool_list.append(function_tool)

        if self.include_resources:
            resources_list = await self.fetch_resources()
            for resource in resources_list:
                if hasattr(resource, "uri"):
                    uri = resource.uri
                elif hasattr(resource, "template"):
                    uri = resource.template
                fn = self._create_resource_fn(uri)
                function_tool_list.append(
                    FunctionTool.from_defaults(
                        async_fn=fn,
                        name=resource.name.replace("/", "_"),
                        description=resource.description,
                    )
                )

        return function_tool_list

    def to_tool_list(self) -> List[FunctionTool]:
        """
        Synchronous interface: Convert MCP Client tools to FunctionTool objects.
        Note: This method should not be called in an asynchronous environment, otherwise an exception will be thrown. Use to_tool_list_async instead.

        Returns:
            A list of FunctionTool objects.

        """
        return patch_sync(self.to_tool_list_async)()

    def create_model_from_json_schema(
        self,
        schema: dict[str, Any],
        model_name: str = "DynamicModel",
    ) -> type[BaseModel]:
        """
        To create a Pydantic model from the JSON Schema of MCP tools.

        Args:
            schema: A JSON Schema dictionary containing properties and required fields.
            model_name: The name of the model.

        Returns:
            A Pydantic model class.

        """
        defs = schema.get("$defs", {})

        # Process all type definitions
        for cls_name, cls_schema in defs.items():
            self.properties_cache[cls_name] = self._create_model(
                cls_schema,
                cls_name,
                defs,
            )

        return self._create_model(schema, model_name)

    def _create_model(
        self,
        schema: dict,
        model_name: str,
        defs: dict = {},
    ) -> type[BaseModel]:
        """Create a Pydantic model from a schema."""
        if model_name in self.properties_cache:
            return self.properties_cache[model_name]

        fields = self._extract_fields(schema, defs)
        model = create_model(model_name, **fields)
        self.properties_cache[model_name] = model
        return model

fetch_tools async

fetch_tools() -> List[Any]

An asynchronous method to get the tools list from MCP Client. If allowed_tools is set, it will filter the tools.

Returns:

Type	Description
List[Any]	A list of tools, each tool object needs to contain name, description, inputSchema properties.

Source code in llama_index/tools/mcp/base.py

async def fetch_tools(self) -> List[Any]:
    """
    An asynchronous method to get the tools list from MCP Client. If allowed_tools is set, it will filter the tools.

    Returns:
        A list of tools, each tool object needs to contain name, description, inputSchema properties.

    """
    response = await self.client.list_tools()
    tools = response.tools if hasattr(response, "tools") else []

    if self.allowed_tools is None:
        # get all tools by default
        return tools

    if any(self.allowed_tools):
        return [tool for tool in tools if tool.name in self.allowed_tools]

    logging.warning(
        "Returning an empty tool list due to the empty `allowed_tools` list. Please ensure `allowed_tools` is set appropriately."
    )
    return []

fetch_resources async

fetch_resources() -> List[Resource]

An asynchronous method to get the resources list from MCP Client.

Source code in llama_index/tools/mcp/base.py

async def fetch_resources(self) -> List[Resource]:
    """
    An asynchronous method to get the resources list from MCP Client.
    """
    static_response = await self.client.list_resources()
    dynamic_response = await self.client.list_resource_templates()
    static_resources = (
        static_response.resources if hasattr(static_response, "resources") else []
    )
    dynamic_resources = (
        dynamic_response.resourceTemplates
        if hasattr(dynamic_response, "resourceTemplates")
        else []
    )
    resources = static_resources + dynamic_resources
    if self.allowed_tools is None:
        return resources

    if any(self.allowed_tools):
        return [
            resource
            for resource in resources
            if resource.name in self.allowed_tools
        ]

    logging.warning(
        "Returning an empty resource list due to the empty `allowed_tools` list. Please ensure `allowed_tools` is set appropriately."
    )
    return []

to_tool_list_async async

to_tool_list_async() -> List[FunctionTool]

Asynchronous method to convert MCP tools to FunctionTool objects.

Returns:

Type	Description
List[FunctionTool]	A list of FunctionTool objects.

Source code in llama_index/tools/mcp/base.py

async def to_tool_list_async(self) -> List[FunctionTool]:
    """
    Asynchronous method to convert MCP tools to FunctionTool objects.

    Returns:
        A list of FunctionTool objects.

    """
    tools_list = await self.fetch_tools()
    function_tool_list: List[FunctionTool] = []
    for tool in tools_list:
        fn = self._create_tool_fn(tool.name)
        # Create a Pydantic model based on the tool inputSchema
        model_schema = self.create_model_from_json_schema(
            tool.inputSchema, model_name=f"{tool.name}_Schema"
        )
        metadata = ToolMetadata(
            name=tool.name,
            description=tool.description,
            fn_schema=model_schema,
        )
        function_tool = FunctionTool.from_defaults(
            async_fn=fn, tool_metadata=metadata
        )
        function_tool_list.append(function_tool)

    if self.include_resources:
        resources_list = await self.fetch_resources()
        for resource in resources_list:
            if hasattr(resource, "uri"):
                uri = resource.uri
            elif hasattr(resource, "template"):
                uri = resource.template
            fn = self._create_resource_fn(uri)
            function_tool_list.append(
                FunctionTool.from_defaults(
                    async_fn=fn,
                    name=resource.name.replace("/", "_"),
                    description=resource.description,
                )
            )

    return function_tool_list

to_tool_list

to_tool_list() -> List[FunctionTool]

Synchronous interface: Convert MCP Client tools to FunctionTool objects. Note: This method should not be called in an asynchronous environment, otherwise an exception will be thrown. Use to_tool_list_async instead.

Returns:

Type	Description
List[FunctionTool]	A list of FunctionTool objects.

Source code in llama_index/tools/mcp/base.py

def to_tool_list(self) -> List[FunctionTool]:
    """
    Synchronous interface: Convert MCP Client tools to FunctionTool objects.
    Note: This method should not be called in an asynchronous environment, otherwise an exception will be thrown. Use to_tool_list_async instead.

    Returns:
        A list of FunctionTool objects.

    """
    return patch_sync(self.to_tool_list_async)()

create_model_from_json_schema

create_model_from_json_schema(schema: dict[str, Any], model_name: str = 'DynamicModel') -> type[BaseModel]

To create a Pydantic model from the JSON Schema of MCP tools.

Parameters:

Name	Type	Description	Default
schema	dict[str, Any]	A JSON Schema dictionary containing properties and required fields.	required
model_name	str	The name of the model.	'DynamicModel'

Returns:

Type	Description
type[BaseModel]	A Pydantic model class.

Source code in llama_index/tools/mcp/base.py

def create_model_from_json_schema(
    self,
    schema: dict[str, Any],
    model_name: str = "DynamicModel",
) -> type[BaseModel]:
    """
    To create a Pydantic model from the JSON Schema of MCP tools.

    Args:
        schema: A JSON Schema dictionary containing properties and required fields.
        model_name: The name of the model.

    Returns:
        A Pydantic model class.

    """
    defs = schema.get("$defs", {})

    # Process all type definitions
    for cls_name, cls_schema in defs.items():
        self.properties_cache[cls_name] = self._create_model(
            cls_schema,
            cls_name,
            defs,
        )

    return self._create_model(schema, model_name)

BasicMCPClient

Bases: ClientSession

Basic MCP client that can be used to connect to an MCP server.

This is useful for connecting to any MCP server.

Parameters:

Name	Type	Description	Default
command_or_url	str	The command to run or the URL to connect to.	required
args	Optional[List[str]]	The arguments to pass to StdioServerParameters.	None
env	Optional[Dict[str, str]]	The environment variables to set for StdioServerParameters.	None
timeout	int	The timeout for the command in seconds.	30
auth	Optional[OAuthClientProvider]	Optional OAuth client provider for authentication.	None
sampling_callback	Optional[Callable[[CreateMessageRequestParams], Awaitable[CreateMessageResult]]]	Optional callback for handling sampling messages.	None
headers	Optional[Dict[str, Any]]	Optional headers to pass by sse client or streamable http client	None
tool_call_logs_callback	Optional[Callable[[List[str]], Awaitable[Any]]]	Async function to store the logs deriving from an MCP tool call: logs are provided as a list of strings, representing log messages. Defaults to None.	None

Source code in llama_index/tools/mcp/client.py

class BasicMCPClient(ClientSession):
    """
    Basic MCP client that can be used to connect to an MCP server.

    This is useful for connecting to any MCP server.

    Args:
        command_or_url: The command to run or the URL to connect to.
        args: The arguments to pass to StdioServerParameters.
        env: The environment variables to set for StdioServerParameters.
        timeout: The timeout for the command in seconds.
        auth: Optional OAuth client provider for authentication.
        sampling_callback: Optional callback for handling sampling messages.
        headers: Optional headers to pass by sse client or streamable http client
        tool_call_logs_callback: Async function to store the logs deriving from an MCP tool call: logs are provided as a list of strings, representing log messages. Defaults to None.

    """

    def __init__(
        self,
        command_or_url: str,
        args: Optional[List[str]] = None,
        env: Optional[Dict[str, str]] = None,
        timeout: int = 30,
        auth: Optional[OAuthClientProvider] = None,
        sampling_callback: Optional[
            Callable[
                [types.CreateMessageRequestParams], Awaitable[types.CreateMessageResult]
            ]
        ] = None,
        headers: Optional[Dict[str, Any]] = None,
        tool_call_logs_callback: Optional[Callable[[List[str]], Awaitable[Any]]] = None,
    ):
        self.command_or_url = command_or_url
        self.args = args or []
        self.env = env or {}
        self.timeout = timeout
        self.auth = auth
        self.sampling_callback = sampling_callback
        self.headers = headers
        self.tool_call_logs_callback = tool_call_logs_callback

    @classmethod
    def with_oauth(
        cls,
        command_or_url: str,
        client_name: str,
        redirect_uris: List[str],
        redirect_handler: Callable[[str], None],
        callback_handler: Callable[[], Tuple[str, Optional[str]]],
        args: Optional[List[str]] = None,
        env: Optional[Dict[str, str]] = None,
        timeout: int = 30,
        token_storage: Optional[TokenStorage] = None,
        tool_call_logs_callback: Optional[Callable[[List[str]], Awaitable[Any]]] = None,
    ) -> "BasicMCPClient":
        """
        Create a client with OAuth authentication.

        Args:
            command_or_url: The command to run or the URL to connect to
            client_name: The name of the OAuth client
            redirect_uris: The redirect URIs for the OAuth flow
            redirect_handler: Function that handles the redirect URL
            callback_handler: Function that returns the auth code and state
            token_storage: Optional token storage for OAuth client. If not provided,
                           a default in-memory storage is used (tokens will be lost on restart).
            args: The arguments to pass to StdioServerParameters.
            env: The environment variables to set for StdioServerParameters.
            timeout: The timeout for the command in seconds.
            tool_call_logs_callback: Async function to store the logs deriving from an MCP tool call: logs are provided as a list of strings, representing log messages. Defaults to None.

        Returns:
            An authenticated MCP client

        """
        # Use default in-memory storage if none provided
        if token_storage is None:
            token_storage = DefaultInMemoryTokenStorage()
            warnings.warn(
                "Using default in-memory token storage. Tokens will be lost on restart.",
                UserWarning,
            )

        oauth_auth = OAuthClientProvider(
            server_url=command_or_url if urlparse(command_or_url).scheme else None,
            client_metadata=OAuthClientMetadata(
                client_name=client_name,
                redirect_uris=redirect_uris,
                grant_types=["authorization_code", "refresh_token"],
                response_types=["code"],
            ),
            redirect_handler=redirect_handler,
            callback_handler=callback_handler,
            storage=token_storage,
        )

        return cls(
            command_or_url,
            auth=oauth_auth,
            args=args,
            env=env,
            timeout=timeout,
            tool_call_logs_callback=tool_call_logs_callback,
        )

    @asynccontextmanager
    async def _run_session(self) -> AsyncIterator[ClientSession]:
        """Create and initialize a session with the MCP server."""
        url = urlparse(self.command_or_url)
        scheme = url.scheme

        if scheme in ("http", "https"):
            # Check if this is a streamable HTTP endpoint (default) or SSE
            if enable_sse(self.command_or_url):
                # SSE transport
                async with sse_client(
                    self.command_or_url, auth=self.auth, headers=self.headers
                ) as streams:
                    async with ClientSession(
                        *streams,
                        read_timeout_seconds=timedelta(seconds=self.timeout),
                        sampling_callback=self.sampling_callback,
                    ) as session:
                        await session.initialize()
                        yield session
            else:
                # Streamable HTTP transport (recommended)
                async with streamablehttp_client(
                    self.command_or_url, auth=self.auth, headers=self.headers
                ) as (read, write, _):
                    async with ClientSession(
                        read,
                        write,
                        read_timeout_seconds=timedelta(seconds=self.timeout),
                        sampling_callback=self.sampling_callback,
                    ) as session:
                        await session.initialize()
                        yield session
        else:
            # stdio transport
            server_parameters = StdioServerParameters(
                command=self.command_or_url, args=self.args, env=self.env
            )
            async with stdio_client(server_parameters) as streams:
                async with ClientSession(
                    *streams,
                    read_timeout_seconds=timedelta(seconds=self.timeout),
                    sampling_callback=self.sampling_callback,
                ) as session:
                    await session.initialize()
                    yield session

    def _configure_tool_call_logs_callback(self) -> io.StringIO:
        handler = io.StringIO()
        stream_handler = logging.StreamHandler(handler)

        # Configure logging to capture all events
        logging.basicConfig(
            level=logging.DEBUG,  # Capture all log levels
            format="%(asctime)s - %(name)s - %(levelname)s - %(message)s\n",
            handlers=[
                stream_handler,
            ],
        )
        # Also enable logging for specific FastMCP components
        fastmcp_logger = logging.getLogger("fastmcp")
        fastmcp_logger.setLevel(logging.DEBUG)

        # Enable HTTP transport logging to see network details
        http_logger = logging.getLogger("httpx")
        http_logger.setLevel(logging.DEBUG)

        return handler

    # Tool methods
    async def call_tool(
        self,
        tool_name: str,
        arguments: Optional[dict] = None,
        progress_callback: Optional[ProgressFnT] = None,
    ) -> types.CallToolResult:
        """Call a tool on the MCP server."""
        if self.tool_call_logs_callback is not None:
            # we use a string stream so that we can recover all logs at the end of the session
            handler = self._configure_tool_call_logs_callback()

            async with self._run_session() as session:
                result = await session.call_tool(
                    tool_name, arguments=arguments, progress_callback=progress_callback
                )

                # get all logs by dividing the string with \n, since the format of the log has an \n at the end of the log message
                extra_values = handler.getvalue().split("\n")

                # pipe the logs list into tool_call_logs_callback
                await self.tool_call_logs_callback(extra_values)

                return result
        else:
            async with self._run_session() as session:
                return await session.call_tool(
                    tool_name, arguments=arguments, progress_callback=progress_callback
                )

    async def list_tools(self) -> types.ListToolsResult:
        """List all available tools on the MCP server."""
        async with self._run_session() as session:
            return await session.list_tools()

    # Resource methods
    async def list_resources(self) -> types.ListToolsResult:
        """List all available resources on the MCP server."""
        async with self._run_session() as session:
            return await session.list_resources()

    async def list_resource_templates(self) -> types.ListToolsResult:
        """List all dynamic available resources on the MCP server."""
        async with self._run_session() as session:
            return await session.list_resource_templates()

    async def read_resource(self, resource_uri: AnyUrl) -> types.ReadResourceResult:
        """
        Read a resource from the MCP server.

        Returns:
            Tuple containing the resource content as bytes and the MIME type

        """
        async with self._run_session() as session:
            return await session.read_resource(resource_uri)

    ## ----- Prompt methods -----

    async def list_prompts(self) -> List[types.Prompt]:
        """List all available prompts on the MCP server."""
        async with self._run_session() as session:
            return await session.list_prompts()

    async def get_prompt(
        self, prompt_name: str, arguments: Optional[Dict[str, str]] = None
    ) -> List[ChatMessage]:
        """
        Get a prompt from the MCP server.

        Args:
            prompt_name: The name of the prompt to get
            arguments: Optional arguments to pass to the prompt

        Returns:
            The prompt as a list of llama-index ChatMessage objects

        """
        async with self._run_session() as session:
            prompt = await session.get_prompt(prompt_name, arguments)
            llama_messages = []
            for message in prompt.messages:
                if isinstance(message.content, types.TextContent):
                    llama_messages.append(
                        ChatMessage(
                            role=message.role,
                            blocks=[TextBlock(text=message.content.text)],
                        )
                    )
                elif isinstance(message.content, types.ImageContent):
                    llama_messages.append(
                        ChatMessage(
                            role=message.role,
                            blocks=[
                                ImageBlock(
                                    image=message.content.data,
                                    image_mimetype=message.content.mimeType,
                                )
                            ],
                        )
                    )
                elif isinstance(message.content, types.EmbeddedResource):
                    raise NotImplementedError(
                        "Embedded resources are not supported yet"
                    )
                else:
                    raise ValueError(
                        f"Unsupported content type: {type(message.content)}"
                    )

            return llama_messages

with_oauth classmethod

with_oauth(command_or_url: str, client_name: str, redirect_uris: List[str], redirect_handler: Callable[[str], None], callback_handler: Callable[[], Tuple[str, Optional[str]]], args: Optional[List[str]] = None, env: Optional[Dict[str, str]] = None, timeout: int = 30, token_storage: Optional[TokenStorage] = None, tool_call_logs_callback: Optional[Callable[[List[str]], Awaitable[Any]]] = None) -> BasicMCPClient

Create a client with OAuth authentication.

Parameters:

Name	Type	Description	Default
command_or_url	str	The command to run or the URL to connect to	required
client_name	str	The name of the OAuth client	required
redirect_uris	List[str]	The redirect URIs for the OAuth flow	required
redirect_handler	Callable[[str], None]	Function that handles the redirect URL	required
callback_handler	Callable[[], Tuple[str, Optional[str]]]	Function that returns the auth code and state	required
token_storage	Optional[TokenStorage]	Optional token storage for OAuth client. If not provided, a default in-memory storage is used (tokens will be lost on restart).	None
args	Optional[List[str]]	The arguments to pass to StdioServerParameters.	None
env	Optional[Dict[str, str]]	The environment variables to set for StdioServerParameters.	None
timeout	int	The timeout for the command in seconds.	30
tool_call_logs_callback	Optional[Callable[[List[str]], Awaitable[Any]]]	Async function to store the logs deriving from an MCP tool call: logs are provided as a list of strings, representing log messages. Defaults to None.	None

Returns:

Type	Description
BasicMCPClient	An authenticated MCP client

Source code in llama_index/tools/mcp/client.py

@classmethod
def with_oauth(
    cls,
    command_or_url: str,
    client_name: str,
    redirect_uris: List[str],
    redirect_handler: Callable[[str], None],
    callback_handler: Callable[[], Tuple[str, Optional[str]]],
    args: Optional[List[str]] = None,
    env: Optional[Dict[str, str]] = None,
    timeout: int = 30,
    token_storage: Optional[TokenStorage] = None,
    tool_call_logs_callback: Optional[Callable[[List[str]], Awaitable[Any]]] = None,
) -> "BasicMCPClient":
    """
    Create a client with OAuth authentication.

    Args:
        command_or_url: The command to run or the URL to connect to
        client_name: The name of the OAuth client
        redirect_uris: The redirect URIs for the OAuth flow
        redirect_handler: Function that handles the redirect URL
        callback_handler: Function that returns the auth code and state
        token_storage: Optional token storage for OAuth client. If not provided,
                       a default in-memory storage is used (tokens will be lost on restart).
        args: The arguments to pass to StdioServerParameters.
        env: The environment variables to set for StdioServerParameters.
        timeout: The timeout for the command in seconds.
        tool_call_logs_callback: Async function to store the logs deriving from an MCP tool call: logs are provided as a list of strings, representing log messages. Defaults to None.

    Returns:
        An authenticated MCP client

    """
    # Use default in-memory storage if none provided
    if token_storage is None:
        token_storage = DefaultInMemoryTokenStorage()
        warnings.warn(
            "Using default in-memory token storage. Tokens will be lost on restart.",
            UserWarning,
        )

    oauth_auth = OAuthClientProvider(
        server_url=command_or_url if urlparse(command_or_url).scheme else None,
        client_metadata=OAuthClientMetadata(
            client_name=client_name,
            redirect_uris=redirect_uris,
            grant_types=["authorization_code", "refresh_token"],
            response_types=["code"],
        ),
        redirect_handler=redirect_handler,
        callback_handler=callback_handler,
        storage=token_storage,
    )

    return cls(
        command_or_url,
        auth=oauth_auth,
        args=args,
        env=env,
        timeout=timeout,
        tool_call_logs_callback=tool_call_logs_callback,
    )

call_tool async

call_tool(tool_name: str, arguments: Optional[dict] = None, progress_callback: Optional[ProgressFnT] = None) -> CallToolResult

Call a tool on the MCP server.

Source code in llama_index/tools/mcp/client.py

async def call_tool(
    self,
    tool_name: str,
    arguments: Optional[dict] = None,
    progress_callback: Optional[ProgressFnT] = None,
) -> types.CallToolResult:
    """Call a tool on the MCP server."""
    if self.tool_call_logs_callback is not None:
        # we use a string stream so that we can recover all logs at the end of the session
        handler = self._configure_tool_call_logs_callback()

        async with self._run_session() as session:
            result = await session.call_tool(
                tool_name, arguments=arguments, progress_callback=progress_callback
            )

            # get all logs by dividing the string with \n, since the format of the log has an \n at the end of the log message
            extra_values = handler.getvalue().split("\n")

            # pipe the logs list into tool_call_logs_callback
            await self.tool_call_logs_callback(extra_values)

            return result
    else:
        async with self._run_session() as session:
            return await session.call_tool(
                tool_name, arguments=arguments, progress_callback=progress_callback
            )

list_tools async

list_tools() -> ListToolsResult

List all available tools on the MCP server.

Source code in llama_index/tools/mcp/client.py

async def list_tools(self) -> types.ListToolsResult:
    """List all available tools on the MCP server."""
    async with self._run_session() as session:
        return await session.list_tools()

list_resources async

list_resources() -> ListToolsResult

List all available resources on the MCP server.

Source code in llama_index/tools/mcp/client.py

async def list_resources(self) -> types.ListToolsResult:
    """List all available resources on the MCP server."""
    async with self._run_session() as session:
        return await session.list_resources()

list_resource_templates async

list_resource_templates() -> ListToolsResult

List all dynamic available resources on the MCP server.

Source code in llama_index/tools/mcp/client.py

async def list_resource_templates(self) -> types.ListToolsResult:
    """List all dynamic available resources on the MCP server."""
    async with self._run_session() as session:
        return await session.list_resource_templates()

read_resource async

read_resource(resource_uri: AnyUrl) -> ReadResourceResult

Read a resource from the MCP server.

Returns:

Type	Description
ReadResourceResult	Tuple containing the resource content as bytes and the MIME type

Source code in llama_index/tools/mcp/client.py

async def read_resource(self, resource_uri: AnyUrl) -> types.ReadResourceResult:
    """
    Read a resource from the MCP server.

    Returns:
        Tuple containing the resource content as bytes and the MIME type

    """
    async with self._run_session() as session:
        return await session.read_resource(resource_uri)

list_prompts async

list_prompts() -> List[Prompt]

List all available prompts on the MCP server.

Source code in llama_index/tools/mcp/client.py

async def list_prompts(self) -> List[types.Prompt]:
    """List all available prompts on the MCP server."""
    async with self._run_session() as session:
        return await session.list_prompts()

get_prompt async

get_prompt(prompt_name: str, arguments: Optional[Dict[str, str]] = None) -> List[ChatMessage]

Get a prompt from the MCP server.

Parameters:

Name	Type	Description	Default
prompt_name	str	The name of the prompt to get	required
arguments	Optional[Dict[str, str]]	Optional arguments to pass to the prompt	None

Returns:

Type	Description
List[ChatMessage]	The prompt as a list of llama-index ChatMessage objects

Source code in llama_index/tools/mcp/client.py

async def get_prompt(
    self, prompt_name: str, arguments: Optional[Dict[str, str]] = None
) -> List[ChatMessage]:
    """
    Get a prompt from the MCP server.

    Args:
        prompt_name: The name of the prompt to get
        arguments: Optional arguments to pass to the prompt

    Returns:
        The prompt as a list of llama-index ChatMessage objects

    """
    async with self._run_session() as session:
        prompt = await session.get_prompt(prompt_name, arguments)
        llama_messages = []
        for message in prompt.messages:
            if isinstance(message.content, types.TextContent):
                llama_messages.append(
                    ChatMessage(
                        role=message.role,
                        blocks=[TextBlock(text=message.content.text)],
                    )
                )
            elif isinstance(message.content, types.ImageContent):
                llama_messages.append(
                    ChatMessage(
                        role=message.role,
                        blocks=[
                            ImageBlock(
                                image=message.content.data,
                                image_mimetype=message.content.mimeType,
                            )
                        ],
                    )
                )
            elif isinstance(message.content, types.EmbeddedResource):
                raise NotImplementedError(
                    "Embedded resources are not supported yet"
                )
            else:
                raise ValueError(
                    f"Unsupported content type: {type(message.content)}"
                )

        return llama_messages

workflow_as_mcp

workflow_as_mcp(workflow: Workflow, workflow_name: Optional[str] = None, workflow_description: Optional[str] = None, start_event_model: Optional[BaseModel] = None, **fastmcp_init_kwargs: Any) -> FastMCP

Convert a workflow to an MCP app.

This will convert any Workflow to an MCP app. It will expose the workflow as a tool within MCP, which will

Parameters:

Name	Type	Description	Default
workflow	Workflow	The workflow to convert.	required
workflow_name	optional	The name of the workflow. Defaults to the workflow class name.	None
workflow_description	optional	The description of the workflow. Defaults to the workflow docstring.	None
start_event_model	optional	The start event model of the workflow. Can be a BaseModel or a StartEvent class. Defaults to the workflow's custom StartEvent class.	None
**fastmcp_init_kwargs	Any	Additional keyword arguments to pass to the FastMCP constructor.	{}

Returns:

Type	Description
FastMCP	The MCP app object.

Source code in llama_index/tools/mcp/utils.py

def workflow_as_mcp(
    workflow: Workflow,
    workflow_name: Optional[str] = None,
    workflow_description: Optional[str] = None,
    start_event_model: Optional[BaseModel] = None,
    **fastmcp_init_kwargs: Any,
) -> FastMCP:
    """
    Convert a workflow to an MCP app.

    This will convert any `Workflow` to an MCP app. It will expose the workflow as a tool
    within MCP, which will

    Args:
        workflow:
            The workflow to convert.
        workflow_name (optional):
            The name of the workflow. Defaults to the workflow class name.
        workflow_description (optional):
            The description of the workflow. Defaults to the workflow docstring.
        start_event_model (optional):
            The start event model of the workflow. Can be a `BaseModel` or a `StartEvent` class.
            Defaults to the workflow's custom `StartEvent` class.
        **fastmcp_init_kwargs:
            Additional keyword arguments to pass to the FastMCP constructor.

    Returns:
        The MCP app object.

    """
    app = FastMCP(**fastmcp_init_kwargs)

    # Dynamically get the start event class -- this is a bit of a hack
    StartEventCLS = start_event_model or workflow._start_event_class
    if StartEventCLS == StartEvent:
        raise ValueError(
            "Must declare a custom StartEvent class in your workflow or provide a start_event_model."
        )

    # Get the workflow name and description
    workflow_name = workflow_name or workflow.__class__.__name__
    workflow_description = workflow_description or workflow.__doc__

    @app.tool(name=workflow_name, description=workflow_description)
    async def _workflow_tool(run_args: StartEventCLS, context: Context) -> Any:
        # Handle edge cases where the start event is an Event or a BaseModel
        # If the workflow does not have a custom StartEvent class, then we need to handle the event differently

        if isinstance(run_args, Event) and workflow._start_event_class != StartEvent:
            handler = workflow.run(start_event=run_args)
        elif isinstance(run_args, BaseModel):
            handler = workflow.run(**run_args.model_dump())
        elif isinstance(run_args, dict):
            start_event = StartEventCLS.model_validate(run_args)
            handler = workflow.run(start_event=start_event)
        else:
            raise ValueError(f"Invalid start event type: {type(run_args)}")

        async for event in handler.stream_events():
            if not isinstance(event, StopEvent):
                await context.log("info", message=event.model_dump_json())

        return await handler

    return app

get_tools_from_mcp_url

get_tools_from_mcp_url(command_or_url: str, client: Optional[ClientSession] = None, allowed_tools: Optional[List[str]] = None, include_resources: bool = False) -> List[FunctionTool]

Get tools from an MCP server or command.

Parameters:

Name	Type	Description	Default
command_or_url	str	The command to run or the URL to connect to.	required
client	optional	The client to use to connect to the MCP server.	None
allowed_tools	optional	The tool names to allow from the MCP server.	None
include_resources	optional	Whether to include resources in the tool list.	False

Source code in llama_index/tools/mcp/utils.py

def get_tools_from_mcp_url(
    command_or_url: str,
    client: Optional[ClientSession] = None,
    allowed_tools: Optional[List[str]] = None,
    include_resources: bool = False,
) -> List[FunctionTool]:
    """
    Get tools from an MCP server or command.

    Args:
        command_or_url: The command to run or the URL to connect to.
        client (optional): The client to use to connect to the MCP server.
        allowed_tools (optional): The tool names to allow from the MCP server.
        include_resources (optional): Whether to include resources in the tool list.

    """
    client = client or BasicMCPClient(command_or_url)
    tool_spec = McpToolSpec(
        client, allowed_tools=allowed_tools, include_resources=include_resources
    )
    return tool_spec.to_tool_list()

aget_tools_from_mcp_url async

aget_tools_from_mcp_url(command_or_url: str, client: Optional[ClientSession] = None, allowed_tools: Optional[List[str]] = None, include_resources: bool = False) -> List[FunctionTool]

Get tools from an MCP server or command.

Parameters:

Name	Type	Description	Default
command_or_url	str	The command to run or the URL to connect to.	required
client	optional	The client to use to connect to the MCP server.	None
allowed_tools	optional	The tool names to allow from the MCP server.	None
include_resources	optional	Whether to include resources in the tool list.	False

Source code in llama_index/tools/mcp/utils.py

async def aget_tools_from_mcp_url(
    command_or_url: str,
    client: Optional[ClientSession] = None,
    allowed_tools: Optional[List[str]] = None,
    include_resources: bool = False,
) -> List[FunctionTool]:
    """
    Get tools from an MCP server or command.

    Args:
        command_or_url: The command to run or the URL to connect to.
        client (optional): The client to use to connect to the MCP server.
        allowed_tools (optional): The tool names to allow from the MCP server.
        include_resources (optional): Whether to include resources in the tool list.

    """
    client = client or BasicMCPClient(command_or_url)
    tool_spec = McpToolSpec(
        client, allowed_tools=allowed_tools, include_resources=include_resources
    )
    return await tool_spec.to_tool_list_async()

options: members: - McpToolSpec